diff --git a/include/opendht/dht.h b/include/opendht/dht.h index c6217357..5c882e6d 100644 --- a/include/opendht/dht.h +++ b/include/opendht/dht.h @@ -630,17 +630,17 @@ private: InfoHash id {}; sockaddr_storage ss; socklen_t sslen {}; - uint16_t tid {}; + size_t rid {}; time_point time {}; /*constexpr*/ Listener() : ss() {} - Listener(const InfoHash& id, const sockaddr *from, socklen_t fromlen, uint16_t ttid, time_point t) : id(id), ss(), sslen(fromlen), tid(ttid), time(t) { + Listener(const InfoHash& id, const sockaddr *from, socklen_t fromlen, uint16_t rid, time_point t) : id(id), ss(), sslen(fromlen), rid(rid), time(t) { memcpy(&ss, from, fromlen); } - void refresh(const sockaddr *from, socklen_t fromlen, uint16_t ttid, time_point t) { + void refresh(const sockaddr *from, socklen_t fromlen, size_t rid, time_point t) { memcpy(&ss, from, fromlen); sslen = fromlen; - tid = ttid; + this->rid = rid; time = t; } }; @@ -769,6 +769,8 @@ private: sockaddr_storage blacklist[BLACKLISTED_MAX] {}; unsigned next_blacklisted = 0; + std::unique_ptr network_engine {}; + // timing time_point now; time_point mybucket_grow_time {time_point::min()}, mybucket6_grow_time {time_point::min()}; @@ -800,7 +802,7 @@ private: }); } - void storageAddListener(const InfoHash& id, const InfoHash& node, const sockaddr *from, socklen_t fromlen, uint16_t tid); + void storageAddListener(const InfoHash& id, const InfoHash& node, const sockaddr *from, socklen_t fromlen, size_t tid); bool storageStore(const InfoHash& id, const std::shared_ptr& value, time_point created); void expireStorage(); void storageChanged(Storage& st, ValueStorage&); @@ -836,7 +838,7 @@ private: void dumpBucket(const Bucket& b, std::ostream& out) const; // Nodes - std::shared_ptr newNode(const InfoHash& id, const sockaddr*, socklen_t, int confirm, const sockaddr* addr=nullptr, socklen_t addr_length=0); + std::shared_ptr newNode(const InfoHash& id, const sockaddr*, socklen_t, int confirm); std::shared_ptr findNode(const InfoHash& id, sa_family_t af); const std::shared_ptr findNode(const InfoHash& id, sa_family_t af) const; bool trySearchInsert(const std::shared_ptr& node); @@ -884,6 +886,19 @@ private: MessageStats in_stats {}, out_stats {}; + /* when our address is reported by a distant peer. */ + void onReportedAddr(const InfoHash& id, sockaddr* sa , socklen_t salen); + /* when we receive a ping request */ + NetworkEngine::RequestAnswer onPing(std::shared_ptr node); + /* when we receive a "find node" request */ + NetworkEngine::RequestAnswer onFindNode(std::shared_ptr node, InfoHash& hash, want_t want); + /* when we receive a "get values" request */ + NetworkEngine::RequestAnswer onGetValues(std::shared_ptr node, InfoHash& hash, want_t want); + /* when we receive a listen request */ + NetworkEngine::RequestAnswer onListen(std::shared_ptr node, InfoHash& hash, Blob& token, size_t rid); + /* when we receive an announce request */ + NetworkEngine::RequestAnswer onAnnounce(std::shared_ptr node, + InfoHash& hash, Blob& token, std::vector> v, time_point created); }; } diff --git a/include/opendht/network_engine.h b/include/opendht/network_engine.h index e7ac8636..752eb4d8 100644 --- a/include/opendht/network_engine.h +++ b/include/opendht/network_engine.h @@ -50,12 +50,12 @@ public: static const constexpr uint16_t TRUNCATED_ID {421}; /* id was truncated. */ static const constexpr uint16_t WRONG_NODE_INFO_BUF_LEN {422}; /* node info length is wrong */ - static const std::string VALUES_BUT_NO_INFOHASH; - static const std::string LISTEN_NO_INFOHASH; - static const std::string LISTEN_WRONG_TOKEN; - static const std::string PUT_NO_INFOHASH; - static const std::string PUT_WRONG_TOKEN; - static const std::string PUT_INVALID_ID; + static const std::string GET_NO_INFOHASH; /* received get request with no infohash */ + static const std::string LISTEN_NO_INFOHASH; /* got listen request without infohash */ + static const std::string LISTEN_WRONG_TOKEN; /* wrong token in listen request */ + static const std::string PUT_NO_INFOHASH; /* no infohash in put request */ + static const std::string PUT_WRONG_TOKEN; /* got put request with wrong token */ + static const std::string PUT_INVALID_ID; /* invalid id in put request */ DhtProtocolException(uint16_t code, const std::string& msg="") : DhtException(msg), code(code), msg(msg) {} @@ -107,13 +107,13 @@ private: * @param saddr (type: sockaddr*) sockaddr* pointer containing address ip information. * @param saddr_len (type: socklen_t) lenght of the sockaddr struct. */ - std::function(InfoHash, sockaddr*, socklen_t, int)> onNewNode; + std::function(const InfoHash&, const sockaddr*, socklen_t, int)> onNewNode; /** * @brief when an addres is reported from a distant node. * * @param arg1 Arg description */ - std::function onReportedAddr; + std::function onReportedAddr; /** * @brief on ping request callback. * @@ -129,7 +129,7 @@ private: * or ipv6. */ std::function, - InfoHash, + InfoHash&, want_t)> onFindNode {}; /** * @brief on "get values" request callback. @@ -140,7 +140,7 @@ private: * or ipv6. */ std::function, - InfoHash, + InfoHash&, want_t)> onGetValues {}; /** * @brief on listen request callback. @@ -151,8 +151,8 @@ private: * @param rid (type: size_t) request id. */ std::function, - InfoHash, - Blob, + InfoHash&, + Blob&, size_t)> onListen {}; /** * @brief on announce request callback. @@ -164,9 +164,9 @@ private: * @param created (type: time_point) time when the value was created. */ std::function, - InfoHash, - Blob, - std::shared_ptr, + InfoHash&, + Blob&, + std::vector>, time_point)> onAnnounce {}; public: @@ -179,23 +179,26 @@ public: // Config from Dht. struct DhtInfo { - InfoHash& myid; - int& dht_socket; - int& dht_socket6; + const InfoHash& myid; + const int& dht_socket; + const int& dht_socket6; }; NetworkEngine(DhtInfo info, + decltype(NetworkEngine::onNewNode) onNewNode, + decltype(NetworkEngine::onReportedAddr) onReportedAddr, decltype(NetworkEngine::onPing) onPing, decltype(NetworkEngine::onFindNode) onFindNode, decltype(NetworkEngine::onGetValues) onGetValues, decltype(NetworkEngine::onListen) onListen, decltype(NetworkEngine::onAnnounce) onAnnounce) : - onPing(onPing), onFindNode(onFindNode), onGetValues(onGetValues), onListen(onListen), onAnnounce(onAnnounce), - myid(info.myid), dht_socket(info.dht_socket), dht_socket6(info.dht_socket6) + onNewNode(onNewNode), onReportedAddr(onReportedAddr), onPing(onPing), onFindNode(onFindNode), + onGetValues(onGetValues), onListen(onListen), onAnnounce(onAnnounce), myid(info.myid), + dht_socket(info.dht_socket), dht_socket6(info.dht_socket6) { req_ids = std::uniform_int_distribution{1}(rd_device); } - virtual ~NetworkEngine(); + virtual ~NetworkEngine() {}; /** @@ -238,7 +241,7 @@ public: RequestCb on_done, RequestCb on_expired); - void processMessage(const uint8_t *buf, size_t buflen, const sockaddr *from, socklen_t fromlen, time_point now); + time_point processMessage(const uint8_t *buf, size_t buflen, const sockaddr *from, socklen_t fromlen, time_point now); private: /*************** diff --git a/src/dht.cpp b/src/dht.cpp index c3a23d3a..02ac0427 100644 --- a/src/dht.cpp +++ b/src/dht.cpp @@ -538,7 +538,8 @@ void Dht::reportedAddr(const sockaddr *sa, socklen_t sa_len) { auto it = std::find_if(reported_addr.begin(), reported_addr.end(), [=](const ReportedAddr& addr){ - return (addr.second.second == sa_len) && std::equal((uint8_t*)&addr.second.first, (uint8_t*)&addr.second.first + addr.second.second, (uint8_t*)sa); + return (addr.second.second == sa_len) && + std::equal((uint8_t*)&addr.second.first, (uint8_t*)&addr.second.first + addr.second.second, (uint8_t*)sa); }); if (it == reported_addr.end()) { if (reported_addr.size() < 32) @@ -550,7 +551,7 @@ Dht::reportedAddr(const sockaddr *sa, socklen_t sa_len) /* We just learnt about a node, not necessarily a new one. Confirm is 1 if the node sent a message, 2 if it sent us a reply. */ std::shared_ptr -Dht::newNode(const InfoHash& id, const sockaddr *sa, socklen_t salen, int confirm, const sockaddr* addr, socklen_t addr_length) +Dht::newNode(const InfoHash& id, const sockaddr *sa, socklen_t salen, int confirm) { if (id == myid || isMartian(sa, salen) || isNodeBlacklisted(sa, salen)) return nullptr; @@ -562,12 +563,6 @@ Dht::newNode(const InfoHash& id, const sockaddr *sa, socklen_t salen, int confir bool mybucket = list.contains(b, myid); - if (confirm == 2) { - b->time = now; - if (addr and addr_length) - reportedAddr(addr, addr_length); - } - for (auto& n : b->nodes) { if (n->id != id) continue; /* Known node. Update stuff. */ @@ -1735,7 +1730,7 @@ Dht::Storage::clear() } void -Dht::storageAddListener(const InfoHash& id, const InfoHash& node, const sockaddr *from, socklen_t fromlen, uint16_t tid) +Dht::storageAddListener(const InfoHash& id, const InfoHash& node, const sockaddr *from, socklen_t fromlen, size_t rid) { auto st = findStorage(id); if (st == store.end()) { @@ -1751,10 +1746,10 @@ Dht::storageAddListener(const InfoHash& id, const InfoHash& node, const sockaddr if (l == st->listeners.end()) { //TODO //sendClosestNodes(from, fromlen, TransId {TransPrefix::GET_VALUES, tid}, st->id, WANT4 | WANT6, makeToken(from, false), st->getValues()); - st->listeners.emplace_back(node, from, fromlen, tid, now); + st->listeners.emplace_back(node, from, fromlen, rid, now); } else - l->refresh(from, fromlen, tid, now); + l->refresh(from, fromlen, rid, now); } void @@ -2118,6 +2113,20 @@ Dht::Dht(int s, int s6, Config config) expireBuckets(buckets); expireBuckets(buckets6); + + using namespace std::placeholders; + network_engine = std::unique_ptr( + new NetworkEngine { + NetworkEngine::DhtInfo {myid, dht_socket, dht_socket6}, + std::bind(&Dht::newNode, this, _1, _2, _3, _4), + std::bind(&Dht::onReportedAddr, this, _1, _2, _3), + std::bind(&Dht::onPing, this, _1), + std::bind(&Dht::onFindNode, this, _1, _2, _3), + std::bind(&Dht::onGetValues, this, _1, _2, _3), + std::bind(&Dht::onListen, this, _1, _2, _3, _4), + std::bind(&Dht::onAnnounce, this, _1, _2, _3, _4, _5) + } + ); DHT_DEBUG("DHT initialised with node ID %s", myid.toString().c_str()); } @@ -2397,4 +2406,161 @@ Dht::pingNode(const sockaddr *sa, socklen_t salen) return 0; } +void +Dht::onReportedAddr(const InfoHash& id, sockaddr* addr , socklen_t addr_length) { + const auto& b = (addr->sa_family == AF_INET ? buckets : buckets6).findBucket(id); + b->time = now; + if (addr and addr_length) + reportedAddr(addr, addr_length); +} + +NetworkEngine::RequestAnswer +Dht::onPing(std::shared_ptr node) { + in_stats.ping++; + //DHT_DEBUG("Sending pong."); + return std::move(NetworkEngine::RequestAnswer {}); +} + +NetworkEngine::RequestAnswer +Dht::onFindNode(std::shared_ptr node, InfoHash& hash, want_t want) { + in_stats.find++; + DHT_DEBUG("[node %s %s] got 'find' request (%d).", node->id.toString().c_str(), print_addr(node->ss, node->sslen).c_str(), want); + Blob ntoken = makeToken((sockaddr*)&node->ss, false); + + NetworkEngine::RequestAnswer answer { + ntoken, + {}, + buckets.findClosestNodes(hash, now, TARGET_NODES), + buckets6.findClosestNodes(hash, now, TARGET_NODES) + }; + return std::move(answer); +} + +NetworkEngine::RequestAnswer +Dht::onGetValues(std::shared_ptr node, InfoHash& hash, want_t want) { + NetworkEngine::RequestAnswer* answer; + in_stats.get++; + DHT_DEBUG("[node %s %s] got 'get' request for %s.", node->id.toString().c_str(), print_addr(node->ss, node->sslen).c_str(), hash.toString().c_str()); + if (hash == zeroes) { + DHT_WARN("[node %s %s] Eek! Got get_values with no info_hash.", node->id.toString().c_str(), print_addr(node->ss, node->sslen).c_str()); + throw DhtProtocolException {DhtProtocolException::NON_AUTHORITATIVE_INFORMATION, DhtProtocolException::GET_NO_INFOHASH}; + } else { + auto st = findStorage(hash); + Blob ntoken = makeToken((sockaddr*)&node->ss, false); + answer = new NetworkEngine::RequestAnswer { + ntoken, + {}, + buckets.findClosestNodes(hash, now, TARGET_NODES), + buckets6.findClosestNodes(hash, now, TARGET_NODES) + }; + if (st != store.end() && not st->empty()) { + DHT_DEBUG("[node %s %s] sending %u values.", node->id.toString().c_str(), print_addr(node->ss, node->sslen).c_str(), st->valueCount()); + //answer->values.insert(answer->values.end(), st->getValues()); + auto values = st->getValues(); + answer->values.resize(values.size()); + std::transform(values.begin(), values.end(), answer->values.begin(), [](const ValueStorage& vs) { + return std::move(vs.data); + }); + } else { + DHT_DEBUG("[node %s %s] sending nodes.", node->id.toString().c_str(), print_addr(node->ss, node->sslen).c_str()); + } + } + return std::move(*answer); +} + +NetworkEngine::RequestAnswer +Dht::onListen(std::shared_ptr node, InfoHash& hash, Blob& token, size_t rid) { + in_stats.listen++; + DHT_DEBUG("[node %s %s] got 'listen' request for %s.", node->id.toString().c_str(), print_addr(node->ss, node->sslen).c_str(), hash.toString().c_str()); + if (hash == zeroes) { + DHT_WARN("Listen with no info_hash."); + throw DhtProtocolException {DhtProtocolException::NON_AUTHORITATIVE_INFORMATION, DhtProtocolException::LISTEN_NO_INFOHASH}; + } + if (!tokenMatch(token, (sockaddr*)&node->ss)) { + DHT_WARN("[node %s %s] incorrect token %s for 'listen'.", + node->id.toString().c_str(), print_addr(node->ss, node->sslen).c_str(), + hash.toString().c_str(), to_hex(token.data(), token.size()).c_str()); + throw DhtProtocolException {DhtProtocolException::UNAUTHORIZED, DhtProtocolException::LISTEN_WRONG_TOKEN}; + } + //TODO in NetworkEngine + //if (!msg.tid.matches(TransPrefix::LISTEN, &ttid)) { + // break; + //} + storageAddListener(hash, node->id, (sockaddr*)&node->ss, node->sslen, rid); + return std::move(NetworkEngine::RequestAnswer {}); +} + +NetworkEngine::RequestAnswer +Dht::onAnnounce(std::shared_ptr node, InfoHash& hash, Blob& token, std::vector> values, time_point created) { + in_stats.put++; + DHT_DEBUG("[node %s %s] got 'put' request for %s.", + node->id.toString().c_str(), print_addr(node->ss, node->sslen).c_str(), + hash.toString().c_str()); + if (hash == zeroes) { + DHT_WARN("Put with no info_hash."); + throw DhtProtocolException {DhtProtocolException::NON_AUTHORITATIVE_INFORMATION, DhtProtocolException::PUT_NO_INFOHASH}; + } + if (!tokenMatch(token, (sockaddr*)&node->ss)) { + DHT_WARN("[node %s %s] incorrect token %s for 'put'.", + node->id.toString().c_str(), print_addr(node->ss, node->sslen).c_str(), + hash.toString().c_str(), to_hex(token.data(), token.size()).c_str()); + throw DhtProtocolException {DhtProtocolException::UNAUTHORIZED, DhtProtocolException::PUT_WRONG_TOKEN}; + } + { + // We store a value only if we think we're part of the + // SEARCH_NODES nodes around the target id. + auto closest_nodes = (((sockaddr*)&node->ss)->sa_family == AF_INET ? buckets : buckets6).findClosestNodes(hash, now, SEARCH_NODES); + if (hash.xorCmp(closest_nodes.back()->id, myid) < 0) { + DHT_WARN("[node %s %s] announce too far node->ss the target id. Dropping value.", + node->id.toString().c_str(), print_addr(node->ss, node->sslen).c_str()); + //TODO: NetworkEngine + //for (auto& v : values) { + // sendValueAnnounced(node->ss, node->sslen, msg.tid, v->id); + //} + return std::move(NetworkEngine::RequestAnswer {}); + } + } + + for (const auto& v : values) { + if (v->id == Value::INVALID_ID) { + DHT_WARN("[value %s %s] incorrect value id", hash.toString().c_str(), v->id); + throw DhtProtocolException { + DhtProtocolException::NON_AUTHORITATIVE_INFORMATION, + DhtProtocolException::PUT_INVALID_ID + }; + } + auto lv = getLocalById(hash, v->id); + std::shared_ptr vc = v; + if (lv) { + if (*lv == *vc) { + DHT_WARN("[value %s %lu] nothing to do.", hash.toString().c_str(), lv->id); + } else { + const auto& type = getType(lv->type); + if (type.editPolicy(hash, lv, vc, node->id, (sockaddr*)&node->ss, node->sslen)) { + DHT_DEBUG("[value %s %lu] editing %s.", hash.toString().c_str(), lv->id, vc->toString().c_str()); + storageStore(hash, vc, created); + } else { + DHT_DEBUG("[value %s %lu] rejecting edition of %s because of storage policy.", hash.toString().c_str(), lv->id, vc->toString().c_str()); + } + } + } else { + // Allow the value to be edited by the storage policy + const auto& type = getType(vc->type); + if (type.storePolicy(hash, vc, node->id, (sockaddr*)&node->ss, node->sslen)) { + DHT_DEBUG("[value %s %lu] storing %s.", hash.toString().c_str(), vc->id, vc->toString().c_str()); + storageStore(hash, vc, created); + } else { + DHT_DEBUG("[value %s %lu] rejecting storage of %s.", hash.toString().c_str(), vc->id, vc->toString().c_str()); + } + } + + /* Note that if storageStore failed, we lie to the requestor. + This is to prevent them node->ss backtracking, and hence + polluting the DHT. */ + //TODO: NetworkEngine + //sendValueAnnounced(node->ss, node->sslen, msg.tid, v->id); + } + return std::move(NetworkEngine::RequestAnswer {}); +} + } diff --git a/src/network_engine.cpp b/src/network_engine.cpp index f1cff2a4..0f30d8b2 100644 --- a/src/network_engine.cpp +++ b/src/network_engine.cpp @@ -30,7 +30,7 @@ THE SOFTWARE. namespace dht { -const std::string DhtProtocolException::VALUES_BUT_NO_INFOHASH {"Get_values with no info_hash"}; +const std::string DhtProtocolException::GET_NO_INFOHASH {"Get_values with no info_hash"}; const std::string DhtProtocolException::LISTEN_NO_INFOHASH {"Listen with no info_hash"}; const std::string DhtProtocolException::LISTEN_WRONG_TOKEN {"Listen with wrong token"}; const std::string DhtProtocolException::PUT_NO_INFOHASH {"Put with no info_hash"};