sockaddr: use more precise exception

msvc: replace write with _write when building on windows

.travis.yml

@@ -1,3 +1,4 @@
+dist: xenial
 sudo: required
 
 services:

CMakeLists.txt

@@ -1,7 +1,7 @@
 cmake_minimum_required (VERSION 3.1)
 project (opendht)
 set (opendht_VERSION_MAJOR 1)
-set (opendht_VERSION_MINOR 7.4)
+set (opendht_VERSION_MINOR 8.0)
 set (opendht_VERSION ${opendht_VERSION_MAJOR}.${opendht_VERSION_MINOR})
 set (PACKAGE_VERSION ${opendht_VERSION})
 set (VERSION "${opendht_VERSION}")
@@ -59,7 +59,7 @@ endif()
 
 if (OPENDHT_PROXY_SERVER OR OPENDHT_PROXY_CLIENT)
     find_package(Restbed REQUIRED)
-    if (!Jsoncpp_FOUND)
+    if (NOT Jsoncpp_FOUND)
         message(SEND_ERROR "Jsoncpp is required for DHT proxy support")
     endif()
 endif()
@@ -99,8 +99,8 @@ endif ()
 if (Nettle_INCLUDE_DIRS)
     include_directories (SYSTEM "${Nettle_INCLUDE_DIRS}")
 endif ()
-if (RESTBED_INCLUDE)
-    include_directories (SYSTEM "${RESTBED_INCLUDE}")
+if (Restbed_INCLUDE_DIR)
+    include_directories (SYSTEM "${Restbed_INCLUDE_DIR}")
 endif ()
 if (Jsoncpp_INCLUDE_DIRS)
     include_directories (SYSTEM "${Jsoncpp_INCLUDE_DIRS}")
@@ -246,16 +246,14 @@ if (OPENDHT_STATIC)
         ${opendht_HEADERS}
     )
     set_target_properties (opendht-static PROPERTIES OUTPUT_NAME "opendht")
-    if (OPENDHT_LTO)
-        target_link_libraries(opendht-static -flto -fuse-linker-plugin)
-    endif ()
     if (OPENDHT_ARGON2)
         target_include_directories(opendht-static SYSTEM PRIVATE argon2)
     else ()
-        target_link_libraries(opendht-static ${argon2_LIBRARIES})
         target_include_directories(opendht-static SYSTEM PRIVATE ${argon2_INCLUDE_DIRS})
     endif ()
-    target_link_libraries(opendht-static ${CMAKE_THREAD_LIBS_INIT} ${GNUTLS_LIBRARIES} ${Nettle_LIBRARIES} ${Restbed_LIBRARIES} ${Jsoncpp_LIBRARIES})
+    target_link_libraries(opendht-static
+        PRIVATE ${Restbed_LIBRARY} ${argon2_LIBRARIES}
+        PUBLIC ${CMAKE_THREAD_LIBS_INIT} ${GNUTLS_LIBRARIES} ${Nettle_LIBRARIES} ${Jsoncpp_LIBRARIES})
     install (TARGETS opendht-static DESTINATION ${CMAKE_INSTALL_LIBDIR} EXPORT opendht)
 endif ()
 
@@ -267,16 +265,13 @@ if (OPENDHT_SHARED)
     set_target_properties (opendht PROPERTIES IMPORT_SUFFIX "_import.lib")
     set_target_properties (opendht PROPERTIES SOVERSION ${opendht_VERSION_MAJOR} VERSION ${opendht_VERSION})
     target_compile_definitions(opendht PRIVATE OPENDHT_BUILD)
-    if (OPENDHT_LTO)
-        target_link_libraries(opendht PRIVATE -flto -fuse-linker-plugin)
-    endif ()
     if (OPENDHT_ARGON2)
         target_include_directories(opendht SYSTEM PRIVATE argon2)
     else ()
         target_link_libraries(opendht PRIVATE ${argon2_LIBRARIES})
         target_include_directories(opendht SYSTEM PRIVATE ${argon2_INCLUDE_DIRS})
     endif ()
-    target_link_libraries(opendht PRIVATE ${CMAKE_THREAD_LIBS_INIT} ${GNUTLS_LIBRARIES} ${Nettle_LIBRARIES} ${Restbed_LIBRARIES} ${Jsoncpp_LIBRARIES})
+    target_link_libraries(opendht PRIVATE ${CMAKE_THREAD_LIBS_INIT} ${GNUTLS_LIBRARIES} ${Nettle_LIBRARIES} ${Restbed_LIBRARY} ${Jsoncpp_LIBRARIES})
     install (TARGETS opendht DESTINATION ${CMAKE_INSTALL_LIBDIR} EXPORT opendht)
 endif ()
 

cmake/FindRestbed.cmake

@@ -1,17 +1,16 @@
-find_path (RESTBED_INCLUDE restbed
-           HINTS
-           "/usr/include"
-           "/usr/local/include"
-           "/opt/local/include"
-)
-
-if (RESTBED_INCLUDE)
-    message(STATUS "${green}Found Restbed: ${RESTBED_INCLUDE}")
-else()
-    message(FATAL_ERROR "${red}Failed to locate Restbed.}")
-endif()
-
-if (RESTBED_INCLUDE)
-    set(RESTBED_FOUND TRUE)
-    set(Restbed_LIBRARIES restbed)
+if(NOT Restbed_FOUND)
+    find_path (Restbed_INCLUDE_DIR restbed
+               HINTS
+               "/usr/include"
+               "/usr/local/include"
+               "/opt/local/include")
+    find_library(Restbed_LIBRARY restbed
+                 HINTS ${Restbed_ROOT_DIR} PATH_SUFFIXES lib)
+    include(FindPackageHandleStandardArgs)
+    find_package_handle_standard_args(Restbed DEFAULT_MSG Restbed_LIBRARY Restbed_INCLUDE_DIR)
+    if (Restbed_INCLUDE_DIR)
+        set(Restbed_FOUND TRUE)
+        set(Restbed_LIBRARIES ${Restbed_LIBRARY})
+        set(Restbed_INCLUDE_DIRS ${Restbed_INCLUDE_DIR})
+    endif()
 endif()

configure.ac

@@ -1,7 +1,7 @@
 dnl define macros
 m4_define([opendht_major_version], 1)
-m4_define([opendht_minor_version], 7)
-m4_define([opendht_patch_version], 4)
+m4_define([opendht_minor_version], 8)
+m4_define([opendht_patch_version], 0)
 m4_define([opendht_version],
 		  [opendht_major_version.opendht_minor_version.opendht_patch_version])
 
@@ -130,7 +130,7 @@ PKG_CHECK_MODULES([GnuTLS], [gnutls >= 3.3])
 PKG_CHECK_MODULES([MsgPack], [msgpack >= 1.2])
 
 AC_ARG_WITH([jsoncpp], AS_HELP_STRING([--without-jsoncpp], [Build without JsonCpp support]))
-AS_IF([test "x$with_jsoncpp" != "xno"], 
+AS_IF([test "x$with_jsoncpp" != "xno"],
       [PKG_CHECK_MODULES([JsonCpp], [jsoncpp >= 1.7.2], [have_jsoncpp=yes], [have_jsoncpp=no])],
       [have_jsoncpp=no])
 AS_IF([test "x$have_jsoncpp" = "xyes"], [

include/opendht/crypto.h

@@ -86,8 +86,16 @@ struct OPENDHT_PUBLIC PublicKey
 
     PublicKey& operator=(PublicKey&& o) noexcept;
 
+    /**
+     * Get public key fingerprint
+     */
     InfoHash getId() const;
+
+    /**
+     * Get public key long fingerprint
+     */
     PkId getLongId() const;
+
     bool checkSignature(const Blob& data, const Blob& signature) const;
     Blob encrypt(const Blob&) const;
 
@@ -133,6 +141,7 @@ struct OPENDHT_PUBLIC PrivateKey
     PrivateKey(const Blob& import, const std::string& password = {});
     ~PrivateKey();
     explicit operator bool() const { return key; }
+
     PublicKey getPublicKey() const;
     Blob serialize(const std::string& password = {}) const;
 
@@ -368,6 +377,7 @@ struct OPENDHT_PUBLIC Certificate {
 
     /** Same as getPublicKey().getId() */
     InfoHash getId() const;
+    /** Same as getPublicKey().getLongId() */
     PkId getLongId() const;
 
     /** Read certificate Common Name (CN) */
@@ -404,8 +414,20 @@ struct OPENDHT_PUBLIC Certificate {
 
     std::string print() const;
 
+    /**
+     * As a CA, revoke a certificate, adding it to
+     * the attached Certificate Revocation List (CRL)
+     */
     void revoke(const PrivateKey&, const Certificate&);
+
+    /**
+     * Get the list of certificates revoked as as CA.
+     */
     std::vector<std::shared_ptr<RevocationList>> getRevocationLists() const;
+
+    /**
+     * Attach existing revocation list.
+     */
     void addRevocationList(RevocationList&&);
     void addRevocationList(std::shared_ptr<RevocationList>);
 
@@ -508,9 +530,13 @@ public:
     ~secure_vector() { clean(); }
 
     static secure_vector<T> getRandom(size_t size) {
-        secure_vector<T> ret(size/sizeof(T));
+        secure_vector<T> ret(size);
         crypto::random_device rdev;
+#ifdef _WIN32
+        std::uniform_int_distribution<int> rand_byte{ 0, std::numeric_limits<uint8_t>::max() };
+#else
         std::uniform_int_distribution<uint8_t> rand_byte;
+#endif
         std::generate_n((uint8_t*)ret.data_.data(), ret.size()*sizeof(T), std::bind(rand_byte, std::ref(rdev)));
         return ret;
     }

include/opendht/dht.h

@@ -411,6 +411,7 @@ private:
     void storageAddListener(const InfoHash& id, const Sp<Node>& node, size_t tid, Query&& = {});
     bool storageStore(const InfoHash& id, const Sp<Value>& value, time_point created, const SockAddr& sa = {}, bool permanent = false);
     bool storageErase(const InfoHash& id, Value::Id vid);
+    bool storageRefresh(const InfoHash& id, Value::Id vid);
     void expireStore();
     void expireStorage(InfoHash h);
     void expireStore(decltype(store)::iterator);
@@ -465,7 +466,7 @@ private:
      * Low-level method that will perform a search on the DHT for the specified
      * infohash (id), using the specified IP version (IPv4 or IPv6).
      */
-    Sp<Search> search(const InfoHash& id, sa_family_t af, GetCallback = {}, QueryCallback = {}, DoneCallback = {}, Value::Filter = {}, Query q = {});
+    Sp<Search> search(const InfoHash& id, sa_family_t af, GetCallback = {}, QueryCallback = {}, DoneCallback = {}, Value::Filter = {}, const Sp<Query>& q = {});
 
     void announce(const InfoHash& id, sa_family_t af, Sp<Value> value, DoneCallback callback, time_point created=time_point::max(), bool permanent = false);
     size_t listenTo(const InfoHash& id, sa_family_t af, ValueCallback cb, Value::Filter f = Value::AllFilter(), const Sp<Query>& q = {});

include/opendht/dht_interface.h

@@ -40,7 +40,7 @@ public:
     virtual NodeStatus getStatus() const = 0;
 
     /**
-     * Get the ID of the node.
+     * Get the ID of the DHT node.
      */
     virtual const InfoHash& getNodeId() const = 0;
 
@@ -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/dhtrunner.h

@@ -158,6 +158,15 @@ public:
         },
         getFilterSet<T>());
     }
+    template <class T>
+    std::future<size_t> listen(InfoHash hash, std::function<bool(std::vector<T>&&, bool)> cb)
+    {
+        return listen(hash, [=](const std::vector<std::shared_ptr<Value>>& vals, bool expired) {
+            return cb(unpackVector<T>(vals), expired);
+        },
+        getFilterSet<T>());
+    }
+
     template <typename T>
     std::future<size_t> listen(InfoHash hash, std::function<bool(T&&)> cb, Value::Filter f = Value::AllFilter(), Where w = {})
     {
@@ -174,6 +183,22 @@ public:
         },
         getFilterSet<T>(f), w);
     }
+    template <typename T>
+    std::future<size_t> listen(InfoHash hash, std::function<bool(T&&, bool)> cb, Value::Filter f = Value::AllFilter(), Where w = {})
+    {
+        return listen(hash, [=](const std::vector<std::shared_ptr<Value>>& vals, bool expired) {
+            for (const auto& v : vals) {
+                try {
+                    if (not cb(Value::unpack<T>(*v), expired))
+                        return false;
+                } catch (const std::exception&) {
+                    continue;
+                }
+            }
+            return true;
+        },
+        getFilterSet<T>(f), w);
+    }
 
     void cancelListen(InfoHash h, size_t token);
     void cancelListen(InfoHash h, std::shared_future<size_t> token);
@@ -248,8 +273,14 @@ public:
 
     void dumpTables() const;
 
+    /**
+     * Get the public key fingerprint if an identity is used with this node, 0 otherwise.
+     */
     InfoHash getId() const;
 
+    /**
+     * Get the ID of the DHT node.
+     */
     InfoHash getNodeId() const;
 
     /**

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/node_cache.h

@@ -40,12 +40,16 @@ struct NodeCache {
     ~NodeCache();
 
 private:
-    class NodeMap : public std::map<InfoHash, std::weak_ptr<Node>> {
+    class NodeMap : private std::map<InfoHash, std::weak_ptr<Node>> {
     public:
         Sp<Node> getNode(const InfoHash& id);
         Sp<Node> getNode(const InfoHash& id, const SockAddr&, time_point now, bool confirmed, bool client);
+        std::vector<Sp<Node>> getCachedNodes(const InfoHash& id, size_t count) const;
         void clearBadNodes();
         void setExpired();
+        void cleanup();
+    private:
+        size_t cleanup_counter {0};
     };
 
     const NodeMap& cache(sa_family_t af) const { return af == AF_INET ? cache_4 : cache_6; }

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/sockaddr.h

@@ -67,7 +67,7 @@ public:
      */
     SockAddr(const sockaddr* sa, socklen_t length) {
         if (length > sizeof(sockaddr_storage))
-            throw std::runtime_error("Socket address length is too large");
+            throw std::invalid_argument("Socket address length is too large");
         set(sa, length);
     }
     SockAddr(const sockaddr* sa) {
@@ -78,7 +78,7 @@ public:
             else if(sa->sa_family == AF_INET6)
                 len = sizeof(sockaddr_in6);
             else
-                throw std::runtime_error("Unknown address family");
+                throw std::invalid_argument("Unknown address family");
         }
         set(sa, len);
     }

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,16 @@ struct OPENDHT_PUBLIC Value
      * Hold encrypted version of the data.
      */
     Blob cypher {};
+
+    bool checkSignature();
+    Sp<Value> decrypt(const crypto::PrivateKey& key);
+
+private:
+    /* Cache for crypto ops */
+    bool signatureChecked {false};
+    bool signatureValid {false};
+    bool decrypted {false};
+    Sp<Value> decryptedValue {};
 };
 
 using ValuesExport = std::pair<InfoHash, Blob>;

src/Makefile.am

@@ -1,6 +1,6 @@
 lib_LTLIBRARIES = libopendht.la
 
-libopendht_la_CPPFLAGS = @CPPFLAGS@ -I$(top_srcdir)/include/opendht @Argon2_CFLAGS@ @JsonCpp_CFLAGS@
+libopendht_la_CPPFLAGS = @CPPFLAGS@ -I$(top_srcdir)/include/opendht @Argon2_CFLAGS@ @JsonCpp_CFLAGS@ @MsgPack_CFLAGS@
 libopendht_la_LIBADD   = @Argon2_LIBS@ @JsonCpp_LIBS@ @GnuTLS_LIBS@ @Nettle_LIBS@
 libopendht_la_LDFLAGS  = @LDFLAGS@ @Argon2_LDFLAGS@
 libopendht_la_SOURCES  = \
@@ -65,8 +65,8 @@ libopendht_la_SOURCES += dht_proxy_client.cpp
 nobase_include_HEADERS += ../include/opendht/dht_proxy_client.h ../include/opendht/dht_interface.h
 endif
 
-if PROXY_CLIENT_OR_SERVER
 libopendht_la_SOURCES += base64.h base64.cpp
+if PROXY_CLIENT_OR_SERVER
 nobase_include_HEADERS += ../include/opendht/proxy.h
 endif
 

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());*/
     }
 
@@ -690,7 +710,7 @@ unsigned Dht::refill(Dht::Search& sr) {
 
 /* Start a search. */
 Sp<Dht::Search>
-Dht::search(const InfoHash& id, sa_family_t af, GetCallback gcb, QueryCallback qcb, DoneCallback dcb, Value::Filter f, Query q)
+Dht::search(const InfoHash& id, sa_family_t af, GetCallback gcb, QueryCallback qcb, DoneCallback dcb, Value::Filter f, const Sp<Query>& q)
 {
     if (!isRunning(af)) {
         DHT_LOG.e(id, "[search %s IPv%c] unsupported protocol", id.toString().c_str(), (af == AF_INET) ? '4' : '6');
@@ -732,26 +752,14 @@ Dht::search(const InfoHash& id, sa_family_t af, GetCallback gcb, QueryCallback q
         sr->expired = false;
         sr->nodes.clear();
         sr->nodes.reserve(SEARCH_NODES+1);
+        sr->nextSearchStep = scheduler.add(time_point::max(), std::bind(&Dht::searchStep, this, sr));
         DHT_LOG.w(id, "[search %s IPv%c] new search", id.toString().c_str(), (af == AF_INET) ? '4' : '6');
         if (search_id == 0)
             search_id++;
     }
 
-    if (gcb or qcb) {
-        auto now = scheduler.time();
-        sr->callbacks.insert(std::make_pair<time_point, Get>(
-            std::move(now),
-            Get { scheduler.time(), f, std::make_shared<Query>(q),
-                qcb ? qcb : QueryCallback {}, gcb ? gcb : GetCallback {}, dcb
-            }
-        ));
-    }
-
+    sr->get(f, q, qcb, gcb, dcb, scheduler);
     refill(*sr);
-    if (sr->nextSearchStep)
-        scheduler.edit(sr->nextSearchStep, scheduler.time());
-    else
-        sr->nextSearchStep = scheduler.add(scheduler.time(), std::bind(&Dht::searchStep, this, sr));
 
     return sr;
 }
@@ -928,7 +936,7 @@ Dht::put(const InfoHash& id, Sp<Value> val, DoneCallback callback, time_point cr
     scheduler.syncTime();
     const auto& now = scheduler.time();
     created = std::min(now, created);
-    storageStore(id, val, created, {}, true);
+    storageStore(id, val, created, {}, permanent);
 
     DHT_LOG.d(id, "put: adding %s -> %s", id.toString().c_str(), val->toString().c_str());
 
@@ -991,10 +999,10 @@ Dht::get(const InfoHash& id, GetCallback getcb, DoneCallback donecb, Value::Filt
 {
     scheduler.syncTime();
 
-    Query q {{}, where};
+    auto q = std::make_shared<Query>(Select {}, std::move(where));
     auto op = std::make_shared<GetStatus<Value>>();
+    auto f = filter.chain(q->where.getFilter());
 
-    auto f = filter.chain(q.where.getFilter());
     auto add_values = [op,f](const std::vector<Sp<Value>>& values) {
         std::vector<Sp<Value>> newvals {};
         for (const auto& v : values) {
@@ -1060,16 +1068,17 @@ void Dht::query(const InfoHash& id, QueryCallback cb, DoneCallback done_cb, Quer
     /* Try to answer this search locally. */
     qcb(local_fields);
 
+    auto sq = std::make_shared<Query>(std::move(q));
     Dht::search(id, AF_INET, {}, qcb, [=](bool ok, const std::vector<Sp<Node>>& nodes) {
         //DHT_LOG_WARN("DHT done IPv4");
         op->status4 = {true, ok};
         doneCallbackWrapper(done_cb, nodes, *op);
-    }, f, q);
+    }, f, sq);
     Dht::search(id, AF_INET6, {}, qcb, [=](bool ok, const std::vector<Sp<Node>>& nodes) {
         //DHT_LOG_WARN("DHT done IPv6");
         op->status6 = {true, ok};
         doneCallbackWrapper(done_cb, nodes, *op);
-    }, f, q);
+    }, f, sq);
 }
 
 std::vector<Sp<Value>>
@@ -1767,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;
@@ -2335,7 +2344,8 @@ Dht::onAnnounce(Sp<Node> n,
         Sp<Value> vc = v;
         if (lv) {
             if (*lv == *vc) {
-                DHT_LOG.d(hash, node.id, "[store %s] nothing to do for %s", hash.toString().c_str(), lv->toString().c_str());
+                storageRefresh(hash, v->id);
+                DHT_LOG.d(hash, node.id, "[store %s] [node %s] refreshed value %s", hash.toString().c_str(), node.toString().c_str(), std::to_string(v->id).c_str());
             } else {
                 const auto& type = getType(lv->type);
                 if (type.editPolicy(hash, lv, vc, node.id, node.getAddr())) {
@@ -2367,17 +2377,11 @@ Dht::onRefresh(Sp<Node> node, const InfoHash& hash, const Blob& token, const Val
 {
     using namespace net;
 
-    const auto& now = scheduler.time();
     if (not tokenMatch(token, node->getAddr())) {
         DHT_LOG.w(hash, node->id, "[node %s] incorrect token %s for 'put'", node->toString().c_str(), hash.toString().c_str());
         throw DhtProtocolException {DhtProtocolException::UNAUTHORIZED, DhtProtocolException::PUT_WRONG_TOKEN};
     }
-
-    auto s = store.find(hash);
-    if (s != store.end()) {
-        auto expiration = s->second.refresh(now, vid, types);
-        if (expiration != time_point::max())
-            scheduler.add(expiration, std::bind(&Dht::expireStorage, this, hash));
+    if (storageRefresh(hash, vid)) {
         DHT_LOG.d(hash, node->id, "[store %s] [node %s] refreshed value %s", hash.toString().c_str(), node->toString().c_str(), std::to_string(vid).c_str());
     } else {
         DHT_LOG.d(hash, node->id, "[store %s] [node %s] got refresh for unknown value",
@@ -2387,6 +2391,20 @@ Dht::onRefresh(Sp<Node> node, const InfoHash& hash, const Blob& token, const Val
     return {};
 }
 
+bool
+Dht::storageRefresh(const InfoHash& id, Value::Id vid)
+{
+    const auto& now = scheduler.time();
+    auto s = store.find(id);
+    if (s != store.end()) {
+        auto expiration = s->second.refresh(now, vid, types);
+        if (expiration != time_point::max())
+            scheduler.add(expiration, std::bind(&Dht::expireStorage, this, id));
+        return true;
+    }
+    return false;
+}
+
 void
 Dht::onAnnounceDone(const Sp<Node>& node, net::RequestAnswer& answer, Sp<Search>& sr)
 {

src/dht_proxy_client.cpp

@@ -254,7 +254,8 @@ DhtProxyClient::get(const InfoHash& key, GetCallback cb, DoneCallback donecb, Va
     o.thread = std::thread([=](){
         // Try to contact the proxy and set the status to connected when done.
         // will change the connectivity status
-        auto ok = std::make_shared<std::atomic_bool>(true);
+        struct GetState{ std::atomic_bool ok {true}; std::atomic_bool stop {false}; };
+        auto state = std::make_shared<GetState>();
         restbed::Http::async(req,
             [=](const std::shared_ptr<restbed::Request>& req,
                 const std::shared_ptr<restbed::Response>& reply) {
@@ -262,9 +263,9 @@ DhtProxyClient::get(const InfoHash& key, GetCallback cb, DoneCallback donecb, Va
 
             if (code == 200) {
                 try {
-                    while (restbed::Http::is_open(req) and not *finished) {
+                    while (restbed::Http::is_open(req) and not *finished and not state->stop) {
                         restbed::Http::fetch("\n", reply);
-                        if (*finished)
+                        if (*finished or state->stop)
                             break;
                         std::string body;
                         reply->get_body(body);
@@ -279,29 +280,30 @@ DhtProxyClient::get(const InfoHash& key, GetCallback cb, DoneCallback donecb, Va
                             auto value = std::make_shared<Value>(json);
                             if ((not filter or filter(*value)) and cb) {
                                 std::lock_guard<std::mutex> lock(lockCallbacks);
-                                callbacks_.emplace_back([cb, value, finished]() {
-                                    if (not *finished and not cb({value}))
-                                        *finished = true;
+                                callbacks_.emplace_back([cb, value, state]() {
+                                    if (not state->stop and not cb({value}))
+                                        state->stop = true;
                                 });
                                 loopSignal_();
                             }
                         } else {
-                            *ok = false;
+                            state->ok = false;
                         }
                     }
                 } catch (std::runtime_error& e) { }
             } else {
-                *ok = false;
+                state->ok = false;
             }
         }).wait();
         if (donecb) {
             std::lock_guard<std::mutex> lock(lockCallbacks);
             callbacks_.emplace_back([=](){
-                donecb(*ok, {});
+                donecb(state->ok, {});
+                state->stop = true;
             });
             loopSignal_();
         }
-        if (!ok) {
+        if (!state->ok) {
             // Connection failed, update connectivity
             opFailed();
         }
@@ -706,11 +708,12 @@ DhtProxyClient::sendListen(const std::shared_ptr<restbed::Request>& req, const V
                     Json::CharReaderBuilder rbuilder;
                     auto reader = std::unique_ptr<Json::CharReader>(rbuilder.newCharReader());
                     if (reader->parse(body.data(), body.data() + body.size(), &json, &err)) {
+                        auto expired = json.get("expired", Json::Value(false)).asBool();
                         auto value = std::make_shared<Value>(json);
                         if ((not filter or filter(*value)) and cb)  {
                             std::lock_guard<std::mutex> lock(lockCallbacks);
-                            callbacks_.emplace_back([cb, value, state]() {
-                                if (not state->cancel and not cb({value}, false))
+                            callbacks_.emplace_back([cb, value, state, expired]() {
+                                if (not state->cancel and not cb({value}, expired))
                                     state->cancel = true;
                             });
                             loopSignal_();

src/dht_proxy_server.cpp

@@ -196,13 +196,13 @@ DhtProxyServer::updateStats() const
 }
 
 void
-DhtProxyServer::getNodeInfo(const std::shared_ptr<restbed::Session>& session) const
+DhtProxyServer::getNodeInfo(const Sp<restbed::Session>& session) const
 {
     requestNum_++;
     const auto request = session->get_request();
     int content_length = std::stoi(request->get_header("Content-Length", "0"));
     session->fetch(content_length,
-        [this](const std::shared_ptr<restbed::Session> s, const restbed::Bytes& /*b*/) mutable
+        [this](const Sp<restbed::Session>& s, const restbed::Bytes& /*b*/) mutable
         {
             try {
                 if (dht_) {
@@ -235,13 +235,13 @@ DhtProxyServer::getNodeInfo(const std::shared_ptr<restbed::Session>& session) co
 }
 
 void
-DhtProxyServer::getStats(const std::shared_ptr<restbed::Session>& session) const
+DhtProxyServer::getStats(const Sp<restbed::Session>& session) const
 {
     requestNum_++;
     const auto request = session->get_request();
     int content_length = std::stoi(request->get_header("Content-Length", "0"));
     session->fetch(content_length,
-        [this](const std::shared_ptr<restbed::Session> s, const restbed::Bytes& /*b*/) mutable
+        [this](const Sp<restbed::Session>& s, const restbed::Bytes& /*b*/) mutable
         {
             try {
                 if (dht_) {
@@ -265,14 +265,14 @@ DhtProxyServer::getStats(const std::shared_ptr<restbed::Session>& session) const
 }
 
 void
-DhtProxyServer::get(const std::shared_ptr<restbed::Session>& session) const
+DhtProxyServer::get(const Sp<restbed::Session>& session) const
 {
     requestNum_++;
     const auto request = session->get_request();
     int content_length = std::stoi(request->get_header("Content-Length", "0"));
     auto hash = request->get_path_parameter("hash");
     session->fetch(content_length,
-        [=](const std::shared_ptr<restbed::Session> s, const restbed::Bytes& /*b* */)
+        [=](const Sp<restbed::Session>& s, const restbed::Bytes& /*b* */)
         {
             try {
                 if (dht_) {
@@ -280,23 +280,21 @@ DhtProxyServer::get(const std::shared_ptr<restbed::Session>& session) const
                     if (!infoHash) {
                         infoHash = InfoHash::get(hash);
                     }
-                    s->yield(restbed::OK, "", [=]( const std::shared_ptr< restbed::Session > s) {
-                        auto cacheSession = std::weak_ptr<restbed::Session>(s);
-
-                        dht_->get(infoHash, [cacheSession](std::shared_ptr<Value> value) {
-                            auto s = cacheSession.lock();
-                            if (!s) return false;
-                            // Send values as soon as we get them
-                            Json::StreamWriterBuilder wbuilder;
-                            wbuilder["commentStyle"] = "None";
-                            wbuilder["indentation"] = "";
-                            auto output = Json::writeString(wbuilder, value->toJson()) + "\n";
-                            s->yield(output, [](const std::shared_ptr<restbed::Session> /*session*/){ });
-                            return true;
-                        }, [s](bool /*ok* */) {
-                            // Communication is finished
+                    s->yield(restbed::OK, "", [=](const Sp<restbed::Session>&) {});
+                    dht_->get(infoHash, [s](const Sp<Value>& value) {
+                        if (s->is_closed()) return false;
+                        // Send values as soon as we get them
+                        Json::StreamWriterBuilder wbuilder;
+                        wbuilder["commentStyle"] = "None";
+                        wbuilder["indentation"] = "";
+                        auto output = Json::writeString(wbuilder, value->toJson()) + "\n";
+                        s->yield(output, [](const Sp<restbed::Session>& /*session*/){ });
+                        return true;
+                    }, [s](bool /*ok* */) {
+                        // Communication is finished
+                        if (not s->is_closed()) {
                             s->close();
-                        });
+                        }
                     });
                 } else {
                     s->close(restbed::SERVICE_UNAVAILABLE, "{\"err\":\"Incorrect DhtRunner\"}");
@@ -309,7 +307,7 @@ DhtProxyServer::get(const std::shared_ptr<restbed::Session>& session) const
 }
 
 void
-DhtProxyServer::listen(const std::shared_ptr<restbed::Session>& session)
+DhtProxyServer::listen(const Sp<restbed::Session>& session)
 {
     requestNum_++;
     const auto request = session->get_request();
@@ -319,7 +317,7 @@ DhtProxyServer::listen(const std::shared_ptr<restbed::Session>& session)
     if (!infoHash)
         infoHash = InfoHash::get(hash);
     session->fetch(content_length,
-        [=](const std::shared_ptr<restbed::Session> s, const restbed::Bytes& /*b* */)
+        [=](const Sp<restbed::Session>& s, const restbed::Bytes& /*b* */)
         {
             try {
                 if (dht_) {
@@ -337,7 +335,7 @@ DhtProxyServer::listen(const std::shared_ptr<restbed::Session>& session)
                     // cache the session to avoid an incrementation of the shared_ptr's counter
                     // else, the session->close() will not close the socket.
                     auto cacheSession = std::weak_ptr<restbed::Session>(s);
-                    listener.token = dht_->listen(infoHash, [cacheSession](std::shared_ptr<Value> value) {
+                    listener.token = dht_->listen(infoHash, [cacheSession](const std::vector<Sp<Value>>& values, bool expired) {
                         auto s = cacheSession.lock();
                         if (!s) return false;
                         // Send values as soon as we get them
@@ -345,8 +343,13 @@ DhtProxyServer::listen(const std::shared_ptr<restbed::Session>& session)
                             Json::StreamWriterBuilder wbuilder;
                             wbuilder["commentStyle"] = "None";
                             wbuilder["indentation"] = "";
-                            auto output = Json::writeString(wbuilder, value->toJson()) + "\n";
-                            s->yield(output, [](const std::shared_ptr<restbed::Session>){ });
+                            for (const auto& value : values) {
+                                auto val = value->toJson();
+                                if (expired)
+                                    val["expired"] = true;
+                                auto output = Json::writeString(wbuilder, val) + "\n";
+                                s->yield(output, [](const Sp<restbed::Session>&){ });
+                            }
                         }
                         return !s->is_closed();
                     });
@@ -482,7 +485,6 @@ DhtProxyServer::subscribe(const std::shared_ptr<restbed::Session>& session)
 void
 DhtProxyServer::unsubscribe(const std::shared_ptr<restbed::Session>& session)
 {
-
     requestNum_++;
     const auto request = session->get_request();
     int content_length = std::stoi(request->get_header("Content-Length", "0"));

src/dhtrunner.cpp

@@ -37,6 +37,7 @@
 #include <winsock2.h>
 #include <ws2tcpip.h>
 #define close(x) closesocket(x)
+#define write(...) _write(__VA_ARGS__)
 #endif
 
 namespace dht {

src/network_engine.cpp

@@ -185,7 +185,7 @@ NetworkEngine::tellListener(Sp<Node> node, Tid socket_id, const InfoHash& hash,
 }
 
 void
-NetworkEngine::tellListenerRefreshed(Sp<Node> n, Tid socket_id, const InfoHash& hash, const Blob& token, const std::vector<Value::Id>& values)
+NetworkEngine::tellListenerRefreshed(Sp<Node> n, Tid socket_id, const InfoHash&, const Blob& token, const std::vector<Value::Id>& values)
 {
     msgpack::sbuffer buffer;
     msgpack::packer<msgpack::sbuffer> pk(&buffer);
@@ -215,7 +215,7 @@ NetworkEngine::tellListenerRefreshed(Sp<Node> n, Tid socket_id, const InfoHash&
 }
 
 void
-NetworkEngine::tellListenerExpired(Sp<Node> n, Tid socket_id, const InfoHash& hash, const Blob& token, const std::vector<Value::Id>& values)
+NetworkEngine::tellListenerExpired(Sp<Node> n, Tid socket_id, const InfoHash&, const Blob& token, const std::vector<Value::Id>& values)
 {
     msgpack::sbuffer buffer;
     msgpack::packer<msgpack::sbuffer> pk(&buffer);
@@ -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;
     }
 
@@ -551,7 +551,7 @@ NetworkEngine::process(std::unique_ptr<ParsedMessage>&& msg, const SockAddr& fro
         }
         case MessageType::Reply:
             if (req) { /* request reply */
-                if (msg->type == MessageType::AnnounceValue or msg->type == MessageType::Listen)
+                if (req->getType() == MessageType::AnnounceValue or req->getType() == MessageType::Listen)
                     req->node->authSuccess();
                 req->reply_time = scheduler.time();
 
@@ -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/node_cache.cpp

@@ -20,6 +20,9 @@
 
 namespace dht {
 
+constexpr size_t CLEANUP_MAX_NODES {1024};
+constexpr size_t CLEANUP_FREQ {1024};
+
 NodeCache::~NodeCache()
 {
     cache_4.setExpired();
@@ -41,29 +44,33 @@ NodeCache::getNode(const InfoHash& id, const SockAddr& addr, time_point now, boo
 std::vector<Sp<Node>>
 NodeCache::getCachedNodes(const InfoHash& id, sa_family_t sa_f, size_t count) const
 {
-    const auto& c = cache(sa_f);
+    return cache(sa_f).getCachedNodes(id, count);
+}
 
+std::vector<Sp<Node>>
+NodeCache::NodeMap::getCachedNodes(const InfoHash& id, size_t count) const
+{
     std::vector<Sp<Node>> nodes;
-    nodes.reserve(std::min(c.size(), count));
-    NodeMap::const_iterator it;
-    auto dec_it = [&c](NodeMap::const_iterator& it) {
+    nodes.reserve(std::min(size(), count));
+    const_iterator it;
+    auto dec_it = [this](const_iterator& it) {
         auto ret = it;
-        it = (it == c.cbegin()) ? c.cend() : std::prev(it);
+        it = (it == cbegin()) ? cend() : std::prev(it);
         return ret;
     };
 
-    auto it_p = c.lower_bound(id),
+    auto it_p = lower_bound(id),
         it_n = it_p;
-    if (not c.empty())
+    if (not empty())
         dec_it(it_p); /* Create 2 separate iterator if we could */
 
-    while (nodes.size() < count and (it_n != c.cend() or it_p != c.cend())) {
+    while (nodes.size() < count and (it_n != cend() or it_p != cend())) {
         /* If one of the iterator is at the end, then take the other one
            If they are both in middle of somewhere comapre both and take
            the closest to the id. */
-        if (it_p == c.cend())       it = it_n++;
-        else if (it_n == c.cend())  it = dec_it(it_p);
-        else                        it = id.xorCmp(it_p->first, it_n->first) < 0 ? dec_it(it_p) : it_n++;
+        if (it_p == cend())      it = it_n++;
+        else if (it_n == cend()) it = dec_it(it_p);
+        else                     it = id.xorCmp(it_p->first, it_n->first) < 0 ? dec_it(it_p) : it_n++;
 
         if (auto n = it->second.lock())
             if ( not n->isExpired() and not n->isClient() )
@@ -104,6 +111,10 @@ NodeCache::NodeMap::getNode(const InfoHash& id, const SockAddr& addr, time_point
     if (not node) {
         node = std::make_shared<Node>(id, addr, client);
         it.first->second = node;
+        if (cleanup_counter++ == CLEANUP_FREQ) {
+            cleanup();
+            cleanup_counter = 0;
+        }
     } else if (confirm or node->isOld(now)) {
         node->update(addr);
     }
@@ -120,6 +131,7 @@ NodeCache::NodeMap::clearBadNodes() {
             erase(it++);
         }
     }
+    cleanup_counter = 0;
 }
 
 void
@@ -128,6 +140,21 @@ NodeCache::NodeMap::setExpired() {
         if (auto n = wn.second.lock())
             n->setExpired();
     clear();
+    cleanup_counter = 0;
+}
+
+void
+NodeCache::NodeMap::cleanup()
+{
+    auto it = lower_bound(InfoHash::getRandom());
+    for (size_t n = 0, maxNodes = std::min(size(), CLEANUP_MAX_NODES); n != maxNodes; n++) {
+        if (it == end())
+            it = begin();
+        if (it->second.expired())
+            erase(it++);
+        else
+            ++it;
+    }
 }
 
 }

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();
     }
 
     /**
@@ -476,6 +475,17 @@ struct Dht::Search {
     bool isAnnounced(Value::Id id) const;
     bool isListening(time_point now) const;
 
+    void get(Value::Filter f, const Sp<Query>& q, const QueryCallback& qcb, const GetCallback& gcb, const DoneCallback& dcb, Scheduler& scheduler) {
+        if (gcb or qcb) {
+            const auto& now = scheduler.time();
+            callbacks.emplace(now, Get { now, f, q, qcb, gcb, dcb });
+            auto values = cache.get(f);
+            if (not values.empty())
+                gcb(values);
+            scheduler.edit(nextSearchStep, now);
+        }
+    }
+
     size_t listen(ValueCallback cb, Value::Filter f, const Sp<Query>& q, Scheduler& scheduler) {
         //DHT_LOG.e(id, "[search %s IPv%c] listen", id.toString().c_str(), (af == AF_INET) ? '4' : '6');
         return cache.listen(cb, q, f, [&](const Sp<Query>& q, ValueCallback vcb){

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);
     }
 }
@@ -68,27 +68,19 @@ ValueType
 SecureDht::secureType(ValueType&& type)
 {
     type.storePolicy = [this,type](InfoHash id, Sp<Value>& v, const InfoHash& nid, const SockAddr& a) {
-        if (v->isSigned()) {
-            if (!v->owner or !v->owner->checkSignature(v->getToSign(), v->signature)) {
-                DHT_LOG.WARN("Signature verification failed");
-                return false;
-            }
-            else
-                DHT_LOG.WARN("Signature verification succeeded");
-        }
+        if (v->isSigned())
+            return v->checkSignature();
         return type.storePolicy(id, v, nid, a);
     };
     type.editPolicy = [this,type](InfoHash id, const Sp<Value>& o, Sp<Value>& n, const InfoHash& nid, const SockAddr& a) {
-        if (!o->isSigned())
+        if (not o->isSigned())
             return type.editPolicy(id, o, n, nid, a);
-        if (o->owner != n->owner) {
+        if (o->owner != n->owner or not n->isSigned()) {
             DHT_LOG.WARN("Edition forbidden: owner changed.");
             return false;
         }
-        if (!o->owner or !o->owner->checkSignature(n->getToSign(), n->signature)) {
-            DHT_LOG.WARN("Edition forbidden: signature verification failed.");
+        if (not n->checkSignature())
             return false;
-        }
         if (o->seq == n->seq) {
             // If the data is exactly the same,
             // it can be reannounced, possibly by someone else.
@@ -139,7 +131,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 +139,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 +156,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 +164,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 +179,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 +197,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;
@@ -236,24 +228,22 @@ SecureDht::checkValue(const Sp<Value>& v)
             return {};
         }
         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));
+            if (auto decrypted_val = v->decrypt(*key_)) {
+                if (decrypted_val->owner)
+                    nodesPubKeys_[decrypted_val->owner->getId()] = decrypted_val->owner;
+                return decrypted_val;
             }
-            // Ignore values belonging to other people
         } catch (const std::exception& e) {
             DHT_LOG.WARN("Could not decrypt value %s : %s", v->toString().c_str(), e.what());
         }
     }
     // Check signed values
     else if (v->isSigned()) {
-        if (v->owner and v->owner->checkSignature(v->getToSign(), v->signature)) {
-            nodesPubKeys_[v->owner->getId()] = v->owner;
+        if (v->checkSignature()) {
+            if (v->owner)
+                nodesPubKeys_[v->owner->getId()] = v->owner;
             return v;
-        }
-        else
+        } else
             DHT_LOG.WARN("Signature verification failed for %s", v->toString().c_str());
     }
     // Forward normal values
@@ -327,14 +317,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 +345,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, {});

src/value.cpp

@@ -233,6 +233,41 @@ Value::toJson() const
     return val;
 }
 
+bool
+Value::checkSignature()
+{
+    if (!signatureChecked) {
+        signatureChecked = true;
+        if (isSigned()) {
+            signatureValid = owner and owner->checkSignature(getToSign(), signature);
+        } else {
+            signatureValid = true;
+        }
+    }
+    return signatureValid;
+}
+
+Sp<Value>
+Value::decrypt(const crypto::PrivateKey& key)
+{
+    if (not decrypted) {
+        decrypted = true;
+        if (isEncrypted()) {
+            auto decryptedBlob = key.decrypt(cypher);
+            std::unique_ptr<Value> v {new Value(id)};
+            auto msg = msgpack::unpack((const char*)decryptedBlob.data(), decryptedBlob.size());
+            v->msgpack_unpack_body(msg.get());
+            if (v->recipient != key.getPublicKey().getId())
+                throw crypto::DecryptError("Recipient mismatch");
+            // Ignore values belonging to other people
+            if (not v->owner or not v->owner->checkSignature(v->getToSign(), v->signature))
+                throw crypto::DecryptError("Signature mismatch");
+            decryptedValue = std::move(v);
+        }
+    }
+    return decryptedValue;
+}
+
 uint64_t
 unpackId(const Json::Value& json, const std::string& key) {
     uint64_t ret = 0;

tools/Makefile.am

@@ -1,7 +1,7 @@
 bin_PROGRAMS = dhtnode dhtchat dhtscanner
 noinst_HEADERS = tools_common.h
 
-AM_CPPFLAGS = -I../include
+AM_CPPFLAGS = -I../include @JsonCpp_CFLAGS@ @MsgPack_CFLAGS@
 
 dhtnode_SOURCES = dhtnode.cpp
 dhtnode_LDFLAGS = -lopendht -lreadline -L@top_builddir@/src/.libs @Argon2_LDFLAGS@ @GnuTLS_LIBS@

tools/proxy_node.html

@@ -63,11 +63,15 @@ $(function() {
                         if (!element || element.length == 0)
                             return;
                         const o = JSON.parse(element);
-                        const d = window.atob(o.data);
-                        const delay = Math.max(0, start-now);
-                        $('<li class="list-group-item"/>').append(valueGetElement(o)).appendTo(group).hide().delay(delay).slideDown(100);
-                        lastAppended = i+1;
-                        start = Math.max(start, now)+25;
+                        if (o.expired) {
+                            $('#value'+o.id).slideUp(100, (e) => $(e.target).remove());
+                        } else {
+                            const d = window.atob(o.data);
+                            const delay = Math.max(0, start-now);
+                            $('<li class="list-group-item" id="value'+o.id+'"/>').append(valueGetElement(o)).appendTo(group).hide().delay(delay).slideDown(100);
+                            lastAppended = i+1;
+                            start = Math.max(start, now)+25;
+                        }
                     }
                 } else if (this.status !== 0) {
                     group.empty().append($('<li class="list-group-item list-group-item-danger"/>').text("Error loading content: " + this.statusText));