value: cleanup

c/opendht.cpp

@@ -3,6 +3,7 @@
 
 using ValueSp = std::shared_ptr<dht::Value>;
 using PrivkeySp = std::shared_ptr<dht::crypto::PrivateKey>;
+using PubkeySp = std::shared_ptr<const dht::crypto::PublicKey>;
 using CertSp = std::shared_ptr<dht::crypto::Certificate>;
 
 #ifdef __cplusplus
@@ -10,6 +11,19 @@ extern "C" {
 #endif
 
 // dht::InfoHash
+
+inline dht_infohash dht_infohash_to_c(const dht::InfoHash& h) {
+    dht_infohash ret;
+    *reinterpret_cast<dht::InfoHash*>(&ret) = h;
+    return ret;
+}
+
+inline dht_pkid dht_pkid_to_c(const dht::PkId& h) {
+    dht_pkid ret;
+    *reinterpret_cast<dht::PkId*>(&ret) = h;
+    return ret;
+}
+
 const char* dht_infohash_print(const dht_infohash* h) {
     return reinterpret_cast<const dht::InfoHash*>(h)->to_c_str();
 }
@@ -26,10 +40,18 @@ void dht_infohash_get(dht_infohash* h, const uint8_t* dat, size_t dat_size) {
     *reinterpret_cast<dht::InfoHash*>(h) = dht::InfoHash::get(dat, dat_size);
 }
 
+void dht_infohash_get_from_string(dht_infohash* h, const char* dat) {
+    *reinterpret_cast<dht::InfoHash*>(h) = dht::InfoHash::get((const uint8_t*)dat, (size_t)strlen(dat));
+}
+
 bool dht_infohash_is_zero(const dht_infohash* h) {
     return !static_cast<bool>(*reinterpret_cast<const dht::InfoHash*>(h));
 }
 
+void dht_infohash_from_hex(dht_infohash* h, const char* dat) {
+    *h = dht_infohash_to_c(dht::InfoHash(std::string(dat, HASH_LEN*2)));
+}
+
 const char* dht_pkid_print(const dht_pkid* h) {
     return reinterpret_cast<const dht::PkId*>(h)->to_c_str();
 }
@@ -60,6 +82,21 @@ dht_value_id dht_value_get_id(const dht_value* data) {
     return vsp->id;
 }
 
+dht_publickey* dht_value_get_owner(const dht_value* data) {
+    const ValueSp& vsp(*reinterpret_cast<const ValueSp*>(data));
+    return vsp->owner ? reinterpret_cast<dht_publickey*>(new PubkeySp(vsp->owner)) : nullptr;
+}
+
+dht_infohash dht_value_get_recipient(const dht_value* data) {
+    const ValueSp& vsp(*reinterpret_cast<const ValueSp*>(data));
+    return dht_infohash_to_c(vsp->recipient);
+}
+
+const char* dht_value_get_user_type(const dht_value* data) {
+    const ValueSp& vsp(*reinterpret_cast<const ValueSp*>(data));
+    return vsp->user_type.c_str();
+}
+
 dht_value* dht_value_new(const uint8_t* data, size_t size) {
     return reinterpret_cast<dht_value*>(new ValueSp(std::make_shared<dht::Value>(data, size)));
 }
@@ -73,51 +110,46 @@ void dht_value_unref(dht_value* v) {
 }
 
 // dht::crypto::PublicKey
-dht_publickey* dht_publickey_new() {
-    return reinterpret_cast<dht_publickey*>(new dht::crypto::PublicKey);
+dht_publickey* dht_publickey_import(const uint8_t* dat, size_t dat_size) {
+    try {
+        return reinterpret_cast<dht_publickey*>(new PubkeySp(std::make_shared<const dht::crypto::PublicKey>(dat, dat_size)));
+    } catch (const dht::crypto::CryptoException& e) {
+        return nullptr;
+    }
 }
 
 void dht_publickey_delete(dht_publickey* pk) {
-    delete reinterpret_cast<dht::crypto::PublicKey*>(pk);
-}
-
-int dht_publickey_unpack(dht_publickey* pk, const uint8_t* dat, size_t dat_size) {
-    try {
-        reinterpret_cast<dht::crypto::PublicKey*>(pk)->unpack(dat, dat_size);
-    } catch (const dht::crypto::CryptoException& e) {
-        return -1;
-    }
-    return 0;
+    delete reinterpret_cast<PubkeySp*>(pk);
 }
 
 int dht_publickey_pack(dht_publickey* pk, char* out, size_t* outlen) {
-    return gnutls_pubkey_export(reinterpret_cast<dht::crypto::PublicKey*>(pk)->pk, GNUTLS_X509_FMT_DER, out, outlen);
+    const auto& pkey = *reinterpret_cast<const PubkeySp*>(pk);
+    return gnutls_pubkey_export(pkey->pk, GNUTLS_X509_FMT_DER, out, outlen);
 }
 
 dht_infohash dht_publickey_get_id(const dht_publickey* pk) {
-    auto pkey = reinterpret_cast<const dht::crypto::PublicKey*>(pk);
-    dht_infohash h;
-    *reinterpret_cast<dht::InfoHash*>(&h) = pkey->getId();
-    return h;
+    const auto& pkey = *reinterpret_cast<const PubkeySp*>(pk);
+    return dht_infohash_to_c(pkey->getId());
 }
 
 dht_pkid dht_publickey_get_long_id(const dht_publickey* pk) {
-    auto pkey = reinterpret_cast<const dht::crypto::PublicKey*>(pk);
-    dht_pkid h;
-    *reinterpret_cast<dht::PkId*>(&h) = pkey->getLongId();
-    return h;
+    const auto& pkey = *reinterpret_cast<const PubkeySp*>(pk);
+    return dht_pkid_to_c(pkey->getLongId());
 }
 
 bool dht_publickey_check_signature(const dht_publickey* pk, const char* data, size_t data_size, const char* signature, size_t signature_size) {
-    return reinterpret_cast<const dht::crypto::PublicKey*>(pk)->checkSignature((const uint8_t*)data, data_size, (const uint8_t*)signature, signature_size);
+    const auto& pkey = *reinterpret_cast<const PubkeySp*>(pk);
+    return pkey->checkSignature((const uint8_t*)data, data_size, (const uint8_t*)signature, signature_size);
 }
 
 dht_blob* dht_publickey_encrypt(const dht_publickey* pk, const char* data, size_t data_size) {
+    const auto& pkey = *reinterpret_cast<const PubkeySp*>(pk);
     auto rdata = new dht::Blob;
-    *rdata = reinterpret_cast<const dht::crypto::PublicKey*>(pk)->encrypt((const uint8_t*)data, data_size);
+    *rdata = pkey->encrypt((const uint8_t*)data, data_size);
     return (dht_blob*)rdata;
 }
 
+// dht::crypto::PrivateKey
 dht_privatekey* dht_privatekey_generate(unsigned key_length_bits) {
     if (key_length_bits == 0)
         key_length_bits = 4096;
@@ -134,13 +166,73 @@ dht_privatekey* dht_privatekey_import(const uint8_t* dat, size_t dat_size, const
 
 dht_publickey* dht_privatekey_get_publickey(const dht_privatekey* k) {
     const auto& key = *reinterpret_cast<const PrivkeySp*>(k);
-    return reinterpret_cast<dht_publickey*>(new dht::crypto::PublicKey(key->getPublicKey()));
+    return reinterpret_cast<dht_publickey*>(new PubkeySp(std::make_shared<dht::crypto::PublicKey>(key->getPublicKey())));
 }
 
 void dht_privatekey_delete(dht_privatekey* pk) {
     delete reinterpret_cast<PrivkeySp*>(pk);
 }
 
+// dht::crypto::Certificate
+dht_certificate* dht_certificate_import(const uint8_t* dat, size_t dat_size) {
+    try {
+        return reinterpret_cast<dht_certificate*>(new CertSp(std::make_shared<dht::crypto::Certificate>(dat, dat_size)));
+    } catch (const dht::crypto::CryptoException& e) {
+        return nullptr;
+    }
+}
+
+void dht_certificate_delete(dht_certificate* c) {
+    delete reinterpret_cast<CertSp*>(c);
+}
+
+dht_infohash dht_certificate_get_id(const dht_certificate* c) {
+    const auto& cert = *reinterpret_cast<const CertSp*>(c);
+    return dht_infohash_to_c(cert->getId());
+}
+
+dht_pkid dht_certificate_get_long_id(const dht_certificate* c) {
+    const auto& cert = *reinterpret_cast<const CertSp*>(c);
+    return dht_pkid_to_c(cert->getLongId());
+}
+
+dht_publickey* dht_certificate_get_publickey(const dht_certificate* c) {
+    const auto& cert = *reinterpret_cast<const CertSp*>(c);
+    return reinterpret_cast<dht_publickey*>(new PubkeySp(std::make_shared<dht::crypto::PublicKey>(cert->getPublicKey())));
+}
+
+// dht::crypto::Identity
+inline dht::crypto::Identity dht_identity_from_c(const dht_identity* cid) {
+    dht::crypto::Identity id {};
+    if (cid and cid->privatekey)
+        id.first = *reinterpret_cast<const PrivkeySp*>(cid->privatekey);
+    if (cid and cid->certificate)
+        id.second = *reinterpret_cast<const CertSp*>(cid->certificate);
+    return id;
+}
+
+inline dht_identity dht_identity_to_c(const dht::crypto::Identity& id) {
+    dht_identity cid {};
+    cid.privatekey = id.first ? reinterpret_cast<dht_privatekey*>(new PrivkeySp(id.first)) : NULL;
+    cid.certificate = id.second ? reinterpret_cast<dht_certificate*>(new CertSp(id.second)) : NULL;
+    return cid;
+}
+
+OPENDHT_C_PUBLIC dht_identity dht_identity_generate(const char* common_name, const dht_identity* ca) {
+    return dht_identity_to_c(dht::crypto::generateIdentity(common_name, dht_identity_from_c(ca)));
+}
+
+OPENDHT_C_PUBLIC void dht_identity_delete(dht_identity* id) {
+    if (id->certificate) {
+        dht_certificate_delete(id->certificate);
+        id->certificate = NULL;
+    }
+    if (id->privatekey) {
+        dht_privatekey_delete(id->privatekey);
+        id->privatekey = NULL;
+    }
+}
+
 // config
 void dht_runner_config_default(dht_runner_config* config) {
     bzero(config, sizeof(dht_runner_config));
@@ -171,8 +263,8 @@ void dht_runner_run_config(dht_runner* r, in_port_t port, const dht_runner_confi
     config.dht_config.node_config.persist_path = conf->dht_config.node_config.persist_path
         ? std::string(conf->dht_config.node_config.persist_path) : std::string{};
 
-    if (conf->dht_config.id.privkey)
-        config.dht_config.id.first = *reinterpret_cast<const PrivkeySp*>(conf->dht_config.id.privkey);
+    if (conf->dht_config.id.privatekey)
+        config.dht_config.id.first = *reinterpret_cast<const PrivkeySp*>(conf->dht_config.id.privatekey);
 
     if (conf->dht_config.id.certificate)
         config.dht_config.id.second = *reinterpret_cast<const CertSp*>(conf->dht_config.id.certificate);
@@ -249,24 +341,35 @@ void dht_op_token_delete(dht_op_token* token) {
     delete reinterpret_cast<std::future<size_t>*>(token);
 }
 
-void dht_runner_put(dht_runner* r, const dht_infohash* h, const dht_value* v, dht_done_cb done_cb, void* cb_user_data) {
+void dht_runner_put(dht_runner* r, const dht_infohash* h, const dht_value* v, dht_done_cb done_cb, void* cb_user_data, bool permanent) {
     auto runner = reinterpret_cast<dht::DhtRunner*>(r);
     auto hash = reinterpret_cast<const dht::InfoHash*>(h);
     auto value = reinterpret_cast<const ValueSp*>(v);
     runner->put(*hash, *value, [done_cb, cb_user_data](bool ok){
         if (done_cb)
             done_cb(ok, cb_user_data);
-    });
+    }, dht::time_point::max(), permanent);
 }
 
-void dht_runner_put_permanent(dht_runner* r, const dht_infohash* h, const dht_value* v, dht_done_cb done_cb, void* cb_user_data) {
+void dht_runner_put_signed(dht_runner* r, const dht_infohash* h, const dht_value* v, dht_done_cb done_cb, void* cb_user_data, bool permanent) {
     auto runner = reinterpret_cast<dht::DhtRunner*>(r);
     auto hash = reinterpret_cast<const dht::InfoHash*>(h);
     auto value = reinterpret_cast<const ValueSp*>(v);
-    runner->put(*hash, *value, [done_cb, cb_user_data](bool ok){
+    runner->putSigned(*hash, *value, [done_cb, cb_user_data](bool ok){
         if (done_cb)
             done_cb(ok, cb_user_data);
-    }, dht::time_point::max(), true);
+    }, permanent);
+}
+
+void dht_runner_put_encrypted(dht_runner* r, const dht_infohash* h, const dht_infohash* to, const dht_value* v, dht_done_cb done_cb, void* cb_user_data, bool permanent) {
+    auto runner = reinterpret_cast<dht::DhtRunner*>(r);
+    auto hash = reinterpret_cast<const dht::InfoHash*>(h);
+    auto toHash = reinterpret_cast<const dht::InfoHash*>(to);
+    auto value = reinterpret_cast<const ValueSp*>(v);
+    runner->putEncrypted(*hash, *toHash, *value, [done_cb, cb_user_data](bool ok){
+        if (done_cb)
+            done_cb(ok, cb_user_data);
+    }, permanent);
 }
 
 void dht_runner_cancel_put(dht_runner* r, const dht_infohash* h, dht_value_id value_id) {

c/opendht_c.h

@@ -18,16 +18,6 @@ struct OPENDHT_C_PUBLIC dht_data_view {
 };
 typedef struct dht_data_view dht_data_view;
 
-// dht::Value
-struct OPENDHT_C_PUBLIC dht_value;
-typedef struct dht_value dht_value;
-typedef uint64_t dht_value_id;
-OPENDHT_C_PUBLIC dht_data_view dht_value_get_data(const dht_value* data);
-OPENDHT_C_PUBLIC dht_value_id dht_value_get_id(const dht_value* data);
-OPENDHT_C_PUBLIC dht_value* dht_value_new(const uint8_t* data, size_t size);
-OPENDHT_C_PUBLIC dht_value* dht_value_ref(const dht_value*);
-OPENDHT_C_PUBLIC void dht_value_unref(dht_value*);
-
 // dht::Blob
 struct OPENDHT_C_PUBLIC dht_blob;
 typedef struct dht_blob dht_blob;
@@ -39,7 +29,9 @@ struct OPENDHT_C_PUBLIC dht_infohash { uint8_t d[HASH_LEN]; };
 typedef struct dht_infohash dht_infohash;
 OPENDHT_C_PUBLIC void dht_infohash_zero(dht_infohash* h);
 OPENDHT_C_PUBLIC void dht_infohash_random(dht_infohash* h);
+OPENDHT_C_PUBLIC void dht_infohash_from_hex(dht_infohash* h, const char* dat);
 OPENDHT_C_PUBLIC void dht_infohash_get(dht_infohash* h, const uint8_t* dat, size_t dat_size);
+OPENDHT_C_PUBLIC void dht_infohash_get_from_string(dht_infohash* h, const char* str);
 OPENDHT_C_PUBLIC const char* dht_infohash_print(const dht_infohash* h);
 OPENDHT_C_PUBLIC bool dht_infohash_is_zero(const dht_infohash* h);
 
@@ -51,9 +43,8 @@ OPENDHT_C_PUBLIC const char* dht_pkid_print(const dht_pkid* h);
 // dht::crypto::PublicKey
 struct OPENDHT_C_PUBLIC dht_publickey;
 typedef struct dht_publickey dht_publickey;
-OPENDHT_C_PUBLIC dht_publickey* dht_publickey_new();
+OPENDHT_C_PUBLIC dht_publickey* dht_publickey_import(const uint8_t* dat, size_t dat_size);
 OPENDHT_C_PUBLIC void dht_publickey_delete(dht_publickey* pk);
-OPENDHT_C_PUBLIC int dht_publickey_unpack(dht_publickey* pk, const uint8_t* dat, size_t dat_size);
 OPENDHT_C_PUBLIC int dht_publickey_pack(dht_publickey* pk, char* out, size_t* out_size);
 OPENDHT_C_PUBLIC dht_infohash dht_publickey_get_id(const dht_publickey* pk);
 OPENDHT_C_PUBLIC dht_pkid dht_publickey_get_long_id(const dht_publickey* pk);
@@ -71,18 +62,38 @@ OPENDHT_C_PUBLIC void dht_privatekey_delete(dht_privatekey*);
 // dht::crypto::Certificate
 struct OPENDHT_C_PUBLIC dht_certificate;
 typedef struct dht_certificate dht_certificate;
+OPENDHT_C_PUBLIC dht_certificate* dht_certificate_import(const uint8_t* dat, size_t dat_size);
+OPENDHT_C_PUBLIC dht_infohash dht_certificate_get_id(const dht_certificate*);
+OPENDHT_C_PUBLIC dht_pkid dht_certificate_get_long_id(const dht_certificate*);
+OPENDHT_C_PUBLIC dht_publickey* dht_certificate_get_publickey(const dht_certificate*);
+OPENDHT_C_PUBLIC void dht_certificate_delete(dht_certificate*);
 
 struct OPENDHT_PUBLIC dht_identity {
-    dht_privatekey* privkey;
+    dht_privatekey* privatekey;
     dht_certificate* certificate;
 };
 typedef struct dht_identity dht_identity;
+OPENDHT_C_PUBLIC dht_identity dht_identity_generate(const char* common_name, const dht_identity* ca);
+OPENDHT_C_PUBLIC void dht_identity_delete(dht_identity*);
+
+// dht::Value
+struct OPENDHT_C_PUBLIC dht_value;
+typedef struct dht_value dht_value;
+typedef uint64_t dht_value_id;
+OPENDHT_C_PUBLIC dht_value* dht_value_new(const uint8_t* data, size_t size);
+OPENDHT_C_PUBLIC dht_value* dht_value_ref(const dht_value*);
+OPENDHT_C_PUBLIC void dht_value_unref(dht_value*);
+OPENDHT_C_PUBLIC dht_data_view dht_value_get_data(const dht_value* data);
+OPENDHT_C_PUBLIC dht_value_id dht_value_get_id(const dht_value* data);
+OPENDHT_C_PUBLIC dht_publickey* dht_value_get_owner(const dht_value* data);
+OPENDHT_C_PUBLIC dht_infohash dht_value_get_recipient(const dht_value* data);
+OPENDHT_C_PUBLIC const char* dht_value_get_user_type(const dht_value* data);
 
 // callbacks
 typedef bool (*dht_get_cb)(const dht_value* value, void* user_data);
 typedef bool (*dht_value_cb)(const dht_value* value, bool expired, void* user_data);
-typedef bool (*dht_done_cb)(bool ok, void* user_data);
-typedef bool (*dht_shutdown_cb)(void* user_data);
+typedef void (*dht_done_cb)(bool ok, void* user_data);
+typedef void (*dht_shutdown_cb)(void* user_data);
 
 struct OPENDHT_C_PUBLIC dht_op_token;
 typedef struct dht_op_token dht_op_token;
@@ -98,8 +109,7 @@ struct OPENDHT_PUBLIC dht_node_config {
 };
 typedef struct dht_node_config dht_node_config;
 
-struct OPENDHT_PUBLIC dht_secure_config
-{
+struct OPENDHT_PUBLIC dht_secure_config {
     dht_node_config node_config;
     dht_identity id;
 };
@@ -131,8 +141,9 @@ OPENDHT_C_PUBLIC void dht_runner_bootstrap(dht_runner* runner, const char* host,
 OPENDHT_C_PUBLIC void dht_runner_get(dht_runner* runner, const dht_infohash* hash, dht_get_cb cb, dht_done_cb done_cb, void* cb_user_data);
 OPENDHT_C_PUBLIC dht_op_token* dht_runner_listen(dht_runner* runner, const dht_infohash* hash, dht_value_cb cb, dht_shutdown_cb done_cb, void* cb_user_data);
 OPENDHT_C_PUBLIC void dht_runner_cancel_listen(dht_runner* runner, const dht_infohash* hash, dht_op_token* token);
-OPENDHT_C_PUBLIC void dht_runner_put(dht_runner* runner, const dht_infohash* hash, const dht_value* value, dht_done_cb done_cb, void* cb_user_data);
-OPENDHT_C_PUBLIC void dht_runner_put_permanent(dht_runner* runner, const dht_infohash* hash, const dht_value* value, dht_done_cb done_cb, void* cb_user_data);
+OPENDHT_C_PUBLIC void dht_runner_put(dht_runner* runner, const dht_infohash* hash, const dht_value* value, dht_done_cb done_cb, void* cb_user_data, bool permanent);
+OPENDHT_C_PUBLIC void dht_runner_put_signed(dht_runner* runner, const dht_infohash* hash, const dht_value* value, dht_done_cb done_cb, void* cb_user_data, bool permanent);
+OPENDHT_C_PUBLIC void dht_runner_put_encrypted(dht_runner* runner, const dht_infohash* hash, const dht_infohash* to, const dht_value* value, dht_done_cb done_cb, void* cb_user_data, bool permanent);
 OPENDHT_C_PUBLIC void dht_runner_cancel_put(dht_runner* runner, const dht_infohash* hash, dht_value_id value_id);
 OPENDHT_C_PUBLIC void dht_runner_shutdown(dht_runner* runner, dht_shutdown_cb done_cb, void* cb_user_data);
 OPENDHT_C_PUBLIC dht_infohash dht_runner_get_node_id(const dht_runner* runner);

docker/DockerfileDeps

@@ -2,7 +2,7 @@ FROM ubuntu:16.04
 MAINTAINER Adrien Béraud <adrien.beraud@savoirfairelinux.com>
 RUN apt-get update && apt-get install -y \
         build-essential cmake git wget libncurses5-dev libreadline-dev nettle-dev \
-        libgnutls28-dev libuv1-dev cython3 python3-dev libcppunit-dev libjsoncpp-dev \
+        libgnutls28-dev libuv1-dev cython3 python3-dev python3-setuptools libcppunit-dev libjsoncpp-dev \
         autotools-dev autoconf libasio-dev \
     && apt-get clean
 
@@ -28,14 +28,11 @@ RUN echo "** Building a recent version of asio ***" \
     && make install \
     && cd ../../ && rm -rf asio*
 
-#patch for https://github.com/Stiffstream/restinio-conan-example/issues/2
-RUN apt-get update && apt-get install -y python3-pip && pip3 install --upgrade cmake
-
 RUN echo "*** Downloading RESTinio ***" \
     && mkdir restinio && cd restinio \
-    && wget https://github.com/Stiffstream/restinio/archive/v.0.5.1.2.tar.gz \
-    && ls -l && tar -xzf v.0.5.1.2.tar.gz \
-    && cd restinio-v.0.5.1.2/dev \
+    && wget https://github.com/aberaud/restinio/archive/8d5d3e8237e0947adb9ba1ffc8281f4ad7cb2a59.tar.gz \
+    && ls -l && tar -xzf 8d5d3e8237e0947adb9ba1ffc8281f4ad7cb2a59.tar.gz \
+    && cd restinio-8d5d3e8237e0947adb9ba1ffc8281f4ad7cb2a59/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/crypto.h

@@ -72,7 +72,8 @@ struct OPENDHT_PUBLIC PublicKey
      * Takes ownership of an existing gnutls_pubkey.
      */
     PublicKey(gnutls_pubkey_t k) : pk(k) {}
-    PublicKey(const Blob& pk);
+    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; };
 
     ~PublicKey();

include/opendht/dht_proxy_client.h

@@ -285,9 +285,7 @@ private:
      */
     struct InfoState;
     void getProxyInfos();
-    void handleProxyStatus(const asio::error_code &ec, std::shared_ptr<InfoState> infoState);
-    void queryProxyInfo(std::shared_ptr<InfoState> infoState, const sa_family_t family,
-                        std::vector<asio::ip::tcp::endpoint>&& endpoints);
+    void queryProxyInfo(std::shared_ptr<InfoState> infoState, sa_family_t family, std::shared_ptr<http::Resolver> resolver);
     void onProxyInfos(const Json::Value& val, const sa_family_t family);
     SockAddr parsePublicAddress(const Json::Value& val);
 
@@ -310,9 +308,8 @@ private:
     void handleResubscribe(const asio::error_code& ec, const InfoHash& key,
                            const size_t token, std::shared_ptr<OperationState> opstate);
 
-    void doPut(const InfoHash&, Sp<Value>, DoneCallback, time_point created, bool permanent);
-    void handleRefreshPut(const asio::error_code& ec, const InfoHash& key, const Value::Id id,
-                          std::shared_ptr<std::atomic_bool> ok);
+    void doPut(const InfoHash&, Sp<Value>, DoneCallbackSimple, time_point created, bool permanent);
+    void handleRefreshPut(const asio::error_code& ec, InfoHash key, Value::Id id);
 
     /**
      * Initialize statusIpvX_
@@ -327,23 +324,12 @@ private:
      */
     void cancelAllOperations();
 
+    std::string proxyUrl_;
     dht::crypto::Identity clientIdentity_;
     std::shared_ptr<dht::crypto::Certificate> serverCertificate_;
-    std::pair<std::string, std::string> serverHostService_;
+    //std::pair<std::string, std::string> serverHostService_;
     std::string pushClientId_;
 
-    /*
-     * ASIO I/O Context for sockets in httpClient_
-     * Note: Each context is used in one thread only
-     */
-    asio::io_context httpContext_;
-    std::shared_ptr<http::Resolver> resolver_;
-    std::map<unsigned int /*id*/, std::shared_ptr<http::Request>> requests_;
-    /*
-     * Thread for executing the http io_context.run() blocking call
-     */
-    std::thread httpClientThread_;
-
     mutable std::mutex lockCurrentProxyInfos_;
     NodeStatus statusIpv4_ {NodeStatus::Disconnected};
     NodeStatus statusIpv6_ {NodeStatus::Disconnected};
@@ -357,24 +343,34 @@ private:
     // registred types
     TypeStore types;
 
+    /*
+     * ASIO I/O Context for sockets in httpClient_
+     * Note: Each context is used in one thread only
+     */
+    asio::io_context httpContext_;
+    std::shared_ptr<http::Resolver> resolver_;
+    std::map<unsigned int /*id*/, std::shared_ptr<http::Request>> requests_;
+    /*
+     * Thread for executing the http io_context.run() blocking call
+     */
+    std::thread httpClientThread_;
+
     /**
      * Store listen requests.
      */
     struct ProxySearch;
 
+    mutable std::mutex searchLock_;
     size_t listenerToken_ {0};
     std::map<InfoHash, ProxySearch> searches_;
-    mutable std::mutex searchLock_;
 
     /**
      * Callbacks should be executed in the main thread.
      */
-    std::vector<std::function<void()>> callbacks_;
     std::mutex lockCallbacks_;
+    std::vector<std::function<void()>> callbacks_;
 
     Sp<InfoState> infoState_;
-    Sp<asio::steady_timer> statusTimer_;
-    mutable std::mutex statusLock_;
 
     /**
      * Retrieve if we can connect to the proxy (update statusIpvX_)
@@ -409,7 +405,11 @@ private:
 
     std::atomic_bool isDestroying_ {false};
 
+    Json::StreamWriterBuilder jsonBuilder_;
+    Json::CharReaderBuilder jsonReaderBuilder_;
     std::shared_ptr<dht::Logger> logger_;
+
+    std::shared_ptr<http::Request> buildRequest(const std::string& target = {});
 };
 
 }

include/opendht/dht_proxy_server.h

@@ -149,6 +149,8 @@ private:
 
     std::unique_ptr<RestRouter> createRestRouter();
 
+    void onConnectionClosed(restinio::connection_id_t);
+
     /**
      * Return the PublicKey id, the node id and node stats
      * Method: GET "/"
@@ -317,6 +319,7 @@ private:
 
     std::shared_ptr<DhtRunner> dht_;
     Json::StreamWriterBuilder jsonBuilder_;
+    Json::CharReaderBuilder jsonReaderBuilder_;
 
     // http server
     std::thread serverThread_;
@@ -335,10 +338,9 @@ private:
     std::unique_ptr<asio::steady_timer> printStatsTimer_;
 
     // Thread-safe access to listeners map.
-    std::shared_ptr<std::mutex> lockListener_;
+    std::mutex lockListener_;
     // Shared with connection listener.
-    std::shared_ptr<std::map<restinio::connection_id_t,
-                             http::ListenerSession>> listeners_;
+    std::map<restinio::connection_id_t, http::ListenerSession> listeners_;
     // Connection Listener observing conn state changes.
     std::shared_ptr<ConnectionListener> connListener_;
 

include/opendht/dhtrunner.h

@@ -230,29 +230,30 @@ public:
     }
     void put(const std::string& key, Value&& value, DoneCallbackSimple cb={}, time_point created=time_point::max(), bool permanent = false);
 
-    void cancelPut(const InfoHash& h, const Value::Id& id);
+    void cancelPut(const InfoHash& h, Value::Id id);
+    void cancelPut(const InfoHash& h, const std::shared_ptr<Value>& value);
 
-    void putSigned(InfoHash hash, std::shared_ptr<Value> value, DoneCallback cb={});
-    void putSigned(InfoHash hash, std::shared_ptr<Value> value, DoneCallbackSimple cb) {
-        putSigned(hash, value, bindDoneCb(cb));
+    void putSigned(InfoHash hash, std::shared_ptr<Value> value, DoneCallback cb={}, bool permanent = false);
+    void putSigned(InfoHash hash, std::shared_ptr<Value> value, DoneCallbackSimple cb, bool permanent = false) {
+        putSigned(hash, value, bindDoneCb(cb), permanent);
     }
 
-    void putSigned(InfoHash hash, Value&& value, DoneCallback cb={});
-    void putSigned(InfoHash hash, Value&& value, DoneCallbackSimple cb) {
-        putSigned(hash, std::forward<Value>(value), bindDoneCb(cb));
+    void putSigned(InfoHash hash, Value&& value, DoneCallback cb={}, bool permanent = false);
+    void putSigned(InfoHash hash, Value&& value, DoneCallbackSimple cb, bool permanent = false) {
+        putSigned(hash, std::forward<Value>(value), bindDoneCb(cb), permanent);
     }
-    void putSigned(const std::string& key, Value&& value, DoneCallbackSimple cb={});
+    void putSigned(const std::string& key, Value&& value, DoneCallbackSimple cb={}, bool permanent = false);
 
-    void putEncrypted(InfoHash hash, InfoHash to, std::shared_ptr<Value> value, DoneCallback cb={});
-    void putEncrypted(InfoHash hash, InfoHash to, std::shared_ptr<Value> value, DoneCallbackSimple cb) {
-        putEncrypted(hash, to, value, bindDoneCb(cb));
+    void putEncrypted(InfoHash hash, InfoHash to, std::shared_ptr<Value> value, DoneCallback cb={}, bool permanent = false);
+    void putEncrypted(InfoHash hash, InfoHash to, std::shared_ptr<Value> value, DoneCallbackSimple cb, bool permanent = false) {
+        putEncrypted(hash, to, value, bindDoneCb(cb), permanent);
     }
 
-    void putEncrypted(InfoHash hash, InfoHash to, Value&& value, DoneCallback cb={});
-    void putEncrypted(InfoHash hash, InfoHash to, Value&& value, DoneCallbackSimple cb) {
-        putEncrypted(hash, to, std::forward<Value>(value), bindDoneCb(cb));
+    void putEncrypted(InfoHash hash, InfoHash to, Value&& value, DoneCallback cb={}, bool permanent = false);
+    void putEncrypted(InfoHash hash, InfoHash to, Value&& value, DoneCallbackSimple cb, bool permanent = false) {
+        putEncrypted(hash, to, std::forward<Value>(value), bindDoneCb(cb), permanent);
     }
-    void putEncrypted(const std::string& key, InfoHash to, Value&& value, DoneCallback cb={});
+    void putEncrypted(const std::string& key, InfoHash to, Value&& value, DoneCallback cb={}, bool permanent = false);
 
     /**
      * Insert known nodes to the routing table, without necessarly ping them.

include/opendht/http.h

@@ -147,7 +147,7 @@ class OPENDHT_PUBLIC Resolver
 {
 public:
     using ResolverCb = std::function<void(const asio::error_code& ec,
-                                          std::vector<asio::ip::tcp::endpoint> endpoints)>;
+                                          const std::vector<asio::ip::tcp::endpoint>& endpoints)>;
 
     Resolver(asio::io_context& ctx, const std::string& url, std::shared_ptr<dht::Logger> logger = {});
     Resolver(asio::io_context& ctx, const std::string& host, const std::string& service,
@@ -156,14 +156,20 @@ public:
     // use already resolved endpoints with classes using this resolver
     Resolver(asio::io_context& ctx, std::vector<asio::ip::tcp::endpoint> endpoints,
              const bool ssl = false, std::shared_ptr<dht::Logger> logger = {});
+    Resolver(asio::io_context& ctx, const std::string& url, std::vector<asio::ip::tcp::endpoint> endpoints,
+            std::shared_ptr<dht::Logger> logger = {});
 
     ~Resolver();
 
     inline const Url& get_url() const {
         return url_;
-    };
+    }
 
-    void add_callback(ResolverCb cb);
+    void add_callback(ResolverCb cb, sa_family_t family = AF_UNSPEC);
+
+    std::shared_ptr<Logger> getLogger() const {
+        return logger_;
+    }
 
 private:
     void resolve(const std::string& host, const std::string& service);
@@ -212,7 +218,8 @@ public:
             const bool ssl = false, std::shared_ptr<dht::Logger> logger = {});
 
     // user defined resolver
-    Request(asio::io_context& ctx, std::shared_ptr<Resolver> resolver, std::shared_ptr<dht::Logger> logger = {});
+    Request(asio::io_context& ctx, std::shared_ptr<Resolver> resolver, sa_family_t family = AF_UNSPEC);
+    Request(asio::io_context& ctx, std::shared_ptr<Resolver> resolver, const std::string& target, sa_family_t family = AF_UNSPEC);
 
     // user defined resolved endpoints
     Request(asio::io_context& ctx, std::vector<asio::ip::tcp::endpoint>&& endpoints,
@@ -307,7 +314,7 @@ private:
     std::string body_;
 
     std::mutex cbs_mutex_;
-    std::unique_ptr<Callbacks> cbs_;
+    Callbacks cbs_;
     State state_;
 
     dht::crypto::Identity client_identity_;
@@ -316,8 +323,9 @@ private:
     std::string host_;
 
     unsigned int id_;
-    static unsigned int ids_;
     asio::io_context& ctx_;
+    sa_family_t family_ = AF_UNSPEC;
+    static unsigned int ids_;
     std::shared_ptr<Connection> conn_;
     std::shared_ptr<Resolver> resolver_;
 

include/opendht/thread_pool.h

@@ -43,9 +43,15 @@ public:
     template<class T>
     std::future<T> get(std::function<T()>&& cb) {
         auto ret = std::make_shared<std::promise<T>>();
-        run(std::bind([=](std::function<T()>& mcb) mutable {
-                ret->set_value(mcb());
-            }, std::move(cb)));
+        run([cb = std::move(cb), ret]() mutable {
+            try {
+                ret->set_value(cb());
+            } catch (...) {
+                try {
+                    ret->set_exception(std::current_exception());
+                } catch(...) {}
+            }
+        });
         return ret->get_future();
     }
     template<class T>

include/opendht/value.h

@@ -167,7 +167,7 @@ struct OPENDHT_PUBLIC Value
         static Filter chain(Filter&& f1, Filter&& f2) {
             if (not f1) return std::move(f2);
             if (not f2) return std::move(f1);
-            return [f1,f2](const Value& v) {
+            return [f1 = std::move(f1), f2 = std::move(f2)](const Value& v) {
                 return f1(v) and f2(v);
             };
         }
@@ -180,12 +180,12 @@ struct OPENDHT_PUBLIC Value
         }
         static Filter chainAll(std::vector<Filter>&& set) {
             if (set.empty()) return {};
-            return std::bind([](const Value& v, std::vector<Filter>& s) {
-                for (const auto& f : s)
+            return [set = std::move(set)](const Value& v) {
+                for (const auto& f : set)
                     if (f and not f(v))
                         return false;
                 return true;
-            }, std::placeholders::_1, std::move(set));
+            };
         }
         static Filter chain(std::initializer_list<Filter> l) {
             return chainAll(std::vector<Filter>(l.begin(), l.end()));

rust/examples/dhtnode.rs

@@ -20,6 +20,7 @@ extern crate opendht;
 use std::{ thread, time };
 
 use opendht::{ InfoHash, DhtRunner, DhtRunnerConfig, Value };
+// use opendht::crypto::*;
 
 fn main() {
     println!("{}", InfoHash::random());
@@ -28,15 +29,24 @@ fn main() {
     println!("{}", InfoHash::get("alice"));
     println!("{}", InfoHash::get("alice").is_zero());
 
+
     let mut dht = DhtRunner::new();
-    let config = DhtRunnerConfig::new();
+    let /*mut*/ config = DhtRunnerConfig::new();
+    //// If you want to inject a certificate, uncomment the following lines and previous mut.
+    //// Note: you can generate a certificate with
+    //// openssl req -x509 -newkey rsa:4096 -sha256 -days 3650 -nodes -keyout example.key -out example.crt -subj /CN=example.com
+    //let cert = DhtCertificate::import("example.crt").ok().expect("Invalid cert file");
+    //let pk = PrivateKey::import("example.key", "");
+    //config.set_identity(cert, pk);
     dht.run_config(1412, config);
     dht.bootstrap("bootstrap.jami.net", 4222);
+    println!("Current node id: {}", dht.node_id());
 
     let /* mut */ data = 42;
     let mut get_cb = |v: Box<Value>| {
         //data += 1;
         println!("GET: VALUE CB - data: {} - v: {}", data, v);
+        true
     };
     let mut done_cb = |ok: bool| {
         println!("GET: DONE CB - data: {} - ok: {}", data, ok);
@@ -47,12 +57,13 @@ fn main() {
     let mut put_done_cb = |ok: bool| {
         println!("PUT: DONE CB - data: {} - ok: {}", data, ok);
     };
-    dht.put(&InfoHash::get("bob"), Value::new("hi!"), &mut put_done_cb);
+    dht.put(&InfoHash::get("bob"), Value::new("hi!"), &mut put_done_cb, false);
 
 
     println!("Start listening /foo");
     let mut value_cb = |v, expired| {
         println!("LISTEN: DONE CB - data: {} - v: {} - expired: {}", data, v, expired);
+        true
     };
     let token = dht.listen(&InfoHash::get("foo"), &mut value_cb);
     let one_min = time::Duration::from_secs(10);

rust/src/crypto.rs

@@ -18,13 +18,17 @@
 
 #![allow(dead_code)]
 
-use crate::ffi::*;
+pub use crate::ffi::*;
 use std::ffi::CString;
+use std::io;
+use std::io::prelude::*;
+use std::fs::File;
+use std::ptr;
 
 impl PublicKey {
-    pub fn new() -> Box<PublicKey> {
+    pub fn new(data: &str) -> Box<PublicKey> {
         unsafe {
-            Box::from_raw(dht_publickey_new())
+            Box::from_raw(dht_publickey_import(data.as_ptr(), data.len()))
         }
     }
 
@@ -89,11 +93,17 @@ impl PrivateKey {
         }
     }
 
-    pub fn import(data: &str, password: &str) -> Box<PrivateKey> {
-        let password = CString::new(password).unwrap();
+    pub fn import(file: &str, password: &str) -> io::Result<Box<PrivateKey>> {
+        let mut f = File::open(file)?;
+        let mut buffer = Vec::new();
+        f.read_to_end(&mut buffer)?;
+        Ok(PrivateKey::from_bytes(&buffer, password))
+    }
+
+    pub fn from_bytes(buffer: &Vec<u8>, password: &str) -> Box<PrivateKey> {
         unsafe {
-            Box::from_raw(dht_privatekey_import(data.as_ptr(),
-                data.as_bytes().len(), password.as_ptr()))
+            Box::from_raw(dht_privatekey_import((&*buffer).as_ptr(), buffer.len(),
+                password.as_ptr() as *const i8))
         }
     }
 
@@ -111,4 +121,74 @@ impl Drop for PrivateKey {
             dht_privatekey_delete(&mut *self)
         }
     }
+}
+
+impl DhtCertificate {
+    pub fn import(file: &str) -> io::Result<Box<DhtCertificate>> {
+        let mut f = File::open(file)?;
+        let mut buffer = Vec::new();
+        f.read_to_end(&mut buffer)?;
+        Ok(DhtCertificate::from_bytes(&buffer))
+    }
+
+    pub fn from_bytes(buffer: &Vec<u8>) -> Box<DhtCertificate> {
+        unsafe {
+            Box::from_raw(dht_certificate_import((&*buffer).as_ptr(), buffer.len()))
+        }
+    }
+
+    pub fn from_slice(buffer: &str) -> Box<DhtCertificate> {
+        unsafe {
+            Box::from_raw(dht_certificate_import((&*buffer).as_ptr(), buffer.len()))
+        }
+    }
+
+    pub fn id(&self) -> InfoHash {
+        unsafe {
+            dht_certificate_get_id(&*self)
+        }
+    }
+
+    pub fn long_id(&self) -> PkId {
+        unsafe {
+            dht_certificate_get_long_id(&*self)
+        }
+    }
+
+    pub fn publickey(&self) -> Box<PublicKey> {
+        unsafe {
+            Box::from_raw(dht_certificate_get_publickey(&*self))
+        }
+    }
+}
+
+impl Drop for DhtCertificate {
+    fn drop(&mut self) {
+        unsafe {
+            dht_certificate_delete(&mut *self)
+        }
+    }
+}
+
+impl DhtIdentity {
+    pub fn new(common_name: &str) -> DhtIdentity {
+        unsafe {
+            DhtIdentity::generate(common_name, Box::from_raw(ptr::null_mut()))
+        }
+    }
+
+    pub fn generate(common_name: &str, ca: Box<DhtIdentity>) -> DhtIdentity {
+        let common_name = CString::new(common_name).unwrap();
+        unsafe {
+            dht_identity_generate(common_name.as_ptr(), &*ca)
+        }
+    }
+}
+
+impl Drop for DhtIdentity {
+    fn drop(&mut self) {
+        unsafe {
+            dht_identity_delete(&mut *self)
+        }
+    }
 }

rust/src/dhtrunner.rs

@@ -39,7 +39,7 @@ impl DhtRunnerConfig {
                     persist_path: ptr::null(),
                 },
                 id: DhtIdentity {
-                    privkey: ptr::null_mut(),
+                    privatekey: ptr::null_mut(),
                     certificate: ptr::null_mut(),
                 },
             },
@@ -51,7 +51,7 @@ impl DhtRunnerConfig {
             peer_publish: false,
             server_ca: ptr::null_mut(),
             client_identity: DhtIdentity {
-                privkey: ptr::null_mut(),
+                privatekey: ptr::null_mut(),
                 certificate: ptr::null_mut(),
             },
         });
@@ -73,6 +73,11 @@ impl DhtRunnerConfig {
         self.push_token = CString::new(push_token).unwrap().as_ptr();
     }
 
+    pub fn set_identity(&mut self, certificate: Box<DhtCertificate>, privatekey: Box<PrivateKey>) {
+        self.dht_config.id.privatekey = Box::into_raw(privatekey);
+        self.dht_config.id.certificate = Box::into_raw(certificate);
+    }
+
 }
 
 impl DhtNodeConfig
@@ -84,13 +89,13 @@ impl DhtNodeConfig
 
 struct GetHandler<'a>
 {
-    get_cb: &'a mut(dyn FnMut(Box<Value>)),
+    get_cb: &'a mut(dyn FnMut(Box<Value>) -> bool),
     done_cb: &'a mut(dyn FnMut(bool))
 }
 
 impl<'a> GetHandler<'a>
 {
-    fn get_cb(&mut self, v: Box<Value>) {
+    fn get_cb(&mut self, v: Box<Value>) -> bool{
         (self.get_cb)(v)
     }
 
@@ -99,9 +104,9 @@ impl<'a> GetHandler<'a>
     }
 }
 
-extern fn get_handler_cb(v: *mut Value, ptr: *mut c_void) {
+extern fn get_handler_cb(v: *mut Value, ptr: *mut c_void) -> bool {
     if ptr.is_null() {
-        return;
+        return true;
     }
     unsafe {
         let handler = ptr as *mut GetHandler;
@@ -137,17 +142,17 @@ extern fn put_handler_done(ok: bool, ptr: *mut c_void) {
 
 struct ListenHandler<'a>
 {
-    cb: &'a mut(dyn FnMut(Box<Value>, bool))
+    cb: &'a mut(dyn FnMut(Box<Value>, bool) -> bool)
 }
 
 impl<'a> ListenHandler<'a>
 {
-    fn cb(&mut self, v: Box<Value>, expired: bool) {
+    fn cb(&mut self, v: Box<Value>, expired: bool) -> bool {
         (self.cb)(v, expired)
     }
 }
 
-extern fn listen_handler(v: *mut Value, expired: bool, ptr: *mut c_void) {
+extern fn listen_handler(v: *mut Value, expired: bool, ptr: *mut c_void) -> bool {
     unsafe {
         let handler = ptr as *mut ListenHandler;
         (*handler).cb((*v).boxed(), expired)
@@ -200,7 +205,7 @@ impl DhtRunner {
     }
 
     pub fn get<'a>(&mut self, h: &InfoHash,
-                get_cb: &'a mut(dyn FnMut(Box<Value>)),
+                get_cb: &'a mut(dyn FnMut(Box<Value>) -> bool),
                 done_cb: &'a mut(dyn FnMut(bool))) {
         let handler = Box::new(GetHandler {
             get_cb,
@@ -213,24 +218,35 @@ impl DhtRunner {
     }
 
     pub fn put<'a>(&mut self, h: &InfoHash, v: Box<Value>,
-                done_cb: &'a mut(dyn FnMut(bool))) {
+                   done_cb: &'a mut(dyn FnMut(bool)), permanent: bool) {
         let handler = Box::new(PutHandler {
             done_cb,
         });
         let handler = Box::into_raw(handler) as *mut c_void;
         unsafe {
-            dht_runner_put(&mut *self, h, &*v, put_handler_done, handler)
+            dht_runner_put(&mut *self, h, &*v, put_handler_done, handler, permanent)
         }
     }
 
-    pub fn permanent_put<'a>(&mut self, h: &InfoHash, v: Box<Value>,
-                done_cb: &'a mut(dyn FnMut(bool))) {
+    pub fn put_signed<'a>(&mut self, h: &InfoHash, v: Box<Value>,
+                                                 done_cb: &'a mut(dyn FnMut(bool)), permanent: bool) {
         let handler = Box::new(PutHandler {
             done_cb,
         });
         let handler = Box::into_raw(handler) as *mut c_void;
         unsafe {
-            dht_runner_put_permanent(&mut *self, h, &*v, put_handler_done, handler)
+            dht_runner_put_signed(&mut *self, h, &*v, put_handler_done, handler, permanent)
+        }
+    }
+
+    pub fn put_encrypted<'a>(&mut self, h: &InfoHash, to: &InfoHash, v: Box<Value>,
+                             done_cb: &'a mut(dyn FnMut(bool)), permanent: bool) {
+        let handler = Box::new(PutHandler {
+            done_cb,
+        });
+        let handler = Box::into_raw(handler) as *mut c_void;
+        unsafe {
+            dht_runner_put_encrypted(&mut *self, h, to, &*v, put_handler_done, handler, permanent)
         }
     }
 
@@ -241,7 +257,7 @@ impl DhtRunner {
     }
 
     pub fn listen<'a>(&mut self, h: &InfoHash,
-                cb: &'a mut(dyn FnMut(Box<Value>, bool))) -> Box<OpToken> {
+                cb: &'a mut(dyn FnMut(Box<Value>, bool) -> bool)) -> Box<OpToken> {
         let handler = Box::new(ListenHandler {
             cb,
         });

rust/src/ffi.rs

@@ -44,6 +44,7 @@ pub struct Blob
 }
 
 #[repr(C)]
+#[derive(PartialEq)]
 pub struct InfoHash
 {
     pub d: [u8; HASH_LEN],
@@ -98,7 +99,7 @@ pub struct DhtCertificate
 #[repr(C)]
 pub struct DhtIdentity
 {
-    pub privkey: *mut PrivateKey,
+    pub privatekey: *mut PrivateKey,
     pub certificate: *mut DhtCertificate,
 }
 
@@ -127,11 +128,14 @@ pub struct DhtRunnerConfig
 #[link(name = "opendht-c")]
 extern {
     // dht::Value
-    pub fn dht_value_get_id(data: *const Value) -> u64;
-    pub fn dht_value_get_data(data: *const Value) -> DataView;
     pub fn dht_value_new(data: *const u8, size: size_t) -> *mut Value;
     pub fn dht_value_ref(data: *const Value) -> *mut Value;
     pub fn dht_value_unref(data: *mut Value);
+    pub fn dht_value_get_data(data: *const Value) -> DataView;
+    pub fn dht_value_get_id(data: *const Value) -> u64;
+    pub fn dht_value_get_owner(data: *const Value) -> *mut PublicKey;
+    pub fn dht_value_get_recipient(data: *const Value) -> InfoHash;
+    pub fn dht_value_get_user_type(data: *const Value) -> *const c_char;
 
     // dht::Blob
     pub fn dht_blob_get_data(data: *const Blob) -> DataView;
@@ -141,13 +145,14 @@ extern {
     pub fn dht_infohash_print(h: *const InfoHash) -> *const c_char;
     pub fn dht_infohash_random(h: *mut InfoHash);
     pub fn dht_infohash_get(h: *mut InfoHash, dat: *mut u8, dat_size: size_t);
+    pub fn dht_infohash_from_hex(h: *mut InfoHash, dat: *const c_char);
     pub fn dht_infohash_is_zero(j: *const InfoHash) -> bool;
 
     // dht::PkId
     pub fn dht_pkid_print(h: *const PkId) -> *const c_char;
 
     // dht::crypto::PublicKey
-    pub fn dht_publickey_new() -> *mut PublicKey;
+    pub fn dht_publickey_import(dat: *const u8, dat_size: size_t) -> *mut PublicKey;
     pub fn dht_publickey_delete(pk: *mut PublicKey);
     pub fn dht_publickey_unpack(pk: *mut PublicKey, dat: *const u8, dat_size: size_t) -> c_int;
     pub fn dht_publickey_pack(pk: *mut PublicKey, out: *const c_char, out_size: size_t) -> c_int;
@@ -162,6 +167,16 @@ extern {
     pub fn dht_privatekey_get_publickey(pk: *const PrivateKey) -> *mut PublicKey;
     pub fn dht_privatekey_delete(pk: *mut PrivateKey);
 
+    // dht::crypto::Certificate
+    pub fn dht_certificate_import(dat: *const u8, dat_size: size_t) -> *mut DhtCertificate;
+    pub fn dht_certificate_get_id(cert: *const DhtCertificate) -> InfoHash;
+    pub fn dht_certificate_get_long_id(cert: *const DhtCertificate) -> PkId;
+    pub fn dht_certificate_get_publickey(cert: *const DhtCertificate) -> *mut PublicKey;
+    pub fn dht_certificate_delete(cert: *mut DhtCertificate);
+
+    pub fn dht_identity_generate(common_name: *const c_char, ca: *const DhtIdentity) -> DhtIdentity;
+    pub fn dht_identity_delete(ca: *mut DhtIdentity);
+
     // dht::OpToken
     pub fn dht_op_token_delete(token: *mut OpToken);
 
@@ -175,18 +190,28 @@ extern {
     pub fn dht_runner_run_config(dht: *mut DhtRunner, port: in_port_t, config: *const DhtRunnerConfig);
     pub fn dht_runner_bootstrap(dht: *mut DhtRunner, host: *const c_char, service: *const c_char);
     pub fn dht_runner_get(dht: *mut DhtRunner, h: *const InfoHash,
-                      get_cb: extern fn(*mut Value, *mut c_void),
-                      done_cb: extern fn(bool, *mut c_void),
-                      cb_user_data: *mut c_void);
+                          get_cb: extern fn(*mut Value, *mut c_void) -> bool,
+                          done_cb: extern fn(bool, *mut c_void),
+                          cb_user_data: *mut c_void);
     pub fn dht_runner_put(dht: *mut DhtRunner, h: *const InfoHash, v: *const Value,
-                      done_cb: extern fn(bool, *mut c_void),
-                      cb_user_data: *mut c_void);
+                          done_cb: extern fn(bool, *mut c_void),
+                          cb_user_data: *mut c_void,
+                          permanent: bool);
+    pub fn dht_runner_put_signed(dht: *mut DhtRunner, h: *const InfoHash, v: *const Value,
+                                 done_cb: extern fn(bool, *mut c_void),
+                                 cb_user_data: *mut c_void,
+                                 permanent: bool);
+    pub fn dht_runner_put_encrypted(dht: *mut DhtRunner, h: *const InfoHash,
+                                    to: *const InfoHash, v: *const Value,
+                                    done_cb: extern fn(bool, *mut c_void),
+                                    cb_user_data: *mut c_void,
+                                    permanent: bool);
     pub fn dht_runner_put_permanent(dht: *mut DhtRunner, h: *const InfoHash, v: *const Value,
                       done_cb: extern fn(bool, *mut c_void),
                       cb_user_data: *mut c_void);
     pub fn dht_runner_cancel_put(dht: *mut DhtRunner, h: *const InfoHash, vid: u64);
     pub fn dht_runner_listen(dht: *mut DhtRunner, h: *const InfoHash,
-                      cb: extern fn(*mut Value, bool, *mut c_void),
+                      cb: extern fn(*mut Value, bool, *mut c_void) -> bool,
                       done_cb: extern fn(*mut c_void),
                       cb_user_data: *mut c_void) -> *mut OpToken;
     pub fn dht_runner_cancel_listen(dht: *mut DhtRunner, h: *const InfoHash,

rust/src/infohash.rs

@@ -47,6 +47,23 @@ impl InfoHash {
         h
     }
 
+    pub fn from_bytes(data: &Vec<u8>) -> InfoHash {
+        let mut h = InfoHash::new();
+        unsafe {
+            dht_infohash_get(&mut h, data.as_ptr() as *mut u8, data.len());
+        }
+        h
+    }
+
+    pub fn from_hex(data: &str) -> InfoHash {
+        let mut h = InfoHash::new();
+        unsafe {
+            let c_str = CString::new(data).unwrap();
+            dht_infohash_from_hex(&mut h, c_str.as_ptr());
+        }
+        h
+    }
+
     pub fn is_zero(&self) -> bool {
         unsafe {
             dht_infohash_is_zero(self)

rust/src/value.rs

@@ -15,9 +15,11 @@
  *  You should have received a copy of the GNU General Public License
  *  along with this program. If not, see <https://www.gnu.org/licenses/>.
  */
+#![allow(dead_code)]
 
 use crate::ffi::*;
 use std::fmt;
+use std::ffi::{ CString, IntoStringError };
 use std::str;
 use std::slice;
 
@@ -62,6 +64,28 @@ impl Value {
             Box::from_raw(dht_value_ref(self))
         }
     }
+
+    pub fn recipient(&self) -> InfoHash {
+        unsafe {
+            dht_value_get_recipient(self)
+        }
+    }
+
+    pub fn owner(&self) -> Option<Box<PublicKey>> {
+        unsafe {
+            let ptr = dht_value_get_owner(self);
+            if ptr.is_null() {
+                return None;
+            }
+            Some(Box::from_raw(ptr))
+        }
+    }
+
+    pub fn user_type(&self) -> Result<String, IntoStringError> {
+        unsafe {
+            CString::from_raw(dht_value_get_user_type(self) as *mut _).into_string()
+        }
+    }
 }
 
 impl Drop for Value {

src/crypto.cpp

@@ -52,6 +52,9 @@ static std::uniform_int_distribution<uint8_t> rand_byte;
 #define GNUTLS_PKCS_PBES2_AES_256 GNUTLS_PKCS_USE_PBES2_AES_256
 #endif
 
+#define DHT_AES_LEGACY_ENCRYPT 1
+#define DHT_AES_LEGACY_DECRYPT 1
+
 namespace dht {
 namespace crypto {
 
@@ -105,7 +108,9 @@ Blob aesEncrypt(const uint8_t* data, size_t data_length, const Blob& key)
     struct gcm_aes_ctx aes;
     gcm_aes_set_key(&aes, key.size(), key.data());
     gcm_aes_set_iv(&aes, GCM_IV_SIZE, ret.data());
+#if DHT_AES_LEGACY_ENCRYPT
     gcm_aes_update(&aes, data_length, data);
+#endif
 
     gcm_aes_encrypt(&aes, data_length, ret.data() + GCM_IV_SIZE, data);
     gcm_aes_digest(&aes, GCM_DIGEST_SIZE, ret.data() + GCM_IV_SIZE + data_length);
@@ -137,21 +142,26 @@ Blob aesDecrypt(const Blob& data, const Blob& key)
 
     size_t data_sz = data.size() - GCM_IV_SIZE - GCM_DIGEST_SIZE;
     Blob ret(data_sz);
-    //gcm_aes_update(&aes, data_sz, data.data() + GCM_IV_SIZE);
     gcm_aes_decrypt(&aes, data_sz, ret.data(), data.data() + GCM_IV_SIZE);
-    //gcm_aes_digest(aes, GCM_DIGEST_SIZE, digest.data());
+    gcm_aes_digest(&aes, GCM_DIGEST_SIZE, digest.data());
 
-    // TODO compute the proper digest directly from the decryption pass
-    Blob ret_tmp(data_sz);
-    struct gcm_aes_ctx aes_d;
-    gcm_aes_set_key(&aes_d, key.size(), key.data());
-    gcm_aes_set_iv(&aes_d, GCM_IV_SIZE, data.data());
-    gcm_aes_update(&aes_d, ret.size(), ret.data());
-    gcm_aes_encrypt(&aes_d, ret.size(), ret_tmp.data(), ret.data());
-    gcm_aes_digest(&aes_d, GCM_DIGEST_SIZE, digest.data());
+    if (not std::equal(digest.begin(), digest.end(), data.end() - GCM_DIGEST_SIZE)) {
+#if DHT_AES_LEGACY_DECRYPT
+        //gcm_aes_decrypt(&aes, data_sz, ret.data(), data.data() + GCM_IV_SIZE);
+        Blob ret_tmp(data_sz);
+        struct gcm_aes_ctx aes_d;
+        gcm_aes_set_key(&aes_d, key.size(), key.data());
+        gcm_aes_set_iv(&aes_d, GCM_IV_SIZE, data.data());
+        gcm_aes_update(&aes_d, ret.size(), ret.data());
+        gcm_aes_encrypt(&aes_d, ret.size(), ret_tmp.data(), ret.data());
+        gcm_aes_digest(&aes_d, GCM_DIGEST_SIZE, digest.data());
 
-    if (not std::equal(digest.begin(), digest.end(), data.end() - GCM_DIGEST_SIZE))
+        if (not std::equal(digest.begin(), digest.end(), data.end() - GCM_DIGEST_SIZE))
+            throw DecryptError("Can't decrypt data");
+#else
         throw DecryptError("Can't decrypt data");
+#endif
+    }
 
     return ret;
 }
@@ -364,9 +374,9 @@ PublicKey::PublicKey()
         throw CryptoException(std::string("Can't initialize public key: ") + gnutls_strerror(err));
 }
 
-PublicKey::PublicKey(const Blob& dat) : PublicKey()
+PublicKey::PublicKey(const uint8_t* dat, size_t dat_size) : PublicKey()
 {
-    unpack(dat.data(), dat.size());
+    unpack(dat, dat_size);
 }
 
 PublicKey::~PublicKey()

src/dht.cpp

@@ -921,8 +921,7 @@ Dht::put(const InfoHash& id, Sp<Value> val, DoneCallback callback, time_point cr
     }
     if (val->id == Value::INVALID_ID) {
         crypto::random_device rdev;
-        std::uniform_int_distribution<Value::Id> rand_id {};
-        val->id = rand_id(rdev);
+        val->id = std::uniform_int_distribution<Value::Id>{1}(rdev);
     }
     scheduler.syncTime();
     const auto& now = scheduler.time();

src/dht_proxy_client.cpp

@@ -80,9 +80,10 @@ DhtProxyClient::DhtProxyClient(
         pushClientId_(pushClientId), loopSignal_(signal), logger_(logger)
 {
     // build http client
-    serverHostService_ = splitPort(serverHost);
-    serverHostService_.second = serverHostService_.second.empty() ? "80" :
-                                serverHostService_.second;
+    proxyUrl_ = serverHost;
+
+    jsonBuilder_["commentStyle"] = "None";
+    jsonBuilder_["indentation"] = "";
     if (logger_){
         if (serverCertificate_)
             logger_->d("[proxy:client] using ca certificate for ssl:\n%s",
@@ -91,8 +92,6 @@ DhtProxyClient::DhtProxyClient(
             logger_->d("[proxy:client] using client certificate for ssl:\n%s",
                        clientIdentity_.second->toString(false/*chain*/).c_str());
     }
-    // resolve once
-    resolver_ = std::make_shared<http::Resolver>(httpContext_, serverHost, logger_);
     // run http client
     httpClientThread_ = std::thread([this](){
         try {
@@ -109,18 +108,18 @@ DhtProxyClient::DhtProxyClient(
                 logger_->e("[proxy:client] run error: %s", ex.what());
         }
     });
-    if (!serverHostService_.first.empty())
+    if (!proxyUrl_.empty())
         startProxy();
 }
 
 void
 DhtProxyClient::startProxy()
 {
-    if (serverHostService_.first.empty())
+    if (proxyUrl_.empty())
         return;
 
     if (logger_)
-        logger_->d("[proxy:client] start proxy with %s", serverHostService_.first.c_str());
+        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_->async_wait(std::bind(&DhtProxyClient::handleProxyConfirm, this, std::placeholders::_1));
@@ -140,7 +139,7 @@ DhtProxyClient::handleProxyConfirm(const asio::error_code &ec)
             logger_->e("[proxy:client] confirm error: %s", ec.message().c_str());
         return;
     }
-    if (serverHostService_.first.empty())
+    if (proxyUrl_.empty())
         return;
     getConnectivityStatus();
 }
@@ -158,8 +157,6 @@ DhtProxyClient::stop()
     cancelAllListeners();
     if (infoState_)
         infoState_->cancel = true;
-    if (statusTimer_)
-        statusTimer_->cancel();
     if (httpClientThread_.joinable())
         httpClientThread_.join();
 }
@@ -263,9 +260,6 @@ DhtProxyClient::periodic(const uint8_t*, size_t, SockAddr)
 
 void
 DhtProxyClient::setHeaderFields(http::Request& request){
-    request.set_header_field(restinio::http_field_t::host,
-                             serverHostService_.first + ":" + serverHostService_.second);
-    request.set_header_field(restinio::http_field_t::user_agent, "RESTinio client");
     request.set_header_field(restinio::http_field_t::accept, "*/*");
     request.set_header_field(restinio::http_field_t::content_type, "application/json");
 }
@@ -276,27 +270,18 @@ DhtProxyClient::get(const InfoHash& key, GetCallback cb, DoneCallback donecb, Va
     if (logger_)
         logger_->d("[proxy:client] [get] [search %s]", key.to_c_str());
 
-    auto request = std::make_shared<http::Request>(httpContext_, resolver_, logger_);
+    auto request = buildRequest("/" + key.toString());
     auto reqid = request->id();
     try {
         request->set_connection_type(restinio::http_connection_header_t::keep_alive);
-        request->set_target("/" + key.toString());
         request->set_method(restinio::http_method_get());
         setHeaderFields(*request);
 
         auto opstate = std::make_shared<OperationState>();
         Value::Filter filter = w.empty() ? f : f.chain(w.getFilter());
 
-        request->add_on_status_callback([this, key, opstate](unsigned int status_code){
-            if (status_code != 200){
-                if (logger_)
-                    logger_->e("[proxy:client] [get %s] status error: %i", key.to_c_str(), status_code);
-                opstate->ok.store(false);
-            }
-        });
         request->add_on_body_callback([this, key, opstate, filter, cb](const char* at, size_t length){
             try {
-                Json::CharReaderBuilder rbuilder;
                 auto body = std::string(at, length);
                 // one value per body line
                 std::string data_line;
@@ -305,7 +290,7 @@ DhtProxyClient::get(const InfoHash& key, GetCallback cb, DoneCallback donecb, Va
                     std::string err;
                     Json::Value json;
                     auto* char_data = static_cast<const char*>(&data_line[0]);
-                    auto reader = std::unique_ptr<Json::CharReader>(rbuilder.newCharReader());
+                    auto reader = std::unique_ptr<Json::CharReader>(jsonReaderBuilder_.newCharReader());
                     if (!reader->parse(char_data, char_data + data_line.size(), &json, &err)){
                         opstate->ok.store(false);
                         return;
@@ -336,6 +321,8 @@ DhtProxyClient::get(const InfoHash& key, GetCallback cb, DoneCallback donecb, Va
                     if (logger_)
                         logger_->e("[proxy:client] [get %s] failed with code=%i", key.to_c_str(), response.status_code);
                     opstate->ok.store(false);
+                    if (response.status_code == 0)
+                        opFailed();
                 }
                 if (donecb){
                     {
@@ -350,10 +337,6 @@ DhtProxyClient::get(const InfoHash& key, GetCallback cb, DoneCallback donecb, Va
                 requests_.erase(reqid);
             }
         });
-        if (serverCertificate_)
-            request->set_certificate_authority(serverCertificate_);
-        if (clientIdentity_.first and clientIdentity_.second)
-            request->set_identity(clientIdentity_);
         request->send();
         requests_[reqid] = request;
     }
@@ -377,35 +360,43 @@ DhtProxyClient::put(const InfoHash& key, Sp<Value> val, DoneCallback cb, time_po
 
     if (val->id == Value::INVALID_ID) {
         crypto::random_device rdev;
-        std::uniform_int_distribution<Value::Id> rand_id {};
-        val->id = rand_id(rdev);
+        val->id = std::uniform_int_distribution<Value::Id>{1}(rdev);
     }
+    std::shared_ptr<std::atomic_bool> ok;
     if (permanent) {
         std::lock_guard<std::mutex> lock(searchLock_);
         auto id = val->id;
         auto& search = searches_[key];
         auto refreshPutTimer = std::make_unique<asio::steady_timer>(httpContext_, std::chrono::steady_clock::now() +
                                                                     proxy::OP_TIMEOUT - proxy::OP_MARGIN);
-        auto ok = std::make_shared<std::atomic_bool>();
-        ok->store(true);
-        refreshPutTimer->async_wait(std::bind(&DhtProxyClient::handleRefreshPut, this,
-                                              std::placeholders::_1, key, id, ok));
+        refreshPutTimer->async_wait(std::bind(&DhtProxyClient::handleRefreshPut, this, std::placeholders::_1, key, id));
         search.puts.erase(id);
+        ok = std::make_shared<std::atomic_bool>(true);
         search.puts.emplace(std::piecewise_construct,
             std::forward_as_tuple(id),
             std::forward_as_tuple(val, std::move(refreshPutTimer), ok));
     }
-    doPut(key, val, std::move(cb), created, permanent);
+    doPut(key, val, [this, cb, ok](bool result){
+        if (ok)
+            *ok = result;
+        if (cb) {
+            std::lock_guard<std::mutex> lock(lockCallbacks_);
+            callbacks_.emplace_back([cb, result](){
+                cb(result, {});
+            });
+        }
+        loopSignal_();
+    }, created, permanent);
 }
+
 void
-DhtProxyClient::handleRefreshPut(const asio::error_code &ec, const InfoHash& key, const Value::Id id,
-                                 std::shared_ptr<std::atomic_bool> ok)
+DhtProxyClient::handleRefreshPut(const asio::error_code &ec, InfoHash key, Value::Id id)
 {
     if (ec == asio::error::operation_aborted)
         return;
     else if (ec){
         if (logger_)
-            logger_->e("[proxy:client] [listener] [refresh %s] %s", key.toString().c_str(), ec.message().c_str());
+            logger_->e("[proxy:client] [put] [refresh %s] %s", key.toString().c_str(), ec.message().c_str());
         return;
     }
     if (logger_)
@@ -415,26 +406,41 @@ DhtProxyClient::handleRefreshPut(const asio::error_code &ec, const InfoHash& key
     if (search != searches_.end()) {
         auto p = search->second.puts.find(id);
         if (p != search->second.puts.end()){
-            doPut(key, p->second.value, [ok](bool result, const std::vector<std::shared_ptr<dht::Node>>&){
+            doPut(key, p->second.value, [ok = p->second.ok](bool result){
                 *ok = result;
             }, time_point::max(), true);
-            p->second.refreshPutTimer->expires_at(std::chrono::steady_clock::now() +
-                                               proxy::OP_TIMEOUT - proxy::OP_MARGIN);
-            p->second.refreshPutTimer->async_wait(std::bind(&DhtProxyClient::handleRefreshPut, this, std::placeholders::_1, key, id, ok));
+            p->second.refreshPutTimer->expires_at(std::chrono::steady_clock::now() + proxy::OP_TIMEOUT - proxy::OP_MARGIN);
+            p->second.refreshPutTimer->async_wait(std::bind(&DhtProxyClient::handleRefreshPut, this, std::placeholders::_1, key, id));
         }
     }
 }
 
+std::shared_ptr<http::Request>
+DhtProxyClient::buildRequest(const std::string& target)
+{
+    auto resolver = resolver_;
+    if (not resolver)
+        resolver = std::make_shared<http::Resolver>(httpContext_, proxyUrl_, logger_);
+    auto request = target.empty() 
+        ? std::make_shared<http::Request>(httpContext_, resolver)
+        : std::make_shared<http::Request>(httpContext_, resolver, target);
+    if (serverCertificate_)
+        request->set_certificate_authority(serverCertificate_);
+    if (clientIdentity_.first and clientIdentity_.second)
+        request->set_identity(clientIdentity_);
+    request->set_header_field(restinio::http_field_t::user_agent, "RESTinio client");
+    return request;
+}
+
 void
-DhtProxyClient::doPut(const InfoHash& key, Sp<Value> val, DoneCallback cb, time_point /*created*/, bool permanent)
+DhtProxyClient::doPut(const InfoHash& key, Sp<Value> val, DoneCallbackSimple cb, time_point /*created*/, bool permanent)
 {
     if (logger_)
         logger_->d("[proxy:client] [put] [search %s] executing for %s", key.to_c_str(), val->toString().c_str());
 
-    auto request = std::make_shared<http::Request>(httpContext_, resolver_, logger_);
+    auto request = buildRequest("/" + key.toString());
     auto reqid = request->id();
     try {
-        request->set_target("/" + key.toString());
         request->set_method(restinio::http_method_post());
         setHeaderFields(*request);
 
@@ -452,45 +458,22 @@ DhtProxyClient::doPut(const InfoHash& key, Sp<Value> val, DoneCallback cb, time_
 #endif
             }
         }
-        Json::StreamWriterBuilder wbuilder;
-        wbuilder["commentStyle"] = "None";
-        wbuilder["indentation"] = "";
-        auto body = Json::writeString(wbuilder, json);
-        request->set_body(body);
-
-        auto ok = std::make_shared<std::atomic_bool>();
-        ok->store(true);
-
-        request->add_on_status_callback([this, key, ok](unsigned int status_code){
-            if (status_code != 200){
-                if (logger_)
-                    logger_->e("[proxy:client] [put %s] status error: %i", key.to_c_str(), status_code);
-                ok->store(false);
-            }
-            return 0;
-        });
-        request->add_on_state_change_callback([this, reqid, ok, cb, key]
+        request->set_body(Json::writeString(jsonBuilder_, json));
+        request->add_on_state_change_callback([this, reqid, cb]
                                               (http::Request::State state, const http::Response& response){
             if (state == http::Request::State::DONE){
-                if (response.status_code != 200)
+                bool ok = response.status_code == 200;
+                if (not ok) {
                     if (logger_)
                         logger_->e("[proxy:client] [status] failed with code=%i", response.status_code);
-                if (cb){
-                    {
-                        std::lock_guard<std::mutex> lock(lockCallbacks_);
-                        callbacks_.emplace_back([cb, ok](){
-                            cb(ok->load(), {});
-                        });
-                    }
-                    loopSignal_();
+                    if (response.status_code == 0)
+                        opFailed();
                 }
+                if (cb)
+                    cb(ok);
                 requests_.erase(reqid);
             }
         });
-        if (serverCertificate_)
-            request->set_certificate_authority(serverCertificate_);
-        if (clientIdentity_.first and clientIdentity_.second)
-            request->set_identity(clientIdentity_);
         request->send();
         requests_[reqid] = request;
     }
@@ -556,127 +539,58 @@ DhtProxyClient::getProxyInfos()
 {
     if (logger_)
         logger_->d("[proxy:client] [info] requesting proxy server node information");
-    std::lock_guard<std::mutex> l(statusLock_);
-
     auto infoState = std::make_shared<InfoState>();
-    if (infoState_)
-        infoState_->cancel = true;
-    infoState_ = infoState;
     {
         std::lock_guard<std::mutex> l(lockCurrentProxyInfos_);
+        if (infoState_)
+            infoState_->cancel = true;
+        infoState_ = infoState;
         if (statusIpv4_ == NodeStatus::Disconnected)
             statusIpv4_ = NodeStatus::Connecting;
         if (statusIpv6_ == NodeStatus::Disconnected)
             statusIpv6_ = NodeStatus::Connecting;
     }
-    // Try to contact the proxy and set the status to connected when done.
-    // will change the connectivity status
-    if (!statusTimer_)
-        statusTimer_ = std::make_shared<asio::steady_timer>(httpContext_);
-    statusTimer_->expires_at(std::chrono::steady_clock::now());
-    statusTimer_->async_wait(std::bind(&DhtProxyClient::handleProxyStatus, this,
-                             std::placeholders::_1, infoState));
-}
-
-void
-DhtProxyClient::handleProxyStatus(const asio::error_code& ec, std::shared_ptr<InfoState> infoState)
-{
-    if (ec == asio::error::operation_aborted)
-        return;
-    else if (ec){
-        if (logger_)
-            logger_->e("[proxy:client] [status] handling error: %s", ec.message().c_str());
-        return;
-    }
     if (logger_)
         logger_->d("[proxy:client] [status] sending request");
 
-    resolver_->add_callback([this, infoState](const asio::error_code& ec,
-                                              std::vector<asio::ip::tcp::endpoint> endpoints)
-    {
-        if (ec){
-            if (logger_)
-                logger_->e("[proxy:client] [status] resolve error: %s", ec.message().c_str());
-            return;
-        }
-        std::vector<asio::ip::tcp::endpoint> endpointsIpv4;
-        std::vector<asio::ip::tcp::endpoint> endpointsIpv6;
-
-        for (auto& endpoint: endpoints){
-            try {
-                if (endpoint.address().is_v6())
-                    endpointsIpv6.push_back(endpoint);
-                if (endpoint.address().is_v4())
-                    endpointsIpv4.push_back(endpoint);
-            } catch (const std::exception&){};
-        }
-        // avoiding nested add_callback call
-        if (!endpointsIpv4.empty())
-            queryProxyInfo(infoState, AF_INET, std::move(endpointsIpv4));
-        if (!endpointsIpv6.empty())
-            queryProxyInfo(infoState, AF_INET6, std::move(endpointsIpv6));
-    });
+    auto resolver = std::make_shared<http::Resolver>(httpContext_, proxyUrl_, logger_);
+    queryProxyInfo(infoState, AF_INET, resolver);
+    queryProxyInfo(infoState, AF_INET6, resolver);
+    resolver_ = resolver;
 }
 
 void
-DhtProxyClient::queryProxyInfo(std::shared_ptr<InfoState> infoState, const sa_family_t family,
-                               std::vector<asio::ip::tcp::endpoint>&& endpoints)
+DhtProxyClient::queryProxyInfo(std::shared_ptr<InfoState> infoState, sa_family_t family, std::shared_ptr<http::Resolver> resolver)
 {
     if (logger_)
         logger_->d("[proxy:client] [status] query ipv%i info", family == AF_INET ? 4 : 6);
-    auto request = std::make_shared<http::Request>(httpContext_, std::move(endpoints),
-        resolver_->get_url().protocol == "https" ? /*ssl*/ true : false, logger_);
+    auto request = std::make_shared<http::Request>(httpContext_, resolver, family);
     auto reqid = request->id();
     try {
-        request->set_connection_type(restinio::http_connection_header_t::keep_alive);
-        request->set_target("/");
-        request->set_method(restinio::http_method_get());
+        request->set_method(restinio::http_method_get()); 
         setHeaderFields(*request);
-
-        auto ok = std::make_shared<std::atomic_bool>();
-        ok->store(true);
-
-        request->add_on_status_callback([this, ok, family](unsigned int status_code){
-            if (status_code != 200){
-                if (logger_)
-                    logger_->e("[proxy:client] [status] ipv%i status code error: %i", family, status_code);
-                ok->store(false);
-            }
-        });
-        request->add_on_body_callback([this, ok, infoState, family](const char* at, size_t length){
-            try{
-                std::string err;
-                Json::Value proxyInfos;
-                Json::CharReaderBuilder rbuilder;
-                auto body = std::string(at, length);
-                auto* char_data = static_cast<const char*>(&body[0]);
-                auto reader = std::unique_ptr<Json::CharReader>(rbuilder.newCharReader());
-                if (!reader->parse(char_data, char_data + body.size(), &proxyInfos, &err)){
-                    ok->store(false);
-                    return;
-                }
-                if (not infoState->cancel)
-                    onProxyInfos(proxyInfos, family);
-            }
-            catch (const std::exception& e) {
-                if (logger_)
-                    logger_->e("[proxy:client] [status] body error: %s", e.what());
-                ok->store(false);
-            }
-        });
         request->add_on_state_change_callback([this, reqid, family, infoState]
                                               (http::Request::State state, const http::Response& response){
-            if (state == http::Request::State::DONE){
-                if (response.status_code != 200)
+            if (state == http::Request::State::DONE) {
+                if (infoState->cancel.load())
+                    return;
+                if (response.status_code != 200) {
                     if (logger_)
                         logger_->e("[proxy:client] [status] ipv%i failed with code=%i",
                                     family == AF_INET ? 4 : 6, response.status_code);
-                if (infoState->cancel.load()){
                     // pass along the failures
-                    if (family == AF_INET and infoState->ipv4 == 0)
-                        onProxyInfos(Json::Value{}, AF_INET);
-                    if (family == AF_INET6 and infoState->ipv6 == 0)
-                    onProxyInfos(Json::Value{}, AF_INET6);
+                    if ((family == AF_INET and infoState->ipv4 == 0) or (family == AF_INET6 and infoState->ipv6 == 0))
+                        onProxyInfos(Json::Value{}, family);
+                } else {
+                    std::string err;
+                    Json::Value proxyInfos;
+                    auto reader = std::unique_ptr<Json::CharReader>(jsonReaderBuilder_.newCharReader());
+                    if (!reader->parse(response.body.data(), response.body.data() + response.body.size(), &proxyInfos, &err)){
+                        onProxyInfos(Json::Value{}, family);
+                        return;
+                    }
+                    if (not infoState->cancel)
+                        onProxyInfos(proxyInfos, family);
                 }
                 requests_.erase(reqid);
             }
@@ -685,12 +599,8 @@ DhtProxyClient::queryProxyInfo(std::shared_ptr<InfoState> infoState, const sa_fa
         if (infoState->cancel.load())
             return;
 
-        if (serverCertificate_)
-            request->set_certificate_authority(serverCertificate_);
-        if (clientIdentity_.first and clientIdentity_.second)
-            request->set_identity(clientIdentity_);
-        request->send();
         requests_[reqid] = request;
+        request->send();
     }
     catch (const std::exception &e){
         if (logger_)
@@ -933,43 +843,29 @@ DhtProxyClient::handleExpireListener(const asio::error_code &ec, const InfoHash&
 
         if (not deviceKey_.empty()) {
             // UNSUBSCRIBE
-            auto request = std::make_shared<http::Request>(httpContext_, resolver_, logger_);
+            auto request = buildRequest("/" + key.toString());
             auto reqid = request->id();
             try {
-                request->set_target("/" + key.toString());
                 request->set_method(restinio::http_method_unsubscribe());
                 setHeaderFields(*request);
 
                 Json::Value body;
                 body["key"] = deviceKey_;
                 body["client_id"] = pushClientId_;
-                Json::StreamWriterBuilder wbuilder;
-                wbuilder["commentStyle"] = "None";
-                wbuilder["indentation"] = "";
-                auto content = Json::writeString(wbuilder, body) + "\n";
-                std::replace(content.begin(), content.end(), '\n', ' ');
-                request->set_body(content);
-
-                request->add_on_status_callback([this, key](unsigned int status_code){
-                    if (status_code != 200)
-                        if (logger_)
-                            logger_->e("[proxy:client] [unsubscribe %s] failed with code=%i",
-                                       key.to_c_str(), status_code);
-                });
+                request->set_body(Json::writeString(jsonBuilder_, body));
                 request->add_on_state_change_callback([this, reqid, key]
                                                       (http::Request::State state, const http::Response& response){
                     if (state == http::Request::State::DONE){
-                        if (response.status_code != 200)
+                        if (response.status_code != 200) {
                             if (logger_)
                                 logger_->e("[proxy:client] [unsubscribe %s] failed with code=%i",
                                            key.to_c_str(), response.status_code);
+                            if (response.status_code == 0)
+                                opFailed();
+                        }
                         requests_.erase(reqid);
                     }
                 });
-                if (serverCertificate_)
-                    request->set_certificate_authority(serverCertificate_);
-                if (clientIdentity_.first and clientIdentity_.second)
-                    request->set_identity(clientIdentity_);
                 request->send();
                 requests_[reqid] = request;
             }
@@ -1003,7 +899,9 @@ DhtProxyClient::sendListen(const restinio::http_request_header_t header,
                            const Sp<OperationState>& opstate,
                            Listener& listener, ListenMethod method)
 {
-    auto request = std::make_shared<http::Request>(httpContext_, resolver_, logger_);
+    if (logger_)
+        logger_->e("[proxy:client] [listen] sendListen: %d", (int)method);
+    auto request = buildRequest();
     listener.request = request;
     auto reqid = request->id();
     try {
@@ -1011,25 +909,16 @@ DhtProxyClient::sendListen(const restinio::http_request_header_t header,
         setHeaderFields(*request);
         if (method == ListenMethod::LISTEN)
             request->set_connection_type(restinio::http_connection_header_t::keep_alive);
-        std::string body;
 #ifdef OPENDHT_PUSH_NOTIFICATIONS
+        std::string body;
         if (method != ListenMethod::LISTEN)
             body = fillBody(method == ListenMethod::RESUBSCRIBE);
-#endif
         request->set_body(body);
+#endif
 
-        request->add_on_status_callback([this, reqid, opstate](unsigned int status_code){
-            if (status_code != 200){
-                if (logger_)
-                    logger_->e("[proxy:client] [listen] send request #%i status error: %i", reqid, status_code);
-                opstate->ok.store(false);
-            }
-        });
         request->add_on_body_callback([this, reqid, opstate, cb](const char* at, size_t length){
             try {
-                Json::CharReaderBuilder rbuilder;
                 auto body = std::string(at, length);
-                std::vector<dht::Sp<dht::Value>> values;
                 // one value per body line
                 std::string data_line;
                 std::stringstream body_stream(body);
@@ -1037,7 +926,7 @@ DhtProxyClient::sendListen(const restinio::http_request_header_t header,
                     std::string err;
                     Json::Value json;
                     auto* char_data = static_cast<const char*>(&data_line[0]);
-                    auto reader = std::unique_ptr<Json::CharReader>(rbuilder.newCharReader());
+                    auto reader = std::unique_ptr<Json::CharReader>(jsonReaderBuilder_.newCharReader());
                     if (!reader->parse(char_data, char_data + data_line.size(), &json, &err)){
                         opstate->ok.store(false);
                         return;
@@ -1065,20 +954,20 @@ DhtProxyClient::sendListen(const restinio::http_request_header_t header,
                 opstate->ok.store(false);
             }
         });
-        request->add_on_state_change_callback([this, reqid]
+        request->add_on_state_change_callback([this, opstate, reqid]
                                               (http::Request::State state, const http::Response& response){
             if (state == http::Request::State::DONE){
-                if (response.status_code != 200)
+                if (response.status_code != 200) {
                     if (logger_)
                         logger_->e("[proxy:client] [listen] send request #%i failed with code=%i",
                                    reqid, response.status_code);
+                    opstate->ok.store(false);
+                    if (response.status_code == 0)
+                        opFailed();
+                }
                 requests_.erase(reqid);
             }
         });
-        if (serverCertificate_)
-            request->set_certificate_authority(serverCertificate_);
-        if (clientIdentity_.first and clientIdentity_.second)
-            request->set_identity(clientIdentity_);
         request->send();
         requests_[reqid] = request;
     }
@@ -1092,6 +981,8 @@ DhtProxyClient::sendListen(const restinio::http_request_header_t header,
 void
 DhtProxyClient::opFailed()
 {
+    if (isDestroying_)
+        return;
     if (logger_)
         logger_->e("[proxy:client] proxy request failed");
     {
@@ -1125,17 +1016,15 @@ DhtProxyClient::restartListeners()
         auto key = search.first;
         for (auto& put : search.second.puts) {
             if (!*put.second.ok) {
-                auto ok = put.second.ok;
-                doPut(key, put.second.value, [ok](bool result, const std::vector<std::shared_ptr<dht::Node>>&){
+                doPut(key, put.second.value, [ok = put.second.ok](bool result){
                     *ok = result;
                 }, time_point::max(), true);
                 if (!put.second.refreshPutTimer){
                     put.second.refreshPutTimer = std::make_unique<asio::steady_timer>(httpContext_);
                 }
-                put.second.refreshPutTimer->expires_at(std::chrono::steady_clock::now() +
-                                                       proxy::OP_TIMEOUT - proxy::OP_MARGIN);
+                put.second.refreshPutTimer->expires_at(std::chrono::steady_clock::now() + proxy::OP_TIMEOUT - proxy::OP_MARGIN);
                 put.second.refreshPutTimer->async_wait(std::bind(&DhtProxyClient::handleRefreshPut, this,
-                                                       std::placeholders::_1, key, put.first, put.second.ok));
+                                                        std::placeholders::_1, key, put.first));
             }
         }
     }
@@ -1207,7 +1096,7 @@ DhtProxyClient::pushNotificationReceived(const std::map<std::string, std::string
                     put.refreshPutTimer = std::make_unique<asio::steady_timer>(httpContext_, std::chrono::steady_clock::now());
                 else
                     put.refreshPutTimer->expires_at(std::chrono::steady_clock::now());
-                put.refreshPutTimer->async_wait(std::bind(&DhtProxyClient::handleRefreshPut, this, std::placeholders::_1, key, vid, put.ok));
+                put.refreshPutTimer->async_wait(std::bind(&DhtProxyClient::handleRefreshPut, this, std::placeholders::_1, key, vid));
             } else {
                 // Refresh listen
                 for (auto& list : search.listeners)
@@ -1333,10 +1222,7 @@ DhtProxyClient::fillBody(bool resubscribe)
         // This is the first listen, we want to retrieve previous values.
         body["refresh"] = true;
     }
-    Json::StreamWriterBuilder wbuilder;
-    wbuilder["commentStyle"] = "None";
-    wbuilder["indentation"] = "";
-    auto content = Json::writeString(wbuilder, body) + "\n";
+    auto content = Json::writeString(jsonBuilder_, body) + "\n";
     std::replace(content.begin(), content.end(), '\n', ' ');
     return content;
 }

src/dht_proxy_server.cpp

@@ -110,10 +110,7 @@ class DhtProxyServer::ConnectionListener
 {
 public:
     ConnectionListener() {};
-    ConnectionListener(std::shared_ptr<dht::DhtRunner> dht,
-        std::shared_ptr<std::map<restinio::connection_id_t, http::ListenerSession>> listeners,
-        std::shared_ptr<std::mutex> lock, std::shared_ptr<dht::Logger> logger):
-        dht_(dht), lock_(lock), listeners_(listeners), logger_(logger) {};
+    ConnectionListener(std::function<void(restinio::connection_id_t)> onClosed) : onClosed_(std::move(onClosed)) {};
     ~ConnectionListener() {};
 
     /**
@@ -123,54 +120,28 @@ public:
     void state_changed(const restinio::connection_state::notice_t& notice) noexcept;
 
 private:
-    std::string to_str( restinio::connection_state::cause_t cause ) noexcept;
-
-    std::shared_ptr<dht::DhtRunner> dht_;
-    std::shared_ptr<std::mutex> lock_;
-    std::shared_ptr<std::map<restinio::connection_id_t, http::ListenerSession>> listeners_;
-
-    std::shared_ptr<dht::Logger> logger_;
+    std::function<void(restinio::connection_id_t)> onClosed_;
 };
 
 void
 DhtProxyServer::ConnectionListener::state_changed(const restinio::connection_state::notice_t& notice) noexcept
 {
-    std::lock_guard<std::mutex> lock(*lock_);
-    auto id = notice.connection_id();
-    auto cause = to_str(notice.cause());
-
-    if (listeners_->find(id) != listeners_->end()){
-        if (notice.cause() == restinio::connection_state::cause_t::closed){
-            if (logger_)
-                logger_->d("[proxy:server] [connection:%li] cancelling listener", id);
-            dht_->cancelListen(listeners_->at(id).hash,
-                               std::move(listeners_->at(id).token));
-            listeners_->erase(id);
-            if (logger_)
-                logger_->d("[proxy:server] %li listeners are connected", listeners_->size());
-        }
+    if (restinio::holds_alternative<restinio::connection_state::closed_t>(notice.cause())) {
+        onClosed_(notice.connection_id());
     }
 }
 
-std::string
-DhtProxyServer::ConnectionListener::to_str(restinio::connection_state::cause_t cause) noexcept
+void
+DhtProxyServer::onConnectionClosed(restinio::connection_id_t id)
 {
-    std::string result;
-    switch(cause)
-    {
-    case restinio::connection_state::cause_t::accepted:
-        result = "accepted";
-        break;
-    case restinio::connection_state::cause_t::closed:
-        result = "closed";
-        break;
-    case restinio::connection_state::cause_t::upgraded_to_websocket:
-        result = "upgraded";
-        break;
-    default:
-        result = "unknown";
+    std::lock_guard<std::mutex> lock(lockListener_);
+    auto it = listeners_.find(id);
+    if (it != listeners_.end()) {
+        dht_->cancelListen(it->second.hash, std::move(it->second.token));
+        listeners_.erase(it);
+        if (logger_)
+            logger_->d("[proxy:server] [connection:%li] listener cancelled, %li still connected", id, listeners_.size());
     }
-    return result;
 }
 
 struct DhtProxyServer::RestRouterTraitsTls : public restinio::default_tls_traits_t
@@ -199,9 +170,8 @@ DhtProxyServer::DhtProxyServer(
     std::shared_ptr<DhtRunner> dht, in_port_t port, const std::string& pushServer,
     std::shared_ptr<dht::Logger> logger
 )
-    :   dht_(dht), logger_(logger), lockListener_(std::make_shared<std::mutex>()),
-        listeners_(std::make_shared<std::map<restinio::connection_id_t, http::ListenerSession>>()),
-        connListener_(std::make_shared<ConnectionListener>(dht, listeners_, lockListener_, logger)),
+    :   dht_(dht), logger_(logger),
+        connListener_(std::make_shared<ConnectionListener>(std::bind(&DhtProxyServer::onConnectionClosed, this, std::placeholders::_1))),
         pushServer_(pushServer)
 {
     if (not dht_)
@@ -365,7 +335,7 @@ DhtProxyServer::updateStats() const
     stats_.pushListenersCount = pushListeners_.size();
 #endif
     stats_.putCount = puts_.size();
-    stats_.listenCount = listeners_->size();
+    stats_.listenCount = listeners_.size();
     stats_.nodeInfo = nodeInfo_;
 }
 
@@ -585,9 +555,9 @@ DhtProxyServer::listen(restinio::request_handle_t request,
         this->initHttpResponse(request->create_response<ResponseByParts>()));
     response->flush();
     try {
-        std::lock_guard<std::mutex> lock(*lockListener_);
+        std::lock_guard<std::mutex> lock(lockListener_);
         // save the listener to handle a disconnect
-        auto &session = (*listeners_)[request->connection_id()];
+        auto &session = listeners_[request->connection_id()];
         session.hash = infoHash;
         session.response = response;
         session.token = dht_->listen(infoHash, [this, response]
@@ -633,9 +603,8 @@ DhtProxyServer::subscribe(restinio::request_handle_t request,
     try {
         std::string err;
         Json::Value root;
-        Json::CharReaderBuilder rbuilder;
         auto* char_data = reinterpret_cast<const char*>(request->body().data());
-        auto reader = std::unique_ptr<Json::CharReader>(rbuilder.newCharReader());
+        auto reader = std::unique_ptr<Json::CharReader>(jsonReaderBuilder_.newCharReader());
         if (!reader->parse(char_data, char_data + request->body().size(), &root, &err)){
             auto response = this->initHttpResponse(
                 request->create_response(restinio::status_bad_request()));
@@ -775,9 +744,8 @@ DhtProxyServer::unsubscribe(restinio::request_handle_t request,
     try {
         std::string err;
         Json::Value root;
-        Json::CharReaderBuilder rbuilder;
         auto* char_data = reinterpret_cast<const char*>(request->body().data());
-        auto reader = std::unique_ptr<Json::CharReader>(rbuilder.newCharReader());
+        auto reader = std::unique_ptr<Json::CharReader>(jsonReaderBuilder_.newCharReader());
 
         if (!reader->parse(char_data, char_data + request->body().size(), &root, &err)){
             auto response = this->initHttpResponse(
@@ -831,7 +799,7 @@ DhtProxyServer::handleCancelPushListen(const asio::error_code &ec, const std::st
     if (logger_)
         logger_->d("[proxy:server] [listen:push %s] cancelled for %s",
                    key.toString().c_str(), clientId.c_str());
-    std::lock_guard<std::mutex> lock(*lockListener_);
+    std::lock_guard<std::mutex> lock(lockListener_);
 
     auto pushListener = pushListeners_.find(pushToken);
     if (pushListener == pushListeners_.end())
@@ -953,14 +921,12 @@ DhtProxyServer::put(restinio::request_handle_t request,
         infoHash = dht::InfoHash::get(params["hash"].to_string());
 
     if (!dht_){
-        auto response = this->initHttpResponse(
-            request->create_response(restinio::status_service_unavailable()));
+        auto response = initHttpResponse(request->create_response(restinio::status_service_unavailable()));
         response.set_body(RESP_MSG_SERVICE_UNAVAILABLE);
         return response.done();
     }
     else if (request->body().empty()){
-        auto response = this->initHttpResponse(
-            request->create_response(restinio::status_bad_request()));
+        auto response = initHttpResponse(request->create_response(restinio::status_bad_request()));
         response.set_body(RESP_MSG_MISSING_PARAMS);
         return response.done();
     }
@@ -968,9 +934,8 @@ DhtProxyServer::put(restinio::request_handle_t request,
     try {
         std::string err;
         Json::Value root;
-        Json::CharReaderBuilder rbuilder;
         auto* char_data = reinterpret_cast<const char*>(request->body().data());
-        auto reader = std::unique_ptr<Json::CharReader>(rbuilder.newCharReader());
+        auto reader = std::unique_ptr<Json::CharReader>(jsonReaderBuilder_.newCharReader());
 
         if (reader->parse(char_data, char_data + request->body().size(), &root, &err)){
             auto value = std::make_shared<dht::Value>(root);
@@ -1082,9 +1047,8 @@ RequestStatus DhtProxyServer::putSigned(restinio::request_handle_t request,
     try {
         std::string err;
         Json::Value root;
-        Json::CharReaderBuilder rbuilder;
         auto* char_data = reinterpret_cast<const char*>(request->body().data());
-        auto reader = std::unique_ptr<Json::CharReader>(rbuilder.newCharReader());
+        auto reader = std::unique_ptr<Json::CharReader>(jsonReaderBuilder_.newCharReader());
 
         if (reader->parse(char_data, char_data + request->body().size(), &root, &err)){
 
@@ -1145,9 +1109,8 @@ DhtProxyServer::putEncrypted(restinio::request_handle_t request,
     try {
         std::string err;
         Json::Value root;
-        Json::CharReaderBuilder rbuilder;
         auto* char_data = reinterpret_cast<const char*>(request->body().data());
-        auto reader = std::unique_ptr<Json::CharReader>(rbuilder.newCharReader());
+        auto reader = std::unique_ptr<Json::CharReader>(jsonReaderBuilder_.newCharReader());
 
         if (reader->parse(char_data, char_data + request->body().size(), &root, &err)){
             InfoHash to(root["to"].asString());

src/dhtrunner.cpp

@@ -708,7 +708,7 @@ DhtRunner::put(const std::string& key, Value&& value, DoneCallbackSimple cb, tim
 }
 
 void
-DhtRunner::cancelPut(const InfoHash& h , const Value::Id& id)
+DhtRunner::cancelPut(const InfoHash& h, Value::Id id)
 {
     {
         std::lock_guard<std::mutex> lck(storage_mtx);
@@ -720,51 +720,63 @@ DhtRunner::cancelPut(const InfoHash& h , const Value::Id& id)
 }
 
 void
-DhtRunner::putSigned(InfoHash hash, std::shared_ptr<Value> value, DoneCallback cb)
+DhtRunner::cancelPut(const InfoHash& h, const std::shared_ptr<Value>& value)
 {
     {
         std::lock_guard<std::mutex> lck(storage_mtx);
         pending_ops.emplace([=](SecureDht& dht) {
-            dht.putSigned(hash, value, cb);
+            dht.cancelPut(h, value->id);
         });
     }
     cv.notify_all();
 }
 
 void
-DhtRunner::putSigned(InfoHash hash, Value&& value, DoneCallback cb)
-{
-    putSigned(hash, std::make_shared<Value>(std::move(value)), std::move(cb));
-}
-
-void
-DhtRunner::putSigned(const std::string& key, Value&& value, DoneCallbackSimple cb)
-{
-    putSigned(InfoHash::get(key), std::forward<Value>(value), std::move(cb));
-}
-
-void
-DhtRunner::putEncrypted(InfoHash hash, InfoHash to, std::shared_ptr<Value> value, DoneCallback cb)
+DhtRunner::putSigned(InfoHash hash, std::shared_ptr<Value> value, DoneCallback cb, bool permanent)
 {
     {
         std::lock_guard<std::mutex> lck(storage_mtx);
         pending_ops.emplace([=](SecureDht& dht) {
-            dht.putEncrypted(hash, to, value, cb);
+            dht.putSigned(hash, value, cb, permanent);
         });
     }
     cv.notify_all();
 }
 
 void
-DhtRunner::putEncrypted(InfoHash hash, InfoHash to, Value&& value, DoneCallback cb)
+DhtRunner::putSigned(InfoHash hash, Value&& value, DoneCallback cb, bool permanent)
 {
-    putEncrypted(hash, to, std::make_shared<Value>(std::move(value)), std::move(cb));
+    putSigned(hash, std::make_shared<Value>(std::move(value)), std::move(cb), permanent);
 }
 
 void
-DhtRunner::putEncrypted(const std::string& key, InfoHash to, Value&& value, DoneCallback cb)
+DhtRunner::putSigned(const std::string& key, Value&& value, DoneCallbackSimple cb, bool permanent)
 {
-    putEncrypted(InfoHash::get(key), to, std::forward<Value>(value), std::move(cb));
+    putSigned(InfoHash::get(key), std::forward<Value>(value), std::move(cb), permanent);
+}
+
+void
+DhtRunner::putEncrypted(InfoHash hash, InfoHash to, std::shared_ptr<Value> value, DoneCallback cb, bool permanent)
+{
+    {
+        std::lock_guard<std::mutex> lck(storage_mtx);
+        pending_ops.emplace([=](SecureDht& dht) {
+            dht.putEncrypted(hash, to, value, cb, permanent);
+        });
+    }
+    cv.notify_all();
+}
+
+void
+DhtRunner::putEncrypted(InfoHash hash, InfoHash to, Value&& value, DoneCallback cb, bool permanent)
+{
+    putEncrypted(hash, to, std::make_shared<Value>(std::move(value)), std::move(cb), permanent);
+}
+
+void
+DhtRunner::putEncrypted(const std::string& key, InfoHash to, Value&& value, DoneCallback cb, bool permanent)
+{
+    putEncrypted(InfoHash::get(key), to, std::forward<Value>(value), std::move(cb), permanent);
 }
 
 void

src/http.cpp

@@ -142,9 +142,9 @@ Connection::close()
     if (!is_open())
         return;
     asio::error_code ec;
-    if (ssl_ctx_)
+    if (ssl_socket_)
         ssl_socket_->close(ec);
-    else
+    else if (socket_)
         socket_->close(ec);
     if (ec and logger_)
         logger_->e("[http:client]  [connection:%i] error closing: %s", id_, ec.message().c_str());
@@ -159,10 +159,12 @@ Connection::id()
 bool
 Connection::is_open()
 {
-    if (ssl_ctx_)
+    if (ssl_socket_)
         return ssl_socket_->is_open();
-    else
+    else if (socket_)
         return socket_->is_open();
+    else
+        return false;
 }
 
 bool
@@ -174,7 +176,7 @@ Connection::is_ssl()
 void
 Connection::set_ssl_verification(const asio::ip::tcp::endpoint& endpoint, const asio::ssl::verify_mode verify_mode)
 {
-    if (ssl_ctx_ and verify_mode != asio::ssl::verify_none){
+    if (ssl_socket_ and verify_mode != asio::ssl::verify_none){
         auto hostname = endpoint.address().to_string();
         ssl_socket_->asio_ssl_stream().set_verify_mode(verify_mode);
         ssl_socket_->asio_ssl_stream().set_verify_callback(
@@ -230,16 +232,18 @@ Connection::read_until(const char delim)
 void
 Connection::async_connect(std::vector<asio::ip::tcp::endpoint>&& endpoints, ConnectHandlerCb cb)
 {
-    if (ssl_ctx_)
+    if (ssl_socket_)
         asio::async_connect(ssl_socket_->lowest_layer(), std::move(endpoints), cb);
-    else
+    else if (socket_)
         asio::async_connect(*socket_, std::move(endpoints), cb);
+    else if (cb)
+        cb(asio::error::operation_aborted, {});
 }
 
 void
 Connection::async_handshake(HandlerCb cb)
 {
-    if (ssl_ctx_)
+    if (ssl_socket_)
         ssl_socket_->async_handshake(asio::ssl::stream<asio::ip::tcp::socket>::client,
                                     [this, cb](const asio::error_code& ec)
         {
@@ -258,6 +262,8 @@ Connection::async_handshake(HandlerCb cb)
         });
     else if (socket_)
         cb(asio::error::no_protocol_option);
+    else if (cb)
+        cb(asio::error::operation_aborted);
 }
 
 void
@@ -265,10 +271,12 @@ Connection::async_write(BytesHandlerCb cb)
 {
     if (!is_open())
         return;
-    if (ssl_ctx_)
+    if (ssl_socket_)
         asio::async_write(*ssl_socket_, write_buf_, cb);
-    else
+    else if (socket_)
         asio::async_write(*socket_, write_buf_, cb);
+    else if (cb)
+        cb(asio::error::operation_aborted, 0);
 }
 
 void
@@ -276,10 +284,12 @@ Connection::async_read_until(const char* delim, BytesHandlerCb cb)
 {
     if (!is_open())
         return;
-    if (ssl_ctx_)
+    if (ssl_socket_)
         asio::async_read_until(*ssl_socket_, read_buf_, delim, cb);
-    else
+    else if (socket_)
         asio::async_read_until(*socket_, read_buf_, delim, cb);
+    else if (cb)
+        cb(asio::error::operation_aborted, 0);
 }
 
 void
@@ -289,8 +299,10 @@ Connection::async_read(const size_t bytes, BytesHandlerCb cb)
         return;
     if (ssl_socket_)
         asio::async_read(*ssl_socket_, read_buf_, asio::transfer_exactly(bytes), cb);
-    else
+    else if (socket_)
         asio::async_read(*socket_, read_buf_, asio::transfer_exactly(bytes), cb);
+    else if (cb)
+        cb(asio::error::operation_aborted, 0);
 }
 
 void
@@ -299,6 +311,8 @@ Connection::timeout(const std::chrono::seconds timeout, HandlerCb cb)
     if (!is_open()){
         if (logger_)
             logger_->e("[http:client]  [connection:%i] closed, can't timeout", id_);
+        if (cb)
+            cb(asio::error::operation_aborted);
         return;
     }
     if (!timeout_timer_)
@@ -359,14 +373,35 @@ Resolver::~Resolver()
     *destroyed_ = true;
 }
 
+inline
+std::vector<asio::ip::tcp::endpoint>
+filter(const std::vector<asio::ip::tcp::endpoint>& epts, sa_family_t family)
+{
+    if (family == AF_UNSPEC)
+        return epts;
+    std::vector<asio::ip::tcp::endpoint> ret;
+    for (const auto& ep : epts) {
+        if (family == AF_INET && ep.address().is_v4())
+            ret.emplace_back(ep);
+        else if (family == AF_INET6 && ep.address().is_v6())
+            ret.emplace_back(ep);
+    }
+    return ret;
+}
+
 void
-Resolver::add_callback(ResolverCb cb)
+Resolver::add_callback(ResolverCb cb, sa_family_t family)
 {
     std::lock_guard<std::mutex> lock(mutex_);
     if (!completed_)
-        cbs_.emplace(std::move(cb));
+        cbs_.emplace(family == AF_UNSPEC ? std::move(cb) : [cb, family](const asio::error_code& ec, const std::vector<asio::ip::tcp::endpoint>& endpoints){
+            if (ec)
+                cb(ec, endpoints);
+            else
+                cb(ec, filter(endpoints, family));
+        });
     else
-        cb(ec_, endpoints_);
+        cb(ec_, family == AF_UNSPEC ? endpoints_ : filter(endpoints_, family));
 }
 
 void
@@ -383,14 +418,6 @@ Resolver::resolve(const std::string& host, const std::string& service)
             if (ec)
                 logger_->e("[http:client] [resolver] error for %s:%s: %s",
                            host.c_str(), service.c_str(), ec.message().c_str());
-            else {
-                for (auto it = endpoints.begin(); it != endpoints.end(); ++it){
-                    asio::ip::tcp::endpoint endpoint = *it;
-                    logger_->d("[http:client] [resolver] %s:%s endpoint (ipv%i): %s",
-                        host.c_str(), service.c_str(), endpoint.address().is_v6() ? 6 : 4,
-                        endpoint.address().to_string().c_str());
-                }
-            }
         }
         decltype(cbs_) cbs;
         {
@@ -416,7 +443,7 @@ unsigned int Request::ids_ = 1;
 
 Request::Request(asio::io_context& ctx, const std::string& url, const Json::Value& json, OnJsonCb jsoncb,
                  std::shared_ptr<dht::Logger> logger)
-    : cbs_(std::make_unique<Callbacks>()), id_(Request::ids_++), ctx_(ctx),
+    : id_(Request::ids_++), ctx_(ctx),
       resolver_(std::make_shared<Resolver>(ctx, url, logger)), logger_(logger)
 {
     init_default_headers();
@@ -439,7 +466,7 @@ Request::Request(asio::io_context& ctx, const std::string& url, const Json::Valu
 }
 
 Request::Request(asio::io_context& ctx, const std::string& url, std::shared_ptr<dht::Logger> logger)
-    : cbs_(std::make_unique<Callbacks>()), id_(Request::ids_++), ctx_(ctx),
+    : id_(Request::ids_++), ctx_(ctx),
       resolver_(std::make_shared<Resolver>(ctx, url, logger)), logger_(logger)
 {
     init_default_headers();
@@ -447,26 +474,33 @@ Request::Request(asio::io_context& ctx, const std::string& url, std::shared_ptr<
 
 Request::Request(asio::io_context& ctx, const std::string& host, const std::string& service,
                  const bool ssl, std::shared_ptr<dht::Logger> logger)
-    : cbs_(std::make_unique<Callbacks>()), id_(Request::ids_++), ctx_(ctx),
+    : id_(Request::ids_++), ctx_(ctx),
       resolver_(std::make_shared<Resolver>(ctx, host, service, ssl, logger)), logger_(logger)
 {
     init_default_headers();
 }
 
-Request::Request(asio::io_context& ctx, std::shared_ptr<Resolver> resolver, std::shared_ptr<dht::Logger> logger)
-    : cbs_(std::make_unique<Callbacks>()), id_(Request::ids_++), ctx_(ctx), resolver_(resolver), logger_(logger)
+Request::Request(asio::io_context& ctx, std::shared_ptr<Resolver> resolver, sa_family_t family)
+    : id_(Request::ids_++), ctx_(ctx), family_(family), resolver_(resolver), logger_(resolver->getLogger())
 {
     init_default_headers();
 }
 
 Request::Request(asio::io_context& ctx, std::vector<asio::ip::tcp::endpoint>&& endpoints, const bool ssl,
                  std::shared_ptr<dht::Logger> logger)
-    : cbs_(std::make_unique<Callbacks>()), id_(Request::ids_++), ctx_(ctx),
+    : id_(Request::ids_++), ctx_(ctx),
       resolver_(std::make_shared<Resolver>(ctx, std::move(endpoints), ssl, logger)), logger_(logger)
 {
     init_default_headers();
 }
 
+Request::Request(asio::io_context& ctx, std::shared_ptr<Resolver> resolver, const std::string& target, sa_family_t family)
+    : id_(Request::ids_++), ctx_(ctx), family_(family), resolver_(resolver), logger_(resolver->getLogger())
+{
+    set_header_field(restinio::http_field_t::host, get_url().host + ":" + get_url().service);
+    set_target(target);
+}
+
 Request::~Request()
 {
 }
@@ -474,14 +508,16 @@ Request::~Request()
 void
 Request::init_default_headers()
 {
-    set_header_field(restinio::http_field_t::host, get_url().host + ":" + get_url().service);
-    set_target(resolver_->get_url().target);
+    const auto& url = resolver_->get_url();
+    set_header_field(restinio::http_field_t::host, url.host + ":" + url.service);
+    set_target(url.target);
 }
 
 void
 Request::cancel()
 {
-    conn_->close();
+    if (auto c = conn_)
+        c->close();
 }
 
 unsigned int
@@ -613,29 +649,29 @@ void
 Request::add_on_status_callback(OnStatusCb cb)
 {
     std::lock_guard<std::mutex> lock(cbs_mutex_);
-    cbs_->on_status = std::move(cb);
+    cbs_.on_status = std::move(cb);
 }
 
 void
 Request::add_on_body_callback(OnDataCb cb)
 {
     std::lock_guard<std::mutex> lock(cbs_mutex_);
-    cbs_->on_body = std::move(cb);
+    cbs_.on_body = std::move(cb);
 }
 
 void
 Request::add_on_state_change_callback(OnStateChangeCb cb)
 {
     std::lock_guard<std::mutex> lock(cbs_mutex_);
-    cbs_->on_state_change = std::move(cb);
+    cbs_.on_state_change = std::move(cb);
 }
 
 void
 Request::notify_state_change(const State state)
 {
     state_ = state;
-    if (cbs_ and cbs_->on_state_change)
-        cbs_->on_state_change(state, response_);
+    if (cbs_.on_state_change)
+        cbs_.on_state_change(state, response_);
 }
 
 void
@@ -644,7 +680,7 @@ Request::init_parser()
     if (!parser_)
         parser_ = std::make_unique<http_parser>();
     http_parser_init(parser_.get(), HTTP_RESPONSE);
-    parser_->data = static_cast<void*>(cbs_.get());
+    parser_->data = static_cast<void*>(&cbs_);
 
     if (!parser_s_)
         parser_s_ = std::make_unique<http_parser_settings>();
@@ -652,36 +688,30 @@ Request::init_parser()
     {
         // user registered callbacks wrappers to store its data in the response
         std::lock_guard<std::mutex> lock(cbs_mutex_);
-        auto on_status_cb = cbs_->on_status;
-        cbs_->on_status = [this, on_status_cb](unsigned int status_code){
+        cbs_.on_status = [this, statusCb = std::move(cbs_.on_status)](unsigned int status_code){
             response_.status_code = status_code;
-            if (on_status_cb)
-                on_status_cb(status_code);
+            if (statusCb)
+                statusCb(status_code);
         };
-        auto header_field = std::make_shared<std::string>("");
-        auto on_header_field_cb = cbs_->on_header_field;
-        cbs_->on_header_field = [header_field, on_header_field_cb](const char* at, size_t length) {
-            header_field->erase();
-            auto field = std::string(at, length);
-            header_field->append(field);
-            if (on_header_field_cb)
-                on_header_field_cb(at, length);
+        auto header_field = std::make_shared<std::string>();
+        cbs_.on_header_field = [header_field, headerFieldCb = std::move(cbs_.on_header_field)](const char* at, size_t length) {
+            *header_field = std::string(at, length);
+            if (headerFieldCb)
+                headerFieldCb(at, length);
         };
-        auto on_header_value_cb = cbs_->on_header_value;
-        cbs_->on_header_value = [this, header_field, on_header_value_cb](const char* at, size_t length) {
+        cbs_.on_header_value = [this, header_field, headerValueCb = std::move(cbs_.on_header_value)](const char* at, size_t length) {
             response_.headers[*header_field] = std::string(at, length);
-            if (on_header_value_cb)
-                on_header_value_cb(at, length);
+            if (headerValueCb)
+                headerValueCb(at, length);
         };
-        cbs_->on_headers_complete = [this](){
+        cbs_.on_headers_complete = [this](){
             notify_state_change(State::HEADER_RECEIVED);
         };
-        auto on_body_cb = cbs_->on_body;
-        cbs_->on_body = [on_body_cb](const char* at, size_t length) {
-            if (on_body_cb)
-                on_body_cb(at, length);
+        cbs_.on_body = [bodyCb = std::move(cbs_.on_body)](const char* at, size_t length) {
+            if (bodyCb)
+                bodyCb(at, length);
         };
-        cbs_->on_message_complete = [this](){
+        cbs_.on_message_complete = [this](){
             if (logger_)
                 logger_->d("[http:client]  [request:%i] response: message complete", id_);
             message_complete_.store(true);
@@ -689,39 +719,27 @@ Request::init_parser()
     }
     // http_parser raw c callback (note: no context can be passed into them)
     parser_s_->on_status = [](http_parser* parser, const char* /*at*/, size_t /*length*/) -> int {
-        auto cbs = static_cast<Callbacks*>(parser->data);
-        if (cbs->on_status)
-            cbs->on_status(parser->status_code);
+        static_cast<Callbacks*>(parser->data)->on_status(parser->status_code);
         return 0;
     };
     parser_s_->on_header_field = [](http_parser* parser, const char* at, size_t length) -> int {
-        auto cbs = static_cast<Callbacks*>(parser->data);
-        if (cbs->on_header_field)
-            cbs->on_header_field(at, length);
+        static_cast<Callbacks*>(parser->data)->on_header_field(at, length);
         return 0;
     };
     parser_s_->on_header_value = [](http_parser* parser, const char* at, size_t length) -> int {
-        auto cbs = static_cast<Callbacks*>(parser->data);
-        if (cbs->on_header_value)
-            cbs->on_header_value(at, length);
+        static_cast<Callbacks*>(parser->data)->on_header_value(at, length);
         return 0;
     };
     parser_s_->on_body = [](http_parser* parser, const char* at, size_t length) -> int {
-        auto cbs = static_cast<Callbacks*>(parser->data);
-        if (cbs->on_body)
-            cbs->on_body(at, length);
+        static_cast<Callbacks*>(parser->data)->on_body(at, length);
         return 0;
     };
     parser_s_->on_headers_complete = [](http_parser* parser) -> int {
-        auto cbs = static_cast<Callbacks*>(parser->data);
-        if (cbs->on_headers_complete)
-            cbs->on_headers_complete();
+        static_cast<Callbacks*>(parser->data)->on_headers_complete();
         return 0;
     };
     parser_s_->on_message_complete = [](http_parser* parser) -> int {
-        auto cbs = static_cast<Callbacks*>(parser->data);
-        if (cbs->on_message_complete)
-            cbs->on_message_complete();
+        static_cast<Callbacks*>(parser->data)->on_message_complete();
         return 0;
     };
 }
@@ -738,7 +756,7 @@ Request::connect(std::vector<asio::ip::tcp::endpoint>&& endpoints, HandlerCb cb)
     }
     if (logger_){
         std::string eps = "";
-        for (auto& endpoint : endpoints)
+        for (const auto& endpoint : endpoints)
             eps.append(endpoint.address().to_string() + " ");
         logger_->d("[http:client]  [request:%i] connect begin: %s", id_, eps.c_str());
     }
@@ -813,7 +831,7 @@ Request::send()
         }
         else
             post();
-    });
+    }, family_);
 }
 
 void
@@ -842,14 +860,12 @@ Request::post()
 void
 Request::terminate(const asio::error_code& ec)
 {
-    if (finishing_.load())
+    if (finishing_.exchange(true))
         return;
 
     if (ec != asio::error::eof and ec != asio::error::operation_aborted and logger_)
         logger_->e("[http:client]  [request:%i] end with error: %s", id_, ec.message().c_str());
 
-    finishing_.store(true);
-
     // set response outcome, ignore end of file and abort
     if (!ec or ec == asio::error::eof or ec == asio::error::operation_aborted)
         response_.status_code = 200;

tests/cryptotester.cpp

@@ -34,16 +34,41 @@ CryptoTester::testSignatureEncryption() {
     auto key = dht::crypto::PrivateKey::generate();
     auto public_key = key.getPublicKey();
 
-    std::vector<uint8_t> data {5, 10};
-    std::vector<uint8_t> signature = key.sign(data);
+    std::vector<uint8_t> data1 {5, 10};
+    std::vector<uint8_t> data2(64 * 1024, 10);
+
+    std::vector<uint8_t> signature1 = key.sign(data1);
+    std::vector<uint8_t> signature2 = key.sign(data2);
 
     // check signature
-    CPPUNIT_ASSERT(public_key.checkSignature(data, signature));
+    CPPUNIT_ASSERT(public_key.checkSignature(data1, signature1));
+    CPPUNIT_ASSERT(public_key.checkSignature(data2, signature2));
 
     // encrypt data
-    std::vector<uint8_t> encrypted = public_key.encrypt(data);
-    std::vector<uint8_t> decrypted = key.decrypt(encrypted);
-    CPPUNIT_ASSERT(data == decrypted);
+    {
+        std::vector<uint8_t> encrypted = public_key.encrypt(data1);
+        std::vector<uint8_t> decrypted = key.decrypt(encrypted);
+        CPPUNIT_ASSERT(data1 == decrypted);
+    }
+
+    {
+        std::vector<uint8_t> encrypted = public_key.encrypt(data2);
+        std::vector<uint8_t> decrypted = key.decrypt(encrypted);
+        CPPUNIT_ASSERT(data2 == decrypted);
+    }
+
+    // encrypt data (invalid)
+    {
+        std::vector<uint8_t> encrypted = public_key.encrypt(data1);
+        encrypted[1]++;
+        CPPUNIT_ASSERT_THROW(key.decrypt(encrypted), dht::crypto::DecryptError);
+    }
+
+    {
+        std::vector<uint8_t> encrypted = public_key.encrypt(data2);
+        encrypted[2]++;
+        CPPUNIT_ASSERT_THROW(key.decrypt(encrypted), dht::crypto::DecryptError);
+    }
 }
 
 void

tests/dhtproxytester.cpp

@@ -55,13 +55,13 @@ DhtProxyTester::setUp() {
     clientConfig.dht_config.node_config.maintain_storage = false;
     clientConfig.threaded = true;
     clientConfig.push_node_id = "dhtnode";
+    clientConfig.proxy_server = "https://127.0.0.1:8080";
+
+    dht::DhtRunner::Context clientContext {};
     clientContext.logger = logger;
 
     nodeClient = std::make_shared<dht::DhtRunner>();
     nodeClient->run(0, clientConfig, std::move(clientContext));
-    nodeClient->bootstrap(nodePeer.getBound());
-    nodeClient->setProxyServer("https://127.0.0.1:8080");
-    nodeClient->enableProxy(true); // creates DhtProxyClient
 }
 
 void
@@ -159,12 +159,13 @@ DhtProxyTester::testResubscribeGetValues() {
     std::unique_lock<std::mutex> lk(cv_m);
     auto key = dht::InfoHash::get("GLaDOs");
 
-    // If a peer send a value, the listen operation from the client
+    // If a peer sent a value, the listen operation from the client
     // should retrieve this value
     dht::Value firstVal {"Hey! It's been a long time. How have you been?"};
     auto firstVal_data = firstVal.data;
-    nodePeer.put(key, std::move(firstVal), [&](bool) {
+    nodePeer.put(key, std::move(firstVal), [&](bool ok) {
         std::lock_guard<std::mutex> lk(cv_m);
+        CPPUNIT_ASSERT(ok);
         done = true;
         cv.notify_all();
     });
@@ -180,11 +181,11 @@ DhtProxyTester::testResubscribeGetValues() {
 
     // Reboot node (to avoid cache)
     nodeClient->join();
+    clientConfig.push_token = "atlas";
+
+    dht::DhtRunner::Context clientContext {};
+    clientContext.logger = logger;
     nodeClient->run(0, clientConfig, std::move(clientContext));
-    nodeClient->bootstrap(nodePeer.getBound());
-    nodeClient->setProxyServer("https://127.0.0.1:8080");
-    nodeClient->enableProxy(true);
-    nodeClient->setPushNotificationToken("atlas");
 
     // For the second subscribe, the proxy will return the value in the body
     auto values = std::vector<dht::Blob>();

tests/dhtproxytester.h

@@ -72,7 +72,6 @@ class DhtProxyTester : public CppUnit::TestFixture {
     std::shared_ptr<dht::DhtRunner> nodeProxy;
     std::unique_ptr<dht::DhtProxyServer> serverProxy;
 
-    dht::DhtRunner::Context clientContext {};
     dht::DhtRunner::Config clientConfig {};
     std::shared_ptr<dht::Logger> logger {};
 };

tools/dhtcnode.c

@@ -15,6 +15,7 @@ bool dht_value_callback(const dht_value* value, bool expired, void* user_data)
 {
     dht_data_view data = dht_value_get_data(value);
     printf("Value callback %s: %.*s.\n", expired ? "expired" : "new", (int)data.size, data.data);
+    return true;
 }
 
 bool dht_get_callback(const dht_value* value, void* user_data)
@@ -22,15 +23,16 @@ bool dht_get_callback(const dht_value* value, void* user_data)
     dht_runner* runner = (dht_runner*)user_data;
     dht_data_view data = dht_value_get_data(value);
     printf("Get callback: %.*s.\n", (int)data.size, data.data);
+    return true;
 }
 
-bool dht_done_callback(bool ok, void* user_data)
+void dht_done_callback(bool ok, void* user_data)
 {
     dht_runner* runner = (dht_runner*)user_data;
     printf("Done callback. %s\n", ok ? "Success !" : "Failure :-(");
 }
 
-bool op_context_free(void* user_data)
+void op_context_free(void* user_data)
 {
     struct op_context* ctx = (struct op_context*)user_data;
     printf("op_context_free %d.\n", ctx->d);
@@ -60,6 +62,22 @@ char* print_addr(const struct sockaddr* addr) {
 
 int main()
 {
+    dht_identity id = dht_identity_generate("testNode", NULL);
+    dht_infohash cert_id = dht_certificate_get_id(id.certificate);
+    printf("Cert ID: %s\n", dht_infohash_print(&cert_id));
+
+    dht_publickey* pk = dht_certificate_get_publickey(id.certificate);
+    dht_infohash pk_id = dht_publickey_get_id(pk);
+    printf("PK ID: %s\n", dht_infohash_print(&pk_id));
+    dht_publickey_delete(pk);
+
+    pk = dht_privatekey_get_publickey(id.privatekey);
+    pk_id = dht_publickey_get_id(pk);
+    printf("Key ID: %s\n", dht_infohash_print(&pk_id));
+    dht_publickey_delete(pk);
+
+    dht_identity_delete(&id);
+
     dht_runner* runner = dht_runner_new();
     dht_runner_run(runner, 4040);
 
@@ -71,7 +89,7 @@ int main()
     // Put data
     const char* data_str = "yo, this is some data";
     dht_value* val = dht_value_new(data_str, strlen(data_str));
-    dht_runner_put(runner, &h, val, dht_done_callback, runner);
+    dht_runner_put(runner, &h, val, dht_done_callback, runner, false);
     dht_value_unref(val);
 
     // Get data