misc: use std::make_unique

If configured to run as a proxy client, no port will be bound and the local dht node
won't be instantiated.

.github/workflows/ccpp.yml

@@ -29,6 +29,24 @@ jobs:
     - name: make check
       run: make check
 
+  build-ubuntu-minimal:
+    name: Ubuntu/GCC minimal build
+    runs-on: ubuntu-latest
+    steps:
+    - uses: actions/checkout@v3
+    - name: deps
+      run: |
+        sudo apt install libncurses5-dev libreadline-dev nettle-dev \
+        libgnutls28-dev libcppunit-dev libmsgpack-dev libargon2-0-dev
+    - name: cmake
+      run: |
+        mkdir build && cd build
+        cmake .. -DCMAKE_INSTALL_PREFIX=/usr/local -DCMAKE_BUILD_TYPE=Debug \
+                 -DOPENDHT_C=Off -DOPENDHT_TESTS=On -DOPENDHT_PEER_DISCOVERY=Off -DOPENDHT_PYTHON=Off \
+                 -DOPENDHT_TOOLS=On -DOPENDHT_PROXY_SERVER=Off -DOPENDHT_PROXY_CLIENT=Off
+    - name: make
+      run: cd build && make
+
   build-macos:
     name: macOS/Clang build
     runs-on: macos-11
@@ -37,7 +55,7 @@ jobs:
     - name: deps
       run: |
         brew install msgpack-cxx asio gnutls nettle readline fmt jsoncpp argon2 openssl http-parser cppunit
-    
+
     - name: restinio
       run: |
         mkdir restinio && cd restinio

.vscode/tasks.json

@@ -20,6 +20,7 @@
             "command": "cmake",
             "args": [
                 "${workspaceRoot}",
+                "-DCMAKE_EXPORT_COMPILE_COMMANDS=On",
                 "-DCMAKE_BUILD_TYPE=Debug",
                 "-DOPENDHT_PROXY_CLIENT=On",
                 "-DOPENDHT_PROXY_SERVER=On",

CMakeLists.txt

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

configure.ac

@@ -1,7 +1,7 @@
 dnl define macros
 m4_define([opendht_major_version], 2)
 m4_define([opendht_minor_version], 4)
-m4_define([opendht_patch_version], 3)
+m4_define([opendht_patch_version], 5)
 m4_define([opendht_version],
 		  [opendht_major_version.opendht_minor_version.opendht_patch_version])
 
@@ -129,8 +129,8 @@ AC_ARG_ENABLE([proxy_client], AS_HELP_STRING([--enable-proxy-client], [Enable pr
 AM_CONDITIONAL(ENABLE_PROXY_CLIENT, test x$proxy_client == xyes)
 
 AC_ARG_ENABLE([tests], AS_HELP_STRING([--enable-tests], [Enable tests]), build_tests=yes, build_tests=no)
-AM_CONDITIONAL(ENABLE_TESTS, test x$build_tests == xyes)
-AM_COND_IF([ENABLE_TESTS], [
+AM_CONDITIONAL(ENABLE_TESTS, test "x$build_tests" == xyes)
+AS_IF([test "x$build_tests" == xyes], [
 	PKG_CHECK_MODULES([CppUnit], [cppunit >= 1.12])
 ])
 

include/opendht/dht_proxy_client.h

@@ -263,7 +263,7 @@ public:
     void dumpTables() const override {}
     std::vector<unsigned> getNodeMessageStats(bool) override { return {}; }
     void setStorageLimit(size_t) override {}
-    virtual size_t getStorageLimit() const { return 0; }
+    virtual size_t getStorageLimit() const override { return 0; }
     void connectivityChanged(sa_family_t) override {
         getProxyInfos();
     }

include/opendht/dhtrunner.h

@@ -437,7 +437,6 @@ public:
     /**
      * Start or stop the proxy
      * @param proxify if we want to use the proxy
-     * @param deviceKey non empty to enable push notifications
      */
     void enableProxy(bool proxify);
 
@@ -474,12 +473,9 @@ private:
     DoneCallback bindOpDoneCallback(DoneCallback&& cb);
     DoneCallbackSimple bindOpDoneCallback(DoneCallbackSimple&& cb);
 
-    /** Local DHT instance */
+    /** DHT instance */
     std::unique_ptr<SecureDht> dht_;
 
-    /** Proxy client instance */
-    std::unique_ptr<SecureDht> dht_via_proxy_;
-
     /** true if we are currently using a proxy */
     std::atomic_bool use_proxy {false};
 
@@ -491,17 +487,6 @@ private:
      * reset dht clients
      */
     void resetDht();
-    /**
-     * @return the current active DHT
-     */
-    SecureDht* activeDht() const;
-
-    /**
-     * Store current listeners and translates global tokens for each client.
-     */
-    struct Listener;
-    std::map<size_t, Listener> listeners_;
-    size_t listener_token_ {1};
 
     mutable std::mutex dht_mtx {};
     std::thread dht_thread {};

include/opendht/value.h

@@ -213,7 +213,7 @@ struct OPENDHT_PUBLIC Value
         }
         static Filter notFilter(Filter&& f) {
             if (not f) return [](const Value&) { return false; };
-            return [f](const Value& v) { return not f(v); };
+            return [f = std::move(f)](const Value& v) { return not f(v); };
         }
         std::vector<Sp<Value>> filter(const std::vector<Sp<Value>>& values) {
             if (not (*this))
@@ -343,10 +343,10 @@ struct OPENDHT_PUBLIC Value
         return unpack<T>(*this);
     }
 
-    bool isEncrypted() const {
+    inline bool isEncrypted() const {
         return not cypher.empty();
     }
-    bool isSigned() const {
+    inline bool isSigned() const {
         return owner and not signature.empty();
     }
 
@@ -355,37 +355,24 @@ struct OPENDHT_PUBLIC Value
      * Afterward, checkSignature() will return true and owner will
      * be set to the corresponding public key.
      */
-    void sign(const crypto::PrivateKey& key) {
-        if (isEncrypted())
-            throw DhtException("Can't sign encrypted data.");
-        owner = key.getSharedPublicKey();
-        signature = key.sign(getToSign());
-    }
+    void sign(const crypto::PrivateKey& key);
 
     /**
      * Check that the value is signed and that the signature matches.
      * If true, the owner field will contain the signer public key.
      */
-    bool checkSignature() const {
+    inline bool checkSignature() const {
         return isSigned() and owner->checkSignature(getToSign(), signature);
     }
 
-    std::shared_ptr<crypto::PublicKey> getOwner() const {
-        return std::static_pointer_cast<crypto::PublicKey>(owner);
+    inline std::shared_ptr<crypto::PublicKey> getOwner() const {
+        return owner;
     }
 
     /**
      * Sign the value with from and returns the encrypted version for to.
      */
-    Value encrypt(const crypto::PrivateKey& from, const crypto::PublicKey& to) {
-        if (isEncrypted())
-            throw DhtException("Data is already encrypted.");
-        setRecipient(to.getId());
-        sign(from);
-        Value nv {id};
-        nv.setCypher(to.encrypt(getToEncrypt()));
-        return nv;
-    }
+    Value encrypt(const crypto::PrivateKey& from, const crypto::PublicKey& to);
 
     Value() {}
 
@@ -453,18 +440,18 @@ struct OPENDHT_PUBLIC Value
         return id == o.id and contentEquals(o);
     }
 
-    void setRecipient(const InfoHash& r) {
+    inline void setRecipient(const InfoHash& r) {
         recipient = r;
     }
 
-    void setCypher(Blob&& c) {
+    inline void setCypher(Blob&& c) {
         cypher = std::move(c);
     }
 
     /**
      * Pack part of the data to be signed (must always be done the same way)
      */
-    Blob getToSign() const {
+    inline Blob getToSign() const {
         msgpack::sbuffer buffer;
         msgpack::packer<msgpack::sbuffer> pk(&buffer);
         msgpack_pack_to_sign(pk);
@@ -474,7 +461,7 @@ struct OPENDHT_PUBLIC Value
     /**
      * Pack part of the data to be encrypted
      */
-    Blob getToEncrypt() const {
+    inline Blob getToEncrypt() const {
         msgpack::sbuffer buffer;
         msgpack::packer<msgpack::sbuffer> pk(&buffer);
         msgpack_pack_to_encrypt(pk);
@@ -484,7 +471,7 @@ struct OPENDHT_PUBLIC Value
     /** print value for debugging */
     OPENDHT_PUBLIC friend std::ostream& operator<< (std::ostream& s, const Value& v);
 
-    std::string toString() const {
+    inline std::string toString() const {
         std::stringstream ss;
         ss << *this;
         return ss.str();

python/tools/README.md

@@ -20,7 +20,7 @@ help).
 
 ## Python dependencies
 
-- pyroute2 >=0.3.14
+- pyroute2 >=0.6.9
 - matplotlib
 - GeoIP (used by `scanner.py` for drawing map of the world)
 - ipaddress

python/tools/dht/network.py

@@ -30,7 +30,7 @@ import traceback
 import ipaddress
 import netifaces
 import numpy as np
-from pyroute2.netns.process.proxy import NSPopen
+from pyroute2 import NSPopen
 import msgpack
 
 from opendht import *

python/tools/dht/virtual_network_builder.py

@@ -15,107 +15,253 @@
 #
 # You should have received a copy of the GNU General Public License
 # along with this program; If not, see <http://www.gnu.org/licenses/>.
+import argparse
+import os
+import subprocess
 
-import argparse, subprocess
+from pyroute2 import NDB, NSPopen
 
-from pyroute2 import IPDB, NetNS
-from pyroute2.netns.process.proxy import NSPopen
 
-if __name__ == "__main__":
-    parser = argparse.ArgumentParser(description='Creates a virtual network topology for testing')
-    parser.add_argument('-i', '--ifname', help='interface name', default='ethdht')
-    parser.add_argument('-n', '--ifnum', type=int, help='number of isolated interfaces to create', default=1)
-    parser.add_argument('-r', '--remove', help='remove instead of adding network interfaces', action="store_true")
-    parser.add_argument('-l', '--loss', help='simulated packet loss (percent)', type=int, default=0)
-    parser.add_argument('-d', '--delay', help='simulated latency (ms)', type=int, default=0)
-    parser.add_argument('-4', '--ipv4', help='Enable IPv4', action="store_true")
-    parser.add_argument('-6', '--ipv6', help='Enable IPv6', action="store_true")
+def int_range(mini, maxi):
+    def check_ifnum(arg):
+        try:
+            ret = int(arg)
+        except ValueError:
+            raise argparse.ArgumentTypeError('must be an integer')
+        if ret > maxi or ret < mini:
+            raise argparse.ArgumentTypeError(
+                f'must be {mini} <= int <= {maxi}'
+            )
+        return ret
+
+    return check_ifnum
+
+
+if __name__ == '__main__':
+    parser = argparse.ArgumentParser(
+        description='Creates a virtual network topology for testing'
+    )
+    parser.add_argument(
+        '-i', '--ifname', help='interface name', default='ethdht'
+    )
+    parser.add_argument(
+        '-n',
+        '--ifnum',
+        type=int_range(1, 245),
+        help='number of isolated interfaces to create',
+        default=1,
+    )
+    parser.add_argument(
+        '-r',
+        '--remove',
+        help='remove instead of adding network interfaces',
+        action='store_true',
+    )
+    parser.add_argument(
+        '-l',
+        '--loss',
+        help='simulated packet loss (percent)',
+        type=int,
+        default=0,
+    )
+    parser.add_argument(
+        '-d', '--delay', help='simulated latency (ms)', type=int, default=0
+    )
+    parser.add_argument(
+        '-4', '--ipv4', help='Enable IPv4', action='store_true'
+    )
+    parser.add_argument(
+        '-6', '--ipv6', help='Enable IPv6', action='store_true'
+    )
+    parser.add_argument(
+        '-b',
+        '--debug',
+        help='Turn on debug logging and dump topology databases',
+        action='store_true',
+    )
+    parser.add_argument(
+        '-v',
+        '--verbose',
+        help='Turn on verbose output on netns and interfaces operations',
+        action='store_true',
+    )
 
     args = parser.parse_args()
 
     local_addr4 = '10.0.42.'
     local_addr6 = '2001:db9::'
-    brige_name = 'br'+args.ifname
+    bripv4 = f'{local_addr4}1/24'
+    bripv6 = f'{local_addr6}1/64'
+    bridge_name = f'br{args.ifname}'
+    tap_name = f'tap{args.ifname}'
+    veth_names = []
+    namespaces = []
+    ipv4addrs = []
+    ipv6addrs = []
+    for ifn in range(args.ifnum):
+        namespaces.append(f'node{ifn}')
+        veth_names.append(f'{args.ifname}{ifn}')
+        ipv4addrs.append(f'{local_addr4}{ifn+8}/24' if args.ipv4 else None)
+        ipv6addrs.append(f'{local_addr6}{ifn+8}/64' if args.ipv6 else None)
 
-    ip = None
-    try:
-        ip = IPDB()
+    with NDB(log='debug' if args.debug else None) as ndb:
         if args.remove:
-            # cleanup interfaces
-            for ifn in range(args.ifnum):
-                iface = args.ifname+str(ifn)
-                if iface in ip.interfaces:
-                    with ip.interfaces[iface] as i:
-                        i.remove()
-            if 'tap'+args.ifname in ip.interfaces:
-                with ip.interfaces['tap'+args.ifname] as i:
-                    i.remove()
-            if brige_name in ip.interfaces:
-                with ip.interfaces[brige_name] as i:
-                    i.remove()
-            for ifn in range(args.ifnum):
-                netns = NetNS('node'+str(ifn))
-                netns.close()
-                netns.remove()
-        else:
-            for ifn in range(args.ifnum):
-                iface = args.ifname+str(ifn)
-                if not iface in ip.interfaces:
-                    ip.create(kind='veth', ifname=iface, peer=iface+'.1').commit()
+            # cleanup interfaces in the main namespace
+            for iface in veth_names + [bridge_name] + [tap_name]:
+                if iface in ndb.interfaces:
+                    ndb.interfaces[iface].remove().commit()
+                    if args.verbose:
+                        print(f'link: del main/{iface}')
 
-            ip.create(kind='tuntap', ifname='tap'+args.ifname, mode='tap').commit()
-
-            with ip.create(kind='bridge', ifname=brige_name) as i:
-                for ifn in range(args.ifnum):
-                    iface = args.ifname+str(ifn)
-                    i.add_port(ip.interfaces[iface])
-                i.add_port(ip.interfaces['tap'+args.ifname])
-                if args.ipv4:
-                    i.add_ip(local_addr4+'1/24')
-                if args.ipv6:
-                    i.add_ip(local_addr6+'1/64')
-                i.up()
-
-            with ip.interfaces['tap'+args.ifname] as tap:
-                tap.up()
-
-            for ifn in range(args.ifnum):
-                iface = args.ifname+str(ifn)
-
-                nns = NetNS('node'+str(ifn))
-                iface1 = iface+'.1'
-                with ip.interfaces[iface1] as i:
-                    i.net_ns_fd = nns.netns
-
-                with ip.interfaces[iface] as i:
-                    i.up()
-
-                ip_ns = IPDB(nl=nns)
+            # cleanup namespaces
+            for nsname in namespaces:
                 try:
-                    with ip_ns.interfaces.lo as lo:
-                        lo.up()
-                    with ip_ns.interfaces[iface1] as i:
-                        if args.ipv4:
-                            i.add_ip(local_addr4+str(ifn+8)+'/24')
-                        if args.ipv6:
-                            i.add_ip(local_addr6+str(ifn+8)+'/64')
-                        i.up()
-                finally:
-                    ip_ns.release()
-
-                nsp = NSPopen(nns.netns, ["tc", "qdisc", "add", "dev", iface1, "root", "netem", "delay", str(args.delay)+"ms", str(int(args.delay/2))+"ms", "loss", str(args.loss)+"%", "25%"], stdout=subprocess.PIPE)
-                #print(nsp.communicate()[0].decode())
+                    ndb.netns[nsname].remove().commit()
+                    if args.verbose:
+                        print(f'netns: del {nsname}')
+                except KeyError:
+                    pass
+        else:
+            # create ports
+            for veth, nsname, ipv4addr, ipv6addr in zip(
+                veth_names, namespaces, ipv4addrs, ipv6addrs
+            ):
+                # create a network namespace and launch NDB for it
+                #
+                # another possible solution could be simply to attach
+                # the namespace to the main NDB instance, but it can
+                # take a lot of memory in case of many interfaces, thus
+                # launch and discard netns NDB instances
+                netns = NDB(
+                    log='debug' if args.debug else None,
+                    sources=[
+                        {
+                            'target': 'localhost',
+                            'netns': nsname,
+                            'kind': 'netns',
+                        }
+                    ],
+                )
+                if args.verbose:
+                    print(f'netns: add {nsname}')
+                # create the port and push the peer into the namespace
+                (
+                    ndb.interfaces.create(
+                        **{
+                            'ifname': veth,
+                            'kind': 'veth',
+                            'state': 'up',
+                            'peer': {'ifname': veth, 'net_ns_fd': nsname},
+                        }
+                    ).commit()
+                )
+                if args.verbose:
+                    print(f'link: add main/{veth} <-> {nsname}/{veth}')
+                # bring up namespace's loopback
+                (
+                    netns.interfaces.wait(ifname='lo', timeout=3)
+                    .set('state', 'up')
+                    .commit()
+                )
+                if args.verbose:
+                    print(f'link: set {nsname}/lo')
+                # bring up the peer
+                with netns.interfaces.wait(ifname=veth, timeout=3) as i:
+                    i.set('state', 'up')
+                    if args.ipv4:
+                        i.add_ip(ipv4addr)
+                    if args.ipv6:
+                        i.add_ip(ipv6addr)
+                if args.verbose:
+                    print(f'link: set {nsname}/{veth}, {ipv4addr}, {ipv6addr}')
+                # disconnect the namespace NDB agent, not removing the NS
+                if args.debug:
+                    fname = f'{nsname}-ndb.db'
+                    print(f'dump: netns topology database {fname}')
+                    netns.schema.backup(fname)
+                netns.close()
+                # set up the emulation QDisc
+                nsp = NSPopen(
+                    nsname,
+                    [
+                        'tc',
+                        'qdisc',
+                        'add',
+                        'dev',
+                        veth,
+                        'root',
+                        'netem',
+                        'delay',
+                        f'{args.delay}ms',
+                        f'{int(args.delay)/2}ms',
+                        'loss',
+                        f'{args.loss}%',
+                        '25%',
+                    ],
+                    stdout=subprocess.PIPE,
+                )
                 nsp.communicate()
                 nsp.wait()
                 nsp.release()
+                if args.verbose:
+                    print(
+                        f'netem: add {nsname}/{veth}, '
+                        f'{args.delay}, {args.loss}'
+                    )
 
-            if args.ipv4:
-                subprocess.call(["sysctl", "-w", "net.ipv4.conf."+brige_name+".forwarding=1"])
-            if args.ipv6:
-                subprocess.call(["sysctl", "-w", "net.ipv6.conf."+brige_name+".forwarding=1"])
+            # create the tap
+            #
+            # for some reason we should create the tap inteface first,
+            # and only then bring it up, thus two commit() calls
+            (
+                ndb.interfaces.create(
+                    kind='tuntap', ifname=tap_name, mode='tap'
+                )
+                .commit()
+                .set('state', 'up')
+                .commit()
+            )
+            if args.verbose:
+                print(f'link: add main/{tap_name}')
 
-    except Exception as e:
-          print('Error',e)
-    finally:
-        if ip:
-            ip.release()
+            # create the bridge and add all the ports
+            with ndb.interfaces.create(
+                ifname=bridge_name, kind='bridge', state='up'
+            ) as i:
+                if args.ipv4:
+                    i.add_ip(bripv4)
+                if args.ipv6:
+                    i.add_ip(bripv6)
+                for iface in veth_names + [tap_name]:
+                    i.add_port(iface)
+            if args.verbose:
+                print(f'link: add main/{bridge_name}, {bripv4}, {bripv6}')
+
+            with open(os.devnull, 'w') as fnull:
+                if args.ipv4:
+                    subprocess.call(
+                        [
+                            'sysctl',
+                            '-w',
+                            f'net.ipv4.conf.{bridge_name}.forwarding=1',
+                        ],
+                        stdout=fnull,
+                    )
+                if args.verbose:
+                    print(f'sysctl: set {bridge_name} ipv4 forwarding')
+                if args.ipv6:
+                    subprocess.call(
+                        [
+                            'sysctl',
+                            '-w',
+                            f'net.ipv6.conf.{bridge_name}.forwarding=1',
+                        ],
+                        stdout=fnull,
+                    )
+                if args.verbose:
+                    print(f'sysctl: set {bridge_name} ipv4 forwarding')
+
+            if args.debug:
+                fname = 'main-ndb.db'
+                print('dump: the main netns topology database')
+                ndb.schema.backup(fname)

src/dhtrunner.cpp

@@ -38,15 +38,6 @@ namespace dht {
 
 static const std::string PEER_DISCOVERY_DHT_SERVICE = "dht";
 
-struct DhtRunner::Listener {
-    size_t tokenClassicDht {0};
-    size_t tokenProxyDht {0};
-    ValueCallback gcb;
-    InfoHash hash {};
-    Value::Filter f;
-    Where w;
-};
-
 struct NodeInsertionPack {
     dht::InfoHash nodeId;
     in_port_t port;
@@ -55,9 +46,6 @@ struct NodeInsertionPack {
 };
 
 DhtRunner::DhtRunner() : dht_()
-#ifdef OPENDHT_PROXY_CLIENT
-, dht_via_proxy_()
-#endif //OPENDHT_PROXY_CLIENT
 {
 #ifdef _WIN32
     WSADATA wsd;
@@ -119,8 +107,9 @@ DhtRunner::run(const Config& config, Context&& context)
             local6.setFamily(AF_INET6);
         }
         auto state_path = config.dht_config.node_config.persist_path;
-        if (not state_path.empty() && (local4.getPort() == 0 || local6.getPort() == 0)) {
+        if (not state_path.empty())
             state_path += "_port.txt";
+        if (not state_path.empty() && (local4.getPort() == 0 || local6.getPort() == 0)) {
             std::ifstream inConfig(state_path);
             if (inConfig.is_open()) {
                 in_port_t port;
@@ -149,7 +138,7 @@ DhtRunner::run(const Config& config, Context&& context)
             context.sock.reset(new net::UdpSocket(local4, local6, context.logger));
         }
 
-        if (not state_path.empty()) {
+        if (not state_path.empty() and config.proxy_server.empty()) {
             std::ofstream outConfig(state_path);
             outConfig << context.sock->getBoundRef(AF_INET).getPort() << std::endl;
             outConfig << context.sock->getBoundRef(AF_INET6).getPort() << std::endl;
@@ -182,30 +171,25 @@ DhtRunner::run(const Config& config, Context&& context)
         config_ = config;
         identityAnnouncedCb_ = context.identityAnnouncedCb;
 #endif
-        auto dht = std::make_unique<Dht>(std::move(context.sock), SecureDht::getConfig(config.dht_config), context.logger);
-        dht_ = std::make_unique<SecureDht>(std::move(dht), config.dht_config, std::move(context.identityAnnouncedCb), context.logger);
-        enableProxy(not config.proxy_server.empty());
+        if (config.proxy_server.empty()) {
+            auto dht = std::make_unique<Dht>(std::move(context.sock), SecureDht::getConfig(config.dht_config), context.logger);
+            dht_ = std::make_unique<SecureDht>(std::move(dht), config.dht_config, std::move(context.identityAnnouncedCb), context.logger);
+        } else {
+            enableProxy(true);
+        }
     } catch(const std::exception& e) {
         config_ = {};
         identityAnnouncedCb_ = {};
         dht_.reset();
-#ifdef OPENDHT_PROXY_CLIENT
-        dht_via_proxy_.reset();
-#endif
         running = State::Idle;
         throw;
     }
 
-    if (context.logger and dht_via_proxy_) {
-        dht_via_proxy_->setLogger(context.logger);
-    }
     if (context.statusChangedCallback) {
         statusCb = std::move(context.statusChangedCallback);
     }
     if (context.certificateStore) {
         dht_->setLocalCertificateStore(std::move(context.certificateStore));
-        if (dht_via_proxy_)
-            dht_via_proxy_->setLocalCertificateStore(std::move(context.certificateStore));
     }
 
     if (not config.threaded)
@@ -288,7 +272,8 @@ DhtRunner::shutdown(ShutdownCallback cb, bool stop) {
     auto expected = State::Running;
     if (not running.compare_exchange_strong(expected, State::Stopping)) {
         if (expected == State::Stopping and ongoing_ops) {
-            shutdownCallbacks_.emplace_back(std::move(cb));
+            if (cb)
+                shutdownCallbacks_.emplace_back(std::move(cb));
         }
         else if (cb) {
             lck.unlock();
@@ -298,20 +283,10 @@ DhtRunner::shutdown(ShutdownCallback cb, bool stop) {
     }
     if (logger_)
         logger_->d("[runner %p] state changed to Stopping, %zu ongoing ops", this, ongoing_ops.load());
-#ifdef OPENDHT_PROXY_CLIENT
-    ongoing_ops += 2;
-#else
     ongoing_ops++;
-#endif
     shutdownCallbacks_.emplace_back(std::move(cb));
     pending_ops.emplace([=](SecureDht&) mutable {
         auto onShutdown = [this]{ opEnded(); };
-#ifdef OPENDHT_PROXY_CLIENT
-        if (dht_via_proxy_)
-            dht_via_proxy_->shutdown(onShutdown, stop);
-        else
-            opEnded();
-#endif
         if (dht_)
             dht_->shutdown(onShutdown, stop);
         else
@@ -418,31 +393,25 @@ void
 DhtRunner::dumpTables() const
 {
     std::lock_guard<std::mutex> lck(dht_mtx);
-    activeDht()->dumpTables();
+    dht_->dumpTables();
 }
 
 InfoHash
 DhtRunner::getId() const
 {
-    if (auto dht = activeDht())
-        return dht->getId();
-    return {};
+    return dht_ ? dht_->getId() : InfoHash{};
 }
 
 std::shared_ptr<crypto::PublicKey>
 DhtRunner::getPublicKey() const
 {
-    if (auto dht = activeDht())
-        return dht->getPublicKey();
-    return {};
+    return dht_ ? dht_->getPublicKey() : std::shared_ptr<crypto::PublicKey>{};
 }
 
 InfoHash
 DhtRunner::getNodeId() const
 {
-    if (auto dht = activeDht())
-        return dht->getNodeId();
-    return {};
+    return dht_ ? dht_->getNodeId() : InfoHash{};
 }
 
 
@@ -484,10 +453,6 @@ DhtRunner::setLogger(const Sp<Logger>& logger) {
     logger_ = logger;
     if (dht_)
         dht_->setLogger(logger);
-#ifdef OPENDHT_PROXY_CLIENT
-    if (dht_via_proxy_)
-        dht_via_proxy_->setLogger(logger);
-#endif
 }
 
 void
@@ -501,16 +466,12 @@ DhtRunner::setLogFilter(const InfoHash& f) {
     std::lock_guard<std::mutex> lck(dht_mtx);
     if (dht_)
         dht_->setLogFilter(f);
-#ifdef OPENDHT_PROXY_CLIENT
-    if (dht_via_proxy_)
-        dht_via_proxy_->setLogFilter(f);
-#endif
 }
 
 void
 DhtRunner::registerType(const ValueType& type) {
     std::lock_guard<std::mutex> lck(dht_mtx);
-    activeDht()->registerType(type);
+    dht_->registerType(type);
 }
 
 void
@@ -523,7 +484,7 @@ unsigned
 DhtRunner::getNodesStats(sa_family_t af, unsigned *good_return, unsigned *dubious_return, unsigned *cached_return, unsigned *incoming_return) const
 {
     std::lock_guard<std::mutex> lck(dht_mtx);
-    const auto stats = activeDht()->getNodesStats(af);
+    const auto stats = dht_->getNodesStats(af);
     if (good_return)
         *good_return = stats.good_nodes;
     if (dubious_return)
@@ -539,19 +500,19 @@ NodeStats
 DhtRunner::getNodesStats(sa_family_t af) const
 {
     std::lock_guard<std::mutex> lck(dht_mtx);
-    return activeDht()->getNodesStats(af);
+    return dht_->getNodesStats(af);
 }
 
 NodeInfo
 DhtRunner::getNodeInfo() const {
     std::lock_guard<std::mutex> lck(dht_mtx);
     NodeInfo info {};
-    if (auto dht = activeDht()) {
-        info.id = dht->getId();
-        info.node_id = dht->getNodeId();
-        info.ipv4 = dht->getNodesStats(AF_INET);
-        info.ipv6 = dht->getNodesStats(AF_INET6);
-        if (auto sock = dht->getSocket()) {
+    if (dht_) {
+        info.id = dht_->getId();
+        info.node_id = dht_->getNodeId();
+        info.ipv4 = dht_->getNodesStats(AF_INET);
+        info.ipv6 = dht_->getNodesStats(AF_INET6);
+        if (auto sock = dht_->getSocket()) {
             info.bound4 = sock->getBoundRef(AF_INET).getPort();
             info.bound6 = sock->getBoundRef(AF_INET6).getPort();
         }
@@ -588,45 +549,45 @@ std::vector<unsigned>
 DhtRunner::getNodeMessageStats(bool in) const
 {
     std::lock_guard<std::mutex> lck(dht_mtx);
-    return activeDht()->getNodeMessageStats(in);
+    return dht_->getNodeMessageStats(in);
 }
 
 std::string
 DhtRunner::getStorageLog() const
 {
     std::lock_guard<std::mutex> lck(dht_mtx);
-    return activeDht()->getStorageLog();
+    return dht_->getStorageLog();
 }
 std::string
 DhtRunner::getStorageLog(const InfoHash& f) const
 {
     std::lock_guard<std::mutex> lck(dht_mtx);
-    return activeDht()->getStorageLog(f);
+    return dht_->getStorageLog(f);
 }
 std::string
 DhtRunner::getRoutingTablesLog(sa_family_t af) const
 {
     std::lock_guard<std::mutex> lck(dht_mtx);
-    return activeDht()->getRoutingTablesLog(af);
+    return dht_->getRoutingTablesLog(af);
 }
 std::string
 DhtRunner::getSearchesLog(sa_family_t af) const
 {
     std::lock_guard<std::mutex> lck(dht_mtx);
-    return activeDht()->getSearchesLog(af);
+    return dht_->getSearchesLog(af);
 }
 std::string
 DhtRunner::getSearchLog(const InfoHash& f, sa_family_t af) const
 {
     std::lock_guard<std::mutex> lck(dht_mtx);
-    return activeDht()->getSearchLog(f, af);
+    return dht_->getSearchLog(f, af);
 }
 std::vector<SockAddr>
 DhtRunner::getPublicAddress(sa_family_t af)
 {
     std::lock_guard<std::mutex> lck(dht_mtx);
-    if (auto dht = activeDht())
-        return dht->getPublicAddress(af);
+    if (dht_)
+        return dht_->getPublicAddress(af);
     return {};
 }
 std::vector<std::string>
@@ -653,15 +614,12 @@ DhtRunner::getPublicAddress(std::function<void(std::vector<SockAddr>&&)> cb, sa_
 void
 DhtRunner::registerCertificate(std::shared_ptr<crypto::Certificate> cert) {
     std::lock_guard<std::mutex> lck(dht_mtx);
-    activeDht()->registerCertificate(cert);
+    dht_->registerCertificate(cert);
 }
+
 void
 DhtRunner::setLocalCertificateStore(CertificateStoreQuery&& query_method) {
     std::lock_guard<std::mutex> lck(dht_mtx);
-#ifdef OPENDHT_PROXY_CLIENT
-    if (dht_via_proxy_)
-        dht_via_proxy_->setLocalCertificateStore(std::forward<CertificateStoreQuery>(query_method));
-#endif
     if (dht_)
         dht_->setLocalCertificateStore(std::forward<CertificateStoreQuery>(query_method));
 }
@@ -669,8 +627,7 @@ DhtRunner::setLocalCertificateStore(CertificateStoreQuery&& query_method) {
 time_point
 DhtRunner::loop_()
 {
-    auto dht = activeDht();
-    if (not dht)
+    if (not dht_)
         return {};
 
     decltype(pending_ops) ops {};
@@ -681,7 +638,7 @@ DhtRunner::loop_()
                std::move(pending_ops) : std::move(pending_ops_prio);
     }
     while (not ops.empty()) {
-        ops.front()(*dht);
+        ops.front()(*dht_);
         ops.pop();
     }
 
@@ -714,13 +671,13 @@ DhtRunner::loop_()
             if (now - pkt.received > net::RX_QUEUE_MAX_DELAY)
                 dropped++;
             else
-                wakeup = dht->periodic(pkt.data.data(), pkt.data.size(), std::move(pkt.from), now);
+                wakeup = dht_->periodic(pkt.data.data(), pkt.data.size(), std::move(pkt.from), now);
             pkt.data.clear();
         }
         received_treated.splice(received_treated.end(), std::move(received));
     } else {
         // Or just run the scheduler
-        wakeup = dht->periodic(nullptr, 0, nullptr, 0, clock::now());
+        wakeup = dht_->periodic(nullptr, 0, nullptr, 0, clock::now());
     }
 
     if (not received_treated.empty()) {
@@ -732,8 +689,8 @@ DhtRunner::loop_()
     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);
+    NodeStatus nstatus4 = dht_->updateStatus(AF_INET);
+    NodeStatus nstatus6 = dht_->updateStatus(AF_INET6);
     if (nstatus4 != status4 || nstatus6 != status6) {
         status4 = nstatus4;
         status6 = nstatus6;
@@ -791,27 +748,7 @@ DhtRunner::listen(InfoHash hash, ValueCallback vcb, Value::Filter f, Where w)
         return ret_token->get_future();
     }
     pending_ops.emplace([=](SecureDht& dht) mutable {
-#ifdef OPENDHT_PROXY_CLIENT
-        auto tokenbGlobal = listener_token_++;
-        auto& listener = listeners_[tokenbGlobal];
-        listener.hash = hash;
-        listener.f = std::move(f);
-        listener.w = std::move(w);
-        listener.gcb = [hash,vcb,tokenbGlobal,this](const std::vector<Sp<Value>>& vals, bool expired) {
-            if (not vcb(vals, expired)) {
-                cancelListen(hash, tokenbGlobal);
-                return false;
-            }
-            return true;
-        };
-        if (auto token = dht.listen(hash, listener.gcb, listener.f, listener.w)) {
-            if (use_proxy)  listener.tokenProxyDht = token;
-            else            listener.tokenClassicDht = token;
-        }
-        ret_token->set_value(tokenbGlobal);
-#else
         ret_token->set_value(dht.listen(hash, std::move(vcb), std::move(f), std::move(w)));
-#endif
     });
     cv.notify_all();
     return ret_token->get_future();
@@ -830,27 +767,10 @@ DhtRunner::cancelListen(InfoHash h, size_t token)
     if (running != State::Running)
         return;
     ongoing_ops++;
-#ifdef OPENDHT_PROXY_CLIENT
-    pending_ops.emplace([=](SecureDht&) {
-        auto it = listeners_.find(token);
-        if (it != listeners_.end()) {
-            if (it->second.tokenClassicDht)
-                dht_->cancelListen(h, it->second.tokenClassicDht);
-            if (it->second.tokenProxyDht and dht_via_proxy_)
-                dht_via_proxy_->cancelListen(h, it->second.tokenProxyDht);
-            listeners_.erase(it);
-        } else {
-            if (logger_)
-                logger_->w("[runner %p] cancelListen: unknown token %zu.", this, token);
-        }
-        opEnded();
-    });
-#else
     pending_ops.emplace([=](SecureDht& dht) {
         dht.cancelListen(h, token);
         opEnded();
     });
-#endif // OPENDHT_PROXY_CLIENT
     cv.notify_all();
 }
 
@@ -861,28 +781,10 @@ DhtRunner::cancelListen(InfoHash h, std::shared_future<size_t> ftoken)
     if (running != State::Running)
         return;
     ongoing_ops++;
-#ifdef OPENDHT_PROXY_CLIENT
-    pending_ops.emplace([this, h, ftoken = std::move(ftoken)](SecureDht&) {
-        auto token = ftoken.get();
-        auto it = listeners_.find(token);
-        if (it != listeners_.end()) {
-            if (it->second.tokenClassicDht)
-                dht_->cancelListen(h, it->second.tokenClassicDht);
-            if (it->second.tokenProxyDht and dht_via_proxy_)
-                dht_via_proxy_->cancelListen(h, it->second.tokenProxyDht);
-            listeners_.erase(it);
-        } else {
-            if (logger_)
-                logger_->w("[runner %p] cancelListen: unknown token %zu.", this, token);
-        }
-        opEnded();
-    });
-#else
     pending_ops.emplace([this, h, ftoken = std::move(ftoken)](SecureDht& dht) {
         dht.cancelListen(h, ftoken.get());
         opEnded();
     });
-#endif // OPENDHT_PROXY_CLIENT
     cv.notify_all();
 }
 
@@ -1171,23 +1073,9 @@ void
 DhtRunner::resetDht()
 {
     peerDiscovery_.reset();
-#ifdef OPENDHT_PROXY_CLIENT
-    listeners_.clear();
-    dht_via_proxy_.reset();
-#endif // OPENDHT_PROXY_CLIENT
     dht_.reset();
 }
 
-SecureDht*
-DhtRunner::activeDht() const
-{
-#ifdef OPENDHT_PROXY_CLIENT
-    return use_proxy? dht_via_proxy_.get() : dht_.get();
-#else
-    return dht_.get();
-#endif // OPENDHT_PROXY_CLIENT
-}
-
 void
 DhtRunner::setProxyServer(const std::string& proxy, const std::string& pushNodeId)
 {
@@ -1200,7 +1088,7 @@ DhtRunner::setProxyServer(const std::string& proxy, const std::string& pushNodeI
     enableProxy(use_proxy and not config_.proxy_server.empty());
 #else
     if (not proxy.empty())
-        std::cerr << "DHT proxy requested but OpenDHT built without proxy support." << std::endl;
+        throw std::runtime_error("DHT proxy requested but OpenDHT built without proxy support.");
 #endif
 }
 
@@ -1208,54 +1096,33 @@ void
 DhtRunner::enableProxy(bool proxify)
 {
 #ifdef OPENDHT_PROXY_CLIENT
-    if (dht_via_proxy_) {
-        dht_via_proxy_->shutdown({});
+    if (dht_) {
+        dht_->shutdown({});
     }
     if (proxify) {
         // Init the proxy client
-        auto dht_via_proxy = std::unique_ptr<DhtInterface>(
-            new DhtProxyClient(
+        auto dht_via_proxy = std::make_unique<DhtProxyClient>(
                 config_.server_ca,
                 config_.client_identity,
                 [this]{
                     if (config_.threaded) {
-                        {
-                            std::lock_guard<std::mutex> lck(storage_mtx);
-                            pending_ops_prio.emplace([=](SecureDht&) mutable {});
-                        }
+                        std::lock_guard<std::mutex> lck(storage_mtx);
+                        pending_ops_prio.emplace([=](SecureDht&) mutable {});
                         cv.notify_all();
                     }
                 },
-                config_.proxy_server, config_.push_node_id, logger_)
-        );
-#ifdef OPENDHT_PUSH_NOTIFICATIONS
+                config_.proxy_server, config_.push_node_id, logger_);
         if (not config_.push_token.empty())
             dht_via_proxy->setPushNotificationToken(config_.push_token);
-#endif
-        dht_via_proxy_ = std::unique_ptr<SecureDht>(new SecureDht(std::move(dht_via_proxy), config_.dht_config, identityAnnouncedCb_, logger_));
-        // add current listeners
-        for (auto& l: listeners_)
-            l.second.tokenProxyDht = dht_via_proxy_->listen(l.second.hash, l.second.gcb, l.second.f, l.second.w);
+        dht_ = std::make_unique<SecureDht>(std::move(dht_via_proxy), config_.dht_config, identityAnnouncedCb_, logger_);
         // and use it
         use_proxy = proxify;
     } else {
         use_proxy = proxify;
-        std::lock_guard<std::mutex> lck(storage_mtx);
-        if (not listeners_.empty()) {
-            pending_ops.emplace([this](SecureDht& /*dht*/) mutable {
-                if (not dht_)
-                    return;
-                for (auto& l : listeners_) {
-                    if (not l.second.tokenClassicDht) {
-                        l.second.tokenClassicDht = dht_->listen(l.second.hash, l.second.gcb, l.second.f, l.second.w);
-                    }
-                }
-            });
-        }
     }
 #else
     if (proxify)
-        std::cerr << "DHT proxy requested but OpenDHT built without proxy support." << std::endl;
+        throw std::runtime_error("DHT proxy requested but OpenDHT built without proxy support.");
 #endif
 }
 
@@ -1263,16 +1130,8 @@ void
 DhtRunner::forwardAllMessages(bool forward)
 {
     std::lock_guard<std::mutex> lck(dht_mtx);
-#ifdef OPENDHT_PROXY_SERVER
-#ifdef OPENDHT_PROXY_CLIENT
-    if (dht_via_proxy_)
-        dht_via_proxy_->forwardAllMessages(forward);
-#endif // OPENDHT_PROXY_CLIENT
     if (dht_)
         dht_->forwardAllMessages(forward);
-#else
-    (void) forward;
-#endif // OPENDHT_PROXY_SERVER
 }
 
 /**
@@ -1283,8 +1142,8 @@ DhtRunner::setPushNotificationToken(const std::string& token) {
     std::lock_guard<std::mutex> lck(dht_mtx);
 #if defined(OPENDHT_PROXY_CLIENT) && defined(OPENDHT_PUSH_NOTIFICATIONS)
     config_.push_token = token;
-    if (dht_via_proxy_)
-        dht_via_proxy_->setPushNotificationToken(token);
+    if (dht_)
+        dht_->setPushNotificationToken(token);
 #else
     (void) token;
 #endif
@@ -1294,13 +1153,11 @@ void
 DhtRunner::pushNotificationReceived(const std::map<std::string, std::string>& data)
 {
 #if defined(OPENDHT_PROXY_CLIENT) && defined(OPENDHT_PUSH_NOTIFICATIONS)
-    {
-        std::lock_guard<std::mutex> lck(storage_mtx);
-        pending_ops_prio.emplace([=](SecureDht&) {
-            if (dht_via_proxy_)
-                dht_via_proxy_->pushNotificationReceived(data);
-        });
-    }
+    std::lock_guard<std::mutex> lck(storage_mtx);
+    pending_ops_prio.emplace([=](SecureDht&) {
+        if (dht_)
+            dht_->pushNotificationReceived(data);
+    });
     cv.notify_all();
 #else
     (void) data;

src/network_engine.cpp

@@ -415,7 +415,7 @@ NetworkEngine::processMessage(const uint8_t *buf, size_t buflen, SockAddr f)
         return;
     }
 
-    std::unique_ptr<ParsedMessage> msg {new ParsedMessage};
+    auto msg = std::make_unique<ParsedMessage>();
     try {
         msgpack::unpacked msg_res = msgpack::unpack((const char*)buf, buflen);
         msg->msgpack_unpack(msg_res.get());

src/securedht.cpp

@@ -71,7 +71,7 @@ SecureDht::~SecureDht() = default;
 ValueType
 SecureDht::secureType(ValueType&& type)
 {
-    type.storePolicy = [this,type](InfoHash id, Sp<Value>& v, const InfoHash& nid, const SockAddr& a) {
+    type.storePolicy = [type](InfoHash id, Sp<Value>& v, const InfoHash& nid, const SockAddr& a) {
         if (v->isSigned())
             return v->checkSignature();
         return type.storePolicy(id, v, nid, a);

src/value.cpp

@@ -239,6 +239,42 @@ Value::toJson() const
     return val;
 }
 
+uint64_t
+unpackId(const Json::Value& json, const std::string& key) {
+    uint64_t ret = 0;
+    try {
+        const auto& t = json[key];
+        if (t.isString()) {
+            ret = std::stoull(t.asString());
+        } else {
+            ret = t.asLargestUInt();
+        }
+    } catch (...) {}
+    return ret;
+}
+#endif
+
+void
+Value::sign(const crypto::PrivateKey& key)
+{
+    if (isEncrypted())
+        throw DhtException("Can't sign encrypted data.");
+    owner = key.getSharedPublicKey();
+    signature = key.sign(getToSign());
+}
+
+Value
+Value::encrypt(const crypto::PrivateKey& from, const crypto::PublicKey& to)
+{
+    if (isEncrypted())
+        throw DhtException("Data is already encrypted.");
+    setRecipient(to.getId());
+    sign(from);
+    Value nv {id};
+    nv.setCypher(to.encrypt(getToEncrypt()));
+    return nv;
+}
+
 bool
 Value::checkSignature()
 {
@@ -260,8 +296,8 @@ Value::decrypt(const crypto::PrivateKey& key)
         decrypted = true;
         if (isEncrypted()) {
             auto decryptedBlob = key.decrypt(cypher);
-            std::unique_ptr<Value> v {new Value(id)};
             auto msg = msgpack::unpack((const char*)decryptedBlob.data(), decryptedBlob.size());
+            auto v = std::make_unique<Value>(id);
             v->msgpack_unpack_body(msg.get());
             if (v->recipient != key.getPublicKey().getId())
                 throw crypto::DecryptError("Recipient mismatch");
@@ -274,21 +310,6 @@ Value::decrypt(const crypto::PrivateKey& key)
     return decryptedValue;
 }
 
-uint64_t
-unpackId(const Json::Value& json, const std::string& key) {
-    uint64_t ret = 0;
-    try {
-        const auto& t = json[key];
-        if (t.isString()) {
-            ret = std::stoull(t.asString());
-        } else {
-            ret = t.asLargestUInt();
-        }
-    } catch (...) {}
-    return ret;
-}
-#endif
-
 bool
 FieldValue::operator==(const FieldValue& vfd) const
 {