From e414ca89c70f667318ec953a8b81bbda72ceb267 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Adrien=20B=C3=A9raud?= Date: Thu, 22 Nov 2018 15:43:38 -0500 Subject: [PATCH 1/3] dht: don't send probe query for non permanent puts --- src/dht.cpp | 50 +++++++++++++++++++++++++++++++++++--------------- src/search.h | 13 ++++++------- 2 files changed, 41 insertions(+), 22 deletions(-) diff --git a/src/dht.cpp b/src/dht.cpp index 5ae5ec9f..dd57eb27 100644 --- a/src/dht.cpp +++ b/src/dht.cpp @@ -381,7 +381,6 @@ void Dht::searchSendAnnounceValue(const Sp& sr) { if (sr->announce.empty()) return; unsigned i = 0; - auto probe_query = std::make_shared(Select {}.field(Value::Field::Id).field(Value::Field::SeqNum)); std::weak_ptr ws = sr; auto onDone = [this,ws](const net::Request& req, net::RequestAnswer&& answer) @@ -471,25 +470,46 @@ void Dht::searchSendAnnounceValue(const Sp& sr) { } }; + Sp probe_query {}; const auto& now = scheduler.time(); for (auto& n : sr->nodes) { if (not n.isSynced(now)) continue; - if (std::find_if(sr->announce.cbegin(), sr->announce.cend(), - [&now,&n](const Announce& a) { - return n.getAnnounceTime(a.value->id) <= now; - }) == sr->announce.cend()) - continue; - DHT_LOG.w(sr->id, n.node->id, "[search %s] [node %s] sending %s", - sr->id.toString().c_str(), n.node->toString().c_str(), probe_query->toString().c_str()); - n.probe_query = probe_query; - n.getStatus[probe_query] = network_engine.sendGetValues(n.node, - sr->id, - *probe_query, - -1, - onSelectDone, - std::bind(&Dht::searchNodeGetExpired, this, _1, _2, ws, probe_query)); + const auto& gs = n.probe_query ? n.getStatus.find(n.probe_query) : n.getStatus.cend(); + if (gs != n.getStatus.cend() and gs->second and gs->second->pending()) { + continue; + } + + bool sendQuery = false; + for (auto& a : sr->announce) { + if (n.getAnnounceTime(a.value->id) <= now) { + if (a.permanent) { + sendQuery = true; + } else { + DHT_LOG.w(sr->id, n.node->id, "[search %s] [node %s] sending 'put' (vid: %d)", + sr->id.toString().c_str(), n.node->toString().c_str(), a.value->id); + n.acked[a.value->id] = { + network_engine.sendAnnounceValue(n.node, sr->id, a.value, a.created, n.token, onDone, onExpired), + now + getType(a.value->type).expiration + }; + } + } + } + + if (sendQuery) { + if (not probe_query) + probe_query = std::make_shared(Select {}.field(Value::Field::Id).field(Value::Field::SeqNum)); + DHT_LOG.w(sr->id, n.node->id, "[search %s] [node %s] sending %s", + sr->id.toString().c_str(), n.node->toString().c_str(), probe_query->toString().c_str()); + n.probe_query = probe_query; + n.getStatus[probe_query] = network_engine.sendGetValues(n.node, + sr->id, + *probe_query, + -1, + onSelectDone, + std::bind(&Dht::searchNodeGetExpired, this, _1, _2, ws, probe_query)); + } if (not n.candidate and ++i == TARGET_NODES) break; } diff --git a/src/search.h b/src/search.h index 6c5086a0..00621809 100644 --- a/src/search.h +++ b/src/search.h @@ -324,14 +324,13 @@ struct Dht::SearchNode { */ time_point getAnnounceTime(Value::Id vid) const { const auto& ack = acked.find(vid); - const auto& gs = probe_query ? getStatus.find(probe_query) : getStatus.cend(); - if ((ack == acked.cend() or not ack->second.first) and (gs == getStatus.cend() - or not gs->second or not gs->second->pending())) + if (ack == acked.cend() or not ack->second.first) { return time_point::min(); - return ((gs != getStatus.cend() and gs->second and gs->second->pending()) - or ack == acked.cend() or not ack->second.first or ack->second.first->pending()) - ? time_point::max() - : (ack->second.first->completed() ? ack->second.second - REANNOUNCE_MARGIN : time_point::min()); + } + if (ack->second.first->completed()) { + return ack->second.second - REANNOUNCE_MARGIN; + } + return ack->second.first->pending() ? time_point::max() : time_point::min(); } /** From 00c1be9cdceebaf56b4829d78228372486355bac Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Adrien=20B=C3=A9raud?= Date: Thu, 22 Nov 2018 10:42:22 -0500 Subject: [PATCH 2/3] logger: use DBG for DEBUG to avoid name conflicts --- include/opendht/dht_interface.h | 2 +- include/opendht/log_enable.h | 8 ++++---- include/opendht/securedht.h | 6 ++---- src/dht.cpp | 6 +++--- src/network_engine.cpp | 4 ++-- src/securedht.cpp | 17 ++++++++--------- 6 files changed, 20 insertions(+), 23 deletions(-) diff --git a/include/opendht/dht_interface.h b/include/opendht/dht_interface.h index 6c51bb5e..bf837492 100644 --- a/include/opendht/dht_interface.h +++ b/include/opendht/dht_interface.h @@ -216,7 +216,7 @@ public: */ virtual void setLoggers(LogMethod error = NOLOG, LogMethod warn = NOLOG, LogMethod debug = NOLOG) { - DHT_LOG.DEBUG = debug; + DHT_LOG.DBG = debug; DHT_LOG.WARN = warn; DHT_LOG.ERR = error; } diff --git a/include/opendht/log_enable.h b/include/opendht/log_enable.h index dac3863d..9bbef46b 100644 --- a/include/opendht/log_enable.h +++ b/include/opendht/log_enable.h @@ -85,7 +85,7 @@ private: }; struct Logger { - LogMethod DEBUG = NOLOG; + LogMethod DBG = NOLOG; LogMethod WARN = NOLOG; LogMethod ERR = NOLOG; void setFilter(const InfoHash& f) { @@ -114,7 +114,7 @@ struct Logger { #if OPENDHT_LOG va_list args; va_start(args, format); - log0(DEBUG, format, args); + log0(DBG, format, args); va_end(args); #endif } @@ -122,7 +122,7 @@ struct Logger { #if OPENDHT_LOG va_list args; va_start(args, format); - log1(DEBUG, f, format, args); + log1(DBG, f, format, args); va_end(args); #endif } @@ -130,7 +130,7 @@ struct Logger { #if OPENDHT_LOG va_list args; va_start(args, format); - log2(DEBUG, f1, f2, format, args); + log2(DBG, f1, f2, format, args); va_end(args); #endif } diff --git a/include/opendht/securedht.h b/include/opendht/securedht.h index e1049cd8..cd01f160 100644 --- a/include/opendht/securedht.h +++ b/include/opendht/securedht.h @@ -310,11 +310,9 @@ public: dht_->pushNotificationReceived(notification); } - void setLoggers(LogMethod error = NOLOG, LogMethod warn = NOLOG, LogMethod debug = NOLOG) + void setLoggers(LogMethod error = NOLOG, LogMethod warn = NOLOG, LogMethod debug = NOLOG) override { - DHT_LOG.DEBUG = debug; - DHT_LOG.WARN = warn; - DHT_LOG.ERR = error; + DhtInterface::setLoggers(error, warn, debug); dht_->setLoggers(error, warn, debug); } diff --git a/src/dht.cpp b/src/dht.cpp index dd57eb27..d62a7f4b 100644 --- a/src/dht.cpp +++ b/src/dht.cpp @@ -635,7 +635,7 @@ Dht::searchStep(Sp sr) // true if this node is part of the target nodes cluter. /*bool in = sr->id.xorCmp(myid, sr->nodes.back().node->id) < 0; - DHT_LOG_DEBUG("[search %s IPv%c] synced%s", + DHT_LOG_DBG("[search %s IPv%c] synced%s", sr->id.toString().c_str(), sr->af == AF_INET ? '4' : '6', in ? ", in" : "");*/ if (not sr->listeners.empty()) { @@ -665,7 +665,7 @@ Dht::searchStep(Sp sr) i++; } while (sent and sr->currentlySolicitedNodeCount() < MAX_REQUESTED_SEARCH_NODES); - /*DHT_LOG_DEBUG("[search %s IPv%c] step: sent %u requests (total %u).", + /*DHT_LOG_DBG("[search %s IPv%c] step: sent %u requests (total %u).", sr->id.toString().c_str(), sr->af == AF_INET ? '4' : '6', i, sr->currentlySolicitedNodeCount());*/ } @@ -1776,7 +1776,7 @@ Dht::Dht(const int& s, const int& s6, Config config) bool Dht::neighbourhoodMaintenance(RoutingTable& list) { - //DHT_LOG_DEBUG("neighbourhoodMaintenance"); + //DHT_LOG_DBG("neighbourhoodMaintenance"); auto b = list.findBucket(myid); if (b == list.end()) return false; diff --git a/src/network_engine.cpp b/src/network_engine.cpp index c38a79ae..f6b601c8 100644 --- a/src/network_engine.cpp +++ b/src/network_engine.cpp @@ -419,7 +419,7 @@ NetworkEngine::processMessage(const uint8_t *buf, size_t buflen, const SockAddr& msg->msgpack_unpack(msg_res.get()); } catch (const std::exception& e) { DHT_LOG.w("Can't parse message of size %lu: %s", buflen, e.what()); - DHT_LOG.DEBUG.logPrintable(buf, buflen); + //DHT_LOG.DBG.logPrintable(buf, buflen); return; } @@ -977,7 +977,7 @@ NetworkEngine::sendNodesValues(const SockAddr& addr, Tid tid, const Blob& nodes, pk.pack(std::string("v")); pk.pack_array(st.size()*fields.size()); for (const auto& v : st) v->msgpack_pack_fields(fields, pk); - //DHT_LOG_DEBUG("sending closest nodes (%d+%d nodes.), %u value headers containing %u fields", + //DHT_LOG_DBG("sending closest nodes (%d+%d nodes.), %u value headers containing %u fields", // nodes.size(), nodes6.size(), st.size(), fields.size()); } } diff --git a/src/securedht.cpp b/src/securedht.cpp index 2fa3254f..4d173170 100644 --- a/src/securedht.cpp +++ b/src/securedht.cpp @@ -56,7 +56,7 @@ SecureDht::SecureDht(std::unique_ptr dht, SecureDht::Config conf) 1 }, [this](bool ok) { if (ok) - DHT_LOG.DEBUG("SecureDht: public key announced successfully"); + DHT_LOG.DBG("SecureDht: public key announced successfully"); }, {}, true); } } @@ -139,7 +139,7 @@ SecureDht::registerCertificate(const InfoHash& node, const Blob& data) } InfoHash h = crt->getPublicKey().getId(); if (node == h) { - DHT_LOG.DEBUG("Registering certificate for %s", h.toString().c_str()); + DHT_LOG.DBG("Registering certificate for %s", h.toString().c_str()); auto it = nodesCertificates_.find(h); if (it == nodesCertificates_.end()) std::tie(it, std::ignore) = nodesCertificates_.emplace(h, std::move(crt)); @@ -147,7 +147,7 @@ SecureDht::registerCertificate(const InfoHash& node, const Blob& data) it->second = std::move(crt); return it->second; } else { - DHT_LOG.DEBUG("Certificate %s for node %s does not match node id !", h.toString().c_str(), node.toString().c_str()); + DHT_LOG.WARN("Certificate %s for node %s does not match node id !", h.toString().c_str(), node.toString().c_str()); return nullptr; } } @@ -164,7 +164,7 @@ SecureDht::findCertificate(const InfoHash& node, std::function b = getCertificate(node); if (b && *b) { - DHT_LOG.DEBUG("Using certificate from cache for %s", node.toString().c_str()); + DHT_LOG.DBG("Using certificate from cache for %s", node.toString().c_str()); if (cb) cb(b); return; @@ -172,7 +172,7 @@ SecureDht::findCertificate(const InfoHash& node, std::functiondata)) { *found = true; - DHT_LOG.DEBUG("Found certificate for %s", node.toString().c_str()); + DHT_LOG.DBG("Found certificate for %s", node.toString().c_str()); if (cb) cb(cert); return false; @@ -205,7 +205,7 @@ SecureDht::findPublicKey(const InfoHash& node, std::function val, DoneCallback callback, // Check if we are already announcing a value auto p = dht_->getPut(hash, val->id); if (p && val->seq <= p->seq) { - DHT_LOG.DEBUG("Found previous value being announced."); val->seq = p->seq + 1; } // Check if data already exists on the dht get(hash, [val,this] (const std::vector>& vals) { - DHT_LOG.DEBUG("Found online previous value being announced."); + DHT_LOG.DBG("Found online previous value being announced."); for (const auto& v : vals) { if (!v->isSigned()) DHT_LOG.ERR("Existing non-signed value seems to exists at this location."); From b2542a57df06525966d2f5550daf0abef5d75599 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Adrien=20B=C3=A9raud?= Date: Thu, 22 Nov 2018 15:37:22 -0500 Subject: [PATCH 3/3] securedht: cache value signature check/decryption results --- include/opendht/value.h | 10 ++++++++++ src/securedht.cpp | 14 ++++++++++++-- 2 files changed, 22 insertions(+), 2 deletions(-) diff --git a/include/opendht/value.h b/include/opendht/value.h index 63795caf..a4a3348a 100644 --- a/include/opendht/value.h +++ b/include/opendht/value.h @@ -122,6 +122,8 @@ private: std::map types {}; }; +struct CryptoValueCache; + /** * A "value" is data potentially stored on the Dht, with some metadata. * @@ -588,6 +590,14 @@ struct OPENDHT_PUBLIC Value * Hold encrypted version of the data. */ Blob cypher {}; + +private: + friend class SecureDht; + /* Cache for crypto ops */ + bool signatureChecked {false}; + bool signatureValid {false}; + bool decrypted {false}; + Sp decryptedValue {}; }; using ValuesExport = std::pair; diff --git a/src/securedht.cpp b/src/securedht.cpp index 4d173170..a0d5a5a2 100644 --- a/src/securedht.cpp +++ b/src/securedht.cpp @@ -235,12 +235,17 @@ SecureDht::checkValue(const Sp& v) #endif return {}; } + if (v->decrypted) { + return v->decryptedValue; + } + v->decrypted = true; try { Value decrypted_val (decrypt(*v)); if (decrypted_val.recipient == getId()) { if (decrypted_val.owner) nodesPubKeys_[decrypted_val.owner->getId()] = decrypted_val.owner; - return std::make_shared(std::move(decrypted_val)); + v->decryptedValue = std::make_shared(std::move(decrypted_val)); + return v->decryptedValue; } // Ignore values belonging to other people } catch (const std::exception& e) { @@ -249,7 +254,12 @@ SecureDht::checkValue(const Sp& v) } // Check signed values else if (v->isSigned()) { + if (v->signatureChecked) { + return v->signatureValid ? v : Sp{}; + } + v->signatureChecked = true; if (v->owner and v->owner->checkSignature(v->getToSign(), v->signature)) { + v->signatureValid = true; nodesPubKeys_[v->owner->getId()] = v->owner; return v; } @@ -355,7 +365,7 @@ SecureDht::putSigned(const InfoHash& hash, Sp val, DoneCallback callback, void SecureDht::putEncrypted(const InfoHash& hash, const InfoHash& to, Sp val, DoneCallback callback, bool permanent) { - findPublicKey(to, [=](const Sp pk) { + findPublicKey(to, [=](const Sp& pk) { if(!pk || !*pk) { if (callback) callback(false, {});