dht: add max bootstrap period

Also starts a bootstrap if disconnected.

.github/workflows/ccpp.yml

@@ -13,12 +13,9 @@ jobs:
       run: |
         sudo apt install libncurses5-dev libreadline-dev nettle-dev \
         libgnutls28-dev libuv1-dev cython3 python3-dev python3-setuptools libcppunit-dev libjsoncpp-dev \
-        autotools-dev autoconf libfmt-dev libhttp-parser-dev libmsgpack-dev
+        autotools-dev autoconf libfmt-dev libhttp-parser-dev libmsgpack-dev libargon2-0-dev
     - name: autogen
       run: ./autogen.sh
-    - name: argon2
-      run: |
-        cd argon2 && make && sudo make install && cd ..
     - name: asio
       run: |
         wget https://github.com/aberaud/asio/archive/b2b7a1c166390459e1c169c8ae9ef3234b361e3f.tar.gz \

CMakeLists.txt

@@ -7,7 +7,7 @@ include(FindPkgConfig)
 include(cmake/CheckAtomic.cmake)
 
 set (opendht_VERSION_MAJOR 2)
-set (opendht_VERSION_MINOR 3.0)
+set (opendht_VERSION_MINOR 3.2)
 set (opendht_VERSION ${opendht_VERSION_MAJOR}.${opendht_VERSION_MINOR})
 set (PACKAGE_VERSION ${opendht_VERSION})
 set (VERSION "${opendht_VERSION}")

c/opendht.cpp

@@ -173,7 +173,7 @@ int dht_privatekey_export(const dht_privatekey* k, char* out, size_t* out_size,
 
 dht_publickey* dht_privatekey_get_publickey(const dht_privatekey* k) {
     const auto& key = *reinterpret_cast<const PrivkeySp*>(k);
-    return reinterpret_cast<dht_publickey*>(new PubkeySp(std::make_shared<dht::crypto::PublicKey>(key->getPublicKey())));
+    return reinterpret_cast<dht_publickey*>(new PubkeySp(key->getSharedPublicKey()));
 }
 
 void dht_privatekey_delete(dht_privatekey* pk) {

configure.ac

@@ -1,7 +1,7 @@
 dnl define macros
 m4_define([opendht_major_version], 2)
 m4_define([opendht_minor_version], 3)
-m4_define([opendht_patch_version], 0)
+m4_define([opendht_patch_version], 2)
 m4_define([opendht_version],
 		  [opendht_major_version.opendht_minor_version.opendht_patch_version])
 

include/opendht/crypto.h

@@ -541,8 +541,8 @@ struct OPENDHT_PUBLIC Certificate {
     void addRevocationList(RevocationList&&);
     void addRevocationList(std::shared_ptr<RevocationList>);
 
-    static Certificate generate(const PrivateKey& key, const std::string& name = "dhtnode", const Identity& ca = {}, bool is_ca = false);
-    static Certificate generate(const CertificateRequest& request, const Identity& ca);
+    static Certificate generate(const PrivateKey& key, const std::string& name = "dhtnode", const Identity& ca = {}, bool is_ca = false, int64_t validity = 0);
+    static Certificate generate(const CertificateRequest& request, const Identity& ca, int64_t validity = 0);
 
     gnutls_x509_crt_t getCopy() const {
         if (not cert)
@@ -592,12 +592,20 @@ struct OPENDHT_PUBLIC Certificate {
      */
     std::pair<std::string, Blob> generateOcspRequest(gnutls_x509_crt_t& issuer);
 
+    /**
+     * Change certificate's expiration
+     */
+    void setValidity(const Identity& ca, int64_t validity);
+    void setValidity(const PrivateKey& key, int64_t validity);
+
     gnutls_x509_crt_t cert {nullptr};
     std::shared_ptr<Certificate> issuer {};
     std::shared_ptr<OcspResponse> ocspResponse;
 private:
     Certificate(const Certificate&) = delete;
     Certificate& operator=(const Certificate&) = delete;
+    mutable InfoHash cachedId_ {};
+    mutable PkId cachedLongId_ {};
 
     struct crlNumberCmp {
         bool operator() (const std::shared_ptr<RevocationList>& lhs, const std::shared_ptr<RevocationList>& rhs) const {

include/opendht/dht.h

@@ -114,7 +114,7 @@ public:
 
     void addBootstrap(const std::string& host, const std::string& service) override {
         bootstrap_nodes.emplace_back(host, service);
-        onDisconnected();
+        startBootstrap();
     }
 
     void clearBootstrap() override {
@@ -356,6 +356,8 @@ private:
 
     static constexpr duration REANNOUNCE_MARGIN {std::chrono::seconds(10)};
 
+    static constexpr std::chrono::seconds BOOTSTRAP_PERIOD {10};
+
     static constexpr size_t TOKEN_SIZE {32};
 
     // internal structures
@@ -393,7 +395,7 @@ private:
     Kad dht6 {};
 
     std::vector<std::pair<std::string,std::string>> bootstrap_nodes {};
-    std::chrono::steady_clock::duration bootstrap_period {std::chrono::seconds(10)};
+    std::chrono::steady_clock::duration bootstrap_period {BOOTSTRAP_PERIOD};
     Sp<Scheduler::Job> bootstrapJob {};
 
     std::map<InfoHash, Storage> store;
@@ -480,6 +482,9 @@ private:
     void sendCachedPing(Bucket& b);
     bool bucketMaintenance(RoutingTable&);
     void dumpBucket(const Bucket& b, std::ostream& out) const;
+    void bootstrap();
+    void startBootstrap();
+    void stopBootstrap();
 
     // Nodes
     void onNewNode(const Sp<Node>& node, int confirm);
@@ -487,7 +492,6 @@ private:
     bool trySearchInsert(const Sp<Node>& node);
 
     // Searches
-
     inline SearchMap& searches(sa_family_t af) { return dht(af).searches; }
     inline const SearchMap& searches(sa_family_t af) const { return dht(af).searches; }
 

include/opendht/dht_proxy_client.h

@@ -337,6 +337,7 @@ private:
     NodeStats stats6_ {};
     SockAddr publicAddressV4_;
     SockAddr publicAddressV6_;
+    std::atomic_bool launchConnectedCbs_ {false};
 
     InfoHash myid {};
 

include/opendht/dhtrunner.h

@@ -456,8 +456,6 @@ public:
     void forwardAllMessages(bool forward);
 
 private:
-    static constexpr std::chrono::seconds BOOTSTRAP_PERIOD {10};
-
     enum class State {
         Idle,
         Running,

include/opendht/thread_pool.h

@@ -27,6 +27,8 @@
 #include <future>
 #include <functional>
 
+#include <ciso646> // fix windows compiler bug
+
 namespace dht {
 
 class OPENDHT_PUBLIC ThreadPool {

src/crypto.cpp

@@ -69,7 +69,7 @@ constexpr gnutls_digest_algorithm_t gnutlsHashAlgo(size_t min_res) {
                                GNUTLS_DIG_SHA1));
 }
 
-constexpr size_t gnutlsHashSize(int algo) {
+constexpr size_t gnutlsHashSize(gnutls_digest_algorithm_t algo) {
     return (algo == GNUTLS_DIG_SHA512) ? 512/8 : (
            (algo == GNUTLS_DIG_SHA256) ? 256/8 : (
            (algo == GNUTLS_DIG_SHA1)   ? 160/8 : 0 ));
@@ -853,12 +853,15 @@ Certificate::getId() const
 {
     if (not cert)
         return {};
+    if (cachedId_)
+        return cachedId_;
     InfoHash id;
     size_t sz = id.size();
     if (auto err = gnutls_x509_crt_get_key_id(cert, 0, id.data(), &sz))
         throw CryptoException(std::string("Can't get certificate public key ID: ") + gnutls_strerror(err));
     if (sz != id.size())
         throw CryptoException("Can't get certificate public key ID: wrong output length.");
+    cachedId_ = id;
     return id;
 }
 
@@ -867,6 +870,8 @@ Certificate::getLongId() const
 {
     if (not cert)
         return {};
+    if (cachedLongId_)
+        return cachedLongId_;
 #if GNUTLS_VERSION_NUMBER < 0x030401
     throw CryptoException("Can't get certificate 256 bits public key ID: GnuTLS 3.4.1 or higher required.");
 #else
@@ -876,6 +881,7 @@ Certificate::getLongId() const
         throw CryptoException(std::string("Can't get certificate 256 bits public key ID: ") + gnutls_strerror(err));
     if (sz != id.size())
         throw CryptoException("Can't get certificate 256 bits public key ID: wrong output length.");
+    cachedLongId_ = id;
     return id;
 #endif
 }
@@ -1157,14 +1163,14 @@ setRandomSerial(gnutls_x509_crt_t cert)
 }
 
 Certificate
-Certificate::generate(const PrivateKey& key, const std::string& name, const Identity& ca, bool is_ca)
+Certificate::generate(const PrivateKey& key, const std::string& name, const Identity& ca, bool is_ca, int64_t validity)
 {
     gnutls_x509_crt_t cert;
     if (not key.x509_key or gnutls_x509_crt_init(&cert) != GNUTLS_E_SUCCESS)
         return {};
     Certificate ret {cert};
 
-    setValidityPeriod(cert, 10 * 365 * 24 * 60 * 60);
+    setValidityPeriod(cert, validity <= 0 ? 10 * 365 * 24 * 60 * 60 : validity);
     if (int err = gnutls_x509_crt_set_key(cert, key.x509_key)) {
         throw CryptoException(std::string("Error when setting certificate key ") + gnutls_strerror(err));
     }
@@ -1210,7 +1216,7 @@ Certificate::generate(const PrivateKey& key, const std::string& name, const Iden
 }
 
 Certificate
-Certificate::generate(const CertificateRequest& request, const Identity& ca)
+Certificate::generate(const CertificateRequest& request, const Identity& ca, int64_t validity)
 {
     gnutls_x509_crt_t cert;
     if (auto err = gnutls_x509_crt_init(&cert))
@@ -1223,7 +1229,7 @@ Certificate::generate(const CertificateRequest& request, const Identity& ca)
         throw CryptoException(std::string("Can't set certificate version: ") + gnutls_strerror(err));
     }
 
-    setValidityPeriod(cert, 10 * 365 * 24 * 60 * 60);
+    setValidityPeriod(cert, validity <= 0 ? 10 * 365 * 24 * 60 * 60 : validity);
     setRandomSerial(cert);
 
     if (auto err = gnutls_x509_crt_privkey_sign(cert, ca.second->cert, ca.first->key, ca.second->getPreferredDigest(), 0)) {
@@ -1234,6 +1240,32 @@ Certificate::generate(const CertificateRequest& request, const Identity& ca)
     return ret.getPacked();
 }
 
+void
+Certificate::setValidity(const Identity& ca, int64_t validity)
+{
+    setValidityPeriod(cert, validity);
+    setRandomSerial(cert);
+    if (ca.first && ca.second) {
+        if (not ca.second->isCA()) {
+            throw CryptoException("Signing certificate must be CA");
+        }
+        if (int err = gnutls_x509_crt_privkey_sign(cert, ca.second->cert, ca.first->key, ca.second->getPreferredDigest(), 0)) {
+            throw CryptoException(std::string("Error when signing certificate ") + gnutls_strerror(err));
+        }
+    }
+}
+
+void
+Certificate::setValidity(const PrivateKey& key, int64_t validity)
+{
+    setValidityPeriod(cert, validity);
+    setRandomSerial(cert);
+    const auto& pk = key.getPublicKey();
+    if (int err = gnutls_x509_crt_privkey_sign(cert, cert, key.key, pk.getPreferredDigest(), 0)) {
+        throw CryptoException(std::string("Error when signing certificate ") + gnutls_strerror(err));
+    }
+}
+
 std::vector<std::shared_ptr<RevocationList>>
 Certificate::getRevocationLists() const
 {

src/dht.cpp

@@ -38,10 +38,12 @@ using namespace std::placeholders;
 
 constexpr std::chrono::minutes Dht::MAX_STORAGE_MAINTENANCE_EXPIRE_TIME;
 constexpr std::chrono::minutes Dht::SEARCH_EXPIRE_TIME;
+constexpr std::chrono::seconds Dht::BOOTSTRAP_PERIOD;
 constexpr duration Dht::LISTEN_EXPIRE_TIME;
 constexpr duration Dht::LISTEN_EXPIRE_TIME_PUBLIC;
 constexpr duration Dht::REANNOUNCE_MARGIN;
 static constexpr size_t MAX_REQUESTS_PER_SEC {8 * 1024};
+static constexpr duration BOOTSTRAP_PERIOD_MAX {std::chrono::hours(24)};
 
 NodeStatus
 Dht::updateStatus(sa_family_t af)
@@ -51,9 +53,9 @@ Dht::updateStatus(sa_family_t af)
     d.status = d.getStatus(scheduler.time());
     if (d.status != old) {
         auto& other = dht(af == AF_INET ? AF_INET6 : AF_INET);
-        if (other.status == NodeStatus::Disconnected && d.status == NodeStatus::Disconnected)
+        if (other.status == NodeStatus::Disconnected && d.status == NodeStatus::Disconnected) {
             onDisconnected();
-        else if (other.status == NodeStatus::Connected || d.status == NodeStatus::Connected) {
+        } else if (other.status == NodeStatus::Connected || d.status == NodeStatus::Connected) {
             onConnected();
         }
     }
@@ -1437,6 +1439,7 @@ Dht::connectivityChanged(sa_family_t af)
     reported_addr.erase(std::remove_if(reported_addr.begin(), reported_addr.end(), [&](const ReportedAddr& addr){
         return addr.second.getFamily() == af;
     }), reported_addr.end());
+    startBootstrap(); // will only happen if disconnected
 }
 
 void
@@ -2018,11 +2021,7 @@ Dht::expire()
 void
 Dht::onConnected()
 {
-    if (bootstrapJob) {
-        bootstrapJob->cancel();
-        bootstrapJob.reset();
-    }
-    bootstrap_period = std::chrono::seconds(10);
+    stopBootstrap();
     auto callbacks = std::move(onConnectCallbacks_);
     while (not callbacks.empty()) {
         callbacks.front()();
@@ -2032,6 +2031,13 @@ Dht::onConnected()
 
 void
 Dht::onDisconnected()
+{
+    if (not bootstrapJob)
+        bootstrap();
+}
+
+void
+Dht::bootstrap()
 {
     if (dht4.status != NodeStatus::Disconnected || dht6.status != NodeStatus::Disconnected)
         return;
@@ -2052,8 +2058,25 @@ Dht::onDisconnected()
     }
     if (bootstrapJob)
         bootstrapJob->cancel();
-    bootstrapJob = scheduler.add(scheduler.time() + bootstrap_period, std::bind(&Dht::onDisconnected, this));
-    bootstrap_period *= 2;
+    bootstrapJob = scheduler.add(scheduler.time() + bootstrap_period, std::bind(&Dht::bootstrap, this));
+    bootstrap_period = std::min(bootstrap_period * 2, BOOTSTRAP_PERIOD_MAX);
+}
+
+void
+Dht::startBootstrap()
+{
+    stopBootstrap();
+    bootstrapJob = scheduler.add(scheduler.time(), std::bind(&Dht::bootstrap, this));
+}
+
+void
+Dht::stopBootstrap()
+{
+    if (bootstrapJob) {
+        bootstrapJob->cancel();
+        bootstrapJob.reset();
+    }
+    bootstrap_period = BOOTSTRAP_PERIOD;
 }
 
 void

src/dht_proxy_client.cpp

@@ -720,7 +720,8 @@ DhtProxyClient::onProxyInfos(const Json::Value& proxyInfos, const sa_family_t fa
     }
     auto newStatus = std::max(statusIpv4_, statusIpv6_);
     if (newStatus == NodeStatus::Connected) {
-        if (oldStatus == NodeStatus::Disconnected || oldStatus == NodeStatus::Connecting) {
+        if (oldStatus == NodeStatus::Disconnected || oldStatus == NodeStatus::Connecting || launchConnectedCbs_) {
+            launchConnectedCbs_ = false;
             listenerRestartTimer_->expires_at(std::chrono::steady_clock::now());
             listenerRestartTimer_->async_wait(std::bind(&DhtProxyClient::restartListeners, this, std::placeholders::_1));
             if (not onConnectCallbacks_.empty()) {
@@ -1160,6 +1161,9 @@ DhtProxyClient::pushNotificationReceived(const std::map<std::string, std::string
     {
         // If a push notification is received, the proxy is up and running
         std::lock_guard<std::mutex> l(lockCurrentProxyInfos_);
+        auto oldStatus = std::max(statusIpv4_, statusIpv6_);
+        if (oldStatus != NodeStatus::Connected)
+            launchConnectedCbs_ = true;
         statusIpv4_ = NodeStatus::Connected;
         statusIpv6_ = NodeStatus::Connected;
     }
@@ -1265,11 +1269,7 @@ DhtProxyClient::resubscribe(const InfoHash& key, const size_t token, Listener& l
         logger_->d("[proxy:client] [resubscribe] [search %s]", key.to_c_str());
 
     auto opstate = listener.opstate;
-    opstate->stop = true;
-    if (listener.request){
-        listener.request.reset();
-    }
-    opstate->stop = false;
+    listener.request.reset(); // This will update ok to false
     opstate->ok = true;
 
     restinio::http_request_header_t header;

src/dht_proxy_server.cpp

@@ -1004,7 +1004,13 @@ DhtProxyServer::sendPushNotification(const std::string& token, Json::Value&& jso
         notification["platform"] = type == PushType::Android ? 2 : 1;
         notification["data"] = std::move(json);
         notification["priority"] = highPriority ? "high" : "normal";
-        notification["time_to_live"] = 600;
+        if (type == PushType::Android)
+            notification["time_to_live"] = 3600 * 24; // 24 hours
+        else {
+            const auto expiration = std::chrono::system_clock::now() + std::chrono::hours(24);
+            uint32_t exp = std::chrono::duration_cast<std::chrono::seconds>(expiration.time_since_epoch()).count();
+            notification["expiration"] = exp;
+        }
 
         Json::Value notifications(Json::arrayValue);
         notifications[0] = notification;

src/dhtrunner.cpp

@@ -36,7 +36,6 @@
 
 namespace dht {
 
-constexpr std::chrono::seconds DhtRunner::BOOTSTRAP_PERIOD;
 static const std::string PEER_DISCOVERY_DHT_SERVICE = "dht";
 
 struct DhtRunner::Listener {
@@ -705,8 +704,8 @@ DhtRunner::loop_()
             rcv_free.splice(rcv_free.end(), std::move(received_treated));
     }
 
-    if (dropped)
-        std::cerr << "Dropped " << dropped << " packets with high delay" << std::endl;
+    if (dropped && logger_)
+        logger_->e("[runner %p] Dropped %zu packets with high delay.", this, dropped);
 
     NodeStatus nstatus4 = dht->updateStatus(AF_INET);
     NodeStatus nstatus6 = dht->updateStatus(AF_INET6);

src/storage.h

@@ -283,7 +283,7 @@ Storage::expire(const InfoHash& id, time_point now)
     });
     std::vector<Sp<Value>> ret;
     ret.reserve(std::distance(r, values.end()));
-    ssize_t size_diff {};
+    ssize_t size_diff {0};
     std::for_each(r, values.end(), [&](const ValueStorage& v) {
         size_diff -= v.data->size();
         if (v.store_bucket)

tests/threadpooltester.cpp

@@ -21,6 +21,7 @@
 
 #include "opendht/thread_pool.h"
 #include <atomic>
+#include <thread>
 
 namespace test {
 CPPUNIT_TEST_SUITE_REGISTRATION(ThreadPoolTester);

tools/perftest.cpp

@@ -102,7 +102,7 @@ benchPingPong(unsigned netSize, unsigned n_parallel) {
 
     auto start = clock::now();
 
-    for (unsigned i=0; i<n_parallel; i++) 
+    for (unsigned i=0; i<n_parallel; i++)
         ping(pong_node, locs[i].first);
 
     {