From f93fb92ff57305288e92e5c9c5764cfca52fd24e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Simon=20D=C3=A9saulniers?= Date: Tue, 23 Feb 2016 12:59:03 -0500 Subject: [PATCH] add NetworkEngine::processMessage --- include/opendht/dht.h | 25 +--- include/opendht/network_engine.h | 93 +++++++++----- include/opendht/utils.h | 24 ++-- src/dht.cpp | 191 ++++++++++++++++------------ src/network_engine.cpp | 209 ++++++++++++++++++++++++------- src/utils.cpp | 7 ++ 6 files changed, 363 insertions(+), 186 deletions(-) diff --git a/include/opendht/dht.h b/include/opendht/dht.h index c3c926d1..672e186a 100644 --- a/include/opendht/dht.h +++ b/include/opendht/dht.h @@ -280,19 +280,15 @@ public: std::vector exportValues() const; void importValues(const std::vector&); - int getNodesStats(sa_family_t af, unsigned *good_return, unsigned *dubious_return, unsigned *cached_return, unsigned *incoming_return) const; + int getNodesStats(sa_family_t af, unsigned *good_return, unsigned *dubious_return, unsigned *cached_return, + unsigned *incoming_return) const; std::string getStorageLog() const; std::string getRoutingTablesLog(sa_family_t) const; std::string getSearchesLog(sa_family_t) const; void dumpTables() const; std::vector getNodeMessageStats(bool in = false) { - auto stats = in ? std::vector{in_stats.ping, in_stats.find, in_stats.get, in_stats.listen, in_stats.put} - : std::vector{out_stats.ping, out_stats.find, out_stats.get, out_stats.listen, out_stats.put}; - if (in) { in_stats = {}; } - else { out_stats = {}; } - - return stats; + return network_engine->getNodeMessageStats(in); } /** @@ -355,8 +351,6 @@ private: reason to increase this value. */ static constexpr unsigned BLACKLISTED_MAX {10}; - static constexpr long unsigned MAX_REQUESTS_PER_SEC {1600}; - static constexpr size_t TOKEN_SIZE {64}; struct NodeCache { @@ -776,7 +770,6 @@ private: time_point search_time {time_point::max()}; time_point confirm_nodes_time {time_point::min()}; time_point rotate_secrets_time {time_point::min()}; - std::queue rate_limit_time {}; using ReportedAddr = std::pair; std::vector reported_addr; @@ -871,19 +864,11 @@ private: void searchStep(Search& sr); void dumpSearch(const Search& sr, std::ostream& out) const; - bool rateLimit(); bool neighbourhoodMaintenance(RoutingTable&); - struct MessageStats { - unsigned ping {0}; - unsigned find {0}; - unsigned get {0}; - unsigned put {0}; - unsigned listen {0}; - }; - - MessageStats in_stats {}, out_stats {}; + void processMessage(const uint8_t *buf, size_t buflen, const sockaddr *from, socklen_t fromlen); + void onError(std::shared_ptr node, DhtProtocolException e); /* 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 */ diff --git a/include/opendht/network_engine.h b/include/opendht/network_engine.h index 8ba7d3f7..99d01efc 100644 --- a/include/opendht/network_engine.h +++ b/include/opendht/network_engine.h @@ -38,6 +38,7 @@ THE SOFTWARE. #include #include #include +#include namespace dht { @@ -47,29 +48,32 @@ public: static const constexpr uint16_t NON_AUTHORITATIVE_INFORMATION {203}; /* incomplete request packet. */ static const constexpr uint16_t UNAUTHORIZED {401}; /* wrong tokens. */ // for internal use (custom). - static const constexpr uint16_t TRUNCATED_ID {421}; /* id was truncated. */ + static const constexpr uint16_t INVALID_REPLY_TID {421}; /* id was truncated. */ static const constexpr uint16_t WRONG_NODE_INFO_BUF_LEN {422}; /* node info length is wrong */ - 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 */ + 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) {} + DhtProtocolException(uint16_t code, const std::string& msg="", InfoHash failing_node_id={}) + : DhtException(msg), code(code), msg(msg), failing_node_id(failing_node_id) {} - std::string getMsg() { return msg; } - uint16_t getCode() { return code; } + std::string getMsg() const { return msg; } + uint16_t getCode() const { return code; } + const InfoHash getNodeId() const { return failing_node_id; } private: uint16_t code; + const InfoHash failing_node_id; std::string msg; }; /*! * @class NetworkEngine - * @brief A protocol abstraction of communication on the network. + * @brief An abstraction of communication protocol on the network. * @details * The NetworkEngine processes all requests to nodes by offering a public * interface for handling sending and receiving packets. The following @@ -98,6 +102,13 @@ public: }; private: + + /** + * @brief when we receive an error message. + * + * @param node (type: std::shared_ptr) the node we received the error from. + */ + std::function, DhtProtocolException e)> onError; /** * @brief when a new node happens. * @@ -160,7 +171,7 @@ private: * @param node (type: std::shared_ptr) the requesting node. * @param vhash (type: InfoHash) hash of the value of interest. * @param token (type: Blob) security token. - * @param value (type: std::shared_ptr) value to send. + * @param values (type: std::vector>) values to store. * @param created (type: time_point) time when the value was created. */ std::function, @@ -186,6 +197,7 @@ public: }; NetworkEngine(DhtInfo info, + decltype(NetworkEngine::onError) onError, decltype(NetworkEngine::onNewNode) onNewNode, decltype(NetworkEngine::onReportedAddr) onReportedAddr, decltype(NetworkEngine::onPing) onPing, @@ -193,7 +205,7 @@ public: decltype(NetworkEngine::onGetValues) onGetValues, decltype(NetworkEngine::onListen) onListen, decltype(NetworkEngine::onAnnounce) onAnnounce) : - onNewNode(onNewNode), onReportedAddr(onReportedAddr), onPing(onPing), onFindNode(onFindNode), + onError(onError), 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), DHT_LOG(info.DHT_LOG) { @@ -215,6 +227,7 @@ public: * Requests * **************/ size_t sendPing(std::shared_ptr n, RequestCb on_done, RequestCb on_expired); + size_t sendPing(sockaddr* n, socklen_t salen, RequestCb on_done, RequestCb on_expired); size_t sendFindNode(std::shared_ptr n, const InfoHash& target, want_t want, @@ -242,12 +255,31 @@ public: RequestCb on_done, RequestCb on_expired); - time_point processMessage(const uint8_t *buf, size_t buflen, const sockaddr *from, socklen_t fromlen, time_point now); + /** + * Parses a message and calls appropriate callbacks. + * + * @param buf The buffer containing the binary message. + * @param buflen The length of the buffer. + * @param from The address info of the sender. + * @param fromlen The length of the corresponding sockaddr structure. + * @param now The time to adjust the clock in the network engine. + */ + void processMessage(const uint8_t *buf, size_t buflen, const sockaddr *from, socklen_t fromlen, time_point now); + + std::vector getNodeMessageStats(bool in) { + auto stats = in ? std::vector{in_stats.ping, in_stats.find, in_stats.get, in_stats.listen, in_stats.put} + : std::vector{out_stats.ping, out_stats.find, out_stats.get, out_stats.listen, out_stats.put}; + if (in) { in_stats = {}; } + else { out_stats = {}; } + + return stats; + } private: /*************** * Constants * ***************/ + static constexpr long unsigned MAX_REQUESTS_PER_SEC {1600}; /* the length of a node info buffer in ipv4 format */ static const constexpr size_t NODE4_INFO_BUF_LEN {26}; /* the length of a node info buffer in ipv6 format */ @@ -262,6 +294,7 @@ private: const Logger& DHT_LOG; + bool rateLimit(); struct TransPrefix : public std::array { TransPrefix(const std::string& str) : std::array({{(uint8_t)str[0], (uint8_t)str[1]}}) {} @@ -334,13 +367,6 @@ private: void msgpack_unpack(msgpack::object o); }; - //TODO: balancer ce code dans NetworkEngine::processMessage - //void onReply(ParsedMessage& msg) { - // if (msg.tid.length != 4) - // throw DhtProtocolException {DhtProtocolException::TRUNCATED_ID}; - // onNewNode(msg.id, msg.); - //} - /** * When a request has expired, i.e when Request::MAX_ATTEMPT_COUNT attempts * of processing the request have been made, we consider a node expired. @@ -369,7 +395,7 @@ private: std::function on_expired) : id(id), node(node), msg(msg), on_done(on_done), on_expired(on_expired) {} - bool expired() { return attempt_count >= Request::MAX_ATTEMPT_COUNT; } + bool expired() const { return attempt_count >= Request::MAX_ATTEMPT_COUNT; } std::function on_done {}; std::function on_expired {}; @@ -392,6 +418,14 @@ private: return req_ids == Request::INVALID_ID ? ++req_ids : req_ids; } + struct MessageStats { + unsigned ping {0}; + unsigned find {0}; + unsigned get {0}; + unsigned put {0}; + unsigned listen {0}; + }; + // basic wrapper for socket sendto function int send(const char *buf, size_t len, int flags, const sockaddr *sa, socklen_t salen); @@ -405,20 +439,14 @@ private: void sendNodesValues(const sockaddr* sa, socklen_t salen, TransId tid, - const uint8_t *nodes, - unsigned nodes_len, - const uint8_t *nodes6, - unsigned nodes6_len, + const Blob nodes, + const Blob nodes6, const std::vector>& st, const Blob& token); unsigned insertClosestNode(uint8_t *nodes, unsigned numnodes, const InfoHash& id, const Node& n); - std::pair - bufferNodes(const sockaddr *sa, - socklen_t salen, - TransId tid, + std::pair bufferNodes(sa_family_t af, const InfoHash& id, want_t want, - const Blob& token, const std::vector>& nodes, const std::vector>& nodes6); /* answer to a listen request */ @@ -435,13 +463,14 @@ private: RequestAnswer deserializeNodesValues(ParsedMessage& msg); + std::queue rate_limit_time {}; static std::mt19937 rd_device; time_point now; // requests handling uint32_t req_ids {1}; std::map> requests; - std::map> timers; /** callbacks in case of expired requests. */ + MessageStats in_stats {}, out_stats {}; }; } diff --git a/include/opendht/utils.h b/include/opendht/utils.h index d6aa1803..3cdcf10b 100644 --- a/include/opendht/utils.h +++ b/include/opendht/utils.h @@ -31,6 +31,19 @@ #include +namespace dht { + +static constexpr InfoHash zeroes; +static constexpr InfoHash ones; +static constexpr unsigned TARGET_NODES {8}; + +using Address = std::pair; +using want_t = int_fast8_t; + +std::string print_addr(const sockaddr* sa, socklen_t slen); +std::string print_addr(const sockaddr_storage& ss, socklen_t sslen); +std::string printAddr(const Address& addr); + template void erase_if(std::map& map, const Condition& condition) { @@ -41,17 +54,6 @@ void erase_if(std::map& map, const Condition& condition) } } -namespace dht { - -static constexpr unsigned TARGET_NODES {8}; - -using Address = std::pair; -using want_t = int_fast8_t; - -std::string print_addr(const sockaddr* sa, socklen_t slen); -std::string print_addr(const sockaddr_storage& ss, socklen_t sslen); -std::string printAddr(const Address& addr); - class DhtException : public std::runtime_error { public: DhtException(const std::string &str = "") : diff --git a/src/dht.cpp b/src/dht.cpp index ec98d8bd..7d1f4332 100644 --- a/src/dht.cpp +++ b/src/dht.cpp @@ -99,15 +99,12 @@ print_dt(DT d) { namespace dht { -static constexpr InfoHash zeroes {}; - constexpr std::chrono::seconds Dht::SEARCH_GET_STEP; constexpr std::chrono::minutes Dht::MAX_STORAGE_MAINTENANCE_EXPIRE_TIME; constexpr std::chrono::minutes Dht::SEARCH_EXPIRE_TIME; constexpr std::chrono::seconds Dht::LISTEN_EXPIRE_TIME; constexpr std::chrono::seconds Dht::REANNOUNCE_MARGIN; constexpr std::chrono::seconds Dht::UDP_REPLY_TIME; -constexpr long unsigned Dht::MAX_REQUESTS_PER_SEC; void Dht::setLoggers(LogMethod&& error, LogMethod&& warn, LogMethod&& debug) @@ -2118,6 +2115,7 @@ Dht::Dht(int s, int s6, Config config) network_engine = std::unique_ptr( new NetworkEngine { NetworkEngine::DhtInfo {myid, dht_socket, dht_socket6, DHT_LOG}, + std::bind(&Dht::onError, this, _1, _2), std::bind(&Dht::newNode, this, _1, _2, _3, _4), std::bind(&Dht::onReportedAddr, this, _1, _2, _3), std::bind(&Dht::onPing, this, _1), @@ -2131,21 +2129,6 @@ Dht::Dht(int s, int s6, Config config) } -/* Rate control for requests we receive. */ -bool -Dht::rateLimit() -{ - using namespace std::chrono; - while (not rate_limit_time.empty() and duration_cast(now - rate_limit_time.front()) > seconds(1)) - rate_limit_time.pop(); - - if (rate_limit_time.size() >= MAX_REQUESTS_PER_SEC) - return false; - - rate_limit_time.emplace(now); - return true; -} - bool Dht::neighbourhoodMaintenance(RoutingTable& list) { @@ -2288,14 +2271,41 @@ Dht::maintainStorage(InfoHash id, bool force, DoneCallback donecb) { return announce_per_af; } +void +Dht::processMessage(const uint8_t *buf, size_t buflen, const sockaddr *from, socklen_t fromlen) { + if (buflen == 0) + return; + + if (isMartian(from, fromlen)) + return; + + if (isNodeBlacklisted(from, fromlen)) { DHT_LOG.DEBUG("Received packet from blacklisted node."); - DHT_LOG.WARN("[node %s %s] token flush (%d searches affected)", + return; + } + + try { + network_engine->processMessage(buf, buflen, from, fromlen, now); + } catch (DhtProtocolException& e) { + auto code = e.getCode(); + if (code == DhtProtocolException::INVALID_REPLY_TID + or code == DhtProtocolException::WRONG_NODE_INFO_BUF_LEN) { + /* This is really annoying, as it means that we will + time-out all our searches that go through this node. + Kill it. */ + const auto& id = e.getNodeId(); + blacklistNode(&id, from, fromlen); + } + } +} + time_point Dht::periodic(const uint8_t *buf, size_t buflen, const sockaddr *from, socklen_t fromlen) { - //TODO - DHT_LOG.DEBUG("next search time : (none)"); - DHT_LOG.DEBUG("next search time : %lf s%s", print_dt(search_time-now), (search_time < now)?" (ASAP)":"");*/ + using namespace std::chrono; + now = clock::now(); + + processMessage(buf, buflen, from, fromlen); } std::vector @@ -2407,11 +2417,35 @@ Dht::pingNode(const sockaddr *sa, socklen_t salen) { DHT_LOG.DEBUG("Sending ping to %s", print_addr(sa, salen).c_str()); //return sendPing(sa, salen, TransId {TransPrefix::PING}); - return 0; + return -1; } void -Dht::onReportedAddr(const InfoHash& id, sockaddr* addr , socklen_t addr_length) { +Dht::onError(std::shared_ptr node, DhtProtocolException err) { + if (err.getCode() == DhtProtocolException::UNAUTHORIZED) { + //TODO: utiliser map de search + /*auto esr = findSearch(ttid, from->sa_family);*/ + //if (!esr) return; + unsigned cleared = 0; + for (auto& sr : searches) { + for (auto& n : sr.nodes) { + if (n.node != node) continue; + cleared++; + n.getStatus.request_time = TIME_INVALID; + n.getStatus.reply_time = TIME_INVALID; + if (searchSendGetValues(sr)) + sr.get_step_time = now; + break; + } + } + DHT_LOG.WARN("[node %s %s] token flush (%d searches affected)", + node->id.toString().c_str(), print_addr((sockaddr*)&node->ss, node->sslen).c_str(), cleared); + } +} + +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) @@ -2419,66 +2453,74 @@ Dht::onReportedAddr(const InfoHash& id, sockaddr* addr , socklen_t addr_length) } NetworkEngine::RequestAnswer -Dht::onPing(std::shared_ptr node) { - in_stats.ping++; - //DHT_DEBUG("Sending pong."); - return std::move(NetworkEngine::RequestAnswer {}); +Dht::onPing(std::shared_ptr node) +{ + return {}; } NetworkEngine::RequestAnswer -Dht::onFindNode(std::shared_ptr node, InfoHash& hash, want_t want) { - in_stats.find++; - DHT_LOG.DEBUG("[node %s %s] got 'find' request (%d).", node->id.toString().c_str(), print_addr(node->ss, node->sslen).c_str(), want); +Dht::onFindNode(std::shared_ptr node, InfoHash& hash, want_t want) +{ Blob ntoken = makeToken((sockaddr*)&node->ss, false); - - NetworkEngine::RequestAnswer answer { - ntoken, - {}, - buckets.findClosestNodes(hash, now, TARGET_NODES), - buckets6.findClosestNodes(hash, now, TARGET_NODES) - }; + std::vector> nodes, nodes6; + if ((want & WANT4)) { + auto tmp = buckets.findClosestNodes(hash, now, TARGET_NODES); + nodes.insert(nodes.begin(), tmp.begin(), tmp.end()); + } + if ((want & WANT6)) { + auto tmp = buckets6.findClosestNodes(hash, now, TARGET_NODES); + nodes6.insert(nodes6.begin(), tmp.begin(), tmp.end()); + } + NetworkEngine::RequestAnswer answer {ntoken, {}, std::move(nodes), std::move(nodes6)}; return std::move(answer); } NetworkEngine::RequestAnswer -Dht::onGetValues(std::shared_ptr node, InfoHash& hash, want_t want) { +Dht::onGetValues(std::shared_ptr node, InfoHash& hash, want_t want) +{ NetworkEngine::RequestAnswer* answer; - in_stats.get++; - DHT_LOG.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()); + DHT_LOG.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_LOG.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()); + DHT_LOG.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) + {}, + buckets.findClosestNodes(hash, now, TARGET_NODES), + buckets6.findClosestNodes(hash, now, TARGET_NODES) }; if (st != store.end() && not st->empty()) { - DHT_LOG.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()); + DHT_LOG.DEBUG("[node %s %s] sending %u values.", + node->id.toString().c_str(), print_addr(node->ss, node->sslen).c_str(), st->valueCount()); 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_LOG.DEBUG("[node %s %s] sending nodes.", node->id.toString().c_str(), print_addr(node->ss, node->sslen).c_str()); + DHT_LOG.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_LOG.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()); +Dht::onListen(std::shared_ptr node, InfoHash& hash, Blob& token, size_t rid) +{ + DHT_LOG.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_LOG.WARN("Listen with no info_hash."); - throw DhtProtocolException {DhtProtocolException::NON_AUTHORITATIVE_INFORMATION, DhtProtocolException::LISTEN_NO_INFOHASH}; + throw DhtProtocolException { + DhtProtocolException::NON_AUTHORITATIVE_INFORMATION, + DhtProtocolException::LISTEN_NO_INFOHASH + }; } if (!tokenMatch(token, (sockaddr*)&node->ss)) { DHT_LOG.WARN("[node %s %s] incorrect token %s for 'listen'.", @@ -2486,23 +2528,23 @@ Dht::onListen(std::shared_ptr node, InfoHash& hash, Blob& token, size_t ri 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 {}); + return {}; } NetworkEngine::RequestAnswer -Dht::onAnnounce(std::shared_ptr node, InfoHash& hash, Blob& token, std::vector> values, time_point created) { - in_stats.put++; +Dht::onAnnounce(std::shared_ptr node, InfoHash& hash, Blob& token, std::vector> values, + time_point created) +{ DHT_LOG.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_LOG.WARN("Put with no info_hash."); - throw DhtProtocolException {DhtProtocolException::NON_AUTHORITATIVE_INFORMATION, DhtProtocolException::PUT_NO_INFOHASH}; + throw DhtProtocolException { + DhtProtocolException::NON_AUTHORITATIVE_INFORMATION, + DhtProtocolException::PUT_NO_INFOHASH + }; } if (!tokenMatch(token, (sockaddr*)&node->ss)) { DHT_LOG.WARN("[node %s %s] incorrect token %s for 'put'.", @@ -2513,15 +2555,13 @@ Dht::onAnnounce(std::shared_ptr node, InfoHash& hash, Blob& token, std::ve { // 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); + 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_LOG.WARN("[node %s %s] announce too far node->ss the target id. Dropping value.", + DHT_LOG.WARN("[node %s %s] announce too far from 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 {}); + return {}; } } @@ -2541,10 +2581,12 @@ Dht::onAnnounce(std::shared_ptr node, InfoHash& hash, Blob& token, std::ve } else { const auto& type = getType(lv->type); if (type.editPolicy(hash, lv, vc, node->id, (sockaddr*)&node->ss, node->sslen)) { - DHT_LOG.DEBUG("[value %s %lu] editing %s.", hash.toString().c_str(), lv->id, vc->toString().c_str()); + DHT_LOG.DEBUG("[value %s %lu] editing %s.", + hash.toString().c_str(), lv->id, vc->toString().c_str()); storageStore(hash, vc, created); } else { - DHT_LOG.DEBUG("[value %s %lu] rejecting edition of %s because of storage policy.", hash.toString().c_str(), lv->id, vc->toString().c_str()); + DHT_LOG.DEBUG("[value %s %lu] rejecting edition of %s because of storage policy.", + hash.toString().c_str(), lv->id, vc->toString().c_str()); } } } else { @@ -2554,17 +2596,12 @@ Dht::onAnnounce(std::shared_ptr node, InfoHash& hash, Blob& token, std::ve DHT_LOG.DEBUG("[value %s %lu] storing %s.", hash.toString().c_str(), vc->id, vc->toString().c_str()); storageStore(hash, vc, created); } else { - DHT_LOG.DEBUG("[value %s %lu] rejecting storage of %s.", hash.toString().c_str(), vc->id, vc->toString().c_str()); + DHT_LOG.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 {}); + return {}; } } diff --git a/src/network_engine.cpp b/src/network_engine.cpp index 0f30d8b2..41b21603 100644 --- a/src/network_engine.cpp +++ b/src/network_engine.cpp @@ -46,11 +46,139 @@ const NetworkEngine::TransPrefix NetworkEngine::TransPrefix::FIND_NODE = {"fn"} const NetworkEngine::TransPrefix NetworkEngine::TransPrefix::GET_VALUES = {"gt"}; const NetworkEngine::TransPrefix NetworkEngine::TransPrefix::ANNOUNCE_VALUES = {"pt"}; const NetworkEngine::TransPrefix NetworkEngine::TransPrefix::LISTEN = {"lt"}; +constexpr long unsigned NetworkEngine::MAX_REQUESTS_PER_SEC; + +/* Rate control for requests we receive. */ +bool +NetworkEngine::rateLimit() +{ + using namespace std::chrono; + while (not rate_limit_time.empty() and duration_cast(now - rate_limit_time.front()) > seconds(1)) + rate_limit_time.pop(); + + if (rate_limit_time.size() >= MAX_REQUESTS_PER_SEC) + return false; + + rate_limit_time.emplace(now); + return true; +} void NetworkEngine::processMessage(const uint8_t *buf, size_t buflen, const sockaddr *from, socklen_t fromlen, time_point now) { - //TODO this->now = now; + ParsedMessage msg; + try { + msgpack::unpacked msg_res = msgpack::unpack((const char*)buf, buflen); + msg.msgpack_unpack(msg_res.get()); + if (msg.type != MessageType::Error && msg.id == zeroes) + throw DhtException("no or invalid InfoHash"); + } catch (const std::exception& e) { + DHT_LOG.WARN("Can't process message of size %lu: %s.", buflen, e.what()); + DHT_LOG.DEBUG.logPrintable(buf, buflen); + return; + } + + if (msg.id == myid) { + DHT_LOG.DEBUG("Received message from self."); + return; + } + + if (msg.type > MessageType::Reply) { + /* Rate limit requests. */ + if (!rateLimit()) { + DHT_LOG.WARN("Dropping request due to rate limiting."); + return; + } + } + + if (msg.tid.length != 4) { + DHT_LOG.ERROR("Broken node truncates transaction ids (len: %d): ", msg.tid.length); + DHT_LOG.ERROR.logPrintable(buf, buflen); + throw DhtProtocolException {DhtProtocolException::INVALID_REPLY_TID}; + } + + uint16_t ttid = 0; + if (msg.type == MessageType::Error or msg.type == MessageType::Reply) { + auto node = onNewNode(msg.id, from, fromlen, 2); + switch (msg.type) { + case MessageType::Error: { + if (msg.error_code == DhtProtocolException::UNAUTHORIZED + && msg.id != zeroes + && (msg.tid.matches(TransPrefix::ANNOUNCE_VALUES, &ttid) + || msg.tid.matches(TransPrefix::LISTEN, &ttid))) { + onError(node, DhtProtocolException {DhtProtocolException::UNAUTHORIZED}); + } else { + DHT_LOG.WARN("[node %s %s] received unknown error message %u", + msg.id.toString().c_str(), print_addr(from, fromlen).c_str(), msg.error_code); + DHT_LOG.WARN.logPrintable(buf, buflen); + } + break; + } + case MessageType::Reply: + const auto& reqp = requests.find(msg.tid[2]); + Request* req = nullptr; + if (reqp != requests.end()) + req = &(*reqp->second); + + if (not req) + throw DhtProtocolException {DhtProtocolException::INVALID_REPLY_TID, "", msg.id}; + + req->on_done(req->id, std::move(msg)); + break; + } + } else { + auto node = onNewNode(msg.id, from, fromlen, 1); + try { + switch (msg.type) { + case MessageType::Ping: + in_stats.ping++; + DHT_LOG.DEBUG("Sending pong."); + onPing(node); + sendPong(from, fromlen, msg.tid); + break; + case MessageType::FindNode: { + in_stats.find++; + DHT_LOG.DEBUG("[node %s %s] got 'find' request (%d).", + msg.id.toString().c_str(), print_addr(from, fromlen).c_str(), msg.want); + RequestAnswer answer = onFindNode(node, msg.info_hash, msg.want); + auto nnodes = bufferNodes(from->sa_family, msg.target, msg.want, answer.nodes, answer.nodes6); + sendNodesValues(from, fromlen, msg.tid, nnodes.first, nnodes.second, {}, answer.ntoken); + break; + } + case MessageType::GetValues: { + in_stats.get++; + RequestAnswer answer = onGetValues(node, msg.info_hash, msg.want); + auto nnodes = bufferNodes(from->sa_family, msg.target, msg.want, answer.nodes, answer.nodes6); + sendNodesValues(from, fromlen, msg.tid, nnodes.first, nnodes.second, answer.values, answer.ntoken); + break; + } + case MessageType::AnnounceValue: { + in_stats.put++; + onAnnounce(node, msg.info_hash, msg.token, msg.values, msg.created); + + /* Note that if storageStore failed, we lie to the requestor. + This is to prevent them from backtracking, and hence + polluting the DHT. */ + for (auto& v : msg.values) { + sendValueAnnounced(from, fromlen, msg.tid, v->id); + } + break; + } + case MessageType::Listen: + if (!msg.tid.matches(TransPrefix::LISTEN, &ttid)) { + break; + } + in_stats.listen++; + RequestAnswer answer = onListen(node, msg.info_hash, msg.token, msg.tid[2]); + sendListenConfirmation(from, fromlen, msg.tid); + break; + } + } catch (const std::overflow_error& e) { + DHT_LOG.ERROR("Can't send value: buffer not large enough !"); + } catch (DhtProtocolException& e) { + sendError(from, fromlen, msg.tid, e.getCode(), e.getMsg().c_str()); + } + } } void @@ -77,12 +205,6 @@ NetworkEngine::send(const char *buf, size_t len, int flags, const sockaddr *sa, if (salen == 0) return -1; - //TODO: dans la DHT - //if (isNodeBlacklisted(sa, salen)) { - // DHT_DEBUG("Attempting to send to blacklisted node."); - // return -1; - //} - int s; if (sa->sa_family == AF_INET) s = dht_socket; @@ -127,7 +249,6 @@ NetworkEngine::clearExpiredRequests(size_t rid) { } } -// TODO: dht out_stats.ping++; size_t NetworkEngine::sendPing(std::shared_ptr n, RequestCb on_done, RequestCb on_expired) { auto tid = TransId {TransPrefix::PING}; @@ -153,6 +274,7 @@ NetworkEngine::sendPing(std::shared_ptr n, RequestCb on_done, RequestCb on }; requests.emplace(std::move(req)); send(buffer.data(), buffer.size(), 0, (sockaddr*)&n->ss, n->sslen); + out_stats.ping++; return tid[2]; } @@ -174,7 +296,6 @@ NetworkEngine::sendPong(const sockaddr* sa, socklen_t salen, TransId tid) { send(buffer.data(), buffer.size(), 0, sa, salen); } -// TODO: dht out_stats.find++; size_t NetworkEngine::sendFindNode(std::shared_ptr n, const InfoHash& target, want_t want, int confirm, RequestCb on_done, RequestCb on_expired) { @@ -212,11 +333,11 @@ NetworkEngine::sendFindNode(std::shared_ptr n, const InfoHash& target, wan }; requests.emplace(std::move(req)); send(buffer.data(), buffer.size(), confirm ? 0 : MSG_CONFIRM, (sockaddr*)&n->ss, n->sslen); + out_stats.find++; return tid[2]; } -// TODO: DHT out_stats.get++; size_t NetworkEngine::sendGetValues(std::shared_ptr n, const InfoHash& target, want_t want, int confirm, RequestCb on_done, RequestCb on_expired) { @@ -244,7 +365,7 @@ NetworkEngine::sendGetValues(std::shared_ptr n, const InfoHash& target, wa Blob b {buffer.data(), buffer.data() + buffer.size()}; Request req {getNewRequestId(), n, std::move(b), [=](size_t rid, ParsedMessage&& msg) { /* on done */ - //TODO + on_done(n, rid, deserializeNodesValues(msg)); }, [=](size_t rid, bool s) { /* on expired */ on_expired(n, rid, {}); @@ -253,6 +374,7 @@ NetworkEngine::sendGetValues(std::shared_ptr n, const InfoHash& target, wa }; requests.emplace(std::move(req)); send(buffer.data(), buffer.size(), confirm ? 0 : MSG_CONFIRM, (sockaddr*)&n->ss, n->sslen); + out_stats.get++; return tid[2]; } @@ -260,7 +382,7 @@ NetworkEngine::RequestAnswer NetworkEngine::deserializeNodesValues(ParsedMessage& msg) { RequestAnswer req_a {msg.token, std::move(msg.values)}; if (msg.nodes4.size() % NODE4_INFO_BUF_LEN != 0 || msg.nodes6.size() % NODE6_INFO_BUF_LEN != 0) { - //TODO: retourner l'erreur dans la couche supérieure pour blacklist + throw DhtProtocolException {DhtProtocolException::WRONG_NODE_INFO_BUF_LEN}; } else { // deserialize nodes for (unsigned i = 0; i < msg.nodes4.size() / NODE4_INFO_BUF_LEN; i++) { @@ -291,28 +413,26 @@ NetworkEngine::deserializeNodesValues(ParsedMessage& msg) { return std::move(req_a); } -//TODO void -NetworkEngine::sendNodesValues(const sockaddr* sa, socklen_t salen, TransId tid, const uint8_t *nodes, - unsigned nodes_len, const uint8_t *nodes6, unsigned nodes6_len, +NetworkEngine::sendNodesValues(const sockaddr* sa, socklen_t salen, TransId tid, const Blob nodes, const Blob nodes6, const std::vector>& st, const Blob& token) { msgpack::sbuffer buffer; msgpack::packer pk(&buffer); pk.pack_map(4); pk.pack(std::string("r")); - pk.pack_map(2 + (not st.empty()?1:0) + (nodes_len>0?1:0) + (nodes6_len>0?1:0) + (not token.empty()?1:0)); + pk.pack_map(2 + (not st.empty()?1:0) + (nodes.size()>0?1:0) + (nodes6.size()>0?1:0) + (not token.empty()?1:0)); pk.pack(std::string("id")); pk.pack(myid); insertAddr(pk, sa, salen); - if (nodes_len > 0) { + if (nodes.size() > 0) { pk.pack(std::string("n4")); - pk.pack_bin(nodes_len); - pk.pack_bin_body((const char*)nodes, nodes_len); + pk.pack_bin(nodes.size()); + pk.pack_bin_body((const char*)nodes.data(), nodes.size()); } - if (nodes6_len > 0) { + if (nodes6.size() > 0) { pk.pack(std::string("n6")); - pk.pack_bin(nodes6_len); - pk.pack_bin_body((const char*)nodes6, nodes6_len); + pk.pack_bin(nodes6.size()); + pk.pack_bin_body((const char*)nodes6.data(), nodes6.size()); } if (not token.empty()) { pk.pack(std::string("token")); packToken(pk, token); @@ -343,6 +463,8 @@ NetworkEngine::sendNodesValues(const sockaddr* sa, socklen_t salen, TransId tid, buffer.write((const char*)b.data(), b.size()); } + DHT_LOG.DEBUG("sending closest nodes (%d+%d nodes.)", nodes.size(), nodes6.size()); + pk.pack(std::string("t")); pk.pack_bin(tid.size()); pk.pack_bin_body((const char*)tid.data(), tid.size()); pk.pack(std::string("y")); pk.pack(std::string("r")); @@ -396,46 +518,42 @@ NetworkEngine::insertClosestNode(uint8_t *nodes, unsigned numnodes, const InfoHa return numnodes; } -std::pair -NetworkEngine::bufferNodes(const sockaddr *sa, socklen_t salen, TransId tid, const InfoHash& id, want_t want, - const Blob& token, const std::vector>& nodes, - const std::vector>& nodes6) +std::pair +NetworkEngine::bufferNodes(sa_family_t af, const InfoHash& id, want_t want, + const std::vector>& nodes, const std::vector>& nodes6) { - uint8_t bnodes[8 * 26]; - uint8_t bnodes6[8 * 38]; + uint8_t bnodes[8 * NODE4_INFO_BUF_LEN]; + uint8_t bnodes6[8 * NODE6_INFO_BUF_LEN]; + size_t numnodes, numnodes6 = 0; if (want < 0) - want = sa->sa_family == AF_INET ? WANT4 : WANT6; + want = af == AF_INET ? WANT4 : WANT6; auto buff = [=](uint8_t* nodes, const InfoHash& id, const std::vector>& closest_nodes) { size_t numnodes = 0; for (const auto& n : closest_nodes) { numnodes = insertClosestNode(nodes, numnodes, id, *n); } + return numnodes; }; if ((want & WANT4)) { - buff(bnodes, id, nodes); + numnodes = buff(bnodes, id, nodes); } if ((want & WANT6)) { - buff(bnodes6, id, nodes6); + numnodes6 = buff(bnodes6, id, nodes6); } - //DHT_DEBUG("sending closest nodes (%d+%d nodes.)", numnodes, numnodes6); - return std::make_pair(bnodes, bnodes6); - //try { - //} catch (const std::overflow_error& e) { - // DHT_ERROR("Can't send value: buffer not large enough !"); - // return -1; - //} + Blob bn4(8 * NODE4_INFO_BUF_LEN), bn6(8 * NODE6_INFO_BUF_LEN); + bn4.insert(bn4.begin(), bnodes, bnodes + (8*NODE4_INFO_BUF_LEN)); + bn6.insert(bn6.begin(), bnodes6, bnodes6 + (8*NODE6_INFO_BUF_LEN)); + return {std::move(bn4), std::move(bn6)}; } -//TODO: out_stats.listen++; size_t NetworkEngine::sendListen(std::shared_ptr n, const InfoHash& infohash, const Blob& token, int confirm, RequestCb on_done, RequestCb on_expired) { - //TODO auto tid = TransId {TransPrefix::LISTEN}; msgpack::sbuffer buffer; msgpack::packer pk(&buffer); @@ -456,7 +574,7 @@ NetworkEngine::sendListen(std::shared_ptr n, const InfoHash& infohash, con Blob b {buffer.data(), buffer.data() + buffer.size()}; Request req {getNewRequestId(), n, std::move(b), [=](size_t rid, ParsedMessage&& msg) { /* on done */ - //TODO + on_done(n, rid, {}); }, [=](size_t rid, bool s) { /* on expired */ on_expired(n, rid, {}); @@ -465,12 +583,12 @@ NetworkEngine::sendListen(std::shared_ptr n, const InfoHash& infohash, con }; requests.emplace(std::move(req)); send(buffer.data(), buffer.size(), confirm ? 0 : MSG_CONFIRM, (sockaddr*)&n->ss, n->sslen); + out_stats.listen++; return tid[2]; } void NetworkEngine::sendListenConfirmation(const sockaddr* sa, socklen_t salen, TransId tid) { - //TODO msgpack::sbuffer buffer; msgpack::packer pk(&buffer); pk.pack_map(4); @@ -487,11 +605,9 @@ NetworkEngine::sendListenConfirmation(const sockaddr* sa, socklen_t salen, Trans send(buffer.data(), buffer.size(), 0, sa, salen); } -//TODO: out_stats.put++; size_t NetworkEngine::sendAnnounceValue(std::shared_ptr n, const InfoHash& infohash, const Value& value, time_point created, const Blob& token, int confirm, RequestCb on_done, RequestCb on_expired) { - //TODO auto tid = TransId {TransPrefix::ANNOUNCE_VALUES}; msgpack::sbuffer buffer; msgpack::packer pk(&buffer); @@ -516,7 +632,7 @@ NetworkEngine::sendAnnounceValue(std::shared_ptr n, const InfoHash& infoha Blob b {buffer.data(), buffer.data() + buffer.size()}; Request req {getNewRequestId(), n, std::move(b), [=](size_t rid, ParsedMessage&& msg) { /* on done */ - //TODO + on_done(n, rid, {}); }, [=](size_t rid, bool s) { /* on expired */ on_expired(n, rid, {}); @@ -525,6 +641,7 @@ NetworkEngine::sendAnnounceValue(std::shared_ptr n, const InfoHash& infoha }; requests.emplace(std::move(req)); send(buffer.data(), buffer.size(), confirm ? 0 : MSG_CONFIRM, (sockaddr*)&n->ss, n->sslen); + out_stats.put++; return tid[2]; } @@ -668,7 +785,7 @@ NetworkEngine::ParsedMessage::msgpack_unpack(msgpack::object msg) try { values.emplace_back(std::make_shared(rvalues->via.array.ptr[i])); } catch (const std::exception& e) { - //DHT_WARN("Error reading value: %s", e.what()); + //DHT_LOG.WARN("Error reading value: %s", e.what()); } } diff --git a/src/utils.cpp b/src/utils.cpp index 5f2ea2ba..49164dd7 100644 --- a/src/utils.cpp +++ b/src/utils.cpp @@ -21,6 +21,13 @@ namespace dht { +static constexpr InfoHash zeroes {}; +static constexpr InfoHash ones = {std::array{{ + 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, + 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, + 0xFF, 0xFF, 0xFF, 0xFF +}}}; + std::string dht::print_addr(const sockaddr* sa, socklen_t slen) {