proxy client: fix leak in sendListen

All encrypted values put on the DHT currently have the default priority
of 0 because the priority information is ignored by the Value::encrypt
function.

.github/workflows/ccpp.yml

@@ -7,7 +7,7 @@ jobs:
     name: Ubuntu/GCC Autotools build
     runs-on: ubuntu-latest
     steps:
-    - uses: actions/checkout@v3
+    - uses: actions/checkout@v4
     - name: Dependencies
       run: |
         sudo apt update && \
@@ -50,7 +50,7 @@ jobs:
     name: Ubuntu/GCC Meson build
     runs-on: ubuntu-latest
     steps:
-    - uses: actions/checkout@v3
+    - uses: actions/checkout@v4
     - name: Dependencies
       run: |
         sudo apt update && \
@@ -68,7 +68,7 @@ jobs:
     name: Ubuntu/GCC minimal build
     runs-on: ubuntu-latest
     steps:
-    - uses: actions/checkout@v3
+    - uses: actions/checkout@v4
     - name: Dependencies
       run: |
         sudo apt update && \
@@ -87,16 +87,16 @@ jobs:
 
   build-macos:
     name: macOS/Clang build
-    runs-on: macos-12
+    runs-on: macOS-15
     steps:
-    - uses: actions/checkout@v3
+    - uses: actions/checkout@v4
     - name: Dependencies
       run: |
         brew install msgpack-cxx asio gnutls nettle readline fmt jsoncpp argon2 openssl cppunit
 
     - name: expected-lite
       run: |
-        sudo mkdir /usr/local/include/nonstd && \
+        sudo mkdir -p /usr/local/include/nonstd && \
         sudo wget https://raw.githubusercontent.com/martinmoene/expected-lite/master/include/nonstd/expected.hpp \
           -O /usr/local/include/nonstd/expected.hpp
 

.github/workflows/clang-analyzer.yml

@@ -6,7 +6,7 @@ jobs:
     name: Clang Static Analysis
     runs-on: ubuntu-latest
     steps:
-      - uses: actions/checkout@v3
+      - uses: actions/checkout@v4
       - name: deps
         run: |
           sudo apt update && \

.github/workflows/codeql-analysis.yml

@@ -26,7 +26,7 @@ jobs:
 
     steps:
     - name: Checkout repository
-      uses: actions/checkout@v3
+      uses: actions/checkout@v4
 
     # Initializes the CodeQL tools for scanning.
     - name: Initialize CodeQL

.github/workflows/release-package.yml

@@ -25,10 +25,10 @@ jobs:
 
     steps:
       - name: Checkout repository
-        uses: actions/checkout@v3
+        uses: actions/checkout@v4
 
       - name: Log in to the Container registry
-        uses: docker/login-action@v2
+        uses: docker/login-action@v3
         with:
           registry: ${{ env.REGISTRY }}
           username: ${{ github.actor }}
@@ -36,12 +36,12 @@ jobs:
 
       - name: Extract metadata (tags, labels) for Docker
         id: meta
-        uses: docker/metadata-action@v4
+        uses: docker/metadata-action@v5
         with:
           images: ${{ env.REGISTRY }}/${{ env.IMAGE_NAME_DEPS }}
 
       - name: Build and push Docker image
-        uses: docker/build-push-action@v3
+        uses: docker/build-push-action@v5
         with:
           context: .
           file: docker/DockerfileDeps
@@ -58,10 +58,10 @@ jobs:
 
     steps:
       - name: Checkout repository
-        uses: actions/checkout@v3
+        uses: actions/checkout@v4
 
       - name: Log in to the Container registry
-        uses: docker/login-action@v2
+        uses: docker/login-action@v3
         with:
           registry: ${{ env.REGISTRY }}
           username: ${{ github.actor }}
@@ -69,13 +69,13 @@ jobs:
 
       - name: Extract metadata (tags, labels) for Docker
         id: meta
-        uses: docker/metadata-action@v4
+        uses: docker/metadata-action@v5
         with:
           images: |
             ${{ env.REGISTRY }}/${{ env.IMAGE_NAME }}
 
       - name: Build and push Docker image
-        uses: docker/build-push-action@v3
+        uses: docker/build-push-action@v5
         with:
           context: .
           file: docker/Dockerfile
@@ -92,10 +92,10 @@ jobs:
 
     steps:
       - name: Checkout repository
-        uses: actions/checkout@v3
+        uses: actions/checkout@v4
 
       - name: Log in to the Container registry
-        uses: docker/login-action@v2
+        uses: docker/login-action@v3
         with:
           registry: ${{ env.REGISTRY }}
           username: ${{ github.actor }}
@@ -103,12 +103,12 @@ jobs:
 
       - name: Extract metadata (tags, labels) for Docker
         id: meta
-        uses: docker/metadata-action@v4
+        uses: docker/metadata-action@v5
         with:
           images: ${{ env.REGISTRY }}/${{ env.IMAGE_NAME_DEPS_LLVM }}
 
       - name: Build and push Docker image
-        uses: docker/build-push-action@v3
+        uses: docker/build-push-action@v5
         with:
           context: .
           file: docker/DockerfileDepsLlvm
@@ -125,10 +125,10 @@ jobs:
 
     steps:
       - name: Checkout repository
-        uses: actions/checkout@v3
+        uses: actions/checkout@v4
 
       - name: Log in to the Container registry
-        uses: docker/login-action@v2
+        uses: docker/login-action@v3
         with:
           registry: ${{ env.REGISTRY }}
           username: ${{ github.actor }}
@@ -136,13 +136,13 @@ jobs:
 
       - name: Extract metadata (tags, labels) for Docker
         id: meta
-        uses: docker/metadata-action@v4
+        uses: docker/metadata-action@v5
         with:
           images: |
             ${{ env.REGISTRY }}/${{ env.IMAGE_NAME_LLVM }}
 
       - name: Build and push Docker image
-        uses: docker/build-push-action@v3
+        uses: docker/build-push-action@v5
         with:
           context: .
           file: docker/DockerfileLlvm
@@ -159,10 +159,10 @@ jobs:
 
     steps:
       - name: Checkout repository
-        uses: actions/checkout@v3
+        uses: actions/checkout@v4
 
       - name: Log in to the Container registry
-        uses: docker/login-action@v2
+        uses: docker/login-action@v3
         with:
           registry: ${{ env.REGISTRY }}
           username: ${{ github.actor }}
@@ -170,13 +170,13 @@ jobs:
 
       - name: Extract metadata (tags, labels) for Docker
         id: meta
-        uses: docker/metadata-action@v4
+        uses: docker/metadata-action@v5
         with:
           images: |
             ${{ env.REGISTRY }}/${{ env.IMAGE_NAME_DHTNODE }}
 
       - name: Build and push Docker image
-        uses: docker/build-push-action@v3
+        uses: docker/build-push-action@v5
         with:
           context: .
           file: docker/DockerfileDhtnode
@@ -193,10 +193,10 @@ jobs:
 
     steps:
       - name: Checkout repository
-        uses: actions/checkout@v3
+        uses: actions/checkout@v4
 
       - name: Log in to the Container registry
-        uses: docker/login-action@v2
+        uses: docker/login-action@v3
         with:
           registry: ${{ env.REGISTRY }}
           username: ${{ github.actor }}
@@ -204,13 +204,13 @@ jobs:
 
       - name: Extract metadata (tags, labels) for Docker
         id: meta
-        uses: docker/metadata-action@v4
+        uses: docker/metadata-action@v5
         with:
           images: |
             ${{ env.REGISTRY }}/${{ env.IMAGE_NAME_ALPINE_DEPS }}
 
       - name: Build and push Docker image
-        uses: docker/build-push-action@v3
+        uses: docker/build-push-action@v5
         with:
           context: .
           file: docker/DockerfileDepsAlpine
@@ -227,10 +227,10 @@ jobs:
 
     steps:
       - name: Checkout repository
-        uses: actions/checkout@v3
+        uses: actions/checkout@v4
 
       - name: Log in to the Container registry
-        uses: docker/login-action@v2
+        uses: docker/login-action@v3
         with:
           registry: ${{ env.REGISTRY }}
           username: ${{ github.actor }}
@@ -238,13 +238,13 @@ jobs:
 
       - name: Extract metadata (tags, labels) for Docker
         id: meta
-        uses: docker/metadata-action@v4
+        uses: docker/metadata-action@v5
         with:
           images: |
             ${{ env.REGISTRY }}/${{ env.IMAGE_NAME_ALPINE }}
 
       - name: Build and push Docker image
-        uses: docker/build-push-action@v3
+        uses: docker/build-push-action@v5
         with:
           context: .
           file: docker/DockerfileAlpine
@@ -262,7 +262,7 @@ jobs:
 
     steps:
       - name: Checkout repository
-        uses: actions/checkout@v3
+        uses: actions/checkout@v4
 
       - name: cmake
         run: |
@@ -280,14 +280,14 @@ jobs:
       - name: build
         run: cd build && make dist
 
-      - uses: actions/upload-artifact@v3
+      - uses: actions/upload-artifact@v4
         with:
           name: opendht-wheels-linux
           path: build/python/dist/*.whl
 
       - name: Create Release
         id: create_release
-        uses: softprops/action-gh-release@v1
+        uses: softprops/action-gh-release@v2
         with:
           draft: true
           files: build/python/dist/*.whl

CMakeLists.txt

@@ -85,6 +85,9 @@ if (NOT MSVC)
 
     if (OPENDHT_HTTP OR OPENDHT_PEER_DISCOVERY)
         find_path(ASIO_INCLUDE_DIR asio.hpp REQUIRED)
+        message(STATUS "Found ASIO ${ASIO_INCLUDE_DIR}")
+    else()
+        message(STATUS "ASIO not required")
     endif ()
 
     find_package(fmt)
@@ -94,7 +97,7 @@ if (NOT MSVC)
 
         # llhttp
         find_path(LLHTTP_INCLUDE_DIR llhttp.h)
-        find_library(LLHTTP_LIBRARY libllhttp.a)
+        find_library(LLHTTP_LIBRARY NAMES llhttp libllhttp)
         if (LLHTTP_INCLUDE_DIR AND LLHTTP_LIBRARY)
             message(STATUS "Found llhttp ${LLHTTP_INCLUDE_DIR} ${LLHTTP_LIBRARY}")
             add_library(llhttp_static STATIC IMPORTED)
@@ -160,6 +163,9 @@ endif ()
 
 if (OPENDHT_HTTP OR OPENDHT_PEER_DISCOVERY)
     add_definitions(-DASIO_STANDALONE)
+    if (UNIX AND NOT APPLE)
+        pkg_search_module(liburing IMPORTED_TARGET liburing)
+    endif ()
 endif()
 
 if (NOT MSVC)
@@ -401,6 +407,11 @@ else()
     if (OPENDHT_PROXY_OPENSSL)
         target_link_libraries(opendht PUBLIC PkgConfig::OPENSSL)
     endif()
+    if (liburing_FOUND)
+        set(iouring_lib ", liburing")
+        target_link_libraries(opendht PUBLIC PkgConfig::liburing)
+        target_compile_definitions(opendht PUBLIC ASIO_HAS_IO_URING ASIO_DISABLE_EPOLL)
+    endif()
 endif()
 
 if (BUILD_SHARED_LIBS)

c/opendht.cpp

@@ -344,6 +344,7 @@ int dht_runner_run_config(dht_runner* r, in_port_t port, const dht_runner_config
         config.push_token = conf->push_token ? std::string(conf->push_token) : std::string{};
         config.push_topic = conf->push_topic ? std::string(conf->push_topic) : std::string{};
         config.push_platform = conf->push_platform ? std::string(conf->push_platform) : std::string{};
+        config.proxy_user_agent = conf->proxy_user_agent ? std::string(conf->proxy_user_agent) : std::string{};
         config.peer_discovery = conf->peer_discovery;
         config.peer_publish = conf->peer_publish;
 

c/opendht_c.h

@@ -148,6 +148,7 @@ struct OPENDHT_PUBLIC dht_runner_config {
     const char* push_token;
     const char* push_topic;
     const char* push_platform;
+    const char* proxy_user_agent;
     bool peer_discovery;
     bool peer_publish;
     dht_certificate* server_ca;

include/opendht/dht_proxy_client.h

@@ -53,19 +53,21 @@ public:
     DhtProxyClient();
 
     explicit DhtProxyClient(
-        std::shared_ptr<crypto::Certificate> serverCA, crypto::Identity clientIdentity,
-        std::function<void()> loopSignal, const std::string& serverHost,
-        const std::string& pushClientId = "", std::shared_ptr<Logger> logger = {});
+        std::shared_ptr<crypto::Certificate> serverCA,
+        crypto::Identity clientIdentity,
+        std::function<void()> loopSignal,
+        const std::string& serverHost,
+        const std::string& userAgent = "",
+        const std::string& pushClientId = "",
+        const std::string& pushToken = "",
+        const std::string& pushTopic = "",
+        const std::string& pushPlatform = "",
+        std::shared_ptr<Logger> logger = {}
+    );
 
     void setHeaderFields(http::Request& request);
 
-    virtual void setPushNotificationToken(const std::string& token) override {
-#ifdef OPENDHT_PUSH_NOTIFICATIONS
-        deviceKey_ = token;
-#else
-        (void) token;
-#endif
-    }
+    virtual void setPushNotificationToken(const std::string& token) override;
 
     virtual void setPushNotificationTopic(const std::string& topic) override {
 #ifdef OPENDHT_PUSH_NOTIFICATIONS
@@ -351,7 +353,7 @@ private:
     std::string proxyUrl_;
     dht::crypto::Identity clientIdentity_;
     std::shared_ptr<dht::crypto::Certificate> serverCertificate_;
-    //std::pair<std::string, std::string> serverHostService_;
+    std::string userAgent_ {"OpenDHT"};
     std::string pushClientId_;
     std::string pushSessionId_;
 

include/opendht/dht_proxy_server.h

@@ -98,6 +98,32 @@ public:
 
     asio::io_context& io_context() const;
 
+    using clock = std::chrono::steady_clock;
+    using time_point = clock::time_point;
+
+    struct PushStats {
+        uint64_t highPriorityCount {0};
+        uint64_t normalPriorityCount {0};
+
+        void increment(bool highPriority) {
+            if (highPriority)
+                highPriorityCount++;
+            else
+                normalPriorityCount++;
+        }
+
+        Json::Value toJson() const {
+            Json::Value val;
+            val["highPriorityCount"] = static_cast<Json::UInt64>(highPriorityCount);
+            val["normalPriorityCount"] = static_cast<Json::UInt64>(normalPriorityCount);
+            return val;
+        }
+
+        std::string toString() const {
+            return fmt::format("{} high priority, {} normal priority", highPriorityCount, normalPriorityCount);
+        }
+    };
+
     struct ServerStats {
         /** Current number of listen operations */
         size_t listenCount {0};
@@ -107,6 +133,17 @@ public:
         size_t totalPermanentPuts {0};
         /** Current number of push tokens with at least one listen operation */
         size_t pushListenersCount {0};
+
+        /** Time at which the server was started */
+        time_point serverStartTime;
+        /** Last time at which the stats were updated */
+        time_point lastUpdated;
+        /** Total number of push notification requests that the server attempted to
+          * send since being started, broken down by type and priority level */
+        PushStats androidPush;
+        PushStats iosPush;
+        PushStats unifiedPush;
+
         /** Average requests per second */
         double requestRate {0};
         /** Node Info **/
@@ -115,6 +152,10 @@ public:
         std::string toString() const {
             std::ostringstream ss;
             ss << "Listens: " << listenCount << " Puts: " << putCount << " PushListeners: " << pushListenersCount << std::endl;
+            ss << "Push requests in the last " << print_duration(lastUpdated - serverStartTime) << ": "
+                                               << "[Android: " << androidPush.toString() << "], "
+                                               << "[iOS: " << iosPush.toString() << "], "
+                                               << "[Unified: " << unifiedPush.toString() << "]" << std::endl;
             ss << "Requests: " << requestRate << " per second." << std::endl;
             if (nodeInfo) {
                 auto& ipv4 = nodeInfo->ipv4;
@@ -136,6 +177,11 @@ public:
             result["putCount"] = static_cast<Json::UInt64>(putCount);
             result["totalPermanentPuts"] = static_cast<Json::UInt64>(totalPermanentPuts);
             result["pushListenersCount"] = static_cast<Json::UInt64>(pushListenersCount);
+            result["serverStartTime"] = static_cast<Json::LargestInt>(to_time_t(serverStartTime));
+            result["lastUpdated"] = static_cast<Json::LargestInt>(to_time_t(lastUpdated));
+            result["androidPush"] = androidPush.toJson();
+            result["iosPush"] = iosPush.toJson();
+            result["unifiedPush"] = unifiedPush.toJson();
             result["requestRate"] = requestRate;
             if (nodeInfo)
                 result["nodeInfo"] = nodeInfo->toJson();
@@ -366,9 +412,6 @@ private:
     template <typename Is>
     void loadState(Is& is, size_t size);
 
-    using clock = std::chrono::steady_clock;
-    using time_point = clock::time_point;
-
     std::shared_ptr<asio::io_context> ioContext_;
     std::shared_ptr<DhtRunner> dht_;
     Json::StreamWriterBuilder jsonBuilder_;
@@ -393,6 +436,11 @@ private:
     std::shared_ptr<ServerStats> stats_;
     std::shared_ptr<NodeInfo> nodeInfo_ {};
     std::unique_ptr<asio::steady_timer> printStatsTimer_;
+    const time_point serverStartTime_;
+    mutable std::mutex pushStatsMutex_;
+    PushStats androidPush_;
+    PushStats iosPush_;
+    PushStats unifiedPush_;
 
     // Thread-safe access to listeners map.
     std::mutex lockListener_;

include/opendht/dhtrunner.h

@@ -64,6 +64,7 @@ public:
         std::string push_token {};
         std::string push_topic {};
         std::string push_platform {};
+        std::string proxy_user_agent {};
         bool peer_discovery {false};
         bool peer_publish {false};
         std::shared_ptr<dht::crypto::Certificate> server_ca;
@@ -409,7 +410,8 @@ public:
     void run(const Config& config, Context&& context);
 
     void setOnStatusChanged(StatusCallback&& cb) {
-        statusCb = std::move(cb);
+        if (cb)
+            statusCbs.emplace_back(std::move(cb));
     }
 
     /**
@@ -519,7 +521,8 @@ private:
 
     NodeStatus status4 {NodeStatus::Disconnected},
                status6 {NodeStatus::Disconnected};
-    StatusCallback statusCb {nullptr};
+
+    std::vector<StatusCallback> statusCbs {};
 
     /** PeerDiscovery Parameters */
     std::shared_ptr<PeerDiscovery> peerDiscovery_;

include/opendht/peer_discovery.h

@@ -23,7 +23,9 @@
 #include "sockaddr.h"
 #include "infohash.h"
 #include "logger.h"
+#include "utils.h"
 
+#include <asio/steady_timer.hpp>
 #include <thread>
 
 namespace asio {
@@ -36,31 +38,34 @@ class OPENDHT_PUBLIC PeerDiscovery
 {
 public:
     static constexpr in_port_t DEFAULT_PORT = 8888;
-    using ServiceDiscoveredCallback = std::function<void(msgpack::object&&, SockAddr&&)>;
+    using ServiceDiscoveredCallback = std::function<void(msgpack::object &&, SockAddr &&)>;
 
-    PeerDiscovery(in_port_t port = DEFAULT_PORT, std::shared_ptr<asio::io_context> ioContext = {}, std::shared_ptr<Logger> logger = {});
+    PeerDiscovery(in_port_t port = DEFAULT_PORT,
+                std::shared_ptr<asio::io_context> ioContext = {},
+                std::shared_ptr<Logger> logger = {});
     ~PeerDiscovery();
 
     /**
-     * startDiscovery - Keep Listening data from the sender until node is joinned or stop is called
-    */
+     * startDiscovery - Keep Listening data from the sender until node is joinned
+     * or stop is called
+     */
     void startDiscovery(const std::string &type, ServiceDiscoveredCallback callback);
 
-    template<typename T>
-    void startDiscovery(const std::string &type, std::function<void(T&&, SockAddr&&)> cb) {
-        startDiscovery(type, [cb](msgpack::object&& ob, SockAddr&& addr) {
+    template <typename T>
+    void startDiscovery(const std::string &type, std::function<void(T &&, SockAddr &&)> cb) {
+        startDiscovery(type, [cb](msgpack::object &&ob, SockAddr &&addr) {
             cb(ob.as<T>(), std::move(addr));
         });
     }
 
     /**
      * startPublish - Keeping sending data until node is joinned or stop is called
-    */
+     */
     void startPublish(const std::string &type, const msgpack::sbuffer &pack_buf);
     void startPublish(sa_family_t domain, const std::string &type, const msgpack::sbuffer &pack_buf);
 
-    template<typename T>
-    void startPublish(const std::string &type, const T& object) {
+    template <typename T>
+    void startPublish(const std::string &type, const T &object) {
         msgpack::sbuffer buf;
         msgpack::pack(buf, object);
         startPublish(type, buf);
@@ -68,22 +73,24 @@ public:
 
     /**
      * Thread Stopper
-    */
+     */
     void stop();
 
     /**
      * Remove possible callBack to discovery
-    */
+     */
     bool stopDiscovery(const std::string &type);
 
     /**
      * Remove different serivce message to send
-    */
+     */
     bool stopPublish(const std::string &type);
     bool stopPublish(sa_family_t domain, const std::string &type);
 
     void connectivityChanged();
 
+    void stopConnectivityChanged();
+
 private:
     class DomainPeerDiscovery;
     std::unique_ptr<DomainPeerDiscovery> peerDiscovery4_;
@@ -92,4 +99,4 @@ private:
     std::thread ioRunnner_;
 };
 
-}
+} // namespace dht

meson.build

@@ -13,10 +13,16 @@ openssl = dependency('openssl', required: get_option('proxy_client'))
 jsoncpp = dependency('jsoncpp', required: get_option('proxy_client'))
 fmt = dependency('fmt')
 llhttp = dependency('llhttp', 'libllhttp', required: get_option('proxy_client'))
+io_uring = dependency('liburing', required: false)
 
-deps = [fmt, gnutls, nettle, msgpack, argon2, openssl, jsoncpp, llhttp]
+deps = [fmt, gnutls, nettle, msgpack, argon2, openssl, jsoncpp, llhttp, io_uring]
+conf_data = configuration_data()
 
-add_project_arguments('-DMSGPACK_NO_BOOST', language : 'cpp')
+add_project_arguments('-DMSGPACK_NO_BOOST', '-DASIO_STANDALONE', language : 'cpp')
+if io_uring.found()
+    add_project_arguments('-DASIO_HAS_IO_URING', '-DASIO_DISABLE_EPOLL', language : 'cpp')
+    conf_data.set('iouring_lib', ', liburing')
+endif
 add_project_arguments(['-Wno-return-type','-Wno-deprecated','-Wnon-virtual-dtor','-pedantic-errors','-fvisibility=hidden'], language : 'cpp')
 
 opendht_inc = include_directories('include/opendht')
@@ -40,8 +46,6 @@ opendht_src = [
     'src/thread_pool.cpp',
 ]
 
-conf_data = configuration_data()
-
 if get_option('indexation').enabled()
     opendht_src += 'src/indexation/pht.cpp'
     add_project_arguments('-DOPENDHT_INDEXATION', language : 'cpp')

opendht.pc.in

@@ -8,5 +8,5 @@ Version: @VERSION@
 Libs: -L${libdir} -lopendht
 Libs.private: @http_lib@ -pthread
 Requires: gnutls >= 3.3@jsoncpp_lib@@openssl_lib@
-Requires.private: nettle >= 2.4@argon2_lib@
+Requires.private: nettle >= 2.4@argon2_lib@@iouring_lib@
 Cflags: -I${includedir}

python/opendht.pyx

@@ -500,12 +500,13 @@ cdef class DhtConfig(object):
     def setRateLimit(self, ssize_t max_req_per_sec, ssize_t max_peer_req_per_sec):
         self._config.dht_config.node_config.max_req_per_sec = max_req_per_sec
         self._config.dht_config.node_config.max_peer_req_per_sec = max_peer_req_per_sec
-    def setProxyInfo(self, str proxy_server, str push_node_id="", str push_token="", str push_topic="", str push_platform="", Certificate server_ca = Certificate()):
+    def setProxyInfo(self, str proxy_server, str push_node_id="", str push_token="", str push_topic="", str push_platform="", str user_agent="", Certificate server_ca = Certificate()):
         self._config.proxy_server = proxy_server.encode()
         self._config.push_node_id = push_node_id.encode()
         self._config.push_token = push_token.encode()
         self._config.push_topic = push_topic.encode()
         self._config.push_platform = push_platform.encode()
+        self._config.proxy_user_agent = user_agent.encode()
         self._config.server_ca = server_ca._cert
     def setPeerDiscovery(self, bool peer_discovery, bool peer_publish):
         self._config.peer_discovery = peer_discovery

python/opendht_cpp.pxd

@@ -246,6 +246,7 @@ cdef extern from "opendht/dhtrunner.h" namespace "dht":
             string push_token
             string push_topic
             string push_platform
+            string proxy_user_agent
             bool peer_discovery
             bool peer_publish
             shared_ptr[Certificate] server_ca

rust/Cargo.toml

@@ -2,10 +2,10 @@
 name = "opendht"
 version = "0.1.0"
 authors = ["Sébastien Blin <sebastien.blin@savoirfairelinux.com>"]
-edition = "2018"
-
-# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
+edition = "2021"
 
 [dependencies]
-libc="0.2.0"
-os_socketaddr="0.2.3"
+libc="0.2"
+os_socketaddr="0.2.3"
+
+# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html

rust/src/dhtrunner.rs

@@ -49,6 +49,7 @@ impl DhtRunnerConfig {
             push_token: ptr::null(),
             push_topic: ptr::null(),
             push_platform: ptr::null(),
+            proxy_user_agent: ptr::null(),
             peer_discovery: false,
             peer_publish: false,
             server_ca: ptr::null_mut(),
@@ -56,6 +57,7 @@ impl DhtRunnerConfig {
                 privatekey: ptr::null_mut(),
                 certificate: ptr::null_mut(),
             },
+            log: true,
         });
         unsafe {
             dht_runner_config_default(&mut *config);
@@ -64,23 +66,27 @@ impl DhtRunnerConfig {
     }
 
     pub fn set_proxy_server(&mut self, proxy_server: &str) {
-        self.proxy_server = CString::new(proxy_server).unwrap().as_ptr();
+        self.proxy_server = CString::new(proxy_server).unwrap().into_raw();
     }
 
     pub fn set_push_node_id(&mut self, push_node_id: &str) {
-        self.push_node_id = CString::new(push_node_id).unwrap().as_ptr();
+        self.push_node_id = CString::new(push_node_id).unwrap().into_raw();
     }
 
     pub fn set_push_token(&mut self, push_token: &str) {
-        self.push_token = CString::new(push_token).unwrap().as_ptr();
+        self.push_token = CString::new(push_token).unwrap().into_raw();
     }
 
     pub fn set_push_topic(&mut self, push_topic: &str) {
-        self.push_topic = CString::new(push_topic).unwrap().as_ptr();
+        self.push_topic = CString::new(push_topic).unwrap().into_raw();
     }
 
     pub fn set_push_platform(&mut self, push_platform: &str) {
-        self.push_platform = CString::new(push_platform).unwrap().as_ptr();
+        self.push_platform = CString::new(push_platform).unwrap().into_raw();
+    }
+
+    pub fn set_proxy_user_agent(&mut self, proxy_user_agent: &str) {
+        self.proxy_user_agent = CString::new(proxy_user_agent).unwrap().into_raw();
     }
 
     pub fn set_identity(&mut self, certificate: Box<DhtCertificate>, privatekey: Box<PrivateKey>) {
@@ -93,7 +99,7 @@ impl DhtRunnerConfig {
 impl DhtNodeConfig
 {
     pub fn set_persist_path(&mut self, persist_path: &str) {
-        self.persist_path = CString::new(persist_path).unwrap().as_ptr();
+        self.persist_path = CString::new(persist_path).unwrap().into_raw();
     }
 }
 
@@ -197,8 +203,8 @@ impl DhtRunner {
     pub fn bootstrap(&mut self, host: &str, service: u16) {
         unsafe {
             dht_runner_bootstrap(&mut *self,
-                CString::new(host).unwrap().as_ptr(),
-                CString::new(service.to_string()).unwrap().as_ptr())
+                CString::new(host).unwrap().into_raw(),
+                CString::new(service.to_string()).unwrap().into_raw())
         }
     }
 

rust/src/ffi.rs

@@ -120,13 +120,14 @@ pub struct DhtRunnerConfig
     pub push_token: *const c_char,
     pub push_topic: *const c_char,
     pub push_platform: *const c_char,
+    pub proxy_user_agent: *const c_char,
     pub peer_discovery: bool,
     pub peer_publish: bool,
     pub server_ca: *mut DhtCertificate,
     pub client_identity: DhtIdentity,
+    pub log: bool,
 }
 
-
 #[link(name = "opendht-c")]
 extern {
     // dht::Value

src/compat/getif_workaround_android.h

@@ -22,7 +22,7 @@ get_interface(){
         if (if1->ifa_addr->sa_family != AF_INET)
             continue;
         if((if1->ifa_flags & IFF_MULTICAST) == IFF_MULTICAST){
-            char* sa_data = if1->ifa_addr->sa_data;
+            auto sa_data = reinterpret_cast<uint8_t*>(if1->ifa_addr->sa_data);
             uint32_t value = sa_data[2] << 24
                 | sa_data[3] << 16
                 | sa_data[4] << 8
@@ -36,4 +36,4 @@ get_interface(){
 }
 
 } /* namespace workaround */
-} /* namespace dht */
+} /* namespace dht */

src/crypto.cpp

@@ -93,11 +93,27 @@ Blob aesEncrypt(const uint8_t* data, size_t data_length, const Blob& key)
         std::random_device rdev;
         std::generate_n(ret.begin(), GCM_IV_SIZE, std::bind(rand_byte, std::ref(rdev)));
     }
-    struct gcm_aes_ctx aes;
-    gcm_aes_set_key(&aes, key.size(), key.data());
-    gcm_aes_set_iv(&aes, GCM_IV_SIZE, ret.data());
-    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);
+
+    if (key.size() == AES_LENGTHS[0]) {
+        struct gcm_aes128_ctx aes;
+        gcm_aes128_set_key(&aes, key.data());
+        gcm_aes128_set_iv(&aes, GCM_IV_SIZE, ret.data());
+        gcm_aes128_encrypt(&aes, data_length, ret.data() + GCM_IV_SIZE, data);
+        gcm_aes128_digest(&aes, GCM_DIGEST_SIZE, ret.data() + GCM_IV_SIZE + data_length);
+    } else if (key.size() == AES_LENGTHS[1]) {
+        struct gcm_aes192_ctx aes;
+        gcm_aes192_set_key(&aes, key.data());
+        gcm_aes192_set_iv(&aes, GCM_IV_SIZE, ret.data());
+        gcm_aes192_encrypt(&aes, data_length, ret.data() + GCM_IV_SIZE, data);
+        gcm_aes192_digest(&aes, GCM_DIGEST_SIZE, ret.data() + GCM_IV_SIZE + data_length);
+    } else if (key.size() == AES_LENGTHS[2]) {
+        struct gcm_aes256_ctx aes;
+        gcm_aes256_set_key(&aes, key.data());
+        gcm_aes256_set_iv(&aes, GCM_IV_SIZE, ret.data());
+        gcm_aes256_encrypt(&aes, data_length, ret.data() + GCM_IV_SIZE, data);
+        gcm_aes256_digest(&aes, GCM_DIGEST_SIZE, ret.data() + GCM_IV_SIZE + data_length);
+    }
+
     return ret;
 }
 
@@ -118,14 +134,28 @@ Blob aesDecrypt(const uint8_t* data, size_t data_length, const Blob& key)
 
     std::array<uint8_t, GCM_DIGEST_SIZE> digest;
 
-    struct gcm_aes_ctx aes;
-    gcm_aes_set_key(&aes, key.size(), key.data());
-    gcm_aes_set_iv(&aes, GCM_IV_SIZE, data);
-
     size_t data_sz = data_length - GCM_IV_SIZE - GCM_DIGEST_SIZE;
     Blob ret(data_sz);
-    gcm_aes_decrypt(&aes, data_sz, ret.data(), data + GCM_IV_SIZE);
-    gcm_aes_digest(&aes, GCM_DIGEST_SIZE, digest.data());
+
+    if (key.size() == AES_LENGTHS[0]) {
+        struct gcm_aes128_ctx aes;
+        gcm_aes128_set_key(&aes, key.data());
+        gcm_aes128_set_iv(&aes, GCM_IV_SIZE, data);
+        gcm_aes128_decrypt(&aes, data_sz, ret.data(), data + GCM_IV_SIZE);
+        gcm_aes128_digest(&aes, GCM_DIGEST_SIZE, digest.data());
+    } else if (key.size() == AES_LENGTHS[1]) {
+        struct gcm_aes192_ctx aes;
+        gcm_aes192_set_key(&aes, key.data());
+        gcm_aes192_set_iv(&aes, GCM_IV_SIZE, data);
+        gcm_aes192_decrypt(&aes, data_sz, ret.data(), data + GCM_IV_SIZE);
+        gcm_aes192_digest(&aes, GCM_DIGEST_SIZE, digest.data());
+    } else if (key.size() == AES_LENGTHS[2]) {
+        struct gcm_aes256_ctx aes;
+        gcm_aes256_set_key(&aes, key.data());
+        gcm_aes256_set_iv(&aes, GCM_IV_SIZE, data);
+        gcm_aes256_decrypt(&aes, data_sz, ret.data(), data + GCM_IV_SIZE);
+        gcm_aes256_digest(&aes, GCM_DIGEST_SIZE, digest.data());
+    }
 
     if (not std::equal(digest.begin(), digest.end(), data + data_length - GCM_DIGEST_SIZE)) {
         throw DecryptError("Can't decrypt data");
@@ -227,7 +257,7 @@ PrivateKey::PrivateKey(const uint8_t* src, size_t src_size, const char* password
         throw CryptoException("Can't initialize private key !");
 
     const gnutls_datum_t dt {(uint8_t*)src, static_cast<unsigned>(src_size)};
-    int flags = password_ptr ? GNUTLS_PKCS_PLAIN
+    int flags = (password_ptr == NULL || strlen(password_ptr)==0) ? GNUTLS_PKCS_PLAIN
                 : ( GNUTLS_PKCS_PBES2_AES_128 | GNUTLS_PKCS_PBES2_AES_192  | GNUTLS_PKCS_PBES2_AES_256
                   | GNUTLS_PKCS_PKCS12_3DES   | GNUTLS_PKCS_PKCS12_ARCFOUR | GNUTLS_PKCS_PKCS12_RC2_40);
 
@@ -1134,11 +1164,17 @@ saveIdentity(const Identity& id, const std::string& path, const std::string& pri
         auto ca_key_data = id.first->serialize(privkey_password);
         std::ofstream key_file(path + ".pem");
         key_file.write((char*)ca_key_data.data(), ca_key_data.size());
+        // Throw error if the file is not written
+        if (!key_file)
+            throw CryptoException("Could not write private key file");
     }
     {
         auto ca_key_data = id.second->getPacked();
         std::ofstream crt_file(path + ".crt");
         crt_file.write((char*)ca_key_data.data(), ca_key_data.size());
+        // Throw error if the file is not written
+        if (!crt_file)
+            throw CryptoException("Could not write certificate file");
     }
 }
 
@@ -1152,12 +1188,19 @@ loadIdentity(const std::string &path,const std::string &privkey_password)
     pkStream.close();
     // Create a certificate
     gnutls_x509_crt_t gnuCert;
-    if (gnutls_x509_crt_init(&gnuCert) != GNUTLS_E_SUCCESS)
+    if (gnutls_x509_crt_init(&gnuCert))
         throw std::runtime_error("Failed to initialize gnutls certificate struct");
     gnutls_datum_t crtContent {nullptr, 0};
+
     // Read the certificate file
-    gnutls_load_file((path + ".crt").c_str(), &crtContent);
-    gnutls_x509_crt_import(gnuCert, &crtContent, GNUTLS_X509_FMT_PEM);
+    int err = gnutls_load_file((path + ".crt").c_str(), &crtContent);
+    if (err)
+        throw CryptoException(gnutls_strerror(err));
+
+    err = gnutls_x509_crt_import(gnuCert, &crtContent, GNUTLS_X509_FMT_PEM);
+    if (err)
+        throw CryptoException(gnutls_strerror(err));
+
     auto cert = std::make_shared<Certificate>(gnuCert);
     return {std::move(key), std::move(cert)};
 }
@@ -1204,31 +1247,49 @@ Certificate::generate(const PrivateKey& key, const std::string& name, const Iden
     auto pk_id = pk.getId();
     const std::string uid_str = pk_id.toString();
 
-    gnutls_x509_crt_set_subject_key_id(cert, &pk_id, sizeof(pk_id));
-    gnutls_x509_crt_set_dn_by_oid(cert, GNUTLS_OID_X520_COMMON_NAME, 0, name.data(), name.length());
-    gnutls_x509_crt_set_dn_by_oid(cert, GNUTLS_OID_LDAP_UID, 0, uid_str.data(), uid_str.length());
+    int err = gnutls_x509_crt_set_subject_key_id(cert, &pk_id, sizeof(pk_id));
+    if(err) {
+        throw CryptoException(std::string("Error when setting subject key id ") + gnutls_strerror(err));
+    }
+
+    err = gnutls_x509_crt_set_dn_by_oid(cert, GNUTLS_OID_X520_COMMON_NAME, 0, name.data(), name.length());
+    if(err) {
+        throw CryptoException(std::string("Error when setting subject key id ") + gnutls_strerror(err));
+    }
+    err = gnutls_x509_crt_set_dn_by_oid(cert, GNUTLS_OID_LDAP_UID, 0, uid_str.data(), uid_str.length());
+    if(err) {
+        throw CryptoException(std::string("Error when setting dn by oid ") + gnutls_strerror(err));
+    }
 
     setRandomSerial(cert);
 
     unsigned key_usage = 0;
     if (is_ca) {
-        gnutls_x509_crt_set_ca_status(cert, 1);
+        err = gnutls_x509_crt_set_ca_status(cert, 1);
+        if(err) {
+            throw CryptoException(std::string("Error when setting ca status ") + gnutls_strerror(err));
+        }
         key_usage |= GNUTLS_KEY_KEY_CERT_SIGN | GNUTLS_KEY_CRL_SIGN;
     } else {
         key_usage |= GNUTLS_KEY_DIGITAL_SIGNATURE | GNUTLS_KEY_DATA_ENCIPHERMENT;
     }
-    gnutls_x509_crt_set_key_usage(cert, key_usage);
+    err = gnutls_x509_crt_set_key_usage(cert, key_usage);
+    if(err) {
+        throw CryptoException(std::string("Error when setting ca status ") + gnutls_strerror(err));
+    }
 
     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, pk.getPreferredDigest(), 0)) {
+        err = gnutls_x509_crt_privkey_sign(cert, ca.second->cert, ca.first->key, pk.getPreferredDigest(), 0);
+        if (err) {
             throw CryptoException(std::string("Error when signing certificate ") + gnutls_strerror(err));
         }
         ret.issuer = ca.second;
     } else {
-        if (int err = gnutls_x509_crt_privkey_sign(cert, cert, key.key, pk.getPreferredDigest(), 0)) {
+        err = gnutls_x509_crt_privkey_sign(cert, cert, key.key, pk.getPreferredDigest(), 0);
+        if (err) {
             throw CryptoException(std::string("Error when signing certificate ") + gnutls_strerror(err));
         }
     }

src/dht.cpp

@@ -93,7 +93,7 @@ Dht::shutdown(ShutdownCallback cb, bool stop)
                     r.second.done_cb(false, {});
                 sr.second->callbacks.clear();
                 for (const auto& a : sr.second->announce) {
-                    if (a.callback) a.callback(false, {});
+                    for (auto& cb : a.callbacks) cb(false, {});
                 }
                 sr.second->announce.clear();
                 sr.second->listeners.clear();

src/dht_proxy_client.cpp

@@ -106,13 +106,26 @@ getRandomSessionId(size_t length = 8) {
 DhtProxyClient::DhtProxyClient() {}
 
 DhtProxyClient::DhtProxyClient(
-        std::shared_ptr<dht::crypto::Certificate> serverCA, dht::crypto::Identity clientIdentity,
-        std::function<void()> signal, const std::string& serverHost,
-        const std::string& pushClientId, std::shared_ptr<dht::Logger> logger)
+        std::shared_ptr<dht::crypto::Certificate> serverCA,
+        dht::crypto::Identity clientIdentity,
+        std::function<void()> signal,
+        const std::string& serverHost,
+        const std::string& userAgent,
+        const std::string& pushClientId,
+        const std::string& pushToken,
+        const std::string& pushTopic,
+        const std::string& pushPlatform,
+        std::shared_ptr<dht::Logger> logger
+)
     : DhtInterface(logger)
     , proxyUrl_(serverHost)
     , clientIdentity_(clientIdentity), serverCertificate_(serverCA)
-    , pushClientId_(pushClientId), pushSessionId_(getRandomSessionId())
+    , userAgent_(userAgent)
+    , pushClientId_(pushClientId)
+    , pushSessionId_(getRandomSessionId())
+    , deviceKey_(pushToken)
+    , notificationTopic_(pushTopic)
+    , platform_(pushPlatform)
     , loopSignal_(signal)
     , jsonReader_(Json::CharReaderBuilder{}.newCharReader())
 {
@@ -484,7 +497,7 @@ DhtProxyClient::buildRequest(const std::string& target)
         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");
+    request->set_header_field(restinio::http_field_t::user_agent, userAgent_);
     return request;
 }
 
@@ -1108,10 +1121,12 @@ DhtProxyClient::sendListen(const restinio::http_request_header_t& header,
             std::lock_guard<std::mutex> l(requestLock_);
             requests_[reqid] = request;
         }
-        request->add_on_status_callback([request, seconds = this->listenKeepIdle()] (unsigned status_code) {
+        request->add_on_status_callback([r=std::weak_ptr(request), seconds = this->listenKeepIdle()] (unsigned status_code) {
             if(status_code == 200) {
                 // increase TCP_KEEPIDLE to save power
-                request->get_connection()->set_keepalive(seconds);
+                if (auto request = r.lock()) {
+                    request->get_connection()->set_keepalive(seconds);
+                }
             }
         });
         request->send();
@@ -1381,4 +1396,24 @@ DhtProxyClient::fillBody(bool resubscribe)
 }
 #endif // OPENDHT_PUSH_NOTIFICATIONS
 
+void
+DhtProxyClient::setPushNotificationToken([[maybe_unused]] const std::string& token) {
+#ifdef OPENDHT_PUSH_NOTIFICATIONS
+    std::unique_lock<std::mutex> l(lockCurrentProxyInfos_);
+    if (deviceKey_ != token) {
+        deviceKey_ = token;
+        if (statusIpv4_ == NodeStatus::Connected || statusIpv6_ == NodeStatus::Connected) {
+            if (logger_)
+                logger_->d("[proxy:client] [push] token changed, resubscribing");
+            for (auto& search : searches_) {
+                for (auto& listener : search.second.listeners) {
+                    resubscribe(search.first, listener.first, listener.second);
+                }
+            }
+        }
+    }
+#endif
+}
+
+
 } // namespace dht

src/dht_proxy_server.cpp

@@ -223,6 +223,7 @@ DhtProxyServer::DhtProxyServer(const std::shared_ptr<DhtRunner>& dht,
     :   ioContext_(std::make_shared<asio::io_context>()),
         dht_(dht), persistPath_(config.persistStatePath), logger_(logger),
         printStatsTimer_(std::make_unique<asio::steady_timer>(*ioContext_, 3s)),
+        serverStartTime_(clock::now()),
         connListener_(std::make_shared<ConnectionListener>(std::bind(&DhtProxyServer::onConnectionClosed, this, std::placeholders::_1))),
         pushServer_(config.pushServer),
         bundleId_(config.bundleId)
@@ -542,7 +543,15 @@ DhtProxyServer::updateStats(std::shared_ptr<NodeInfo> info) const
     stats.requestRate = count / dt.count();
 #ifdef OPENDHT_PUSH_NOTIFICATIONS
     stats.pushListenersCount = pushListeners_.size();
+    {
+        std::lock_guard lk(pushStatsMutex_);
+        stats.androidPush = androidPush_;
+        stats.iosPush = iosPush_;
+        stats.unifiedPush = unifiedPush_;
+    }
 #endif
+    stats.serverStartTime = serverStartTime_;
+    stats.lastUpdated = now;
     stats.totalPermanentPuts = 0;
     std::for_each(puts_.begin(), puts_.end(), [&stats](const auto& put) {
         stats.totalPermanentPuts += put.second.puts.size();
@@ -1107,10 +1116,14 @@ DhtProxyServer::sendPushNotification(const std::string& token, Json::Value&& jso
             notification["tokens"] = std::move(tokens);
             notification["platform"] = type == PushType::Android ? 2 : 1;
             notification["data"] = std::move(json);
-            notification["priority"] = highPriority ? "high" : "normal";
-            if (type == PushType::Android)
-                notification["time_to_live"] = 3600 * 24; // 24 hours
-            else {
+            auto priority = highPriority ? "high" : "normal";
+            if (type == PushType::Android) {
+                Json::Value androidConfig(Json::objectValue);
+                androidConfig["priority"] = priority;
+                androidConfig["ttl"] = "86400s"; // time to live = 24 hours
+                notification["android"] = std::move(androidConfig);
+            } else {
+                notification["priority"] = priority;
                 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;
@@ -1151,6 +1164,21 @@ DhtProxyServer::sendPushNotification(const std::string& token, Json::Value&& jso
             requests_[reqid] = request;
         }
         request->send();
+        // For monitoring purposes
+        std::lock_guard lk(pushStatsMutex_);
+        switch (type) {
+        case PushType::Android:
+            androidPush_.increment(highPriority);
+            break;
+        case PushType::iOS:
+            iosPush_.increment(highPriority);
+            break;
+        case PushType::UnifiedPush:
+            unifiedPush_.increment(highPriority);
+            break;
+        default:
+            break;
+        }
     }
     catch (const std::exception &e){
         if (logger_)

src/dhtrunner.cpp

@@ -180,9 +180,9 @@ DhtRunner::run(const Config& config, Context&& context)
         throw;
     }
 
-    if (context.statusChangedCallback) {
-        statusCb = std::move(context.statusChangedCallback);
-    }
+    statusCbs.clear();
+    if (context.statusChangedCallback)
+        statusCbs.emplace_back(std::move(context.statusChangedCallback));
     if (context.certificateStore) {
         dht_->setLocalCertificateStore(std::move(context.certificateStore));
     }
@@ -264,6 +264,17 @@ DhtRunner::run(const Config& config, Context&& context)
                 }
             }
         }
+        if (peerDiscovery_) {
+            statusCbs.emplace_back([this](NodeStatus status4, NodeStatus status6) {
+                if (status4 == NodeStatus::Disconnected && status6 == NodeStatus::Disconnected) {
+                    peerDiscovery_->connectivityChanged();
+                }
+                else if (status4 != NodeStatus::Connected || status6 != NodeStatus::Connected) {
+                    peerDiscovery_->stopConnectivityChanged();
+                }
+
+            });
+        }
 #endif
     }
 }
@@ -690,8 +701,9 @@ DhtRunner::loop_()
     if (nstatus4 != status4 || nstatus6 != status6) {
         status4 = nstatus4;
         status6 = nstatus6;
-        if (statusCb)
-            statusCb(status4, status6);
+        for (auto& cb : statusCbs){
+            cb(status4, status6);
+        }
     }
 
     return wakeup;
@@ -1108,13 +1120,13 @@ DhtRunner::enableProxy(bool proxify)
                         cv.notify_all();
                     }
                 },
-                config_.proxy_server, config_.push_node_id, logger_);
-        if (not config_.push_token.empty())
-            dht_via_proxy->setPushNotificationToken(config_.push_token);
-        if (not config_.push_topic.empty())
-            dht_via_proxy->setPushNotificationTopic(config_.push_topic);
-        if (not config_.push_platform.empty())
-            dht_via_proxy->setPushNotificationPlatform(config_.push_platform);
+                config_.proxy_server,
+                config_.proxy_user_agent,
+                config_.push_node_id,
+                config_.push_token,
+                config_.push_topic,
+                config_.push_platform,
+                logger_);
         dht_ = std::make_unique<SecureDht>(std::move(dht_via_proxy), config_.dht_config, identityAnnouncedCb_, logger_);
         // and use it
         use_proxy = proxify;

src/http.cpp

@@ -461,7 +461,7 @@ Connection::set_ssl_verification(const std::string& hostname, const asio::ssl::v
                     bool verified = verifier(preverified, ctx);
                     auto verify_ec = X509_STORE_CTX_get_error(ctx.native_handle());
                     if (verify_ec != 0 /*X509_V_OK*/ and logger)
-                        logger->error("[http::connection:{:d}] ssl verification error={:d} %d", id, verify_ec, verified);
+                        logger->error("[http::connection:{:d}] ssl verification error={:d} {}", id, verify_ec, verified);
                     if (verified and checkOcsp) {
                         std::unique_ptr<stack_st_X509, void(*)(stack_st_X509*)> chain(
                             X509_STORE_CTX_get1_chain(ctx.native_handle()),

src/peer_discovery.cpp

@@ -52,6 +52,7 @@ public:
 
     void connectivityChanged();
 
+    void stopConnectivityChanged();
 private:
     Sp<Logger> logger_;
     //dmtx_ for callbackmap_ and drunning_ (write)
@@ -59,6 +60,14 @@ private:
     //mtx_ for messages_ and lrunning (listen)
     std::mutex mtx_;
     std::shared_ptr<asio::io_context> ioContext_;
+
+    static constexpr dht::duration PeerDiscovery_PERIOD_MAX{
+        std::chrono::minutes(1)};
+    static constexpr std::chrono::seconds PeerDiscovery_PERIOD{10};
+    asio::steady_timer peerDiscoveryTimer;
+    std::chrono::steady_clock::duration peerDiscovery_period{
+        PeerDiscovery_PERIOD};
+
     asio::ip::udp::socket sockFd_;
     asio::ip::udp::endpoint sockAddrSend_;
 
@@ -86,6 +95,7 @@ private:
 PeerDiscovery::DomainPeerDiscovery::DomainPeerDiscovery(asio::ip::udp domain, in_port_t port, Sp<asio::io_context> ioContext, Sp<Logger> logger)
     : logger_(logger)
     , ioContext_(ioContext)
+    , peerDiscoveryTimer(*ioContext_)
     , sockFd_(*ioContext_, domain)
     , sockAddrSend_(asio::ip::address::from_string(domain.family() == AF_INET ? MULTICAST_ADDRESS_IPV4
                                                                               : MULTICAST_ADDRESS_IPV6), port)
@@ -317,8 +327,30 @@ PeerDiscovery::DomainPeerDiscovery::reDiscover()
 void
 PeerDiscovery::DomainPeerDiscovery::connectivityChanged()
 {
-    reDiscover();
-    publish(sockAddrSend_);
+    ioContext_->post([this] () {
+        reDiscover();
+        publish(sockAddrSend_);
+    });
+    if (logger_)
+        logger_->d("PeerDiscovery: connectivity changed");
+        
+    if (peerDiscovery_period == PeerDiscovery_PERIOD_MAX ){
+        peerDiscovery_period = PeerDiscovery_PERIOD;
+    }
+    else{
+        peerDiscoveryTimer.expires_after(peerDiscovery_period);
+        peerDiscoveryTimer.async_wait([this](const asio::error_code& ec) {
+            if (ec == asio::error::operation_aborted)
+                return;
+            connectivityChanged();
+        });
+        peerDiscovery_period = std::min(peerDiscovery_period * 2, PeerDiscovery_PERIOD_MAX);
+    }
+}
+
+void PeerDiscovery::DomainPeerDiscovery::stopConnectivityChanged() {
+    peerDiscoveryTimer.cancel();
+    peerDiscovery_period = PeerDiscovery_PERIOD;
 }
 
 PeerDiscovery::PeerDiscovery(in_port_t port, Sp<asio::io_context> ioContext, Sp<Logger> logger)
@@ -431,4 +463,11 @@ PeerDiscovery::connectivityChanged()
         peerDiscovery6_->connectivityChanged();
 }
 
+void PeerDiscovery::stopConnectivityChanged() {
+   if (peerDiscovery4_)
+       peerDiscovery4_->stopConnectivityChanged();
+    if (peerDiscovery6_)
+        peerDiscovery6_->stopConnectivityChanged();
+}
+
 } /* namespace dht */

src/search.h

@@ -45,7 +45,7 @@ struct Dht::Announce {
     bool permanent;
     Sp<Value> value;
     time_point created;
-    DoneCallback callback;
+    std::vector<DoneCallback> callbacks;
 };
 
 struct Dht::SearchNode {
@@ -442,8 +442,9 @@ struct Dht::Search {
             g.second.done_cb = {};
         }
         for (auto& a : announce) {
-            a.callback(false, {});
-            a.callback = {};
+            for (auto& cb : a.callbacks)
+                cb(false, {});
+            a.callbacks.clear();
         }
     }
 
@@ -627,7 +628,9 @@ struct Dht::Search {
             return a.value->id == value->id;
         });
         if (a_sr == announce.end()) {
-            announce.emplace_back(Announce {permanent, value, created, callback});
+            auto& a = announce.emplace_back(Announce {permanent, value, created, {}} );
+            if (callback)
+                a.callbacks.emplace_back(std::move(callback));
             for (auto& n : nodes) {
                 n->probe_query.reset();
                 n->acked[value->id].req.reset();
@@ -635,24 +638,26 @@ struct Dht::Search {
         } else {
             a_sr->permanent = permanent;
             a_sr->created = created;
-            if (a_sr->value != value) {
+            if (a_sr->value != value && *a_sr->value != *value) {
+                // Value is updated, previous ops are failed
                 a_sr->value = value;
                 for (auto& n : nodes) {
                     n->acked[value->id].req.reset();
                     n->probe_query.reset();
                 }
-            }
-            if (isAnnounced(value->id)) {
-                if (a_sr->callback)
-                    a_sr->callback(true, {});
-                a_sr->callback = {};
+                for (auto& cb: a_sr->callbacks)
+                    cb(false, {});
+                a_sr->callbacks.clear();
+                if (callback)
+                    a_sr->callbacks.emplace_back(std::move(callback));
+            } else if (isAnnounced(value->id)) {
+                // Same value, already announced
                 if (callback)
                     callback(true, {});
-                return;
             } else {
-                if (a_sr->callback)
-                    a_sr->callback(false, {});
-                a_sr->callback = callback;
+                // Same value, not announced yet
+                if (callback)
+                    a_sr->callbacks.emplace_back(std::move(callback));
             }
         }
     }
@@ -722,8 +727,8 @@ struct Dht::Search {
             std::vector<DoneCallback> a_cbs;
             a_cbs.reserve(announce.size());
             for (auto ait = announce.begin() ; ait != announce.end(); ) {
-                if (ait->callback)
-                    a_cbs.emplace_back(std::move(ait->callback));
+                a_cbs.insert(a_cbs.end(), std::make_move_iterator(ait->callbacks.begin()), std::make_move_iterator(ait->callbacks.end()));
+                ait->callbacks.clear();
                 if (not ait->permanent)
                     ait = announce.erase(ait);
                 else
@@ -747,9 +752,11 @@ struct Dht::Search {
                 if (vid != Value::INVALID_ID and (!a.value || a.value->id != vid))
                     return true;
                 if (isAnnounced(a.value->id)) {
-                    if (a.callback) {
-                        a.callback(true, getNodes());
-                        a.callback = nullptr;
+                    if (!a.callbacks.empty()) {
+                        const auto& nodes = getNodes();
+                        for (auto& cb : a.callbacks)
+                            cb(true, nodes);
+                        a.callbacks.clear();
                     }
                     if (not a.permanent)
                         return false;

src/value.cpp

@@ -271,6 +271,7 @@ Value::encrypt(const crypto::PrivateKey& from, const crypto::PublicKey& to)
     setRecipient(to.getId());
     sign(from);
     Value nv {id};
+    nv.priority = priority;
     nv.setCypher(to.encrypt(getToEncrypt()));
     return nv;
 }

tests/cryptotester.cpp

@@ -239,6 +239,27 @@ void CryptoTester::testAesEncryption() {
     CPPUNIT_ASSERT(data2 == decrypted2);
 }
 
+void CryptoTester::testAesEncryptionWithMultipleKeySizes() {
+    auto data = std::vector<uint8_t>(rand(), rand());
+
+    // Valid key sizes
+    for (auto key_length : {16, 24, 32}) {
+        auto key = std::vector<uint8_t>(key_length, rand());
+
+        auto encrypted_data = dht::crypto::aesEncrypt(data, key);
+        auto decrypted_data = dht::crypto::aesDecrypt(encrypted_data, key);
+
+        CPPUNIT_ASSERT(data == decrypted_data);
+    }
+
+    // Invalid key sizes
+    for (auto key_length : {12, 28, 36}) {
+        auto key = std::vector<uint8_t>(key_length, rand());
+
+        CPPUNIT_ASSERT_THROW(dht::crypto::aesEncrypt(data, key), dht::crypto::DecryptError);
+    }
+}
+
 void
 CryptoTester::tearDown() {
 

tests/cryptotester.h

@@ -34,6 +34,7 @@ class CryptoTester : public CppUnit::TestFixture {
     CPPUNIT_TEST(testCertificateSerialNumber);
     CPPUNIT_TEST(testOcsp);
     CPPUNIT_TEST(testAesEncryption);
+    CPPUNIT_TEST(testAesEncryptionWithMultipleKeySizes);
     CPPUNIT_TEST_SUITE_END();
 
  public:
@@ -69,6 +70,7 @@ class CryptoTester : public CppUnit::TestFixture {
      * Test key streching and aes encryption/decryption
      */
     void testAesEncryption();
+    void testAesEncryptionWithMultipleKeySizes();
 };
 
 }  // namespace test

tests/dhtrunnertester.cpp

@@ -90,6 +90,59 @@ DhtRunnerTester::testGetPut() {
     CPPUNIT_ASSERT(vals.front()->data == val_data);
 }
 
+void
+DhtRunnerTester::testPutDuplicate() {
+    auto key = dht::InfoHash::get("123");
+    auto val = std::make_shared<dht::Value>("hey");
+    val->id = 42;
+    auto val_data = val->data;
+    std::promise<bool> p1;
+    std::promise<bool> p2;
+    node2.put(key, val, [&](bool ok){
+        p1.set_value(ok);
+    });
+    node2.put(key, val, [&](bool ok){
+        p2.set_value(ok);
+    });
+    auto p1ret = p1.get_future().get();
+    auto p2ret = p2.get_future().get();
+    CPPUNIT_ASSERT(p1ret);
+    CPPUNIT_ASSERT(p2ret);
+    auto vals = node1.get(key).get();
+    CPPUNIT_ASSERT(not vals.empty());
+    CPPUNIT_ASSERT(vals.size() == 1);
+    CPPUNIT_ASSERT(vals.front()->data == val_data);
+}
+
+
+void
+DhtRunnerTester::testPutOverride() {
+    auto key = dht::InfoHash::get("123");
+    auto val = std::make_shared<dht::Value>("meh");
+    val->id = 42;
+    auto val2 = std::make_shared<dht::Value>("hey");
+    val2->id = 42;
+    CPPUNIT_ASSERT_EQUAL(val->id, val2->id);
+    auto val_data = val->data;
+    std::promise<bool> p1;
+    std::promise<bool> p2;
+    node2.put(key, val, [&](bool ok){
+        p1.set_value(ok);
+    });
+    node2.put(key, val2, [&](bool ok){
+        p2.set_value(ok);
+    });
+    auto p1ret = p1.get_future().get();
+    auto p2ret = p2.get_future().get();
+    CPPUNIT_ASSERT(!p1ret);
+    CPPUNIT_ASSERT(p2ret);
+    auto vals = node1.get(key).get();
+    CPPUNIT_ASSERT(not vals.empty());
+    CPPUNIT_ASSERT(vals.size() == 1);
+    CPPUNIT_ASSERT(vals.front()->data == val_data);
+}
+
+
 void
 DhtRunnerTester::testListen() {
     std::mutex mutex;
@@ -289,7 +342,7 @@ DhtRunnerTester::testIdOps() {
         valueCountEdit++;
         cv.notify_all();
     });
-    node2.listen(key2, [&](const std::vector<std::shared_ptr<dht::Value>>& values, bool expired){
+    node2.listen(key2, [&](const std::vector<std::shared_ptr<dht::Value>>& values, bool /*expired*/){
         for (const auto& v : values) {
             if (v->seq == 0)
                 CPPUNIT_ASSERT_EQUAL("v1"s, dht::unpackMsg<std::string>(v->data));

tests/dhtrunnertester.h

@@ -31,6 +31,7 @@ class DhtRunnerTester : public CppUnit::TestFixture {
     CPPUNIT_TEST_SUITE(DhtRunnerTester);
     CPPUNIT_TEST(testConstructors);
     CPPUNIT_TEST(testGetPut);
+    CPPUNIT_TEST(testPutDuplicate);
     CPPUNIT_TEST(testListen);
     CPPUNIT_TEST(testListenLotOfBytes);
     CPPUNIT_TEST(testIdOps);
@@ -55,6 +56,14 @@ class DhtRunnerTester : public CppUnit::TestFixture {
      * Test get and put methods
      */
     void testGetPut();
+    /**
+     * Test get and multiple put
+     */
+    void testPutDuplicate();
+    /**
+     * Test get and multiple put with changing value
+     */
+    void testPutOverride();
     /**
      * Test listen method
      */