Merge pull request #348 from savoirfairelinux/feature/probepermanent

dht: don't send probe query for non permanent puts

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;
     }

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
     }

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);
     }
 

include/opendht/value.h

@@ -122,6 +122,8 @@ private:
     std::map<ValueType::Id, ValueType> 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<Value> decryptedValue {};
 };
 
 using ValuesExport = std::pair<InfoHash, Blob>;

src/dht.cpp

@@ -381,7 +381,6 @@ void Dht::searchSendAnnounceValue(const Sp<Search>& sr) {
     if (sr->announce.empty())
         return;
     unsigned i = 0;
-    auto probe_query = std::make_shared<Query>(Select {}.field(Value::Field::Id).field(Value::Field::SeqNum));
     std::weak_ptr<Search> ws = sr;
 
     auto onDone = [this,ws](const net::Request& req, net::RequestAnswer&& answer)
@@ -471,25 +470,46 @@ void Dht::searchSendAnnounceValue(const Sp<Search>& sr) {
         }
     };
 
+    Sp<Query> 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<Query>(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;
     }
@@ -615,7 +635,7 @@ Dht::searchStep(Sp<Search> 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()) {
@@ -645,7 +665,7 @@ Dht::searchStep(Sp<Search> 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());*/
     }
 
@@ -1756,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;

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());
         }
     }

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();
     }
 
     /**

src/securedht.cpp

@@ -56,7 +56,7 @@ SecureDht::SecureDht(std::unique_ptr<DhtInterface> 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<void(const Sp<cry
 {
     Sp<crypto::Certificate> 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::function<void(const Sp<cry
     if (localQueryMethod_) {
         auto res = localQueryMethod_(node);
         if (not res.empty()) {
-            DHT_LOG.DEBUG("Registering certificate from local store for %s", node.toString().c_str());
+            DHT_LOG.DBG("Registering certificate from local store for %s", node.toString().c_str());
             nodesCertificates_.emplace(node, res.front());
             if (cb)
                 cb(res.front());
@@ -187,7 +187,7 @@ SecureDht::findCertificate(const InfoHash& node, std::function<void(const Sp<cry
         for (const auto& v : vals) {
             if (auto cert = registerCertificate(node, v->data)) {
                 *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<void(const Sp<const
 {
     auto pk = getPublicKey(node);
     if (pk && *pk) {
-        DHT_LOG.DEBUG("Found public key from cache for %s", node.toString().c_str());
+        DHT_LOG.DBG("Found public key from cache for %s", node.toString().c_str());
         if (cb)
             cb(pk);
         return;
@@ -235,12 +235,17 @@ SecureDht::checkValue(const Sp<Value>& 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<Value>(std::move(decrypted_val));
+                v->decryptedValue = std::make_shared<Value>(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<Value>& v)
     }
     // Check signed values
     else if (v->isSigned()) {
+        if (v->signatureChecked) {
+            return v->signatureValid ? v : Sp<Value>{};
+        }
+        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;
         }
@@ -327,14 +337,13 @@ SecureDht::putSigned(const InfoHash& hash, Sp<Value> 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<Sp<Value>>& 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.");
@@ -356,7 +365,7 @@ SecureDht::putSigned(const InfoHash& hash, Sp<Value> val, DoneCallback callback,
 void
 SecureDht::putEncrypted(const InfoHash& hash, const InfoHash& to, Sp<Value> val, DoneCallback callback, bool permanent)
 {
-    findPublicKey(to, [=](const Sp<const crypto::PublicKey> pk) {
+    findPublicKey(to, [=](const Sp<const crypto::PublicKey>& pk) {
         if(!pk || !*pk) {
             if (callback)
                 callback(false, {});