dhtrunner: add putEncrypted with long ID or pulic key

Signed-off-by: JUAN MÉNDEZ <vpsink@gmail.com>

CMakeLists.txt

@@ -71,8 +71,11 @@ if (NOT MSVC)
         )
     endif()
 
-    if (OPENDHT_HTTP)
+    if (OPENDHT_HTTP OR OPENDHT_PEER_DISCOVERY)
         find_path(ASIO_INCLUDE_DIR asio.hpp REQUIRED)
+    endif ()
+
+    if (OPENDHT_HTTP)
         find_package(Restinio REQUIRED)
         find_library(FMT_LIBRARY fmt)
         add_library(fmt SHARED IMPORTED)

docker/DockerfileDeps

@@ -17,9 +17,9 @@ RUN apt-get update && apt-get install -y \
 
 RUN echo "*** Downloading RESTinio ***" \
     && mkdir restinio && cd restinio \
-    && wget https://github.com/aberaud/restinio/archive/2c0b6f5e5ba04d7a74e8406a3df1fd433680599d.tar.gz \
-    && ls -l && tar -xzf 2c0b6f5e5ba04d7a74e8406a3df1fd433680599d.tar.gz \
-    && cd restinio-2c0b6f5e5ba04d7a74e8406a3df1fd433680599d/dev \
+    && wget https://github.com/aberaud/restinio/archive/2224ffedef52cb2b74645d63d871d61dbd0f165e.tar.gz \
+    && ls -l && tar -xzf 2224ffedef52cb2b74645d63d871d61dbd0f165e.tar.gz \
+    && cd restinio-2224ffedef52cb2b74645d63d871d61dbd0f165e/dev \
     && cmake -DCMAKE_INSTALL_PREFIX=/usr -DRESTINIO_TEST=OFF -DRESTINIO_SAMPLE=OFF \
              -DRESTINIO_INSTALL_SAMPLES=OFF -DRESTINIO_BENCH=OFF -DRESTINIO_INSTALL_BENCHES=OFF \
              -DRESTINIO_FIND_DEPS=ON -DRESTINIO_ALLOW_SOBJECTIZER=Off -DRESTINIO_USE_BOOST_ASIO=none . \

docker/DockerfileDepsLlvm

@@ -18,9 +18,9 @@ ENV CXX c++
 
 RUN echo "*** Downloading RESTinio ***" \
     && mkdir restinio && cd restinio \
-    && wget https://github.com/aberaud/restinio/archive/2c0b6f5e5ba04d7a74e8406a3df1fd433680599d.tar.gz \
-    && ls -l && tar -xzf 2c0b6f5e5ba04d7a74e8406a3df1fd433680599d.tar.gz \
-    && cd restinio-2c0b6f5e5ba04d7a74e8406a3df1fd433680599d/dev \
+    && wget https://github.com/aberaud/restinio/archive/2224ffedef52cb2b74645d63d871d61dbd0f165e.tar.gz \
+    && ls -l && tar -xzf 2224ffedef52cb2b74645d63d871d61dbd0f165e.tar.gz \
+    && cd restinio-2224ffedef52cb2b74645d63d871d61dbd0f165e/dev \
     && cmake -DCMAKE_INSTALL_PREFIX=/usr -DRESTINIO_TEST=OFF -DRESTINIO_SAMPLE=OFF \
              -DRESTINIO_INSTALL_SAMPLES=OFF -DRESTINIO_BENCH=OFF -DRESTINIO_INSTALL_BENCHES=OFF \
              -DRESTINIO_FIND_DEPS=ON -DRESTINIO_ALLOW_SOBJECTIZER=Off -DRESTINIO_USE_BOOST_ASIO=none . \

include/opendht/callbacks.h

@@ -78,6 +78,7 @@ struct OPENDHT_PUBLIC NodeStats {
 
 struct OPENDHT_PUBLIC NodeInfo {
     InfoHash id;
+    PkId pkid;
     InfoHash node_id;
     NodeStats ipv4 {};
     NodeStats ipv6 {};
@@ -94,7 +95,7 @@ struct OPENDHT_PUBLIC NodeInfo {
     explicit NodeInfo(const Json::Value& v);
 #endif
 
-    MSGPACK_DEFINE_MAP(id, node_id, ipv4, ipv6)
+    MSGPACK_DEFINE_MAP(id, pkid, node_id, ipv4, ipv6)
 };
 
 /**
@@ -164,8 +165,10 @@ using GetCallback = std::function<bool(const std::vector<std::shared_ptr<Value>>
 using ValueCallback = std::function<bool(const std::vector<std::shared_ptr<Value>>& values, bool expired)>;
 using GetCallbackSimple = std::function<bool(std::shared_ptr<Value> value)>;
 using ShutdownCallback = std::function<void()>;
+using IdentityAnnouncedCb = std::function<void(bool)>;
 
 using CertificateStoreQuery = std::function<std::vector<std::shared_ptr<crypto::Certificate>>(const InfoHash& pk_id)>;
+using CertificateStoreLongQuery = std::function<std::vector<std::shared_ptr<crypto::Certificate>>(const PkId& pk_id)>;
 
 typedef bool (*GetCallbackRaw)(std::shared_ptr<Value>, void *user_data);
 typedef bool (*ValueCallbackRaw)(std::shared_ptr<Value>, bool expired, void *user_data);

include/opendht/crypto.h

@@ -80,6 +80,8 @@ struct OPENDHT_PUBLIC PublicKey
      * Takes ownership of an existing gnutls_pubkey.
      */
     PublicKey(gnutls_pubkey_t k) : pk(k) {}
+
+    /** Import public key from serialized data */
     PublicKey(const uint8_t* dat, size_t dat_size);
     PublicKey(const Blob& pk) : PublicKey(pk.data(), pk.size()) {}
     PublicKey(PublicKey&& o) noexcept : pk(o.pk) { o.pk = nullptr; };

include/opendht/default_types.h

@@ -66,8 +66,11 @@ private:
 
 public:
     virtual void unpackValue(const Value& v) override {
-        if (v.owner)
+        if (v.owner) {
+            owner = v.owner;
             from = v.owner->getId();
+            fromId = v.owner->getLongId();
+        }
         BaseClass::unpackValue(v);
     }
 
@@ -75,7 +78,9 @@ public:
         return [](const Value& v){ return v.isSigned(); };
     }
 
+    Sp<crypto::PublicKey> owner;
     dht::InfoHash from;
+    dht::PkId fromId;
 };
 
 template <typename T>

include/opendht/dht.h

@@ -512,6 +512,8 @@ private:
 
     void confirmNodes();
     void expire();
+
+    void onConnected();
     void onDisconnected();
 
     /**

include/opendht/dht_interface.h

@@ -21,6 +21,8 @@
 #include "infohash.h"
 #include "log_enable.h"
 
+#include <queue>
+
 namespace dht {
 
 namespace net {
@@ -46,6 +48,10 @@ public:
     virtual NodeStatus getStatus(sa_family_t af) const = 0;
     virtual NodeStatus getStatus() const = 0;
 
+    void addOnConnectedCallback(std::function<void()> cb) {
+        onConnectCallbacks_.emplace(std::move(cb));
+    }
+
     virtual net::DatagramSocket* getSocket() const { return {}; };
 
     /**
@@ -264,6 +270,7 @@ public:
 
 protected:
     std::shared_ptr<Logger> logger_ {};
+    std::queue<std::function<void()>> onConnectCallbacks_ {};
 };
 
 } // namespace dht

include/opendht/dht_proxy_client.h

@@ -378,8 +378,8 @@ private:
      * Retrieve if we can connect to the proxy (update statusIpvX_)
      */
     void handleProxyConfirm(const asio::error_code &ec);
-    Sp<asio::steady_timer> nextProxyConfirmationTimer_;
-    Sp<asio::steady_timer> listenerRestartTimer_;
+    std::unique_ptr<asio::steady_timer> nextProxyConfirmationTimer_;
+    std::unique_ptr<asio::steady_timer> listenerRestartTimer_;
 
     /**
      * Relaunch LISTEN requests if the client disconnect/reconnect.

include/opendht/dht_proxy_server.h

@@ -445,7 +445,6 @@ private:
         std::map<InfoHash, std::vector<Listener>> listeners;
         MSGPACK_DEFINE_ARRAY(listeners)
     };
-    std::mutex lockPushListeners_;
     std::map<std::string, PushListener> pushListeners_;
     proxy::ListenToken tokenPushNotif_ {0};
 #endif //OPENDHT_PUSH_NOTIFICATIONS

include/opendht/dhtrunner.h

@@ -65,6 +65,7 @@ public:
         bool peer_publish {false};
         std::shared_ptr<dht::crypto::Certificate> server_ca;
         dht::crypto::Identity client_identity;
+        SockAddr bind4 {}, bind6 {};
     };
 
     struct Context {
@@ -73,6 +74,8 @@ public:
         std::shared_ptr<PeerDiscovery> peerDiscovery {};
         StatusCallback statusChangedCallback {};
         CertificateStoreQuery certificateStore {};
+        CertificateStoreLongQuery certificateStoreSha256 {};
+        IdentityAnnouncedCb identityAnnouncedCb {};
         Context() {}
     };
 
@@ -255,6 +258,27 @@ public:
     }
     void putEncrypted(const std::string& key, InfoHash to, Value&& value, DoneCallback cb={}, bool permanent = false);
 
+    void putEncrypted(InfoHash hash, PkId to, std::shared_ptr<Value> value, DoneCallback cb={}, bool permanent = false);
+    void putEncrypted(InfoHash hash, PkId to, std::shared_ptr<Value> value, DoneCallbackSimple cb, bool permanent = false) {
+        putEncrypted(hash, to, value, bindDoneCb(cb), permanent);
+    }
+
+    void putEncrypted(InfoHash hash, PkId to, Value&& value, DoneCallback cb={}, bool permanent = false);
+    void putEncrypted(InfoHash hash, PkId to, Value&& value, DoneCallbackSimple cb, bool permanent = false) {
+        putEncrypted(hash, to, std::forward<Value>(value), bindDoneCb(cb), permanent);
+    }
+
+    void putEncrypted(InfoHash hash, const std::shared_ptr<crypto::PublicKey>& to, std::shared_ptr<Value> value, DoneCallback cb={}, bool permanent = false);
+    void putEncrypted(InfoHash hash, const std::shared_ptr<crypto::PublicKey>& to, std::shared_ptr<Value> value, DoneCallbackSimple cb, bool permanent = false) {
+        putEncrypted(hash, to, value, bindDoneCb(cb), permanent);
+    }
+
+    void putEncrypted(InfoHash hash, const std::shared_ptr<crypto::PublicKey>& to, Value&& value, DoneCallback cb={}, bool permanent = false);
+    void putEncrypted(InfoHash hash, const std::shared_ptr<crypto::PublicKey>& to, Value&& value, DoneCallbackSimple cb, bool permanent = false) {
+        putEncrypted(hash, to, std::forward<Value>(value), bindDoneCb(cb), permanent);
+    }
+
+
     /**
      * Insert known nodes to the routing table, without necessarly ping them.
      * Usefull to restart a node and get things running fast without putting load on the network.
@@ -301,6 +325,7 @@ public:
      * Get the public key fingerprint if an identity is used with this node, 0 otherwise.
      */
     InfoHash getId() const;
+    PkId getLongId() const;
 
     /**
      * Get the ID of the DHT node.
@@ -359,12 +384,13 @@ public:
     std::string getSearchLog(const InfoHash&, sa_family_t af = AF_UNSPEC) const;
     std::vector<SockAddr> getPublicAddress(sa_family_t af = AF_UNSPEC);
     std::vector<std::string> getPublicAddressStr(sa_family_t af = AF_UNSPEC);
+    void getPublicAddress(std::function<void(std::vector<SockAddr>&&)>, sa_family_t af = AF_UNSPEC);
 
     // securedht methods
 
     void findCertificate(InfoHash hash, std::function<void(const std::shared_ptr<crypto::Certificate>&)>);
     void registerCertificate(std::shared_ptr<crypto::Certificate> cert);
-    void setLocalCertificateStore(CertificateStoreQuery&& query_method);
+    void setLocalCertificateStore(CertificateStoreQuery&& query_method, CertificateStoreLongQuery&& query_sha256_method = {});
 
     /**
      * @param port: Local port to bind. Both IPv4 and IPv6 will be tried (ANY).
@@ -379,22 +405,12 @@ public:
         config.threaded = threaded;
         run(port, config);
     }
-    void run(in_port_t port, const Config& config, Context&& context = {});
-
-    /**
-     * @param local4: Local IPv4 address and port to bind. Can be null.
-     * @param local6: Local IPv6 address and port to bind. Can be null.
-     *         You should allways bind to a global IPv6 address.
-     * @param identity: RSA key pair to use for cryptographic operations.
-     * @param threaded: If false, loop() must be called periodically. Otherwise a thread is launched.
-     * @param cb: Optional callback to receive general state information.
-     */
-    void run(SockAddr& local4, SockAddr& local6, const Config& config, Context&& context = {});
+    void run(in_port_t port, Config& config, Context&& context = {});
 
     /**
      * Same as @run(sockaddr_in, sockaddr_in6, Identity, bool, StatusCallback), but with string IP addresses and service (port).
      */
-    void run(const char* ip4, const char* ip6, const char* service, const Config& config, Context&& context = {});
+    void run(const char* ip4, const char* ip6, const char* service, Config& config, Context&& context = {});
 
     void run(const Config& config, Context&& context);
 
@@ -481,6 +497,7 @@ private:
 
     /** Current configuration */
     Config config_;
+    IdentityAnnouncedCb identityAnnouncedCb_;
 
     /**
      * reset dht clients

include/opendht/securedht.h

@@ -45,15 +45,13 @@ public:
         return c;
     }
 
-    SecureDht() {}
-
     /**
      * s, s6: bound socket descriptors for IPv4 and IPv6, respectively.
      *        For the Dht to be initialised, at least one of them must be >= 0.
      * id:    the identity to use for the crypto layer and to compute
      *        our own hash on the Dht.
      */
-    SecureDht(std::unique_ptr<DhtInterface> dht, Config config);
+    SecureDht(std::unique_ptr<DhtInterface> dht, Config config, IdentityAnnouncedCb iacb = {}, const std::shared_ptr<Logger>& l = {});
 
     virtual ~SecureDht();
 
@@ -117,6 +115,14 @@ public:
     void putEncrypted(const InfoHash& hash, const InfoHash& to, Value&& v, DoneCallback callback, bool permanent = false) {
         putEncrypted(hash, to, std::make_shared<Value>(std::move(v)), callback, permanent);
     }
+    void putEncrypted(const InfoHash& hash, const PkId& to, Sp<Value> val, DoneCallback callback, bool permanent = false);
+    void putEncrypted(const InfoHash& hash, const PkId& to, Value&& v, DoneCallback callback, bool permanent = false) {
+        putEncrypted(hash, to, std::make_shared<Value>(std::move(v)), callback, permanent);
+    }
+    void putEncrypted(const InfoHash& hash, const crypto::PublicKey& to, Sp<Value> val, DoneCallback callback, bool permanent = false);
+    void putEncrypted(const InfoHash& hash, const crypto::PublicKey& to, Value&& v, DoneCallback callback, bool permanent = false) {
+        putEncrypted(hash, to, std::make_shared<Value>(std::move(v)), callback, permanent);
+    }
 
     /**
      * Take ownership of the value and sign it using our private key.
@@ -128,21 +134,28 @@ public:
     Value decrypt(const Value& v);
 
     void findCertificate(const InfoHash& node, const std::function<void(const Sp<crypto::Certificate>)>& cb);
-    void findPublicKey(const InfoHash& node, const std::function<void(const Sp<const crypto::PublicKey>)>& cb);
+    void findPublicKey(const InfoHash& node, const std::function<void(const Sp<crypto::PublicKey>)>& cb);
 
-    const Sp<crypto::Certificate> registerCertificate(const InfoHash& node, const Blob& cert);
+    Sp<crypto::Certificate> registerCertificate(const InfoHash& node, const Blob& cert);
     void registerCertificate(Sp<crypto::Certificate>& cert);
 
-    const Sp<crypto::Certificate> getCertificate(const InfoHash& node) const;
-    const Sp<const crypto::PublicKey> getPublicKey(const InfoHash& node) const;
+    Sp<crypto::Certificate> getCertificate(const InfoHash& node) const;
+    Sp<crypto::PublicKey> getPublicKey(const InfoHash& node) const;
+
+    void findCertificate(const PkId& node, const std::function<void(const Sp<crypto::Certificate>)>& cb);
+    void findPublicKey(const PkId& node, const std::function<void(const Sp<crypto::PublicKey>)>& cb);
+    Sp<crypto::Certificate> registerCertificate(const PkId& node, const Blob& cert);
+    Sp<crypto::Certificate> getCertificate(const PkId& node) const;
+    Sp<crypto::PublicKey> getPublicKey(const PkId& node) const;
 
     /**
      * Allows to set a custom callback called by the library to find a locally-stored certificate.
      * The search key used is the public key ID, so there may be multiple certificates retured, signed with
      * the same private key.
      */
-    void setLocalCertificateStore(CertificateStoreQuery&& query_method) {
+    void setLocalCertificateStore(CertificateStoreQuery&& query_method, CertificateStoreLongQuery&& query_sha256_method) {
         localQueryMethod_ = std::move(query_method);
+        localQuerySha256Method_ = std::move(query_sha256_method);
     }
 
     /**
@@ -353,10 +366,13 @@ private:
 
     // method to query the local certificate store
     CertificateStoreQuery localQueryMethod_ {};
+    CertificateStoreLongQuery localQuerySha256Method_ {};
 
     // our certificate cache
     std::map<InfoHash, Sp<crypto::Certificate>> nodesCertificates_ {};
-    std::map<InfoHash, Sp<const crypto::PublicKey>> nodesPubKeys_ {};
+    std::map<InfoHash, Sp<crypto::PublicKey>> nodesPubKeys_ {};
+    std::map<PkId, Sp<crypto::Certificate>> nodesCertificatesLong_ {};
+    std::map<PkId, Sp<crypto::PublicKey>> nodesPubKeysLong_ {};
 
     std::atomic_bool forward_all_ {false};
     bool enableCache_ {false};

include/opendht/sockaddr.h

@@ -68,7 +68,7 @@ public:
      * Build from existing address.
      */
     SockAddr(const sockaddr* sa, socklen_t length) {
-        if (length > sizeof(sockaddr_storage))
+        if (length > (socklen_t)sizeof(sockaddr_storage))
             throw std::runtime_error("Socket address length is too large");
         set(sa, length);
     }
@@ -144,7 +144,7 @@ public:
             if (len) addr.reset((sockaddr*)::calloc(len, 1));
             else     addr.reset();
         }
-        if (len > sizeof(sa_family_t))
+        if (len)
             addr->sa_family = af;
     }
 
@@ -223,16 +223,16 @@ public:
      */
     sockaddr* get() { return addr.get(); }
 
-    const sockaddr_in& getIPv4() const {
+    inline const sockaddr_in& getIPv4() const {
         return *reinterpret_cast<const sockaddr_in*>(get());
     }
-    const sockaddr_in6& getIPv6() const {
+    inline const sockaddr_in6& getIPv6() const {
         return *reinterpret_cast<const sockaddr_in6*>(get());
     }
-    sockaddr_in& getIPv4() {
+    inline sockaddr_in& getIPv4() {
         return *reinterpret_cast<sockaddr_in*>(get());
     }
-    sockaddr_in6& getIPv6() {
+    inline sockaddr_in6& getIPv6() {
         return *reinterpret_cast<sockaddr_in6*>(get());
     }
 

include/opendht/value.h

@@ -356,7 +356,7 @@ struct OPENDHT_PUBLIC Value
     void sign(const crypto::PrivateKey& key) {
         if (isEncrypted())
             throw DhtException("Can't sign encrypted data.");
-        owner = std::make_shared<const crypto::PublicKey>(key.getPublicKey());
+        owner = std::make_shared<crypto::PublicKey>(key.getPublicKey());
         signature = key.sign(getToSign());
     }
 
@@ -368,8 +368,8 @@ struct OPENDHT_PUBLIC Value
         return isSigned() and owner->checkSignature(getToSign(), signature);
     }
 
-    std::shared_ptr<const crypto::PublicKey> getOwner() const {
-        return std::static_pointer_cast<const crypto::PublicKey>(owner);
+    std::shared_ptr<crypto::PublicKey> getOwner() const {
+        return std::static_pointer_cast<crypto::PublicKey>(owner);
     }
 
     /**
@@ -594,7 +594,7 @@ struct OPENDHT_PUBLIC Value
     /**
      * Public key of the signer.
      */
-    std::shared_ptr<const crypto::PublicKey> owner {};
+    std::shared_ptr<crypto::PublicKey> owner {};
 
     /**
      * Hash of the recipient (optional).

python/opendht.pyx

@@ -36,6 +36,7 @@ from libcpp.memory cimport shared_ptr
 from cython.parallel import parallel, prange
 from cython.operator cimport dereference as deref, preincrement as inc, predecrement as dec
 from cpython cimport ref
+from datetime import timedelta
 
 cimport opendht_cpp as cpp
 
@@ -235,8 +236,9 @@ cdef class Where(object):
 
 cdef class Value(object):
     cdef shared_ptr[cpp.Value] _value
-    def __init__(self, bytes val=b''):
-        self._value.reset(new cpp.Value(val, len(val)))
+    def __init__(self, bytes val=b'', cpp.uint16_t id=0):
+        self._value.reset(new cpp.Value(id, val, len(val)))
+
     def __str__(self):
         return self._value.get().toString().decode()
     property owner:
@@ -270,6 +272,14 @@ cdef class Value(object):
         def __get__(self):
             return self._value.get().size()
 
+cdef class ValueType(object):
+    cdef cpp.ValueType * _value
+    def __init__(self, cpp.uint16_t id, str name, expiration: timedelta):
+        if not isinstance(expiration, timedelta):
+            raise TypeError("expiration argument must be of type timedelta")
+        cdef cpp.seconds duration = cpp.seconds(int(expiration.total_seconds()))
+        self._value =  new cpp.ValueType(id, name.encode(), duration)
+
 cdef class NodeSetIter(object):
     cdef map[cpp.InfoHash, shared_ptr[cpp.Node]]* _nodes
     cdef map[cpp.InfoHash, shared_ptr[cpp.Node]].iterator _curIter
@@ -513,6 +523,8 @@ cdef class DhtRunner(_WithID):
         cb_obj = {'shutdown':shutdown_cb}
         ref.Py_INCREF(cb_obj)
         self.thisptr.get().shutdown(cpp.bindShutdownCb(shutdown_callback, <void*>cb_obj))
+    def registerType(self, ValueType type):
+        self.thisptr.get().registerType(deref(type._value))
     def enableLogging(self):
         cpp.enableLogging(self.thisptr.get()[0])
     def disableLogging(self):

python/opendht_cpp.pxd

@@ -26,6 +26,14 @@ from libc.string cimport const_char, const_uchar
 ctypedef uint16_t in_port_t
 ctypedef unsigned short int sa_family_t;
 
+cdef extern from "<iostream>" namespace "std::chrono" nogil:
+    cdef cppclass duration[ulong]:
+        duration() except +
+
+    cdef cppclass seconds:
+        duration seconds(uint64_t) except +
+        duration seconds()
+
 cdef extern from "<memory>" namespace "std" nogil:
     cdef cppclass shared_ptr[T]:
         shared_ptr() except +
@@ -131,7 +139,6 @@ cdef extern from "opendht/value.h" namespace "dht::Value":
 cdef extern from "opendht/value.h" namespace "dht::Value::Field":
     cdef Field None
     cdef Field Id
-    cdef Field ValueType
     cdef Field OwnerPk
     cdef Field SeqNum
     cdef Field UserType
@@ -141,7 +148,7 @@ cdef extern from "opendht/value.h" namespace "dht":
     cdef cppclass Value:
         Value() except +
         Value(vector[uint8_t]) except +
-        Value(const uint8_t* dat_ptr, size_t dat_len) except +
+        Value(const uint16_t t, const uint8_t* dat_ptr, size_t dat_len) except +
         string toString() const
         size_t size() const
         uint64_t id
@@ -150,6 +157,12 @@ cdef extern from "opendht/value.h" namespace "dht":
         vector[uint8_t] data
         string user_type
 
+    cdef cppclass ValueType:
+        ValueType(uint16_t id, string name, seconds expiration) except +
+        uint16_t id
+        string name
+        seconds expiration
+
     cdef cppclass Query:
         Query() except +
         Query(Select s, Where w) except +
@@ -234,6 +247,7 @@ cdef extern from "opendht/dhtrunner.h" namespace "dht":
         void run(const_char*, const_char*, const_char*, Config config)
         void join()
         void shutdown(ShutdownCallback)
+        void registerType(ValueType& value);
         bool isRunning()
         SockAddr getBound(sa_family_t af) const
         string getStorageLog() const

src/callbacks.cpp

@@ -141,6 +141,8 @@ NodeInfo::toJson() const
     Json::Value val;
     if (id)
         val["id"] = id.toString();
+    if (pkid)
+        val["pkid"] = pkid.toString();
     val["node_id"] = node_id.toString();
     val["ipv4"] = ipv4.toJson();
     val["ipv6"] = ipv6.toJson();
@@ -152,6 +154,8 @@ NodeInfo::NodeInfo(const Json::Value& v)
 {
     if (v.isMember("id"))
         id = InfoHash(v["id"].asString());
+    if (v.isMember("pkid"))
+        pkid = PkId(v["pkid"].asString());
     node_id = InfoHash(v["node_id"].asString());
     ipv4 = NodeStats(v["ipv4"]);
     ipv6 = NodeStats(v["ipv6"]);

src/crypto.cpp

@@ -710,13 +710,13 @@ CertificateRequest::sign(const PrivateKey& key, const std::string& password)
 }
 
 bool
-CertificateRequest::verify() const 
+CertificateRequest::verify() const
 {
     return gnutls_x509_crq_verify(request, 0) >= 0;
 }
 
-Blob 
-CertificateRequest::pack() const 
+Blob
+CertificateRequest::pack() const
 {
     gnutls_datum_t dat {nullptr, 0};
     if (auto err = gnutls_x509_crq_export2(request, GNUTLS_X509_FMT_PEM, &dat))
@@ -727,7 +727,7 @@ CertificateRequest::pack() const
 }
 
 std::string
-CertificateRequest::toString() const 
+CertificateRequest::toString() const
 {
     gnutls_datum_t dat {nullptr, 0};
     if (auto err = gnutls_x509_crq_export2(request, GNUTLS_X509_FMT_PEM, &dat))
@@ -1147,8 +1147,8 @@ void
 setRandomSerial(gnutls_x509_crt_t cert)
 {
     random_device rdev;
-    std::uniform_int_distribution<uint64_t> dist{};
-    uint64_t cert_serial = dist(rdev);
+    std::uniform_int_distribution<int64_t> dist{1};
+    int64_t cert_serial = dist(rdev);
     gnutls_x509_crt_set_serial(cert, &cert_serial, sizeof(cert_serial));
 }
 

src/dht.cpp

@@ -52,12 +52,7 @@ Dht::updateStatus(sa_family_t af)
         if (other.status == NodeStatus::Disconnected && d.status == NodeStatus::Disconnected)
             onDisconnected();
         else if (other.status == NodeStatus::Connected || d.status == NodeStatus::Connected) {
-            // On connected
-            if (bootstrapJob) {
-                bootstrapJob->cancel();
-                bootstrapJob.reset();
-            }
-            bootstrap_period = std::chrono::seconds(10);
+            onConnected();
         }
     }
     return d.status;
@@ -491,9 +486,9 @@ void Dht::searchSendAnnounceValue(const Sp<Search>& sr) {
                         sr->id.toString().c_str(), sn->node->toString().c_str(), a.value->id);
                 sn->acked[a.value->id] = {
                     network_engine.sendRefreshValue(sn->node, sr->id, a.value->id, sn->token, onDone,
-                    [this, ws, node=sn->node, v=a.value, 
-                     onDone, 
-                     onExpired, 
+                    [this, ws, node=sn->node, v=a.value,
+                     onDone,
+                     onExpired,
                      created = a.permanent ? time_point::max() : a.created,
                      next_refresh_time
                     ](const net::Request& /*req*/, net::DhtProtocolException&& e){
@@ -591,7 +586,7 @@ Dht::searchSynchedNodeListen(const Sp<Search>& sr, SearchNode& n)
     std::weak_ptr<Search> ws = sr;
     for (const auto& l : sr->listeners) {
         const auto& query = l.second.query;
-        
+
         auto r = n.listenStatus.find(query);
         if (n.getListenTime(r, listenExp) > scheduler.time())
             continue;
@@ -725,7 +720,7 @@ Dht::searchStep(Sp<Search> sr)
 
     while (sr->currentlySolicitedNodeCount() < MAX_REQUESTED_SEARCH_NODES and searchSendGetValues(sr));
 
-    
+
     if (sr->getNumberOfConsecutiveBadNodes() >= std::min<size_t>(sr->nodes.size(), SEARCH_MAX_BAD_NODES))
     {
         if (logger_)
@@ -1981,6 +1976,21 @@ Dht::expire()
     scheduler.add(expire_stuff_time, std::bind(&Dht::expire, this));
 }
 
+void
+Dht::onConnected()
+{
+    if (bootstrapJob) {
+        bootstrapJob->cancel();
+        bootstrapJob.reset();
+    }
+    bootstrap_period = std::chrono::seconds(10);
+    auto callbacks = std::move(onConnectCallbacks_);
+    while (not callbacks.empty()) {
+        callbacks.front()();
+        callbacks.pop();
+    }
+}
+
 void
 Dht::onDisconnected()
 {
@@ -2526,7 +2536,7 @@ Dht::saveState(const std::string& path) const
     state.nodes = exportNodes();
     state.values = exportValues();
     std::ofstream file(path);
-    msgpack::pack(file, state);    
+    msgpack::pack(file, state);
 }
 
 void

src/dht_proxy_client.cpp

@@ -155,10 +155,10 @@ DhtProxyClient::startProxy()
     if (logger_)
         logger_->d("[proxy:client] start proxy with %s", proxyUrl_.c_str());
 
-    nextProxyConfirmationTimer_ = std::make_shared<asio::steady_timer>(httpContext_, std::chrono::steady_clock::now());
+    nextProxyConfirmationTimer_ = std::make_unique<asio::steady_timer>(httpContext_, std::chrono::steady_clock::now());
     nextProxyConfirmationTimer_->async_wait(std::bind(&DhtProxyClient::handleProxyConfirm, this, std::placeholders::_1));
 
-    listenerRestartTimer_ = std::make_shared<asio::steady_timer>(httpContext_);
+    listenerRestartTimer_ = std::make_unique<asio::steady_timer>(httpContext_);
 
     loopSignal_();
 }
@@ -723,6 +723,15 @@ DhtProxyClient::onProxyInfos(const Json::Value& proxyInfos, const sa_family_t fa
         if (oldStatus == NodeStatus::Disconnected || oldStatus == NodeStatus::Connecting) {
             listenerRestartTimer_->expires_at(std::chrono::steady_clock::now());
             listenerRestartTimer_->async_wait(std::bind(&DhtProxyClient::restartListeners, this, std::placeholders::_1));
+            if (not onConnectCallbacks_.empty()) {
+                std::lock_guard<std::mutex> lock(lockCallbacks_);
+                callbacks_.emplace_back([cbs = std::move(onConnectCallbacks_)]() mutable {
+                    while (not cbs.empty()) {
+                        cbs.front()();
+                        cbs.pop();
+                    }
+                });
+            }
         }
         nextProxyConfirmationTimer_->expires_at(std::chrono::steady_clock::now() + std::chrono::minutes(15));
         nextProxyConfirmationTimer_->async_wait(std::bind(&DhtProxyClient::handleProxyConfirm, this, std::placeholders::_1));

src/dht_proxy_server.cpp

@@ -822,7 +822,7 @@ DhtProxyServer::subscribe(restinio::request_handle_t request,
             logger_->d("[proxy:server] [subscribe %s] [client %s] [session %s]", infoHash.toString().c_str(), clientId.c_str(), sessionId.c_str());
 
         // Insert new or return existing push listeners of a token
-        std::lock_guard<std::mutex> lock(lockPushListeners_);
+        std::lock_guard<std::mutex> lock(lockListener_);
         auto& pushListener = pushListeners_[pushToken];
         auto& pushListeners = pushListener.listeners[infoHash];
 

src/dhtrunner.cpp

@@ -76,19 +76,17 @@ DhtRunner::~DhtRunner()
 }
 
 void
-DhtRunner::run(in_port_t port, const Config& config, Context&& context)
+DhtRunner::run(in_port_t port, Config& config, Context&& context)
 {
-    SockAddr sin4;
-    sin4.setFamily(AF_INET);
-    sin4.setPort(port);
-    SockAddr sin6;
-    sin6.setFamily(AF_INET6);
-    sin6.setPort(port);
-    run(sin4, sin6, config, std::move(context));
+    config.bind4.setFamily(AF_INET);
+    config.bind4.setPort(port);
+    config.bind6.setFamily(AF_INET6);
+    config.bind6.setPort(port);
+    run(config, std::move(context));
 }
 
 void
-DhtRunner::run(const char* ip4, const char* ip6, const char* service, const Config& config, Context&& context)
+DhtRunner::run(const char* ip4, const char* ip6, const char* service, Config& config, Context&& context)
 {
     auto res4 = SockAddr::resolve(ip4, service);
     auto res6 = SockAddr::resolve(ip6, service);
@@ -96,44 +94,9 @@ DhtRunner::run(const char* ip4, const char* ip6, const char* service, const Conf
         res4.emplace_back();
     if (res6.empty())
         res6.emplace_back();
-    run(res4.front(), res6.front(), config, std::move(context));
-}
-
-void
-DhtRunner::run(SockAddr& local4, SockAddr& local6, const Config& config, Context&& context)
-{
-    if (running == State::Idle) {
-        auto state_path = config.dht_config.node_config.persist_path;
-        if (not state_path.empty()) {
-            state_path += "_port.txt";
-            std::ifstream inConfig(state_path);
-            if (inConfig.is_open()) {
-                in_port_t port;
-                if (inConfig >> port) {
-                    if (context.logger)
-                        context.logger->d("[runner %p] Using IPv4 port %hu from saved configuration", this, port);
-                    if (local4.getPort() == 0)
-                        local4.setPort(port);
-                }
-                if (inConfig >> port) {
-                    if (context.logger)
-                        context.logger->d("[runner %p] Using IPv6 port %hu from saved configuration", this, port);
-                    if (local6.getPort() == 0)
-                        local6.setPort(port);
-                }
-            }
-        }
-
-        if (not context.sock)
-            context.sock.reset(new net::UdpSocket(local4, local6, context.logger));
-
-        if (not state_path.empty()) {
-            std::ofstream outConfig(state_path);
-            outConfig << context.sock->getBoundRef(AF_INET).getPort() << std::endl;
-            outConfig << context.sock->getBoundRef(AF_INET6).getPort() << std::endl;
-        }
-        run(config, std::move(context));
-    }
+    config.bind4 = std::move(res4.front());
+    config.bind6 = std::move(res6.front());
+    run(config, std::move(context));
 }
 
 void
@@ -141,8 +104,49 @@ DhtRunner::run(const Config& config, Context&& context)
 {
     std::lock_guard<std::mutex> lck(dht_mtx);
     auto expected = State::Idle;
-    if (not running.compare_exchange_strong(expected, State::Running))
+    if (not running.compare_exchange_strong(expected, State::Running)) {
+        if (context.logger)
+            context.logger->w("[runner %p] Node is already running. Call join() first before calling run() again.");
         return;
+    }
+
+    auto local4 = config.bind4;
+    auto local6 = config.bind6;
+    if (not local4 and not local6) {
+        if (context.logger)
+            context.logger->w("[runner %p] No address to bind specified in the configuration, using default addresses");
+        local4.setFamily(AF_INET);
+        local6.setFamily(AF_INET6);
+    }
+    auto state_path = config.dht_config.node_config.persist_path;
+    if (not state_path.empty()) {
+        state_path += "_port.txt";
+        std::ifstream inConfig(state_path);
+        if (inConfig.is_open()) {
+            in_port_t port;
+            if (inConfig >> port) {
+                if (context.logger)
+                    context.logger->d("[runner %p] Using IPv4 port %hu from saved configuration", this, port);
+                if (local4.getPort() == 0)
+                    local4.setPort(port);
+            }
+            if (inConfig >> port) {
+                if (context.logger)
+                    context.logger->d("[runner %p] Using IPv6 port %hu from saved configuration", this, port);
+                if (local6.getPort() == 0)
+                    local6.setPort(port);
+            }
+        }
+    }
+
+    if (not context.sock)
+        context.sock.reset(new net::UdpSocket(local4, local6, context.logger));
+
+    if (not state_path.empty()) {
+        std::ofstream outConfig(state_path);
+        outConfig << context.sock->getBoundRef(AF_INET).getPort() << std::endl;
+        outConfig << context.sock->getBoundRef(AF_INET6).getPort() << std::endl;
+    }
 
     if (context.logger) {
         logger_ = context.logger;
@@ -167,12 +171,13 @@ DhtRunner::run(const Config& config, Context&& context)
         return ret;
     });
 
-    auto dht = std::unique_ptr<DhtInterface>(new Dht(std::move(context.sock), SecureDht::getConfig(config.dht_config), context.logger));
-    dht_ = std::unique_ptr<SecureDht>(new SecureDht(std::move(dht), config.dht_config));
-
 #ifdef OPENDHT_PROXY_CLIENT
     config_ = config;
+    identityAnnouncedCb_ = context.identityAnnouncedCb;
 #endif
+    auto dht = std::unique_ptr<DhtInterface>(new Dht(std::move(context.sock), SecureDht::getConfig(config.dht_config), context.logger));
+    dht_ = std::unique_ptr<SecureDht>(new SecureDht(std::move(dht), config.dht_config, std::move(context.identityAnnouncedCb), context.logger));
+
     enableProxy(not config.proxy_server.empty());
     if (context.logger and dht_via_proxy_) {
         dht_via_proxy_->setLogger(context.logger);
@@ -180,10 +185,10 @@ DhtRunner::run(const Config& config, Context&& context)
     if (context.statusChangedCallback) {
         statusCb = std::move(context.statusChangedCallback);
     }
-    if (context.certificateStore) {
-        dht_->setLocalCertificateStore(std::move(context.certificateStore));
+    if (context.certificateStore || context.certificateStoreSha256) {
+        dht_->setLocalCertificateStore(std::move(context.certificateStore), std::move(context.certificateStoreSha256));
         if (dht_via_proxy_)
-            dht_via_proxy_->setLocalCertificateStore(std::move(context.certificateStore));
+            dht_via_proxy_->setLocalCertificateStore(std::move(context.certificateStore), std::move(context.certificateStoreSha256));
     }
 
     if (not config.threaded)
@@ -393,6 +398,15 @@ DhtRunner::getId() const
     return {};
 }
 
+PkId
+DhtRunner::getLongId() const
+{
+    if (auto dht = activeDht())
+        return dht->getLongId();
+    return {};
+}
+
+
 InfoHash
 DhtRunner::getNodeId() const
 {
@@ -504,6 +518,7 @@ DhtRunner::getNodeInfo() const {
     NodeInfo info {};
     if (auto dht = activeDht()) {
         info.id = dht->getId();
+        info.pkid = dht->getLongId();
         info.node_id = dht->getNodeId();
         info.ipv4 = dht->getNodesStats(AF_INET);
         info.ipv6 = dht->getNodesStats(AF_INET6);
@@ -525,6 +540,7 @@ DhtRunner::getNodeInfo(std::function<void(std::shared_ptr<NodeInfo>)> cb)
         auto sinfo = std::make_shared<NodeInfo>();
         auto& info = *sinfo;
         info.id = dht.getId();
+        info.pkid = dht.getLongId();
         info.node_id = dht.getNodeId();
         info.ipv4 = dht.getNodesStats(AF_INET);
         info.ipv6 = dht.getNodesStats(AF_INET6);
@@ -593,20 +609,32 @@ DhtRunner::getPublicAddressStr(sa_family_t af)
     return ret;
 }
 
+void
+DhtRunner::getPublicAddress(std::function<void(std::vector<SockAddr>&&)> cb, sa_family_t af)
+{
+    std::lock_guard<std::mutex> lck(storage_mtx);
+    ongoing_ops++;
+    pending_ops_prio.emplace([cb = std::move(cb), this, af](SecureDht& dht){
+        cb(dht.getPublicAddress(af));
+        opEnded();
+    });
+    cv.notify_all();
+}
+
 void
 DhtRunner::registerCertificate(std::shared_ptr<crypto::Certificate> cert) {
     std::lock_guard<std::mutex> lck(dht_mtx);
     activeDht()->registerCertificate(cert);
 }
 void
-DhtRunner::setLocalCertificateStore(CertificateStoreQuery&& query_method) {
+DhtRunner::setLocalCertificateStore(CertificateStoreQuery&& query_method, CertificateStoreLongQuery&& query_sha256_method) {
     std::lock_guard<std::mutex> lck(dht_mtx);
 #ifdef OPENDHT_PROXY_CLIENT
     if (dht_via_proxy_)
-        dht_via_proxy_->setLocalCertificateStore(std::forward<CertificateStoreQuery>(query_method));
+        dht_via_proxy_->setLocalCertificateStore(std::move(query_method), std::move(query_sha256_method));
 #endif
     if (dht_)
-        dht_->setLocalCertificateStore(std::forward<CertificateStoreQuery>(query_method));
+        dht_->setLocalCertificateStore(std::move(query_method), std::move(query_sha256_method));
 }
 
 time_point
@@ -926,6 +954,55 @@ DhtRunner::putEncrypted(const std::string& key, InfoHash to, Value&& value, Done
     putEncrypted(InfoHash::get(key), to, std::forward<Value>(value), std::move(cb), permanent);
 }
 
+
+void
+DhtRunner::putEncrypted(InfoHash hash, PkId to, std::shared_ptr<Value> value, DoneCallback cb, bool permanent)
+{
+    if (running != State::Running) {
+        if (cb) cb(false, {});
+        return;
+    }
+    std::lock_guard<std::mutex> lck(storage_mtx);
+    ongoing_ops++;
+    pending_ops.emplace([=,
+        cb = std::move(cb),
+        value = std::move(value)
+    ] (SecureDht& dht) mutable {
+        dht.putEncrypted(hash, to, value, bindOpDoneCallback(std::move(cb)), permanent);
+    });
+    cv.notify_all();
+}
+
+void
+DhtRunner::putEncrypted(InfoHash hash, PkId to, Value&& value, DoneCallback cb, bool permanent)
+{
+    putEncrypted(hash, to, std::make_shared<Value>(std::move(value)), std::move(cb), permanent);
+}
+
+void
+DhtRunner::putEncrypted(InfoHash hash, const std::shared_ptr<crypto::PublicKey>& to, std::shared_ptr<Value> value, DoneCallback cb, bool permanent)
+{
+    if (running != State::Running) {
+        if (cb) cb(false, {});
+        return;
+    }
+    std::lock_guard<std::mutex> lck(storage_mtx);
+    ongoing_ops++;
+    pending_ops.emplace([=,
+        cb = std::move(cb),
+        value = std::move(value)
+    ] (SecureDht& dht) mutable {
+        dht.putEncrypted(hash, *to, value, bindOpDoneCallback(std::move(cb)), permanent);
+    });
+    cv.notify_all();
+}
+
+void
+DhtRunner::putEncrypted(InfoHash hash, const std::shared_ptr<crypto::PublicKey>& to, Value&& value, DoneCallback cb, bool permanent)
+{
+    putEncrypted(hash, to, std::make_shared<Value>(std::move(value)), std::move(cb), permanent);
+}
+
 void
 DhtRunner::bootstrap(const std::string& host, const std::string& service)
 {
@@ -1122,7 +1199,7 @@ DhtRunner::enableProxy(bool proxify)
         if (not config_.push_token.empty())
             dht_via_proxy->setPushNotificationToken(config_.push_token);
 #endif
-        dht_via_proxy_ = std::unique_ptr<SecureDht>(new SecureDht(std::move(dht_via_proxy), config_.dht_config));
+        dht_via_proxy_ = std::unique_ptr<SecureDht>(new SecureDht(std::move(dht_via_proxy), config_.dht_config, identityAnnouncedCb_, logger_));
         // add current listeners
         for (auto& l: listeners_)
             l.second.tokenProxyDht = dht_via_proxy_->listen(l.second.hash, l.second.gcb, l.second.f, l.second.w);

src/http.cpp

@@ -561,12 +561,63 @@ void
 Connection::async_connect(std::vector<asio::ip::tcp::endpoint>&& endpoints, ConnectHandlerCb cb)
 {
     std::lock_guard<std::mutex> lock(mutex_);
-    if (ssl_socket_)
-        asio::async_connect(ssl_socket_->lowest_layer(), std::move(endpoints), wrapCallabck(std::move(cb)));
-    else if (socket_)
-        asio::async_connect(*socket_, std::move(endpoints), wrapCallabck(std::move(cb)));
-    else if (cb)
+    if (!ssl_socket_ && !socket_) {
         cb(asio::error::operation_aborted, {});
+        return;
+    }
+    auto& base = ssl_socket_? ssl_socket_->lowest_layer() : *socket_;
+
+    ConnectHandlerCb wcb = [&base, cb=std::move(cb)](const asio::error_code& ec, const asio::ip::tcp::endpoint& endpoint) {
+        if (!ec) {
+            auto socket = base.native_handle();
+            // Once connected, set a keep alive on the TCP socket with 30 seconds delay
+            // This will generate broken pipes as soon as possible.
+            // Note this needs to be done once connected to have a valid native_handle()
+            uint32_t start = 30;
+            uint32_t interval = 30;
+            uint32_t cnt = 1;
+#ifdef _WIN32
+            std::string val = "1";
+            setsockopt(socket, SOL_SOCKET, SO_KEEPALIVE, val.c_str(), sizeof(val));
+
+            // TCP_KEEPIDLE and TCP_KEEPINTVL are available since Win 10 version 1709
+            // TCP_KEEPCNT since Win 10 version 1703
+#ifdef TCP_KEEPIDLE
+            std::string start_str = std::to_string(start);
+            setsockopt(socket, IPPROTO_TCP, TCP_KEEPIDLE,
+                    start_str.c_str(), sizeof(start_str));
+#endif
+#ifdef TCP_KEEPINTVL
+            std::string interval_str = std::to_string(interval);
+            setsockopt(socket, IPPROTO_TCP, TCP_KEEPINTVL,
+                    interval_str.c_str(), sizeof(interval_str));
+#endif
+#ifdef TCP_KEEPCNT
+            std::string cnt_str = std::to_string(cnt);
+            setsockopt(socket, IPPROTO_TCP, TCP_KEEPCNT,
+                    cnt_str.c_str(), sizeof(cnt_str));
+#endif
+#else
+            uint32_t val = 1;
+            setsockopt(socket, SOL_SOCKET, SO_KEEPALIVE, &val, sizeof(uint32_t));
+#ifdef __APPLE__
+            // Apple devices only have one parameter
+            setsockopt(socket, IPPROTO_TCP, TCP_KEEPALIVE, &start, sizeof(uint32_t));
+#else
+            // Linux based systems
+            setsockopt(socket, SOL_TCP, TCP_KEEPIDLE, &start, sizeof(uint32_t));
+            setsockopt(socket, SOL_TCP, TCP_KEEPINTVL, &interval, sizeof(uint32_t));
+            setsockopt(socket, SOL_TCP, TCP_KEEPCNT, &cnt, sizeof(uint32_t));
+#endif
+#endif
+        }
+        if (cb)
+            cb(ec, endpoint);
+    };
+    if (ssl_socket_)
+        asio::async_connect(ssl_socket_->lowest_layer(), std::move(endpoints), wrapCallabck(std::move(wcb)));
+    else
+        asio::async_connect(*socket_, std::move(endpoints), wrapCallabck(std::move(wcb)));
 }
 
 void
@@ -1254,9 +1305,11 @@ Request::terminate(const asio::error_code& ec)
         return;
 
     response_.aborted = ec == asio::error::operation_aborted;
+    if (ec == asio::error::basic_errors::broken_pipe)
+        response_.status_code = 0U; // Avoid to give a successful answer (happen with a broken pipe, takes the last status)
 
     if (logger_) {
-        if (ec and ec != asio::error::eof and ec != asio::error::operation_aborted)
+        if (ec and ec != asio::error::eof and !response_.aborted)
             logger_->e("[http:request:%i] end with error: %s", id_, ec.message().c_str());
         else
             logger_->d("[http:request:%i] done with status code %u", id_, response_.status_code);
@@ -1415,7 +1468,7 @@ Request::await()
     std::unique_lock<std::mutex> lock(mtx);
     std::condition_variable cv;
     bool ok {false};
-    add_on_done_callback([&](const Response& resp){
+    add_on_done_callback([&](const Response&){
         std::lock_guard<std::mutex> lk(mtx);
         ok = true;
         cv.notify_all();

src/securedht.cpp

@@ -36,8 +36,8 @@ extern "C" {
 
 namespace dht {
 
-SecureDht::SecureDht(std::unique_ptr<DhtInterface> dht, SecureDht::Config conf)
-: dht_(std::move(dht)), key_(conf.id.first), certificate_(conf.id.second), enableCache_(conf.cert_cache_all)
+SecureDht::SecureDht(std::unique_ptr<DhtInterface> dht, SecureDht::Config conf, IdentityAnnouncedCb iacb, const std::shared_ptr<Logger>& l)
+: DhtInterface(l), dht_(std::move(dht)), key_(conf.id.first), certificate_(conf.id.second), enableCache_(conf.cert_cache_all)
 {
     if (!dht_) return;
     for (const auto& type : DEFAULT_TYPES)
@@ -52,16 +52,17 @@ SecureDht::SecureDht(std::unique_ptr<DhtInterface> dht, SecureDht::Config conf)
         auto certId = certificate_->getId();
         if (key_ and certId != key_->getPublicKey().getId())
             throw DhtException("SecureDht: provided certificate doesn't match private key.");
-
-        dht_->put(certId, Value {
-            CERTIFICATE_TYPE,
-            *certificate_,
-            1
-        }, [this, certId](bool ok) {
-            if (ok)
+        dht_->addOnConnectedCallback([this, certId, cb=std::move(iacb)]{
+            dht_->put(certId, Value {
+                CERTIFICATE_TYPE,
+                *certificate_,
+                1
+            }, [this, certId, cb=std::move(cb)](bool ok) {
+                if (cb) cb(ok);
                 if (logger_)
-                    logger_->d(certId, "SecureDht: public key announced successfully");
-        }, {}, true);
+                    logger_->d(certId, "SecureDht: certificate announcement %s", ok ? "succeeded" : "failed");
+            }, {}, true);
+        });
     }
 }
 
@@ -117,7 +118,7 @@ SecureDht::secureType(ValueType&& type)
     return type;
 }
 
-const Sp<crypto::Certificate>
+Sp<crypto::Certificate>
 SecureDht::getCertificate(const InfoHash& node) const
 {
     if (node == getId())
@@ -129,7 +130,7 @@ SecureDht::getCertificate(const InfoHash& node) const
         return it->second;
 }
 
-const Sp<const crypto::PublicKey>
+Sp<crypto::PublicKey>
 SecureDht::getPublicKey(const InfoHash& node) const
 {
     if (node == getId())
@@ -141,7 +142,7 @@ SecureDht::getPublicKey(const InfoHash& node) const
         return it->second;
 }
 
-const Sp<crypto::Certificate>
+Sp<crypto::Certificate>
 SecureDht::registerCertificate(const InfoHash& node, const Blob& data)
 {
     Sp<crypto::Certificate> crt;
@@ -154,12 +155,9 @@ SecureDht::registerCertificate(const InfoHash& node, const Blob& data)
     if (node == h) {
         if (logger_)
             logger_->d("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));
-        else
-            it->second = std::move(crt);
-        return it->second;
+        nodesCertificates_[h] = crt;
+        nodesCertificatesLong_[crt->getLongId()] = crt;
+        return crt;
     } else {
         if (logger_)
             logger_->w("Certificate %s for node %s does not match node id !", h.toString().c_str(), node.toString().c_str());
@@ -170,8 +168,10 @@ SecureDht::registerCertificate(const InfoHash& node, const Blob& data)
 void
 SecureDht::registerCertificate(Sp<crypto::Certificate>& cert)
 {
-    if (cert)
+    if (cert) {
         nodesCertificates_[cert->getId()] = cert;
+        nodesCertificatesLong_[cert->getLongId()] = cert;
+    }
 }
 
 void
@@ -191,6 +191,7 @@ SecureDht::findCertificate(const InfoHash& node, const std::function<void(const
             if (logger_)
                 logger_->d("Registering certificate from local store for %s", node.to_c_str());
             nodesCertificates_.emplace(node, res.front());
+            nodesCertificatesLong_.emplace(res.front()->getLongId(), res.front());
             if (cb)
                 cb(res.front());
             return;
@@ -199,8 +200,6 @@ SecureDht::findCertificate(const InfoHash& node, const std::function<void(const
 
     auto found = std::make_shared<bool>(false);
     dht_->get(node, [cb,node,found,this](const std::vector<Sp<Value>>& vals) {
-        if (*found)
-            return false;
         for (const auto& v : vals) {
             if (auto cert = registerCertificate(node, v->data)) {
                 *found = true;
@@ -211,7 +210,7 @@ SecureDht::findCertificate(const InfoHash& node, const std::function<void(const
                 return false;
             }
         }
-        return true;
+        return !*found;
     }, [cb,found](bool) {
         if (!*found and cb)
             cb(nullptr);
@@ -219,7 +218,7 @@ SecureDht::findCertificate(const InfoHash& node, const std::function<void(const
 }
 
 void
-SecureDht::findPublicKey(const InfoHash& node, const std::function<void(const Sp<const crypto::PublicKey>)>& cb)
+SecureDht::findPublicKey(const InfoHash& node, const std::function<void(const Sp<crypto::PublicKey>)>& cb)
 {
     auto pk = getPublicKey(node);
     if (pk && *pk) {
@@ -234,6 +233,7 @@ SecureDht::findPublicKey(const InfoHash& node, const std::function<void(const Sp
             auto pk = std::make_shared<crypto::PublicKey>(crt->getPublicKey());
             if (*pk) {
                 nodesPubKeys_[pk->getId()] = pk;
+                nodesPubKeysLong_[pk->getLongId()] = pk;
                 if (cb) cb(pk);
                 return;
             }
@@ -242,6 +242,106 @@ SecureDht::findPublicKey(const InfoHash& node, const std::function<void(const Sp
     });
 }
 
+void
+SecureDht::findCertificate(const PkId& node, const std::function<void(const Sp<crypto::Certificate>)>& cb)
+{
+    Sp<crypto::Certificate> b = getCertificate(node);
+    if (b && *b) {
+        if (logger_)
+            logger_->d("Using certificate from cache for %s", node.to_c_str());
+        if (cb)
+            cb(b);
+        return;
+    }
+    if (localQueryMethod_) {
+        auto res = localQuerySha256Method_(node);
+        if (not res.empty()) {
+            if (logger_)
+                logger_->d("Registering certificate from local store for %s", node.to_c_str());
+            nodesCertificates_.emplace(res.front()->getId(), res.front());
+            nodesCertificatesLong_.emplace(node, res.front());
+            if (cb)
+                cb(res.front());
+            return;
+        }
+    }
+    if (cb)
+        cb(nullptr);
+}
+
+void
+SecureDht::findPublicKey(const PkId& node, const std::function<void(const Sp<crypto::PublicKey>)>& cb)
+{
+    auto pk = getPublicKey(node);
+    if (pk && *pk) {
+        if (logger_)
+            logger_->d("Found public key from cache for %s", node.to_c_str());
+        if (cb)
+            cb(pk);
+        return;
+    }
+    findCertificate(node, [=](const Sp<crypto::Certificate> crt) {
+        if (crt && *crt) {
+            auto pk = std::make_shared<crypto::PublicKey>(crt->getPublicKey());
+            if (*pk) {
+                nodesPubKeys_[pk->getId()] = pk;
+                nodesPubKeysLong_[pk->getLongId()] = pk;
+                if (cb) cb(pk);
+                return;
+            }
+        }
+        if (cb) cb(nullptr);
+    });
+}
+
+Sp<crypto::Certificate>
+SecureDht::registerCertificate(const PkId& pkId, const Blob& data)
+{
+    Sp<crypto::Certificate> crt;
+    try {
+        crt = std::make_shared<crypto::Certificate>(data);
+    } catch (const std::exception& e) {
+        return nullptr;
+    }
+    PkId id = crt->getPublicKey().getLongId();
+    if (pkId == id) {
+        if (logger_)
+            logger_->d("Registering certificate for %s", id.toString().c_str());
+        nodesCertificates_[crt->getId()] = crt;
+        nodesCertificatesLong_[id] = crt;
+        return crt;
+    } else {
+        if (logger_)
+            logger_->w("Certificate %s for node %s does not match node id !", id.toString().c_str(), pkId.toString().c_str());
+        return nullptr;
+    }
+}
+
+Sp<crypto::Certificate>
+SecureDht::getCertificate(const PkId& pkId) const
+{
+    if (pkId == getLongId())
+        return certificate_;
+    auto it = nodesCertificatesLong_.find(pkId);
+    if (it == nodesCertificatesLong_.end())
+        return nullptr;
+    else
+        return it->second;
+}
+
+Sp<crypto::PublicKey>
+SecureDht::getPublicKey(const PkId& pkId) const
+{
+    if (pkId == getLongId())
+        return std::make_shared<crypto::PublicKey>(certificate_->getPublicKey());
+    auto it = nodesPubKeysLong_.find(pkId);
+    if (it == nodesPubKeysLong_.end())
+        return nullptr;
+    else
+        return it->second;
+}
+
+
 Sp<Value>
 SecureDht::checkValue(const Sp<Value>& v)
 {
@@ -261,8 +361,10 @@ SecureDht::checkValue(const Sp<Value>& v)
         try {
             Value decrypted_val (decrypt(*v));
             if (decrypted_val.recipient == getId()) {
-                if (decrypted_val.owner)
+                if (decrypted_val.owner) {
                     nodesPubKeys_[decrypted_val.owner->getId()] = decrypted_val.owner;
+                    nodesPubKeysLong_[decrypted_val.owner->getLongId()] = decrypted_val.owner;
+                }
                 v->decryptedValue = std::make_shared<Value>(std::move(decrypted_val));
                 return v->decryptedValue;
             }
@@ -280,8 +382,10 @@ SecureDht::checkValue(const Sp<Value>& v)
         v->signatureChecked = true;
         if (v->owner and v->owner->checkSignature(v->getToSign(), v->signature)) {
             v->signatureValid = true;
-            if (enableCache_)
+            if (enableCache_) {
                 nodesPubKeys_[v->owner->getId()] = v->owner;
+                nodesPubKeysLong_[v->owner->getLongId()] = v->owner;
+            }
             return v;
         }
         else if (logger_)
@@ -404,7 +508,7 @@ SecureDht::putEncrypted(const InfoHash& hash, const InfoHash& to, Sp<Value> val,
             callback(false, {});
         return;
     }
-    findPublicKey(to, [=](const Sp<const crypto::PublicKey>& pk) {
+    findPublicKey(to, [this, hash, val = std::move(val), callback = std::move(callback), permanent](const Sp<crypto::PublicKey>& pk) {
         if(!pk || !*pk) {
             if (callback)
                 callback(false, {});
@@ -423,6 +527,53 @@ SecureDht::putEncrypted(const InfoHash& hash, const InfoHash& to, Sp<Value> val,
     });
 }
 
+void
+SecureDht::putEncrypted(const InfoHash& hash, const PkId& to, Sp<Value> val, DoneCallback callback, bool permanent)
+{
+    if (not key_)  {
+        if (callback)
+            callback(false, {});
+        return;
+    }
+    findPublicKey(to, [this, hash, val = std::move(val), callback = std::move(callback), permanent](const Sp<crypto::PublicKey>& pk) {
+        if(!pk || !*pk) {
+            if (callback)
+                callback(false, {});
+            return;
+        }
+        if (logger_)
+            logger_->w("Encrypting data for PK: %s", pk->getLongId().toString().c_str());
+        try {
+            dht_->put(hash, encrypt(*val, *pk), callback, time_point::max(), permanent);
+        } catch (const std::exception& e) {
+            if (logger_)
+                logger_->e("Error putting encrypted data: %s", e.what());
+            if (callback)
+                callback(false, {});
+        }
+    });
+}
+
+void
+SecureDht::putEncrypted(const InfoHash& hash, const crypto::PublicKey& pk, Sp<Value> val, DoneCallback callback, bool permanent)
+{
+    if (not key_)  {
+        if (callback)
+            callback(false, {});
+        return;
+    }
+    if (logger_)
+        logger_->w("Encrypting data for PK: %s", pk.getLongId().to_c_str());
+    try {
+        dht_->put(hash, encrypt(*val, pk), callback, time_point::max(), permanent);
+    } catch (const std::exception& e) {
+        if (logger_)
+            logger_->e("Error putting encrypted data: %s", e.what());
+        if (callback)
+            callback(false, {});
+    }
+}
+
 void
 SecureDht::sign(Value& v) const
 {

src/value.cpp

@@ -163,7 +163,7 @@ Value::msgpack_unpack_body(const msgpack::object& o)
                 throw msgpack::type_error();
             crypto::PublicKey new_owner;
             new_owner.msgpack_unpack(*rowner);
-            owner = std::make_shared<const crypto::PublicKey>(std::move(new_owner));
+            owner = std::make_shared<crypto::PublicKey>(std::move(new_owner));
             if (auto rrecipient = findMapValue(*rbody, VALUE_KEY_TO)) {
                 recipient = rrecipient->as<InfoHash>();
             }
@@ -193,7 +193,7 @@ Value::Value(const Json::Value& json)
     if (jowner.isString()) {
         auto ownerStr = jowner.asString();
         auto ownerBlob = std::vector<unsigned char>(ownerStr.begin(), ownerStr.end());
-        owner = std::make_shared<const crypto::PublicKey>(ownerBlob);
+        owner = std::make_shared<crypto::PublicKey>(ownerBlob);
     }
     const auto& jto = json[VALUE_KEY_TO];
     if (jto.isString())

tests/dhtrunnertester.cpp

@@ -23,6 +23,7 @@
 #include <mutex>
 #include <condition_variable>
 using namespace std::chrono_literals;
+using namespace std::literals;
 
 namespace test {
 CPPUNIT_TEST_SUITE_REGISTRATION(DhtRunnerTester);
@@ -60,6 +61,12 @@ void
 DhtRunnerTester::testConstructors() {
     CPPUNIT_ASSERT(node1.getBoundPort() == 42222);
     CPPUNIT_ASSERT(node2.getBoundPort() == 42232);
+
+    dht::DhtRunner::Config config {};
+    dht::DhtRunner::Context context {};
+    dht::DhtRunner testNode;
+    testNode.run(config, std::move(context));
+    CPPUNIT_ASSERT(testNode.getBoundPort());
 }
 
 void
@@ -167,6 +174,76 @@ DhtRunnerTester::testListen() {
     node1.cancelListen(d, tokend);
 }
 
+void
+DhtRunnerTester::testIdOps() {
+    std::mutex mutex;
+    std::condition_variable cv;
+    unsigned valueCount(0);
+
+    dht::DhtRunner::Config config2;
+    config2.dht_config.node_config.max_peer_req_per_sec = -1;
+    config2.dht_config.node_config.max_req_per_sec = -1;
+    config2.dht_config.id = dht::crypto::generateIdentity();
+
+    dht::DhtRunner::Context context2;
+    context2.identityAnnouncedCb = [&](bool ok) {
+        CPPUNIT_ASSERT(ok);
+        std::lock_guard<std::mutex> lk(mutex);
+        valueCount++;
+        cv.notify_all();
+    };
+
+    node2.join();
+    node2.run(42232, config2, std::move(context2));
+    node2.bootstrap(node1.getBound());
+
+    node1.findCertificate(node2.getId(), [&](const std::shared_ptr<dht::crypto::Certificate>& crt){
+        CPPUNIT_ASSERT(crt);
+        std::lock_guard<std::mutex> lk(mutex);
+        valueCount++;
+        cv.notify_all();
+    });
+
+    {
+        std::unique_lock<std::mutex> lk(mutex);
+        CPPUNIT_ASSERT(cv.wait_for(lk, 20s, [&]{ return valueCount == 2; }));
+    }
+
+    dht::DhtRunner::Context context1;
+    context1.identityAnnouncedCb = [&](bool ok) {
+        CPPUNIT_ASSERT(ok);
+        std::lock_guard<std::mutex> lk(mutex);
+        valueCount++;
+        cv.notify_all();
+    };
+
+    config2.dht_config.id = dht::crypto::generateIdentity();
+    node1.join();
+    node1.run(42222, config2, std::move(context1));
+    node1.bootstrap(node2.getBound());
+
+    auto key = dht::InfoHash::get("key");
+    node1.putEncrypted(key, node2.getId(), dht::Value("yo"), [&](bool ok){
+        CPPUNIT_ASSERT(ok);
+        std::lock_guard<std::mutex> lk(mutex);
+        valueCount++;
+        cv.notify_all();
+    });
+
+    node2.listen<std::string>(key, [&](std::string&& value){
+        CPPUNIT_ASSERT_EQUAL("yo"s, value);
+        std::lock_guard<std::mutex> lk(mutex);
+        valueCount++;
+        cv.notify_all();
+        return false;
+    });
+
+    {
+        std::unique_lock<std::mutex> lk(mutex);
+        CPPUNIT_ASSERT(cv.wait_for(lk, 20s, [&]{ return valueCount == 4; }));
+    }
+}
+
 void
 DhtRunnerTester::testListenLotOfBytes() {
     std::mutex mutex;

tests/dhtrunnertester.h

@@ -33,6 +33,7 @@ class DhtRunnerTester : public CppUnit::TestFixture {
     CPPUNIT_TEST(testGetPut);
     CPPUNIT_TEST(testListen);
     CPPUNIT_TEST(testListenLotOfBytes);
+    CPPUNIT_TEST(testIdOps);
     CPPUNIT_TEST_SUITE_END();
 
     dht::DhtRunner node1 {};
@@ -58,6 +59,10 @@ class DhtRunnerTester : public CppUnit::TestFixture {
      * Test listen method
      */
     void testListen();
+    /**
+     * Test methods requiring a node identity
+     */
+    void testIdOps();
     /**
      * Test listen method with lot of datas
      */

tools/tools_common.h

@@ -200,6 +200,8 @@ void print_node_info(const dht::NodeInfo& info) {
         std::cout << "IPv4 port " << info.bound4 << ", IPv6 port " << info.bound6 << std::endl;
     if (info.id)
         std::cout << "Public key ID " << info.id << std::endl;
+    if (info.pkid)
+        std::cout << "Public key long ID " << info.pkid << std::endl;
 }
 
 static const constexpr struct option long_options[] = {