Compare commits

..
25 Commits
Author SHA1 Message Date
Adrien Béraud 83f8a92e5e travis: add llvm docker build 2016-07-29 21:16:43 -04:00
Adrien Béraud f41685b864 const shared_ptr 2016-07-29 20:25:34 -04:00
Adrien Béraud 84adbddd6f travis with docker 2016-07-29 20:21:06 -04:00
Simon Désaulniers fa9730dd07 add readme help text for benchmark 2016-06-15 15:57:56 -04:00
Adrien Béraud 33f29e95f4 value: add common crypto ops 2016-06-15 13:35:44 -04:00
Adrien Béraud fdfa10bd2e dht: clear search before expiring it 2016-06-13 15:55:34 -04:00
Adrien Béraud f200f9dfc2 pkgconfig: add private dependencies 2016-06-11 21:50:34 -04:00
Adrien Béraud 9615774114 cmake: build versioned shared library 2016-06-11 21:22:27 -04:00
Adrien Béraud c6b0e531c2 dht: add code documentation 2016-06-11 18:54:18 -04:00
Adrien Béraud 8e5f12dbf1 autotools: handle case where DESTDIR is empty 2016-06-11 16:10:52 -04:00
Adrien Béraud 5bcf1da505 autotools: cleanup 2016-06-11 16:10:27 -04:00
Adrien Béraud 97d88fcfed autotools: disable doxygen by default 2016-06-11 16:04:35 -04:00
Adrien Béraud 963dd6779e benchmark: don't block while putting values 2016-06-03 17:20:58 -04:00
Simon Désaulniers 54a4327039 dht: handle search expired before permanent put 2016-06-03 17:20:58 -04:00
Simon Désaulniers 10e5439048 securedht: permanent put of a certificate. 2016-06-03 17:20:58 -04:00
Simon Désaulniers dcca94aab7 dht,api: enable optional permanent put on dht
A permanent put is going to be executed once and every time the data expires on
each SearchNode.
2016-06-03 17:20:58 -04:00
Simon Désaulniers 173a9ee642 revert attempt to avoid securedht cert put fail 2016-06-03 17:20:58 -04:00
Simon Désaulniers ed1ca6db8f Revert "dht: wait for connection before running searches"
This reverts commit bd08506146.
2016-06-03 17:20:58 -04:00
Adrien Béraud 7367158284 dht: new insertNode algorithm 2016-06-03 17:20:58 -04:00
Adrien Béraud cf8d9ccd0d infohash: merge Windows includes 2016-06-02 17:52:07 -04:00
Adrien Béraud fdbdc91d55 Merge pull request #74 from sim590/python-destdir
[packaging] support python installation with make DESTDIR=$DIR
2016-06-02 15:41:55 -04:00
Simon Désaulniers cf9c63abb7 enable python make DESTDIR=$DIR install support 2016-06-02 14:48:05 -04:00
Simon Désaulniers 475b26048f cmake: pass $DESTDIR environnement var to setup.py 2016-06-02 12:36:21 -04:00
Andreas Traczyk df1615e8d1 Update README.md 2016-06-02 12:22:34 -04:00
Andreas Traczyk 9e461e0960 microsoft compiler support 2016-06-02 12:17:09 -04:00
44 changed files with 414 additions and 643 deletions

No files matched your search

+6 -8
View File
@@ -1,7 +1,7 @@
cmake_minimum_required (VERSION 2.8.6)
project (opendht)
set (opendht_VERSION_MAJOR 0)
set (opendht_VERSION_MINOR 6.4)
set (opendht_VERSION_MINOR 6.1)
set (opendht_VERSION ${opendht_VERSION_MAJOR}.${opendht_VERSION_MINOR})
set (PACKAGE_VERSION ${opendht_VERSION})
set (VERSION "${opendht_VERSION}")
@@ -21,15 +21,14 @@ option (OPENDHT_LTO "Build with LTO" OFF)
option (OPENDHT_PYTHON "Build Python bindings" OFF)
option (OPENDHT_TOOLS "Build DHT tools" ON)
set (CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -pthread -std=c++11 -Wno-return-type -Wall -Wextra -Wnon-virtual-dtor -pedantic-errors")
set (CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -pthread -std=c++11 -Wno-return-type -Wall -Wextra -Wnon-virtual-dtor")
set (CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -DMSGPACK_DISABLE_LEGACY_NIL -DMSGPACK_DISABLE_LEGACY_CONVERT")
if (OPENDHT_LTO)
set (CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -flto")
set (CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -flto -fuse-linker-plugin")
if (CMAKE_COMPILER_IS_GNUCC)
set (CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -fuse-linker-plugin")
set (CMAKE_AR "gcc-ar")
set (CMAKE_NM "gcc-nm")
set (CMAKE_RANLIB "gcc-ranlib")
SET (CMAKE_AR "gcc-ar")
SET (CMAKE_NM "gcc-nm")
SET (CMAKE_RANLIB "gcc-ranlib")
endif ()
endif ()
@@ -133,7 +132,6 @@ endif ()
if (OPENDHT_TOOLS)
add_subdirectory(tools)
endif ()
add_subdirectory(doc)
if (OPENDHT_PYTHON)
add_subdirectory(python)
+2
View File
@@ -10,7 +10,9 @@ if USE_CYTHON
SUBDIRS += python
endif
if HAVE_DOXYGEN
SUBDIRS += doc
endif
ACLOCAL_AMFLAGS = -I m4
+1 -5
View File
@@ -97,11 +97,7 @@ Licence
-
Copyright (C) 2014-2016 Savoir-faire Linux Inc.
OpenDHT is free software; you can redistribute it and/or modify it under the terms of the GNU General Public License as published by the Free Software Foundation; either version 3 of the License, or (at your option) any later version.
This program is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for more details.
See COPYING or https://www.gnu.org/licenses/gpl-3.0.en.html for the full GPLv3 licence.
Licenced under the GNU General Public License version 3 (see COPYING for the full licence).
This project is independent from another project called OpenDHT (Sean Rhea. Ph.D. Thesis, 2005), now extinct.
+2 -13
View File
@@ -1,21 +1,10 @@
dnl define macros
m4_define([opendht_major_version], 0)
m4_define([opendht_minor_version], 6)
m4_define([opendht_patch_version], 4)
m4_define([opendht_version],
[opendht_major_version.opendht_minor_version.opendht_patch_version])
AC_INIT(opendht, [opendht_version])
AC_INIT(opendht, 0.6.1)
AC_CONFIG_AUX_DIR(ac)
AM_INIT_AUTOMAKE([foreign subdir-objects])
AC_CONFIG_HEADERS([config.h])
AC_CONFIG_MACRO_DIR([m4])
AC_CANONICAL_HOST
AC_SUBST(OPENDHT_MAJOR_VERSION, opendht_major_version)
AC_SUBST(OPENDHT_MINOR_VERSION, opendht_minor_version)
AC_SUBST(OPENDHT_PATCH_VERSION, opendht_patch_version)
AC_ARG_ENABLE([debug], AS_HELP_STRING([--enable-debug], [Build in debug mode, adds stricter warnings, disables optimization]))
AS_IF([test "x$enable_debug" = "xyes"],
[CXXFLAGS="${CXXFLAGS} -g -Wno-return-type -Wall -Wextra -Wnon-virtual-dtor -O0"],
@@ -99,7 +88,7 @@ AM_COND_IF([ENABLE_TOOLS], [
])
])
AC_CONFIG_FILES([doc/Doxyfile doc/Makefile])
AM_COND_IF([HAVE_DOXYGEN], [AC_CONFIG_FILES([doc/Doxyfile doc/Makefile])])
dnl Configure setup.py if we build the python module
AM_COND_IF([USE_CYTHON], [
-3
View File
@@ -1,3 +0,0 @@
if (OPENDHT_TOOLS)
INSTALL(FILES ${CMAKE_CURRENT_SOURCE_DIR}/dhtnode.1 DESTINATION ${CMAKE_INSTALL_PREFIX}/share/man/man1)
endif ()
-4
View File
@@ -1,6 +1,3 @@
dist_man1_MANS = dhtnode.1
if HAVE_DOXYGEN
doxyfile.stamp:
$(DOXYGEN) Doxyfile
echo stamp > doxyfile.stamp
@@ -10,4 +7,3 @@ CLEANFILES = doxyfile.stamp
all-local: doxyfile.stamp
clean-local:
rm -rf $(top_srcdir)/doc/man $(top_srcdir)/doc/html
endif
-87
View File
@@ -1,87 +0,0 @@
.TH DHTNODE 1 2016-07-29
.SH NAME
.B dhtnode
- a simple OpenDHT command line node runner.
.SH SYNOPSIS
.B dhtnode [-h]
.B dhtnode [-v [-l \fIlogfile\fP]] [-i] [-d] [-n \fInetwork_id\fP] [-p \fIlocal_port\fP] [-b \fIbootstrap_host\fP[:\fIport\fP]]
.SH DESCRIPTION
This program runs a simple OpenDHT node in an interactive way. If you rather
want to run the node in daemon mode, option \fB'-d'\fP is provided. When running
in the interactive shell, you benefit from the readline capabilities for writing
your commands such as command history. Here are the available commands in the
interactive shell:
.EE
h, help Print this help message.
q, quit Quit the program.
log Start/stop printing DHT logs.
Node information:
ll Print basic information and stats about the current node.
ls Print basic information about current searches.
ld Print basic information about currenty stored values on this node.
lr Print the full current routing table of this node
Operations on the DHT:
b ip:port Ping potential node at given IP address/port.
g [key] Get values at [key].
l [key] Listen for value changes at [key].
p [key] [str] Put string value at [key].
s [key] [str] Put string value at [key], signed with our generated
private key.
e [key] [dest] [str] Put string value at [key], encrypted for [dest] with
its public key (if found).
.SH OPTIONS
.TP
\fB-h\fP
Prints some help.
.TP
\fB-v\fP
Enable the verbose mode (log to stdout by default)
.TP
\fB-l\fP \fIlog_file\fP
Write log to file instead of stdout
.TP
\fB-i\fP
Generate cryptographic identity for the node.
.TP
\fB-d\fP
Run the program in daemon mode (will fork in the background).
.TP
\fB-n\fP \fInetwork_id\fP
Specify the network id. This let you connect to distinct networks and prevents
the merge of two different networks (available since OpenDHT v0.6.1).
.TP
\fB-p\fP \fIlocal_port\fP
Use port \fIlocal_port\fP for the program to bind to.
.TP
\fB-b\fP \fIbootstrap_host\fP[:\fIport\fP]
The program needs to be given a node to connect to the network. You use this
option to provide the ip address of that node.
.SH AUTHORS
.TP
Program written by
.IP \(bu
.\}
Adrien Béraud <adrien.beraud@savoirfairelinux.com>
.TP
Man page written by
.IP \(bu
.\}
Simon Désaulniers <sim.desaulniers@gmail.com>
+2 -2
View File
@@ -2,5 +2,5 @@ FROM aberaud/opendht-deps
MAINTAINER Adrien Béraud <adrien.beraud@savoirfairelinux.com>
RUN git clone https://github.com/savoirfairelinux/opendht.git \
&& cd opendht && mkdir build && cd build \
&& cmake .. -DCMAKE_INSTALL_PREFIX=/usr -DOPENDHT_PYTHON=On -DOPENDHT_LTO=On && make -j8 && make install \
&& cd ../.. && rm -rf opendht
&& cmake .. -DCMAKE_INSTALL_PREFIX=/usr && make -j8 && make install \
&& cd ../.. && rm -rf opendht
+5 -5
View File
@@ -1,9 +1,9 @@
FROM ubuntu:16.04
MAINTAINER Adrien Béraud <adrien.beraud@savoirfairelinux.com>
RUN apt-get update && apt-get install -y build-essential cmake git wget libncurses5-dev libreadline-dev nettle-dev libgnutls-dev cython3 python3-dev python3-setuptools && apt-get clean
RUN wget https://github.com/msgpack/msgpack-c/releases/download/cpp-2.0.0/msgpack-2.0.0.tar.gz \
&& tar -xzf msgpack-2.0.0.tar.gz \
&& cd msgpack-2.0.0 && mkdir build && cd build \
RUN apt-get update && apt-get install -y build-essential cmake git wget libncurses5-dev libreadline-dev nettle-dev libgnutls-dev cython3 python3-dev python3-setuptools
RUN wget https://github.com/msgpack/msgpack-c/releases/download/cpp-1.4.2/msgpack-1.4.2.tar.gz \
&& tar -xzf msgpack-1.4.2.tar.gz \
&& cd msgpack-1.4.2 && mkdir build && cd build \
&& cmake -DMSGPACK_CXX11=ON -DMSGPACK_BUILD_EXAMPLES=OFF -DCMAKE_INSTALL_PREFIX=/usr .. \
&& make -j8 && make install \
&& cd ../.. && rm -rf msgpack-2.0.0 msgpack-2.0.0.tar.gz
&& cd ../.. && rm -rf msgpack-1.4.2
+5 -7
View File
@@ -1,13 +1,11 @@
FROM ubuntu:16.04
MAINTAINER Adrien Béraud <adrien.beraud@savoirfairelinux.com>
RUN apt-get update \
&& apt-get install -y llvm llvm-dev clang make cmake git wget libncurses5-dev libreadline-dev nettle-dev libgnutls-dev cython3 python3-dev python3-setuptools \
&& apt-get install -y llvm clang make cmake git wget libncurses5-dev libreadline-dev nettle-dev libgnutls-dev cython3 python3-dev python3-setuptools \
&& apt-get remove -y gcc g++ && apt-get autoremove -y && apt-get clean
ENV CC cc
ENV CXX c++
RUN wget https://github.com/msgpack/msgpack-c/releases/download/cpp-2.0.0/msgpack-2.0.0.tar.gz \
&& tar -xzf msgpack-2.0.0.tar.gz \
&& cd msgpack-2.0.0 && mkdir build && cd build \
RUN wget https://github.com/msgpack/msgpack-c/releases/download/cpp-1.4.2/msgpack-1.4.2.tar.gz \
&& tar -xzf msgpack-1.4.2.tar.gz \
&& cd msgpack-1.4.2 && mkdir build && cd build \
&& cmake -DMSGPACK_CXX11=ON -DMSGPACK_BUILD_EXAMPLES=OFF -DCMAKE_INSTALL_PREFIX=/usr .. \
&& make -j8 && make install \
&& cd ../.. && rm -rf msgpack-2.0.0 msgpack-2.0.0.tar.gz
&& cd ../.. && rm -rf msgpack-1.4.2
+2 -2
View File
@@ -2,5 +2,5 @@ FROM aberaud/opendht-deps-llvm
MAINTAINER Adrien Béraud <adrien.beraud@savoirfairelinux.com>
RUN git clone https://github.com/savoirfairelinux/opendht.git \
&& cd opendht && mkdir build && cd build \
&& cmake .. -DCMAKE_INSTALL_PREFIX=/usr -DOPENDHT_PYTHON=On && make -j8 && make install \
&& cd ../.. && rm -rf opendht
&& cmake .. -DCMAKE_INSTALL_PREFIX=/usr && make -j8 && make install \
&& cd ../.. && rm -rf opendht
+1 -1
View File
@@ -2,5 +2,5 @@ FROM aberaud/opendht-deps
MAINTAINER Adrien Béraud <adrien.beraud@savoirfairelinux.com>
COPY . /root/opendht
RUN cd /root/opendht && mkdir build && cd build \
&& cmake -DCMAKE_INSTALL_PREFIX=/usr -DOPENDHT_PYTHON=On -DOPENDHT_LTO=On .. && make -j8 && make install \
&& cmake -DCMAKE_INSTALL_PREFIX=/usr .. && make -j8 && make install \
&& cd ../.. && rm -rf opendht
+1 -1
View File
@@ -2,5 +2,5 @@ FROM aberaud/opendht-deps-llvm
MAINTAINER Adrien Béraud <adrien.beraud@savoirfairelinux.com>
COPY . /root/opendht
RUN cd /root/opendht && mkdir build && cd build \
&& cmake -DCMAKE_INSTALL_PREFIX=/usr -DOPENDHT_PYTHON=On .. && make -j8 && make install \
&& cmake -DCMAKE_INSTALL_PREFIX=/usr .. && make -j8 && make install \
&& cd ../.. && rm -rf opendht
+2 -21
View File
@@ -30,44 +30,23 @@ namespace dht {
struct Node;
/**
* Current status of a DHT node.
*/
enum class NodeStatus {
Disconnected, // 0 nodes
Connecting, // 1+ nodes
Connected // 1+ good nodes
};
/**
* Dht configuration.
*/
struct Config {
/** DHT node ID */
InfoHash node_id;
/**
* DHT network ID. A node will only talk with other nodes having
* the same network ID.
* Network ID 0 (default) represents the main public network.
*/
NetId network;
/** For testing purposes only, enables bootstrap mode */
bool is_bootstrap;
};
/**
* SecureDht configuration.
*/
struct SecureDhtConfig
{
Config node_config;
crypto::Identity id;
};
static constexpr size_t DEFAULT_STORAGE_LIMIT {1024 * 1024 * 64};
using ValuesExport = std::pair<InfoHash, Blob>;
using GetCallback = std::function<bool(const std::vector<std::shared_ptr<Value>>& values)>;
@@ -78,6 +57,8 @@ using CertificateStoreQuery = std::function<std::vector<std::shared_ptr<crypto::
typedef bool (*GetCallbackRaw)(std::shared_ptr<Value>, void *user_data);
static constexpr size_t DEFAULT_STORAGE_LIMIT {1024 * 1024 * 64};
GetCallbackSimple bindGetCb(GetCallbackRaw raw_cb, void* user_data);
GetCallback bindGetCb(GetCallbackSimple cb);
+2 -2
View File
@@ -331,7 +331,7 @@ Identity generateIdentity(const std::string& name = "dhtnode", Identity ca = {},
/**
* SHA512
*/
Blob hash(const Blob& data, size_t hash_len = 512/8);
Blob hash(const Blob& data);
/**
* Generates an encryption key from a text password,
@@ -340,7 +340,7 @@ Blob hash(const Blob& data, size_t hash_len = 512/8);
* that can be transmitted in clear, and will be generated if
* not provided (32 bytes).
*/
Blob stretchKey(const std::string& password, Blob& salt, size_t key_length = 512/8);
Blob stretchKey(const std::string& password, Blob& salt);
/**
* AES-GCM encryption. Key must be 128, 192 or 256 bits long (16, 24 or 32 bytes).
+4 -4
View File
@@ -29,7 +29,7 @@ enum class ImStatus : uint8_t {
READ
};
}
MSGPACK_ADD_ENUM(dht::ImStatus)
MSGPACK_ADD_ENUM(dht::ImStatus);
namespace dht {
@@ -55,7 +55,7 @@ public:
std::string service;
Blob data;
MSGPACK_DEFINE(service, data)
MSGPACK_DEFINE(service, data);
};
template <typename T>
@@ -126,7 +126,7 @@ public:
long date {0};
ImStatus status {ImStatus::NONE};
MSGPACK_DEFINE_MAP(id, msg, date, status)
MSGPACK_DEFINE_MAP(id, msg, date, status);
};
class TrustRequest : public EncryptedValue<TrustRequest>
@@ -147,7 +147,7 @@ public:
std::string service;
Blob payload;
MSGPACK_DEFINE(service, payload)
MSGPACK_DEFINE(service, payload);
};
class IceCandidates : public EncryptedValue<IceCandidates>
+10 -18
View File
@@ -117,20 +117,14 @@ public:
* The node is not pinged, so this should be
* used to bootstrap efficiently from previously known nodes.
*/
bool insertNode(const InfoHash& id, const SockAddr&);
bool insertNode(const InfoHash& id, const sockaddr* sa, socklen_t salen) {
return insertNode(id, SockAddr(sa, salen));
}
bool insertNode(const InfoHash& id, const sockaddr*, socklen_t);
bool insertNode(const NodeExport& n) {
return insertNode(n.id, SockAddr(n.ss, n.sslen));
return insertNode(n.id, reinterpret_cast<const sockaddr*>(&n.ss), n.sslen);
}
int pingNode(const sockaddr*, socklen_t);
time_point periodic(const uint8_t *buf, size_t buflen, const SockAddr&);
time_point periodic(const uint8_t *buf, size_t buflen, const sockaddr* from, socklen_t fromlen) {
return periodic(buf, buflen, SockAddr(from, fromlen));
}
time_point periodic(const uint8_t *buf, size_t buflen, const sockaddr *from, socklen_t fromlen);
/**
* Get a value by searching on all available protocols (IPv4, IPv6),
@@ -250,7 +244,7 @@ public:
std::vector<ValuesExport> exportValues() const;
void importValues(const std::vector<ValuesExport>&);
unsigned getNodesStats(sa_family_t af, unsigned *good_return, unsigned *dubious_return, unsigned *cached_return,
int getNodesStats(sa_family_t af, unsigned *good_return, unsigned *dubious_return, unsigned *cached_return,
unsigned *incoming_return) const;
std::string getStorageLog() const;
std::string getRoutingTablesLog(sa_family_t) const;
@@ -276,7 +270,7 @@ public:
return {total_store_size, total_values};
}
std::vector<SockAddr> getPublicAddress(sa_family_t family = 0);
std::vector<Address> getPublicAddress(sa_family_t family = 0);
protected:
Logger DHT_LOG;
@@ -398,7 +392,7 @@ private:
RoutingTable buckets {};
RoutingTable buckets6 {};
std::vector<std::unique_ptr<Storage>> store;
std::vector<Storage> store;
size_t total_values {0};
size_t total_store_size {0};
size_t max_store_size {DEFAULT_STORAGE_LIMIT};
@@ -418,10 +412,8 @@ private:
time_point mybucket_grow_time {time_point::min()}, mybucket6_grow_time {time_point::min()};
NetworkEngine network_engine;
unsigned pending_pings4 {0};
unsigned pending_pings6 {0};
using ReportedAddr = std::pair<unsigned, SockAddr>;
using ReportedAddr = std::pair<unsigned, Address>;
std::vector<ReportedAddr> reported_addr;
void rotateSecrets();
@@ -429,7 +421,7 @@ private:
Blob makeToken(const sockaddr *sa, bool old) const;
bool tokenMatch(const Blob& token, const sockaddr *sa) const;
void reportedAddr(const SockAddr&);
void reportedAddr(const sockaddr *sa, socklen_t sa_len);
// Storage
decltype(Dht::store)::iterator findStorage(const InfoHash& id);
@@ -505,11 +497,11 @@ private:
bool neighbourhoodMaintenance(RoutingTable&);
void processMessage(const uint8_t *buf, size_t buflen, const SockAddr&);
void processMessage(const uint8_t *buf, size_t buflen, const sockaddr *from, socklen_t fromlen);
void onError(std::shared_ptr<Request> node, DhtProtocolException e);
/* when our address is reported by a distant peer. */
void onReportedAddr(const InfoHash& id, const SockAddr&);
void onReportedAddr(const InfoHash& id, sockaddr* sa , socklen_t salen);
/* when we receive a ping request */
NetworkEngine::RequestAnswer onPing(std::shared_ptr<Node> node);
/* when we receive a "find node" request */
+14 -15
View File
@@ -156,16 +156,16 @@ public:
void cancelListen(InfoHash h, size_t token);
void cancelListen(InfoHash h, std::shared_future<size_t> token);
void put(InfoHash hash, std::shared_ptr<Value> value, DoneCallback cb={}, time_point created=time_point::max(), bool permanent = false);
void put(InfoHash hash, std::shared_ptr<Value> value, DoneCallbackSimple cb, time_point created=time_point::max(), bool permanent = false) {
put(hash, value, bindDoneCb(cb), created, permanent);
void put(InfoHash hash, std::shared_ptr<Value> value, DoneCallback cb={}, bool permanent = false);
void put(InfoHash hash, std::shared_ptr<Value> value, DoneCallbackSimple cb, bool permanent = false) {
put(hash, value, bindDoneCb(cb), permanent);
}
void put(InfoHash hash, Value&& value, DoneCallback cb={}, time_point created=time_point::max(), bool permanent = false);
void put(InfoHash hash, Value&& value, DoneCallbackSimple cb, time_point created=time_point::max(), bool permanent = false) {
put(hash, std::forward<Value>(value), bindDoneCb(cb), created, permanent);
void put(InfoHash hash, Value&& value, DoneCallback cb={}, bool permanent = false);
void put(InfoHash hash, Value&& value, DoneCallbackSimple cb, bool permanent = false) {
put(hash, std::forward<Value>(value), bindDoneCb(cb), permanent);
}
void put(const std::string& key, Value&& value, DoneCallbackSimple cb={}, time_point created=time_point::max(), bool permanent = false);
void put(const std::string& key, Value&& value, DoneCallbackSimple cb={}, bool permanent = false);
void cancelPut(const InfoHash& h, const Value::Id& id);
@@ -212,7 +212,7 @@ public:
* Returns the currently bound address.
* @param f: address family of the bound address to retreive.
*/
const SockAddr& getBound(sa_family_t f = AF_INET) const {
const Address& getBound(sa_family_t f = AF_INET) const {
return (f == AF_INET) ? bound4 : bound6;
}
@@ -248,7 +248,7 @@ public:
std::string getStorageLog() const;
std::string getRoutingTablesLog(sa_family_t af) const;
std::string getSearchesLog(sa_family_t af = 0) const;
std::vector<SockAddr> getPublicAddress(sa_family_t af = 0);
std::vector<Address> getPublicAddress(sa_family_t af = 0);
std::vector<std::string> getPublicAddressStr(sa_family_t af = 0);
// securedht methods
@@ -268,13 +268,12 @@ public:
* @param threaded: If false, ::loop() must be called periodically. Otherwise a thread is launched.
* @param cb: Optional callback to receive general state information.
*/
void run(in_port_t port, const crypto::Identity identity, bool threaded = false, NetId network = 0) {
void run(in_port_t port, const crypto::Identity identity, bool threaded = false, bool is_bootstrap = false) {
run(port, {
/*.dht_config = */{
/*.node_config = */{
/*.node_id = */{},
/*.network = */network,
/*.is_bootstrap = */false
/*.is_bootstrap = */is_bootstrap
},
/*.id = */identity
},
@@ -342,7 +341,7 @@ private:
std::thread rcv_thread {};
std::mutex sock_mtx {};
std::vector<std::pair<Blob, SockAddr>> rcv {};
std::vector<std::pair<Blob, std::pair<sockaddr_storage, socklen_t>>> rcv {};
std::queue<std::function<void(SecureDht&)>> pending_ops_prio {};
std::queue<std::function<void(SecureDht&)>> pending_ops {};
@@ -354,8 +353,8 @@ private:
status6 {NodeStatus::Disconnected};
StatusCallback statusCb {nullptr};
SockAddr bound4 {};
SockAddr bound6 {};
Address bound4 {};
Address bound6 {};
};
}
-1
View File
@@ -197,7 +197,6 @@ public:
static InfoHash getRandom();
friend std::ostream& operator<< (std::ostream& s, const InfoHash& h);
friend std::istream& operator>> (std::istream& s, InfoHash& h);
std::string toString() const;
+26 -23
View File
@@ -195,7 +195,7 @@ private:
* @param saddr (type: sockaddr*) sockaddr* pointer containing address ip information.
* @param saddr_len (type: socklen_t) lenght of the sockaddr struct.
*/
std::function<void(const InfoHash&, const SockAddr&)> onReportedAddr;
std::function<void(const InfoHash&, sockaddr*, socklen_t)> onReportedAddr;
/**
* @brief on ping request callback.
*
@@ -256,7 +256,7 @@ public:
using RequestExpiredCb = std::function<void(const Request&, bool)>;
NetworkEngine(Logger& log, Scheduler& scheduler) : myid(zeroes), DHT_LOG(log), scheduler(scheduler) {}
NetworkEngine(InfoHash& myid, NetId net, int s, int s6, Logger& log, Scheduler& scheduler,
NetworkEngine(InfoHash& myid, int s, int s6, Logger& log, Scheduler& scheduler,
decltype(NetworkEngine::onError) onError,
decltype(NetworkEngine::onNewNode) onNewNode,
decltype(NetworkEngine::onReportedAddr) onReportedAddr,
@@ -266,7 +266,7 @@ public:
decltype(NetworkEngine::onListen) onListen,
decltype(NetworkEngine::onAnnounce) onAnnounce) :
onError(onError), onNewNode(onNewNode), onReportedAddr(onReportedAddr), onPing(onPing), onFindNode(onFindNode),
onGetValues(onGetValues), onListen(onListen), onAnnounce(onAnnounce), myid(myid), network(net),
onGetValues(onGetValues), onListen(onListen), onAnnounce(onAnnounce), myid(myid),
dht_socket(s), dht_socket6(s6), DHT_LOG(log), scheduler(scheduler)
{
transaction_id = std::uniform_int_distribution<decltype(transaction_id)>{1}(rd_device);
@@ -342,10 +342,10 @@ public:
* @param fromlen The length of the corresponding sockaddr structure.
* @param now The time to adjust the clock in the network engine.
*/
void processMessage(const uint8_t *buf, size_t buflen, const SockAddr& addr);
void processMessage(const uint8_t *buf, size_t buflen, const sockaddr* from, socklen_t fromlen);
std::shared_ptr<Node> insertNode(const InfoHash& myid, const SockAddr& addr) {
return cache.getNode(myid, addr, scheduler.time(), 0);
std::shared_ptr<Node> insertNode(const InfoHash& myid, const sockaddr* from, socklen_t fromlen) {
return cache.getNode(myid, from, fromlen, scheduler.time(), 0);
}
std::vector<unsigned> getNodeMessageStats(bool in) {
@@ -374,10 +374,20 @@ private:
/* TODO */
static const std::string my_v;
/* DHT info */
const InfoHash& myid;
const int dht_socket {-1};
const int dht_socket6 {-1};
const Logger& DHT_LOG;
NodeCache cache {};
sockaddr_storage blacklist[BLACKLISTED_MAX] {};
unsigned next_blacklisted = 0;
bool rateLimit();
static bool isMartian(const SockAddr& addr);
bool isNodeBlacklisted(const SockAddr& addr) const;
static bool isMartian(const sockaddr* sa, socklen_t len);
bool isNodeBlacklisted(const sockaddr*, socklen_t) const;
void requestStep(std::shared_ptr<Request> req);
@@ -407,15 +417,16 @@ private:
// basic wrapper for socket sendto function
int send(const char *buf, size_t len, int flags, const SockAddr& addr);
int send(const char *buf, size_t len, int flags, const sockaddr *sa, socklen_t salen);
/*************
* Answers *
*************/
/* answer to a ping request */
void sendPong(const SockAddr& addr, TransId tid);
void sendPong(const sockaddr* sa, socklen_t salen, TransId tid);
/* answer to findnodes/getvalues request */
void sendNodesValues(const SockAddr& addr,
void sendNodesValues(const sockaddr* sa,
socklen_t salen,
TransId tid,
const Blob& nodes,
const Blob& nodes6,
@@ -429,11 +440,12 @@ private:
std::vector<std::shared_ptr<Node>>& nodes,
std::vector<std::shared_ptr<Node>>& nodes6);
/* answer to a listen request */
void sendListenConfirmation(const SockAddr& addr, TransId tid);
void sendListenConfirmation(const sockaddr* sa, socklen_t salen, TransId tid);
/* answer to put request */
void sendValueAnnounced(const SockAddr& addr, TransId, Value::Id);
void sendValueAnnounced(const sockaddr* sa, socklen_t salen, TransId, Value::Id);
/* answer in case of error */
void sendError(const SockAddr& addr,
void sendError(const sockaddr* sa,
socklen_t salen,
TransId tid,
uint16_t code,
const std::string& message,
@@ -441,14 +453,6 @@ private:
void deserializeNodesValues(ParsedMessage& msg);
/* DHT info */
const InfoHash& myid;
const NetId network {0};
const int dht_socket {-1};
const int dht_socket6 {-1};
const Logger& DHT_LOG;
NodeCache cache {};
std::queue<time_point> rate_limit_time {};
static std::mt19937 rd_device;
@@ -456,7 +460,6 @@ private:
uint16_t transaction_id {1};
std::map<uint16_t, std::shared_ptr<Request>> requests {};
MessageStats in_stats {}, out_stats {};
std::set<SockAddr> blacklist {};
Scheduler& scheduler;
};
+17 -9
View File
@@ -30,34 +30,40 @@ namespace dht {
struct Request;
struct Node {
friend class NetworkEngine;
InfoHash id;
SockAddr addr;
socklen_t sslen {0};
sockaddr_storage ss;
time_point time {time_point::min()}; /* last time eared about */
time_point reply_time {time_point::min()}; /* time of last correct reply received */
Node(const InfoHash& id, const sockaddr* sa, socklen_t salen)
: id(id), addr(sa, salen) {}
Node(const InfoHash& id, const SockAddr& addr) : id(id), addr(addr) {}
: id(id), sslen(salen), ss() {
std::copy_n((const uint8_t*)sa, salen, (uint8_t*)&ss);
if ((unsigned)salen < sizeof(ss))
std::fill_n((uint8_t*)&ss+salen, sizeof(ss)-salen, 0);
}
InfoHash getId() const {
return id;
}
std::pair<const sockaddr*, socklen_t> getAddr() const {
return {(const sockaddr*)&addr.first, addr.second};
return {(const sockaddr*)&ss, sslen};
}
std::string getAddrStr() const {
return addr.toString();
return print_addr(ss, sslen);
}
bool isExpired() const { return expired_; }
bool isGood(time_point now) const;
bool isPendingMessage() const;
size_t getPendingMessageCount() const;
NodeExport exportNode() const { return NodeExport {id, addr.first, addr.second}; }
sa_family_t getFamily() const { return addr.getFamily(); }
NodeExport exportNode() const { return NodeExport {id, ss, sslen}; }
sa_family_t getFamily() const { return ss.ss_family; }
void update(const SockAddr&);
void update(const sockaddr* sa, socklen_t salen);
void requested(std::shared_ptr<Request>& req);
void received(time_point now, std::shared_ptr<Request> req);
@@ -82,6 +88,7 @@ struct Node {
static constexpr const std::chrono::seconds MAX_RESPONSE_TIME {1};
private:
std::list<std::weak_ptr<Request>> requests_ {};
bool expired_ {false};
@@ -90,6 +97,7 @@ private:
return w.expired();
});
}
};
}
+2 -2
View File
@@ -28,7 +28,7 @@ namespace dht {
struct NodeCache {
std::shared_ptr<Node> getNode(const InfoHash& id, sa_family_t family);
std::shared_ptr<Node> getNode(const InfoHash& id, const SockAddr&, time_point now, int confirmed);
std::shared_ptr<Node> getNode(const InfoHash& id, const sockaddr* sa, socklen_t sa_len, time_point now, int confirmed);
/**
* Reset the connectivity state of every node,
@@ -41,7 +41,7 @@ private:
class NodeMap : public std::map<InfoHash, std::weak_ptr<Node>> {
public:
std::shared_ptr<Node> getNode(const InfoHash& id);
std::shared_ptr<Node> getNode(const InfoHash& id, const SockAddr&, time_point now, int confirmed);
std::shared_ptr<Node> getNode(const InfoHash& id, const sockaddr* sa, socklen_t sa_len, time_point now, int confirmed);
void clearBadNodes();
};
-2
View File
@@ -23,8 +23,6 @@
namespace dht {
static constexpr unsigned TARGET_NODES {8};
struct Bucket {
Bucket() : cached() {}
Bucket(sa_family_t af, const InfoHash& f = {}, time_point t = time_point::min())
+5 -40
View File
@@ -40,49 +40,14 @@
namespace dht {
std::string print_addr(const sockaddr* sa, socklen_t slen);
std::string print_addr(const sockaddr_storage& ss, socklen_t sslen);
static constexpr unsigned TARGET_NODES {8};
struct SockAddr : public std::pair<sockaddr_storage, socklen_t> {
public:
using std::pair<sockaddr_storage, socklen_t>::pair;
SockAddr() : pair<sockaddr_storage, socklen_t>::pair({},0) {}
SockAddr(const SockAddr& o) : pair<sockaddr_storage, socklen_t>::pair({},o.second) {
std::copy_n((uint8_t*)&o.first, o.second, (uint8_t*)&first);
}
SockAddr(const sockaddr* sa, socklen_t len) : pair<sockaddr_storage, socklen_t>::pair({},len) {
if (len > sizeof(sockaddr_storage))
throw std::runtime_error("Socket address length is too large");
std::copy_n((uint8_t*)sa, len, (uint8_t*)&first);
}
bool operator<(const SockAddr& o) const {
if (second != o.second)
return second < o.second;
return std::memcmp((uint8_t*)&first, (uint8_t*)&o.first, second) < 0;
}
bool operator==(const SockAddr& o) const {
return second == o.second
&& std::memcmp((uint8_t*)&first, (uint8_t*)&o.first, second) == 0;
}
SockAddr& operator=(const SockAddr& o) {
std::copy_n((const uint8_t*)&o.first, o.second, (uint8_t*)&first);
second = o.second;
return *this;
}
std::string toString() const {
return print_addr(first, second);
}
sa_family_t getFamily() const { return second > sizeof(sa_family_t) ? first.ss_family : AF_UNSPEC; }
};
using NetId = uint32_t;
using Address = std::pair<sockaddr_storage, socklen_t>;
using want_t = int_fast8_t;
std::string printAddr(const SockAddr& addr);
std::string print_addr(const sockaddr* sa, socklen_t slen);
std::string print_addr(const sockaddr_storage& ss, socklen_t sslen);
std::string printAddr(const Address& addr);
template <typename Key, typename Item, typename Condition>
void erase_if(std::map<Key, Item>& map, const Condition& condition)
+8 -11
View File
@@ -68,7 +68,7 @@ using StorePolicy = std::function<bool(InfoHash key, std::shared_ptr<Value>& val
*/
using EditPolicy = std::function<bool(InfoHash key, const std::shared_ptr<Value>& old_val, std::shared_ptr<Value>& new_val, InfoHash from, const sockaddr* from_addr, socklen_t from_len)>;
static constexpr const size_t MAX_VALUE_SIZE {1024 * 16};
static constexpr const size_t MAX_VALUE_SIZE {1024 * 14};
struct ValueType {
typedef uint16_t Id;
@@ -115,7 +115,7 @@ struct ValueType {
struct Value
{
typedef uint64_t Id;
static const constexpr Id INVALID_ID {0};
static const Id INVALID_ID {0};
class Filter : public std::function<bool(const Value&)> {
using std::function<bool(const Value&)>::function;
@@ -135,17 +135,14 @@ struct Value
return f1(v) and f2(v);
};
}
static Filter chainAll(std::vector<Filter>&& set) {
if (set.empty()) return {};
return std::bind([](const Value& v, std::vector<Filter>& s) {
for (const auto& f : s)
static Filter chain(std::initializer_list<Filter> l) {
const std::vector<Filter> list(l.begin(), l.end());
return [list](const Value& v){
for (const auto& f : list)
if (f and not f(v))
return false;
return true;
}, std::placeholders::_1, std::move(set));
}
static Filter chain(std::initializer_list<Filter> l) {
return chainAll(std::vector<Filter>(l.begin(), l.end()));
};
}
static Filter chainOr(Filter&& f1, Filter&& f2) {
if (not f1 or not f2) return AllFilter();
@@ -156,7 +153,7 @@ struct Value
};
static const Filter AllFilter() {
return {};
return [](const Value&){return true;};
}
static Filter TypeFilter(const ValueType& t) {
+10 -11
View File
@@ -31,7 +31,6 @@ from libcpp.map cimport map as map
from libcpp cimport bool
from libcpp.utility cimport pair
from libcpp.string cimport string
from libcpp.memory cimport shared_ptr
from cython.parallel import parallel, prange
from cython.operator cimport dereference as deref, preincrement as inc, predecrement as dec
@@ -47,13 +46,13 @@ cdef inline void shutdown_callback(void* user_data) with gil:
cbs['shutdown']()
ref.Py_DECREF(cbs)
cdef inline bool get_callback(shared_ptr[cpp.Value] value, void *user_data) with gil:
cdef inline bool get_callback(cpp.shared_ptr[cpp.Value] value, void *user_data) with gil:
cb = (<object>user_data)['get']
pv = Value()
pv._value = value
return cb(pv)
cdef inline void done_callback(bool done, cpp.vector[shared_ptr[cpp.Node]]* nodes, void *user_data) with gil:
cdef inline void done_callback(bool done, cpp.vector[cpp.shared_ptr[cpp.Node]]* nodes, void *user_data) with gil:
node_ids = []
for n in deref(nodes):
h = NodeEntry()
@@ -110,7 +109,7 @@ cdef class InfoHash(_WithID):
return h
cdef class Node(_WithID):
cdef shared_ptr[cpp.Node] _node
cdef cpp.shared_ptr[cpp.Node] _node
def getId(self):
h = InfoHash()
h._infohash = self._node.get().getId()
@@ -121,7 +120,7 @@ cdef class Node(_WithID):
return self._node.get().isExpired()
cdef class NodeEntry(_WithID):
cdef cpp.pair[cpp.InfoHash, shared_ptr[cpp.Node]] _v
cdef cpp.pair[cpp.InfoHash, cpp.shared_ptr[cpp.Node]] _v
def getId(self):
h = InfoHash()
h._infohash = self._v.first
@@ -132,7 +131,7 @@ cdef class NodeEntry(_WithID):
return n
cdef class Value(object):
cdef shared_ptr[cpp.Value] _value
cdef cpp.shared_ptr[cpp.Value] _value
def __init__(self, bytes val=b''):
self._value.reset(new cpp.Value(val, len(val)))
def __str__(self):
@@ -164,8 +163,8 @@ cdef class Value(object):
return self._value.get().size()
cdef class NodeSetIter(object):
cdef map[cpp.InfoHash, shared_ptr[cpp.Node]]* _nodes
cdef map[cpp.InfoHash, shared_ptr[cpp.Node]].iterator _curIter
cdef map[cpp.InfoHash, cpp.shared_ptr[cpp.Node]]* _nodes
cdef map[cpp.InfoHash, cpp.shared_ptr[cpp.Node]].iterator _curIter
def __init__(self, NodeSet s):
self._nodes = &s._nodes
self._curIter = self._nodes.begin()
@@ -178,7 +177,7 @@ cdef class NodeSetIter(object):
return h
cdef class NodeSet(object):
cdef map[cpp.InfoHash, shared_ptr[cpp.Node]] _nodes
cdef map[cpp.InfoHash, cpp.shared_ptr[cpp.Node]] _nodes
def size(self):
return self._nodes.size()
def insert(self, NodeEntry l):
@@ -200,7 +199,7 @@ cdef class NodeSet(object):
return h
def __str__(self):
s = ''
cdef map[cpp.InfoHash, shared_ptr[cpp.Node]].iterator it = self._nodes.begin()
cdef map[cpp.InfoHash, cpp.shared_ptr[cpp.Node]].iterator it = self._nodes.begin()
while it != self._nodes.end():
s += deref(it).first.toString().decode() + ' ' + deref(it).second.get().getAddrStr().decode() + '\n'
inc(it)
@@ -216,7 +215,7 @@ cdef class PublicKey(_WithID):
return h
cdef class Certificate(_WithID):
cdef shared_ptr[cpp.Certificate] _cert
cdef cpp.shared_ptr[cpp.Certificate] _cert
def getId(self):
h = InfoHash()
h._infohash = self._cert.get().getId()
+7 -1
View File
@@ -20,11 +20,17 @@ from libcpp cimport bool
from libcpp.string cimport string
from libcpp.vector cimport vector
from libcpp.utility cimport pair
from libcpp.memory cimport shared_ptr
ctypedef uint16_t in_port_t
ctypedef unsigned short int sa_family_t;
cdef extern from "<memory>" namespace "std" nogil:
cdef cppclass shared_ptr[T]:
shared_ptr() except +
T* get()
T operator*()
void reset(T*)
cdef extern from "<functional>" namespace "std" nogil:
cdef cppclass hash[T]:
hash() except +
+13 -10
View File
@@ -208,23 +208,23 @@ class DhtFeatureTest(FeatureTest):
FeatureTest.lock.notify()
def _dhtPut(self, producer, _hash, *values):
for val in values:
with FeatureTest.lock:
vstr = val.__str__()[:100]
DhtNetwork.log('[PUT]:', _hash.toString(), '->', vstr + ("..." if len(vstr) > 100 else ""))
FeatureTest.done += 1
while FeatureTest.done > 0:
with FeatureTest.lock:
for val in values:
vstr = val.__str__()[:100]
DhtNetwork.log('[PUT]:', _hash.toString(), '->', vstr + ("..." if len(vstr) > 100 else ""))
FeatureTest.done += 1
producer.put(_hash, val, DhtFeatureTest.putDoneCb)
FeatureTest.lock.wait()
while FeatureTest.done > 0:
FeatureTest.lock.wait()
def _dhtGet(self, consumer, _hash):
DhtFeatureTest.foreignValues = []
DhtFeatureTest.foreignNodes = []
with FeatureTest.lock:
FeatureTest.done += 1
DhtNetwork.log('[GET]:', _hash.toString())
consumer.get(_hash, DhtFeatureTest.getcb, DhtFeatureTest.getDoneCb)
while FeatureTest.done > 0:
DhtNetwork.log('[GET]:', _hash.toString())
consumer.get(_hash, DhtFeatureTest.getcb, DhtFeatureTest.getDoneCb)
FeatureTest.lock.wait()
def _gottaGetThemAllPokeNodes(self, consumer, hashes, nodes=None):
@@ -356,7 +356,7 @@ class PersistenceTest(DhtFeatureTest):
bootstrap = self.bootstrap
# Value representing an ICE packet. Each ICE packet is around 1KB.
VALUE_SIZE = 1024
NUM_VALUES = self._num_values/wb.node_num if self._num_values else 50
NUM_VALUES = self._num_values/wb.node_num if self._num_values else 5
nr_values = NUM_VALUES * wb.node_num
nr_nodes = wb.node_num
nr_nodes_cv = threading.Condition()
@@ -365,8 +365,11 @@ class PersistenceTest(DhtFeatureTest):
hashes = [random_hash() for _ in range(wb.node_num)]
# initial set of values
i = 0
for h in hashes:
self._dhtPut(bootstrap.front(), h, *[Value(v) for v in values])
print("at: ", i)
i += 1
def normalBehavior(do, t, log=None):
nonlocal nr_values
+1 -1
View File
@@ -3,7 +3,7 @@ SUBDIRS = argon2
lib_LTLIBRARIES = libopendht.la
AM_CPPFLAGS = -I../include/opendht
libopendht_la_LDFLAGS = @LDFLAGS@ @GnuTLS_LIBS@ @Nettle_LIBS@ -version-number @OPENDHT_MAJOR_VERSION@:@OPENDHT_MINOR_VERSION@:@OPENDHT_PATCH_VERSION@
libopendht_la_LDFLAGS = @LDFLAGS@ @GnuTLS_LIBS@ @Nettle_LIBS@
libopendht_la_LIBADD = ./argon2/libargon2.la
libopendht_la_SOURCES = \
+10 -12
View File
@@ -88,12 +88,18 @@ int argon2_hash(const uint32_t t_cost, const uint32_t m_cost,
int result;
uint8_t *out;
/* Detect and reject overflowing sizes */
/* TODO: This should probably be fixed in the function signature */
if (pwdlen > ARGON2_MAX_PWD_LENGTH) {
return ARGON2_PWD_TOO_LONG;
}
if (hashlen > ARGON2_MAX_OUTLEN) {
return ARGON2_OUTPUT_TOO_LONG;
}
if (hashlen < ARGON2_MIN_OUTLEN) {
return ARGON2_OUTPUT_TOO_SHORT;
if (saltlen > ARGON2_MAX_SALT_LENGTH) {
return ARGON2_SALT_TOO_LONG;
}
out = malloc(hashlen);
@@ -206,19 +212,13 @@ int argon2_verify(const char *encoded, const void *pwd, const size_t pwdlen,
int ret;
int decode_result;
uint32_t encoded_len;
size_t encoded_len_tmp;
if(encoded == NULL) {
return ARGON2_DECODING_FAIL;
}
encoded_len_tmp = strlen(encoded);
/* max values, to be updated in decode_string */
if (UINT32_MAX < encoded_len_tmp) {
return ARGON2_DECODING_FAIL;
}
encoded_len = (uint32_t)encoded_len_tmp;
encoded_len = strlen(encoded);
ctx.adlen = encoded_len;
ctx.saltlen = encoded_len;
ctx.outlen = encoded_len;
@@ -226,8 +226,6 @@ int argon2_verify(const char *encoded, const void *pwd, const size_t pwdlen,
ctx.free_cbk = NULL;
ctx.secret = NULL;
ctx.secretlen = 0;
ctx.pwdlen = 0;
ctx.pwd = NULL;
ctx.ad = malloc(ctx.adlen);
ctx.salt = malloc(ctx.saltlen);
ctx.out = malloc(ctx.outlen);
@@ -394,5 +392,5 @@ size_t argon2_encodedlen(uint32_t t_cost, uint32_t m_cost, uint32_t parallelism,
uint32_t saltlen, uint32_t hashlen) {
return strlen("$argon2x$v=$m=,t=,p=$$") + numlen(t_cost) + numlen(m_cost)
+ numlen(parallelism) + b64len(saltlen) + b64len(hashlen)
+ numlen(ARGON2_VERSION_NUMBER) + 1;
+ numlen(ARGON2_VERSION_NUMBER);
}
+1 -12
View File
@@ -412,18 +412,7 @@ int encode_string(char *dst, size_t dst_len, argon2_context *ctx,
}
size_t b64len(uint32_t len) {
size_t olen = ((size_t)len / 3) << 2;
switch (len % 3) {
case 2:
olen++;
/* fall through */
case 1:
olen += 2;
break;
}
return olen;
return (((size_t)len + 2) / 3) * 4;
}
size_t numlen(uint32_t num) {
+7 -22
View File
@@ -83,18 +83,6 @@ namespace crypto {
static constexpr std::array<size_t, 3> AES_LENGTHS {{128/8, 192/8, 256/8}};
static constexpr size_t PASSWORD_SALT_LENGTH {16};
constexpr gnutls_digest_algorithm_t gnutlsHashAlgo(size_t min_res) {
return (min_res > 256/8) ? GNUTLS_DIG_SHA512 : (
(min_res > 128/8) ? GNUTLS_DIG_SHA256 : (
GNUTLS_DIG_SHA1));
}
constexpr size_t gnutlsHashSize(int algo) {
return (algo == GNUTLS_DIG_SHA512) ? 512/8 : (
(algo == GNUTLS_DIG_SHA256) ? 256/8 : (
(algo == GNUTLS_DIG_SHA1) ? 160/8 : 0 ));
}
size_t aesKeySize(size_t max)
{
size_t aes_key_len = 0;
@@ -193,7 +181,7 @@ Blob aesDecrypt(const Blob& data, const std::string& password)
return aesDecrypt(encrypted, key);
}
Blob stretchKey(const std::string& password, Blob& salt, size_t key_length)
Blob stretchKey(const std::string& password, Blob& salt)
{
if (salt.empty()) {
salt.resize(PASSWORD_SALT_LENGTH);
@@ -205,19 +193,18 @@ Blob stretchKey(const std::string& password, Blob& salt, size_t key_length)
auto ret = argon2i_hash_raw(16, 64*1024, 1, password.data(), password.size(), salt.data(), salt.size(), res.data(), res.size());
if (ret != ARGON2_OK)
throw CryptoException("Can't compute argon2i !");
return hash(res, key_length);
return hash(res);
}
Blob hash(const Blob& data, size_t hash_len)
Blob hash(const Blob& data)
{
auto algo = gnutlsHashAlgo(hash_len);
size_t res_size = gnutlsHashSize(algo);
Blob res;
size_t res_size = 64;
res.resize(res_size);
const gnutls_datum_t gdat {(uint8_t*)data.data(), (unsigned)data.size()};
if (gnutls_fingerprint(algo, &gdat, res.data(), &res_size))
if (gnutls_fingerprint(GNUTLS_DIG_SHA512, &gdat, res.data(), &res_size))
throw CryptoException("Can't compute hash !");
res.resize(std::min(hash_len, res_size));
res.resize(res_size);
return res;
}
@@ -286,7 +273,7 @@ PrivateKey::PrivateKey(PrivateKey&& o) noexcept : key(o.key), x509_key(o.x509_ke
#endif
o.key = nullptr;
o.x509_key = nullptr;
}
};
PrivateKey::~PrivateKey()
{
@@ -324,8 +311,6 @@ PrivateKey::sign(const Blob& data) const
{
if (!key)
throw CryptoException("Can't sign data: no private key set !");
if (std::numeric_limits<unsigned>::max() < data.size())
throw CryptoException("Can't sign data: too large !");
gnutls_datum_t sig;
const gnutls_datum_t dat {(unsigned char*)data.data(), (unsigned)data.size()};
if (gnutls_privkey_sign_data(key, GNUTLS_DIG_SHA512, 0, &dat, &sig) != GNUTLS_E_SUCCESS)
+86 -98
View File
@@ -170,10 +170,9 @@ private:
size_t total_size {};
};
struct Dht::SearchNode {
SearchNode() : node() {}
SearchNode(std::shared_ptr<Node> node) : node(node) {}
SearchNode(const std::shared_ptr<Node>& node) : node(node) {}
using AnnounceStatusMap = std::map<Value::Id, std::shared_ptr<Request>>;
@@ -373,13 +372,12 @@ NodeStatus
Dht::getStatus(sa_family_t af) const
{
unsigned good = 0, dubious = 0, cached = 0, incoming = 0;
unsigned tot = getNodesStats(af, &good, &dubious, &cached, &incoming);
auto& ping = af == AF_INET ? pending_pings4 : pending_pings6;
if (good)
return NodeStatus::Connected;
if (ping or tot)
int tot = getNodesStats(af, &good, &dubious, &cached, &incoming);
if (tot < 1)
return NodeStatus::Disconnected;
else if (good < 1)
return NodeStatus::Connecting;
return NodeStatus::Disconnected;
return NodeStatus::Connected;
}
void
@@ -397,7 +395,7 @@ Dht::shutdown(ShutdownCallback cb) {
};
for (const auto& str : store) {
*remaining += maintainStorage(str->id, true, str_donecb);
*remaining += maintainStorage(str.id, true, str_donecb);
}
DHT_LOG.WARN("Shuting down node: %u ops remaining.", *remaining);
if (!*remaining && cb) { cb(); }
@@ -443,13 +441,13 @@ Dht::sendCachedPing(Bucket& b)
return 0;
}
std::vector<SockAddr>
std::vector<Address>
Dht::getPublicAddress(sa_family_t family)
{
std::sort(reported_addr.begin(), reported_addr.end(), [](const ReportedAddr& a, const ReportedAddr& b) {
return a.first > b.first;
return a.first < b.first;
});
std::vector<SockAddr> ret;
std::vector<Address> ret;
for (const auto& addr : reported_addr)
if (!family || family == addr.second.first.ss_family)
ret.emplace_back(addr.second);
@@ -476,14 +474,15 @@ Dht::trySearchInsert(const std::shared_ptr<Node>& node)
}
void
Dht::reportedAddr(const SockAddr& addr)
Dht::reportedAddr(const sockaddr *sa, socklen_t sa_len)
{
auto it = std::find_if(reported_addr.begin(), reported_addr.end(), [&](const ReportedAddr& a){
return a.second == addr;
auto it = std::find_if(reported_addr.begin(), reported_addr.end(), [=](const ReportedAddr& addr){
return (addr.second.second == sa_len) &&
std::equal((uint8_t*)&addr.second.first, (uint8_t*)&addr.second.first + addr.second.second, (uint8_t*)sa);
});
if (it == reported_addr.end()) {
if (reported_addr.size() < 32)
reported_addr.emplace_back(1, addr);
reported_addr.emplace_back(1, std::make_pair(*((sockaddr_storage*)sa), sa_len));
} else
it->first++;
}
@@ -1370,12 +1369,12 @@ Dht::listen(const InfoHash& id, GetCallback cb, Value::Filter&& f)
auto st = findStorage(id);
size_t tokenlocal = 0;
if (st == store.end() && store.size() < MAX_HASHES) {
store.emplace_back(new Storage(id, scheduler.time()));
store.emplace_back(id, scheduler.time());
st = std::prev(store.end());
}
if (st != store.end()) {
if (not (*st)->empty()) {
std::vector<std::shared_ptr<Value>> newvals = (*st)->get(f);
if (not st->empty()) {
std::vector<std::shared_ptr<Value>> newvals = st->get(f);
if (not newvals.empty()) {
if (!cb(newvals))
return 0;
@@ -1386,8 +1385,8 @@ Dht::listen(const InfoHash& id, GetCallback cb, Value::Filter&& f)
}
}
}
tokenlocal = ++(*st)->listener_token;
(*st)->local_listeners.emplace(tokenlocal, LocalListener{f, gcb});
tokenlocal = ++st->listener_token;
st->local_listeners.emplace(tokenlocal, LocalListener{f, gcb});
}
auto token4 = Dht::listenTo(id, AF_INET, gcb, f);
@@ -1412,7 +1411,7 @@ Dht::cancelListen(const InfoHash& id, size_t token)
auto st = findStorage(id);
auto tokenlocal = std::get<0>(it->second);
if (st != store.end() && tokenlocal)
(*st)->local_listeners.erase(tokenlocal);
st->local_listeners.erase(tokenlocal);
auto searches_cancel_listen = [&](std::map<InfoHash, std::shared_ptr<Search>> srs) {
for (auto& sp : srs) {
@@ -1488,6 +1487,7 @@ Dht::get(const InfoHash& id, GetCallback getcb, DoneCallback donecb, Value::Filt
auto status = std::make_shared<OpStatus>();
auto status4 = std::make_shared<OpStatus>();
auto status6 = std::make_shared<OpStatus>();
auto vals = std::make_shared<std::vector<std::shared_ptr<Value>>>();
auto all_nodes = std::make_shared<std::vector<std::shared_ptr<Node>>>();
@@ -1545,7 +1545,7 @@ Dht::getLocal(const InfoHash& id, Value::Filter f) const
{
auto s = findStorage(id);
if (s == store.end()) return {};
return (*s)->get(f);
return s->get(f);
}
std::shared_ptr<Value>
@@ -1553,7 +1553,7 @@ Dht::getLocalById(const InfoHash& id, Value::Id vid) const
{
auto s = findStorage(id);
if (s != store.end())
return (*s)->getById(vid);
return s->getById(vid);
return {};
}
@@ -1626,15 +1626,15 @@ Dht::cancelPut(const InfoHash& id, const Value::Id& vid)
decltype(Dht::store)::iterator
Dht::findStorage(const InfoHash& id)
{
return std::find_if(store.begin(), store.end(), [&](const std::unique_ptr<Storage>& st) {
return st->id == id;
return std::find_if(store.begin(), store.end(), [&](const Storage& st) {
return st.id == id;
});
}
decltype(Dht::store)::const_iterator
Dht::findStorage(const InfoHash& id) const
{
return std::find_if(store.cbegin(), store.cend(), [&](const std::unique_ptr<Storage>& st) {
return st->id == id;
return std::find_if(store.cbegin(), store.cend(), [&](const Storage& st) {
return st.id == id;
});
}
@@ -1660,7 +1660,7 @@ Dht::storageChanged(Storage& st, ValueStorage& v)
DHT_LOG.DEBUG("Storage changed. Sending update to %s.", l.first->toString().c_str());
std::vector<std::shared_ptr<Value>> vals;
vals.push_back(v.data);
Blob ntoken = makeToken((const sockaddr*)&l.first->addr.first, false);
Blob ntoken = makeToken((const sockaddr*)&l.first->ss, false);
network_engine.tellListener(l.first, l.second.rid, st.id, 0, ntoken, {}, {}, vals);
}
}
@@ -1670,25 +1670,19 @@ Dht::storageStore(const InfoHash& id, const std::shared_ptr<Value>& value, time_
{
const auto& now = scheduler.time();
created = std::min(created, now);
if (created + getType(value->type).expiration < now) {
DHT_LOG.WARN("[store %s] won't store already expired value", id.toString().c_str());
return false;
}
auto st = findStorage(id);
if (st == store.end()) {
if (store.size() >= MAX_HASHES)
return false;
store.emplace_back(new Storage(id, now));
store.emplace_back(id, now);
st = std::prev(store.end());
}
auto store = (*st)->store(value, created, max_store_size - total_store_size);
auto store = st->store(value, created, max_store_size - total_store_size);
if (std::get<0>(store)) {
total_store_size += std::get<1>(store);
total_values += std::get<2>(store);
storageChanged(*(*st), *std::get<0>(store));
storageChanged(*st, *std::get<0>(store));
}
return std::get<0>(store);
}
@@ -1737,21 +1731,21 @@ Dht::storageAddListener(const InfoHash& id, const std::shared_ptr<Node>& node, s
if (st == store.end()) {
if (store.size() >= MAX_HASHES)
return;
store.emplace_back(new Storage(id, now));
store.emplace_back(id, now);
st = std::prev(store.end());
}
auto l = (*st)->listeners.find(node);
if (l == (*st)->listeners.end()) {
const auto& stvalues = (*st)->getValues();
auto l = st->listeners.find(node);
if (l == st->listeners.end()) {
const auto& stvalues = st->getValues();
if (not stvalues.empty()) {
std::vector<std::shared_ptr<Value>> values(stvalues.size());
std::transform(stvalues.begin(), stvalues.end(), values.begin(), [=](const ValueStorage& vs) { return vs.data; });
network_engine.tellListener(node, rid, id, WANT4 | WANT6, makeToken((sockaddr*)&node->addr.first, false),
network_engine.tellListener(node, rid, id, WANT4 | WANT6, makeToken((sockaddr*)&node->ss, false),
buckets.findClosestNodes(id, now, TARGET_NODES), buckets6.findClosestNodes(id, now, TARGET_NODES),
values);
}
(*st)->listeners.emplace(node, Listener {rid, now});
st->listeners.emplace(node, Listener {rid, now});
}
else
l->second.refresh(rid, now);
@@ -1763,21 +1757,21 @@ Dht::expireStorage()
const auto& now = scheduler.time();
auto i = store.begin();
while (i != store.end()) {
for (auto l = (*i)->listeners.cbegin(); l != (*i)->listeners.cend();){
for (auto l = i->listeners.cbegin(); l != i->listeners.cend();){
bool expired = l->second.time + Node::NODE_EXPIRE_TIME < now;
if (expired) {
DHT_LOG.DEBUG("Discarding expired listener %s", l->first->id.toString().c_str());
(*i)->listeners.erase(l++);
i->listeners.erase(l++);
} else
++l;
}
auto stats = (*i)->expire(types, now);
auto stats = i->expire(types, now);
total_store_size += stats.first;
total_values += stats.second;
if ((*i)->empty() && (*i)->listeners.empty() && (*i)->local_listeners.empty()) {
DHT_LOG.DEBUG("Discarding expired value %s", (*i)->id.toString().c_str());
if (i->empty() && i->listeners.empty() && i->local_listeners.empty()) {
DHT_LOG.DEBUG("Discarding expired value %s", i->id.toString().c_str());
i = store.erase(i);
}
else
@@ -1890,7 +1884,7 @@ Dht::tokenMatch(const Blob& token, const sockaddr *sa) const
return false;
}
unsigned
int
Dht::getNodesStats(sa_family_t af, unsigned *good_return, unsigned *dubious_return, unsigned *cached_return, unsigned *incoming_return) const
{
const auto& now = scheduler.time();
@@ -2020,7 +2014,8 @@ Dht::dumpSearch(const Search& sr, std::ostream& out) const
out << "] ";
}
}
out << n.node->addr.toString() << std::endl;
out << print_addr(n.node->ss, n.node->sslen);
out << std::endl;
}
}
@@ -2057,10 +2052,10 @@ Dht::getStorageLog() const
using namespace std::chrono;
std::stringstream out;
for (const auto& st : store) {
out << "Storage " << (*st).id << " " << (*st).listeners.size() << " list., " << (*st).valueCount() << " values (" << (*st).totalSize() << " bytes)" << std::endl;
if (not (*st).local_listeners.empty())
out << " " << (*st).local_listeners.size() << " local listeners" << std::endl;
for (const auto& l : (*st).listeners) {
out << "Storage " << st.id << " " << st.listeners.size() << " list., " << st.valueCount() << " values (" << st.totalSize() << " bytes)" << std::endl;
if (not st.local_listeners.empty())
out << " " << st.local_listeners.size() << " local listeners" << std::endl;
for (const auto& l : st.listeners) {
out << " " << "Listener " << l.first->toString();
auto since = duration_cast<seconds>(now - l.second.time);
auto expires = duration_cast<seconds>(l.second.time + Node::NODE_EXPIRE_TIME - now);
@@ -2107,10 +2102,10 @@ Dht::Dht() : store(), network_engine(DHT_LOG, scheduler) {}
Dht::Dht(int s, int s6, Config config)
: myid(config.node_id), is_bootstrap(config.is_bootstrap), store(),
network_engine(myid, config.network, s, s6, DHT_LOG, scheduler,
network_engine(myid, s, s6, DHT_LOG, scheduler,
std::bind(&Dht::onError, this, _1, _2),
std::bind(&Dht::onNewNode, this, _1, _2),
std::bind(&Dht::onReportedAddr, this, _1, _2),
std::bind(&Dht::onReportedAddr, this, _1, _2, _3),
std::bind(&Dht::onPing, this, _1),
std::bind(&Dht::onFindNode, this, _1, _2, _3),
std::bind(&Dht::onGetValues, this, _1, _2, _3),
@@ -2243,11 +2238,11 @@ Dht::dataPersistence() {
const auto& now = scheduler.time();
auto storage_maintenance_time = time_point::max();
for (auto &str : store) {
if (now > str->maintenance_time) {
maintainStorage(str->id);
str->maintenance_time = now + MAX_STORAGE_MAINTENANCE_EXPIRE_TIME;
if (now > str.maintenance_time) {
maintainStorage(str.id);
str.maintenance_time = now + MAX_STORAGE_MAINTENANCE_EXPIRE_TIME;
}
storage_maintenance_time = std::min(storage_maintenance_time, str->maintenance_time);
storage_maintenance_time = std::min(storage_maintenance_time, str.maintenance_time);
}
scheduler.add(storage_maintenance_time, std::bind(&Dht::dataPersistence, this));
}
@@ -2264,7 +2259,7 @@ Dht::maintainStorage(InfoHash id, bool force, DoneCallback donecb) {
auto nodes = buckets.findClosestNodes(id, now);
if (!nodes.empty()) {
if (force || id.xorCmp(nodes.back()->id, myid) < 0) {
for (auto &value : (*local_storage)->getValues()) {
for (auto &value : local_storage->getValues()) {
const auto& vt = getType(value.data->type);
if (force || value.time + vt.expiration > now + MAX_STORAGE_MAINTENANCE_EXPIRE_TIME) {
// gotta put that value there
@@ -2280,7 +2275,7 @@ Dht::maintainStorage(InfoHash id, bool force, DoneCallback donecb) {
auto nodes6 = buckets6.findClosestNodes(id, now);
if (!nodes6.empty()) {
if (force || id.xorCmp(nodes6.back()->id, myid) < 0) {
for (auto &value : (*local_storage)->getValues()) {
for (auto &value : local_storage->getValues()) {
const auto& vt = getType(value.data->type);
if (force || value.time + vt.expiration > now + MAX_STORAGE_MAINTENANCE_EXPIRE_TIME) {
// gotta put that value there
@@ -2295,22 +2290,22 @@ Dht::maintainStorage(InfoHash id, bool force, DoneCallback donecb) {
if (not want4 and not want6) {
DHT_LOG.DEBUG("Discarding storage values %s", id.toString().c_str());
(*local_storage)->clear();
local_storage->clear();
}
return announce_per_af;
}
void
Dht::processMessage(const uint8_t *buf, size_t buflen, const SockAddr& from)
Dht::processMessage(const uint8_t *buf, size_t buflen, const sockaddr *from, socklen_t fromlen)
{
if (buflen == 0)
return;
try {
network_engine.processMessage(buf, buflen, from);
network_engine.processMessage(buf, buflen, from, fromlen);
} catch (const std::exception& e) {
DHT_LOG.ERR("Can't parse message from %s: %s", from.toString().c_str(), e.what());
DHT_LOG.ERR("Can't parse message from %s: %s", print_addr(from, fromlen).c_str(), e.what());
//auto code = e.getCode();
//if (code == DhtProtocolException::INVALID_TID_SIZE or code == DhtProtocolException::WRONG_NODE_INFO_BUF_LEN) {
/* This is really annoying, as it means that we will
@@ -2323,10 +2318,10 @@ Dht::processMessage(const uint8_t *buf, size_t buflen, const SockAddr& from)
}
time_point
Dht::periodic(const uint8_t *buf, size_t buflen, const SockAddr& from)
Dht::periodic(const uint8_t *buf, size_t buflen, const sockaddr *from, socklen_t fromlen)
{
scheduler.syncTime();
processMessage(buf, buflen, from);
processMessage(buf, buflen, from, fromlen);
return scheduler.run();
}
@@ -2350,11 +2345,11 @@ Dht::confirmNodes()
bool soon = false;
const auto& now = scheduler.time();
if (searches4.empty() and getStatus(AF_INET) == NodeStatus::Connected) {
if (searches4.empty() and getStatus(AF_INET) != NodeStatus::Disconnected) {
DHT_LOG.DEBUG("[confirm nodes] initial IPv4 'get' for my id (%s).", myid.toString().c_str());
search(myid, AF_INET);
}
if (searches6.empty() and getStatus(AF_INET6) == NodeStatus::Connected) {
if (searches6.empty() and getStatus(AF_INET6) != NodeStatus::Disconnected) {
DHT_LOG.DEBUG("[confirm nodes] initial IPv6 'get' for my id (%s).", myid.toString().c_str());
search(myid, AF_INET6);
}
@@ -2388,12 +2383,12 @@ Dht::exportValues() const
e.reserve(store.size());
for (const auto& h : store) {
ValuesExport ve;
ve.first = h->id;
ve.first = h.id;
msgpack::sbuffer buffer;
msgpack::packer<msgpack::sbuffer> pk(&buffer);
pk.pack_array(h->getValues().size());
for (const auto& v : h->getValues()) {
pk.pack_array(h.getValues().size());
for (const auto& v : h.getValues()) {
pk.pack_array(2);
pk.pack(v.time.time_since_epoch().count());
v.data->msgpack_pack(pk);
@@ -2477,28 +2472,21 @@ Dht::exportNodes()
}
bool
Dht::insertNode(const InfoHash& id, const SockAddr& addr)
Dht::insertNode(const InfoHash& id, const sockaddr* sa, socklen_t salen)
{
if (addr.getFamily() != AF_INET && addr.getFamily() != AF_INET6)
if (sa->sa_family != AF_INET && sa->sa_family != AF_INET6)
return false;
scheduler.syncTime();
auto n = network_engine.insertNode(id, addr);
auto n = network_engine.insertNode(id, sa, salen);
return !!n;
}
int
Dht::pingNode(const sockaddr* sa, socklen_t salen)
Dht::pingNode(const sockaddr *sa, socklen_t salen)
{
scheduler.syncTime();
DHT_LOG.DEBUG("Sending ping to %s", print_addr(sa, salen).c_str());
auto& count = sa->sa_family == AF_INET ? pending_pings4 : pending_pings6;
count++;
network_engine.sendPing(sa, salen, [&](const Request&, NetworkEngine::RequestAnswer&&){
count--;
}, [&](const Request&, bool last){
if (last)
count--;
});
network_engine.sendPing(sa, salen, nullptr, nullptr);
return -1;
}
@@ -2523,12 +2511,12 @@ Dht::onError(std::shared_ptr<Request> req, DhtProtocolException e) {
}
void
Dht::onReportedAddr(const InfoHash& id, const SockAddr& addr)
Dht::onReportedAddr(const InfoHash& id, sockaddr* addr , socklen_t addr_length)
{
const auto& b = (addr.getFamily() == AF_INET ? buckets : buckets6).findBucket(id);
const auto& b = (addr->sa_family == AF_INET ? buckets : buckets6).findBucket(id);
b->time = scheduler.time();
if (addr.second)
reportedAddr(addr);
if (addr and addr_length)
reportedAddr(addr, addr_length);
}
NetworkEngine::RequestAnswer
@@ -2542,7 +2530,7 @@ Dht::onFindNode(std::shared_ptr<Node> node, InfoHash& target, want_t want)
{
const auto& now = scheduler.time();
NetworkEngine::RequestAnswer answer;
answer.ntoken = makeToken((sockaddr*)&node->addr.first, false);
answer.ntoken = makeToken((sockaddr*)&node->ss, false);
if (want & WANT4)
answer.nodes4 = buckets.findClosestNodes(target, now, TARGET_NODES);
if (want & WANT6)
@@ -2560,11 +2548,11 @@ Dht::onGetValues(std::shared_ptr<Node> node, InfoHash& hash, want_t)
const auto& now = scheduler.time();
NetworkEngine::RequestAnswer answer {};
auto st = findStorage(hash);
answer.ntoken = makeToken((sockaddr*)&node->addr.first, false);
answer.ntoken = makeToken((sockaddr*)&node->ss, false);
answer.nodes4 = buckets.findClosestNodes(hash, now, TARGET_NODES);
answer.nodes6 = buckets6.findClosestNodes(hash, now, TARGET_NODES);
if (st != store.end() && not (*st)->empty()) {
auto values = (*st)->getValues();
if (st != store.end() && not st->empty()) {
auto values = st->getValues();
answer.values.resize(values.size());
std::transform(values.begin(), values.end(), answer.values.begin(), [](const ValueStorage& vs) {
return vs.data;
@@ -2641,7 +2629,7 @@ Dht::onListen(std::shared_ptr<Node> node, InfoHash& hash, Blob& token, size_t ri
DhtProtocolException::LISTEN_NO_INFOHASH
};
}
if (!tokenMatch(token, (sockaddr*)&node->addr.first)) {
if (!tokenMatch(token, (sockaddr*)&node->ss)) {
DHT_LOG.WARN("[node %s] incorrect token %s for 'listen'.", node->toString().c_str(), hash.toString().c_str());
throw DhtProtocolException {DhtProtocolException::UNAUTHORIZED, DhtProtocolException::LISTEN_WRONG_TOKEN};
}
@@ -2679,7 +2667,7 @@ Dht::onAnnounce(std::shared_ptr<Node> node, InfoHash& hash, Blob& token, std::ve
DhtProtocolException::PUT_NO_INFOHASH
};
}
if (!tokenMatch(token, (sockaddr*)&node->addr.first)) {
if (!tokenMatch(token, (sockaddr*)&node->ss)) {
DHT_LOG.WARN("[node %s] incorrect token %s for 'put'.", node->toString().c_str(), hash.toString().c_str());
throw DhtProtocolException {DhtProtocolException::UNAUTHORIZED, DhtProtocolException::PUT_WRONG_TOKEN};
}
@@ -2709,7 +2697,7 @@ Dht::onAnnounce(std::shared_ptr<Node> node, InfoHash& hash, Blob& token, std::ve
DHT_LOG.WARN("[value %s %lu] nothing to do.", hash.toString().c_str(), lv->id);
} else {
const auto& type = getType(lv->type);
if (type.editPolicy(hash, lv, vc, node->id, (sockaddr*)&node->addr.first, node->addr.second)) {
if (type.editPolicy(hash, lv, vc, node->id, (sockaddr*)&node->ss, node->sslen)) {
DHT_LOG.DEBUG("[value %s %lu] editing %s.",
hash.toString().c_str(), lv->id, vc->toString().c_str());
storageStore(hash, vc, created);
@@ -2721,7 +2709,7 @@ Dht::onAnnounce(std::shared_ptr<Node> node, InfoHash& hash, Blob& token, std::ve
} else {
// Allow the value to be edited by the storage policy
const auto& type = getType(vc->type);
if (type.storePolicy(hash, vc, node->id, (sockaddr*)&node->addr.first, node->addr.second)) {
if (type.storePolicy(hash, vc, node->id, (sockaddr*)&node->ss, node->sslen)) {
DHT_LOG.DEBUG("[value %s %lu] storing %s.", hash.toString().c_str(), vc->id, vc->toString().c_str());
storageStore(hash, vc, created);
} else {
+15 -15
View File
@@ -104,7 +104,7 @@ DhtRunner::run(const sockaddr_in* local4, const sockaddr_in6* local6, DhtRunner:
std::lock_guard<std::mutex> lck(storage_mtx);
if (not pending_ops_prio.empty())
return true;
if (not pending_ops.empty() and getStatus() != NodeStatus::Connecting)
if (not pending_ops.empty() and getStatus() >= NodeStatus::Connecting)
return true;
}
return false;
@@ -252,7 +252,7 @@ DhtRunner::getSearchesLog(sa_family_t af) const
std::lock_guard<std::mutex> lck(dht_mtx);
return dht_->getSearchesLog(af);
}
std::vector<SockAddr>
std::vector<Address>
DhtRunner::getPublicAddress(sa_family_t af)
{
std::lock_guard<std::mutex> lck(dht_mtx);
@@ -293,7 +293,7 @@ DhtRunner::loop_()
ops.front()(*dht_);
ops.pop();
}
if (getStatus() != NodeStatus::Connecting) {
if (getStatus() >= NodeStatus::Connecting) {
{
std::lock_guard<std::mutex> lck(storage_mtx);
ops = std::move(pending_ops);
@@ -315,7 +315,7 @@ DhtRunner::loop_()
for (const auto& pck : received) {
auto& buf = pck.first;
auto& from = pck.second;
wakeup = dht_->periodic(buf.data(), buf.size()-1, from);
wakeup = dht_->periodic(buf.data(), buf.size()-1, (sockaddr*)&from.first, from.second);
}
received.clear();
} else {
@@ -400,19 +400,19 @@ DhtRunner::doRun(const sockaddr_in* sin4, const sockaddr_in6* sin6, SecureDht::C
if(rc > 0) {
std::array<uint8_t, 1024 * 64> buf;
SockAddr from;
from.second = sizeof(from.first);
sockaddr_storage from;
socklen_t fromlen {sizeof(from)};
if(s4 >= 0 && FD_ISSET(s4, &readfds))
rc = recvfrom(s4, (char*)buf.data(), buf.size(), 0, (struct sockaddr*)&from.first, &from.second);
rc = recvfrom(s4, (char*)buf.data(), buf.size(), 0, (struct sockaddr*)&from, &fromlen);
else if(s6 >= 0 && FD_ISSET(s6, &readfds))
rc = recvfrom(s6, (char*)buf.data(), buf.size(), 0, (struct sockaddr*)&from.first, &from.second);
rc = recvfrom(s6, (char*)buf.data(), buf.size(), 0, (struct sockaddr*)&from, &fromlen);
else
break;
if (rc > 0) {
{
std::lock_guard<std::mutex> lck(sock_mtx);
rcv.emplace_back(Blob {buf.begin(), buf.begin()+rc+1}, from);
rcv.emplace_back(Blob {buf.begin(), buf.begin()+rc+1}, std::make_pair(from, fromlen));
}
cv.notify_all();
}
@@ -484,30 +484,30 @@ DhtRunner::cancelListen(InfoHash h, std::shared_future<size_t> token)
}
void
DhtRunner::put(InfoHash hash, Value&& value, DoneCallback cb, time_point created, bool permanent)
DhtRunner::put(InfoHash hash, Value&& value, DoneCallback cb, bool permanent)
{
std::lock_guard<std::mutex> lck(storage_mtx);
auto sv = std::make_shared<Value>(std::move(value));
pending_ops.emplace([=](SecureDht& dht) {
dht.put(hash, sv, cb, created, permanent);
dht.put(hash, sv, cb, {}, permanent);
});
cv.notify_all();
}
void
DhtRunner::put(InfoHash hash, std::shared_ptr<Value> value, DoneCallback cb, time_point created, bool permanent)
DhtRunner::put(InfoHash hash, std::shared_ptr<Value> value, DoneCallback cb, bool permanent)
{
std::lock_guard<std::mutex> lck(storage_mtx);
pending_ops.emplace([=](SecureDht& dht) {
dht.put(hash, value, cb, created, permanent);
dht.put(hash, value, cb, {}, permanent);
});
cv.notify_all();
}
void
DhtRunner::put(const std::string& key, Value&& value, DoneCallbackSimple cb, time_point created, bool permanent)
DhtRunner::put(const std::string& key, Value&& value, DoneCallbackSimple cb, bool permanent)
{
put(InfoHash::get(key), std::forward<Value>(value), cb, created, permanent);
put(InfoHash::get(key), std::forward<Value>(value), cb, permanent);
}
void
-13
View File
@@ -91,17 +91,4 @@ std::ostream& operator<< (std::ostream& s, const InfoHash& h)
return s;
}
std::istream& operator>> (std::istream& s, InfoHash& h)
{
std::array<char, HASH_LEN*2> dat;
s.exceptions(std::istream::eofbit | std::istream::failbit);
s.read(&(*dat.begin()), dat.size());
for (size_t i = 0; i < HASH_LEN; i++) {
unsigned res = 0;
sscanf(dat.data() + 2*i, "%02x", &res);
h[i] = res;
}
return s;
}
}
+99 -128
View File
@@ -49,7 +49,6 @@ static const uint8_t v4prefix[16] = {
0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0xFF, 0xFF, 0, 0, 0, 0
};
constexpr unsigned SEND_NODES {8};
enum class MessageType {
Error = 0,
@@ -64,7 +63,6 @@ enum class MessageType {
struct ParsedMessage {
MessageType type;
InfoHash id; /* the id of the sender */
NetId network {0}; /* network id */
InfoHash info_hash; /* hash for which values are requested */
InfoHash target; /* target id around which to find nodes */
NetworkEngine::TransId tid; /* transaction id */
@@ -77,7 +75,7 @@ struct ParsedMessage {
want_t want; /* states if ipv4 or ipv6 request */
uint16_t error_code; /* error code in case of error */
std::string ua;
SockAddr addr; /* reported address by the distant node */
Address addr; /* reported address by the distant node */
void msgpack_unpack(msgpack::object o);
};
@@ -91,7 +89,7 @@ NetworkEngine::tellListener(std::shared_ptr<Node> node, uint16_t rid, InfoHash h
{
auto nnodes = bufferNodes(node->getFamily(), hash, want, nodes, nodes6);
try {
sendNodesValues(node->addr, TransId {TransPrefix::GET_VALUES, (uint16_t)rid}, nnodes.first, nnodes.second,
sendNodesValues((const sockaddr*)&node->ss, node->sslen, TransId {TransPrefix::GET_VALUES, (uint16_t)rid}, nnodes.first, nnodes.second,
values, ntoken);
} catch (const std::overflow_error& e) {
DHT_LOG.ERR("Can't send value: buffer not large enough !");
@@ -157,7 +155,7 @@ NetworkEngine::requestStep(std::shared_ptr<Request> req)
send((char*)req->msg.data(), req->msg.size(),
(req->node->reply_time >= now - UDP_REPLY_TIME) ? 0 : MSG_CONFIRM,
req->node->addr);
(sockaddr*)&req->node->ss, req->node->sslen);
++req->attempt_count;
req->last_try = now;
std::weak_ptr<Request> wreq = req;
@@ -202,15 +200,15 @@ NetworkEngine::rateLimit()
}
bool
NetworkEngine::isMartian(const SockAddr& addr)
NetworkEngine::isMartian(const sockaddr* sa, socklen_t len)
{
// Check that sa_family can be accessed safely
if (addr.second < sizeof(sockaddr_in))
if (!sa || len < sizeof(sockaddr_in))
return true;
switch(addr.first.ss_family) {
switch(sa->sa_family) {
case AF_INET: {
sockaddr_in *sin = (sockaddr_in*)&addr.first;
sockaddr_in *sin = (sockaddr_in*)sa;
const uint8_t *address = (const uint8_t*)&sin->sin_addr;
return sin->sin_port == 0 ||
(address[0] == 0) ||
@@ -218,9 +216,9 @@ NetworkEngine::isMartian(const SockAddr& addr)
((address[0] & 0xE0) == 0xE0);
}
case AF_INET6: {
if (addr.second < sizeof(sockaddr_in6))
if (len < sizeof(sockaddr_in6))
return true;
sockaddr_in6 *sin6 = (sockaddr_in6*)&addr.first;
sockaddr_in6 *sin6 = (sockaddr_in6*)sa;
const uint8_t *address = (const uint8_t*)&sin6->sin6_addr;
return sin6->sin6_port == 0 ||
(address[0] == 0xFF) ||
@@ -249,25 +247,37 @@ NetworkEngine::blacklistNode(const std::shared_ptr<Node>& n)
++rit;
}
}
blacklist.emplace(n->addr);
memcpy(&blacklist[next_blacklisted], &n->ss, n->sslen);
next_blacklisted = (next_blacklisted + 1) % BLACKLISTED_MAX;
}
bool
NetworkEngine::isNodeBlacklisted(const SockAddr& addr) const
NetworkEngine::isNodeBlacklisted(const sockaddr *sa, socklen_t salen) const
{
return blacklist.find(addr) != blacklist.end();
if (salen > sizeof(sockaddr_storage))
return true;
/*if (isBlacklisted(sa, salen))
return true;*/
for (unsigned i = 0; i < BLACKLISTED_MAX; i++) {
if (memcmp(&blacklist[i], sa, salen) == 0)
return true;
}
return false;
}
void
NetworkEngine::processMessage(const uint8_t *buf, size_t buflen, const SockAddr& from)
NetworkEngine::processMessage(const uint8_t *buf, size_t buflen, const sockaddr* from, socklen_t fromlen)
{
if (isMartian(from)) {
DHT_LOG.WARN("Received packet from martian node %s", from.toString().c_str());
if (isMartian(from, fromlen)) {
DHT_LOG.WARN("Received packet from martian node %s", print_addr(from, fromlen).c_str());
return;
}
if (isNodeBlacklisted(from)) {
DHT_LOG.WARN("Received packet from blacklisted node %s", from.toString().c_str());
if (isNodeBlacklisted(from, fromlen)) {
DHT_LOG.WARN("Received packet from blacklisted node %s", print_addr(from, fromlen).c_str());
return;
}
@@ -283,11 +293,6 @@ NetworkEngine::processMessage(const uint8_t *buf, size_t buflen, const SockAddr&
return;
}
if (msg.network != network) {
DHT_LOG.DEBUG("Received message from other network %u.", msg.network);
return;
}
if (msg.id == myid || msg.id == zeroes) {
DHT_LOG.DEBUG("Received message from self.");
return;
@@ -306,7 +311,7 @@ NetworkEngine::processMessage(const uint8_t *buf, size_t buflen, const SockAddr&
if (msg.tid.length != 4) {
DHT_LOG.ERR("Broken node truncates transaction ids (len: %d): ", msg.tid.length);
DHT_LOG.ERR.logPrintable(buf, buflen);
blacklistNode(cache.getNode(msg.id, from, now, 1));
blacklistNode(cache.getNode(msg.id, from, fromlen, now, 1));
return;
}
@@ -321,7 +326,7 @@ NetworkEngine::processMessage(const uint8_t *buf, size_t buflen, const SockAddr&
auto node = req->node;
if (node->id != msg.id) {
bool unknown_node = node->id == zeroes;
node = cache.getNode(msg.id, from, now, 2);
node = cache.getNode(msg.id, from, fromlen, now, 2);
if (unknown_node) {
// received reply to a message sent when we didn't know the node ID.
req->node = node;
@@ -333,11 +338,11 @@ NetworkEngine::processMessage(const uint8_t *buf, size_t buflen, const SockAddr&
return;
}
} else
node->update(from);
node->update(from, fromlen);
node->received(now, req);
onNewNode(node, 2);
onReportedAddr(msg.id, msg.addr);
onReportedAddr(msg.id, (sockaddr*)&msg.addr.first, msg.addr.second);
if (req->cancelled() or req->expired() or (req->completed() and not req->persistent)) {
DHT_LOG.WARN("[node %s] response to expired, cancelled or completed request", node->toString().c_str());
@@ -357,7 +362,7 @@ NetworkEngine::processMessage(const uint8_t *buf, size_t buflen, const SockAddr&
onError(req, DhtProtocolException {DhtProtocolException::UNAUTHORIZED});
} else {
DHT_LOG.WARN("[node %s %s] received unknown error message %u",
msg.id.toString().c_str(), from.toString().c_str(), msg.error_code);
msg.id.toString().c_str(), print_addr(from, fromlen).c_str(), msg.error_code);
DHT_LOG.WARN.logPrintable(buf, buflen);
}
break;
@@ -375,7 +380,7 @@ NetworkEngine::processMessage(const uint8_t *buf, size_t buflen, const SockAddr&
break;
}
} else {
auto node = cache.getNode(msg.id, from, now, 1);
auto node = cache.getNode(msg.id, from, fromlen, now, 1);
node->received(now, {});
onNewNode(node, 1);
try {
@@ -384,29 +389,29 @@ NetworkEngine::processMessage(const uint8_t *buf, size_t buflen, const SockAddr&
++in_stats.ping;
DHT_LOG.DEBUG("Sending pong.");
onPing(node);
sendPong(from, msg.tid);
sendPong(from, fromlen, msg.tid);
break;
case MessageType::FindNode: {
DHT_LOG.DEBUG("[node %s %s] got 'find' request (%d).",
msg.id.toString().c_str(), from.toString().c_str(), msg.want);
msg.id.toString().c_str(), print_addr(from, fromlen).c_str(), msg.want);
++in_stats.find;
RequestAnswer answer = onFindNode(node, msg.target, msg.want);
auto nnodes = bufferNodes(from.getFamily(), msg.target, msg.want, answer.nodes4, answer.nodes6);
sendNodesValues(from, msg.tid, nnodes.first, nnodes.second, {}, answer.ntoken);
auto nnodes = bufferNodes(from->sa_family, msg.target, msg.want, answer.nodes4, answer.nodes6);
sendNodesValues(from, fromlen, msg.tid, nnodes.first, nnodes.second, {}, answer.ntoken);
break;
}
case MessageType::GetValues: {
DHT_LOG.DEBUG("[node %s %s] got 'get' request for %s.",
msg.id.toString().c_str(), from.toString().c_str(), msg.info_hash.toString().c_str());
msg.id.toString().c_str(), print_addr(from, fromlen).c_str(), msg.info_hash.toString().c_str());
++in_stats.get;
RequestAnswer answer = onGetValues(node, msg.info_hash, msg.want);
auto nnodes = bufferNodes(from.getFamily(), msg.info_hash, msg.want, answer.nodes4, answer.nodes6);
sendNodesValues(from, msg.tid, nnodes.first, nnodes.second, answer.values, answer.ntoken);
auto nnodes = bufferNodes(from->sa_family, msg.info_hash, msg.want, answer.nodes4, answer.nodes6);
sendNodesValues(from, fromlen, msg.tid, nnodes.first, nnodes.second, answer.values, answer.ntoken);
break;
}
case MessageType::AnnounceValue: {
DHT_LOG.DEBUG("[node %s %s] got 'put' request for %s.",
msg.id.toString().c_str(), from.toString().c_str(),
msg.id.toString().c_str(), print_addr(from, fromlen).c_str(),
msg.info_hash.toString().c_str());
++in_stats.put;
onAnnounce(node, msg.info_hash, msg.token, msg.values, msg.created);
@@ -415,16 +420,16 @@ NetworkEngine::processMessage(const uint8_t *buf, size_t buflen, const SockAddr&
This is to prevent them from backtracking, and hence
polluting the DHT. */
for (auto& v : msg.values) {
sendValueAnnounced(from, msg.tid, v->id);
sendValueAnnounced(from, fromlen, msg.tid, v->id);
}
break;
}
case MessageType::Listen: {
DHT_LOG.DEBUG("[node %s %s] got 'listen' request for %s.",
msg.id.toString().c_str(), from.toString().c_str(), msg.info_hash.toString().c_str());
msg.id.toString().c_str(), print_addr(from, fromlen).c_str(), msg.info_hash.toString().c_str());
++in_stats.listen;
RequestAnswer answer = onListen(node, msg.info_hash, msg.token, msg.tid.getTid());
sendListenConfirmation(from, msg.tid);
sendListenConfirmation(from, fromlen, msg.tid);
break;
}
default:
@@ -433,7 +438,7 @@ NetworkEngine::processMessage(const uint8_t *buf, size_t buflen, const SockAddr&
} catch (const std::overflow_error& e) {
DHT_LOG.ERR("Can't send value: buffer not large enough !");
} catch (DhtProtocolException& e) {
sendError(from, msg.tid, e.getCode(), e.getMsg().c_str(), true);
sendError(from, fromlen, msg.tid, e.getCode(), e.getMsg().c_str(), true);
}
}
}
@@ -446,34 +451,33 @@ packToken(msgpack::packer<msgpack::sbuffer>& pk, Blob token)
}
void
insertAddr(msgpack::packer<msgpack::sbuffer>& pk, const SockAddr& addr)
insertAddr(msgpack::packer<msgpack::sbuffer>& pk, const sockaddr *sa, socklen_t)
{
size_t addr_len = std::min<size_t>(addr.second,
(addr.getFamily() == AF_INET) ? sizeof(in_addr) : sizeof(in6_addr));
void* addr_ptr = (addr.getFamily() == AF_INET) ? (void*)&((sockaddr_in*)&addr.first)->sin_addr
: (void*)&((sockaddr_in6*)&addr.first)->sin6_addr;
size_t addr_len = (sa->sa_family == AF_INET) ? sizeof(in_addr) : sizeof(in6_addr);
void* addr_ptr = (sa->sa_family == AF_INET) ? (void*)&((sockaddr_in*)sa)->sin_addr
: (void*)&((sockaddr_in6*)sa)->sin6_addr;
pk.pack("sa");
pk.pack_bin(addr_len);
pk.pack_bin_body((char*)addr_ptr, addr_len);
}
int
NetworkEngine::send(const char *buf, size_t len, int flags, const SockAddr& addr)
NetworkEngine::send(const char *buf, size_t len, int flags, const sockaddr *sa, socklen_t salen)
{
if (addr.second == 0)
if (salen == 0)
return -1;
int s;
if (addr.getFamily() == AF_INET)
if (sa->sa_family == AF_INET)
s = dht_socket;
else if (addr.getFamily() == AF_INET6)
else if (sa->sa_family == AF_INET6)
s = dht_socket6;
else
s = -1;
if (s < 0)
return -1;
return sendto(s, buf, len, flags, (const sockaddr*)&addr.first, addr.second);
return sendto(s, buf, len, flags, sa, salen);
}
std::shared_ptr<Request>
@@ -481,7 +485,7 @@ NetworkEngine::sendPing(std::shared_ptr<Node> node, RequestCb on_done, RequestEx
auto tid = TransId {TransPrefix::PING, getNewTid()};
msgpack::sbuffer buffer;
msgpack::packer<msgpack::sbuffer> pk(&buffer);
pk.pack_map(5+(network?1:0));
pk.pack_map(5);
pk.pack(std::string("a")); pk.pack_map(1);
pk.pack(std::string("id")); pk.pack(myid);
@@ -491,9 +495,6 @@ NetworkEngine::sendPing(std::shared_ptr<Node> node, RequestCb on_done, RequestEx
pk.pack_bin_body((const char*)tid.data(), tid.size());
pk.pack(std::string("y")); pk.pack(std::string("q"));
pk.pack(std::string("v")); pk.pack(my_v);
if (network) {
pk.pack(std::string("n")); pk.pack(network);
}
Blob b {buffer.data(), buffer.data() + buffer.size()};
std::shared_ptr<Request> req(new Request {tid.getTid(), node, std::move(b),
@@ -515,24 +516,21 @@ NetworkEngine::sendPing(std::shared_ptr<Node> node, RequestCb on_done, RequestEx
}
void
NetworkEngine::sendPong(const SockAddr& addr, TransId tid) {
NetworkEngine::sendPong(const sockaddr* sa, socklen_t salen, TransId tid) {
msgpack::sbuffer buffer;
msgpack::packer<msgpack::sbuffer> pk(&buffer);
pk.pack_map(4+(network?1:0));
pk.pack_map(4);
pk.pack(std::string("r")); pk.pack_map(2);
pk.pack(std::string("id")); pk.pack(myid);
insertAddr(pk, addr);
insertAddr(pk, sa, salen);
pk.pack(std::string("t")); pk.pack_bin(tid.size());
pk.pack_bin_body((const char*)tid.data(), tid.size());
pk.pack(std::string("y")); pk.pack(std::string("r"));
pk.pack(std::string("v")); pk.pack(my_v);
if (network) {
pk.pack(std::string("n")); pk.pack(network);
}
send(buffer.data(), buffer.size(), 0, addr);
send(buffer.data(), buffer.size(), 0, sa, salen);
}
std::shared_ptr<Request>
@@ -541,7 +539,7 @@ NetworkEngine::sendFindNode(std::shared_ptr<Node> n, const InfoHash& target, wan
auto tid = TransId {TransPrefix::FIND_NODE, getNewTid()};
msgpack::sbuffer buffer;
msgpack::packer<msgpack::sbuffer> pk(&buffer);
pk.pack_map(5+(network?1:0));
pk.pack_map(5);
pk.pack(std::string("a")); pk.pack_map(2 + (want>0?1:0));
pk.pack(std::string("id")); pk.pack(myid);
@@ -558,9 +556,7 @@ NetworkEngine::sendFindNode(std::shared_ptr<Node> n, const InfoHash& target, wan
pk.pack_bin_body((const char*)tid.data(), tid.size());
pk.pack(std::string("y")); pk.pack(std::string("q"));
pk.pack(std::string("v")); pk.pack(my_v);
if (network) {
pk.pack(std::string("n")); pk.pack(network);
}
Blob b {buffer.data(), buffer.data() + buffer.size()};
std::shared_ptr<Request> req(new Request {tid.getTid(), n, std::move(b),
@@ -587,7 +583,7 @@ NetworkEngine::sendGetValues(std::shared_ptr<Node> n, const InfoHash& info_hash,
auto tid = TransId {TransPrefix::GET_VALUES, getNewTid()};
msgpack::sbuffer buffer;
msgpack::packer<msgpack::sbuffer> pk(&buffer);
pk.pack_map(5+(network?1:0));
pk.pack_map(5);
pk.pack(std::string("a")); pk.pack_map(2 + (want>0?1:0));
pk.pack(std::string("id")); pk.pack(myid);
@@ -604,9 +600,6 @@ NetworkEngine::sendGetValues(std::shared_ptr<Node> n, const InfoHash& info_hash,
pk.pack_bin_body((const char*)tid.data(), tid.size());
pk.pack(std::string("y")); pk.pack(std::string("q"));
pk.pack(std::string("v")); pk.pack(my_v);
if (network) {
pk.pack(std::string("n")); pk.pack(network);
}
Blob b {buffer.data(), buffer.data() + buffer.size()};
std::shared_ptr<Request> req(new Request {tid.getTid(), n, std::move(b),
@@ -633,21 +626,19 @@ NetworkEngine::deserializeNodesValues(ParsedMessage& msg) {
} else {
// deserialize nodes
const auto& now = scheduler.time();
SockAddr addr;
for (unsigned i = 0; i < msg.nodes4_raw.size() / NODE4_INFO_BUF_LEN; i++) {
uint8_t *ni = msg.nodes4_raw.data() + i * NODE4_INFO_BUF_LEN;
const InfoHash& ni_id = *reinterpret_cast<InfoHash*>(ni);
if (ni_id == myid)
continue;
auto sin = (sockaddr_in*)&addr.first;
std::fill_n((uint8_t*)sin, sizeof(sockaddr_in), 0);
sin->sin_family = AF_INET;
memcpy(&sin->sin_addr, ni + ni_id.size(), 4);
memcpy(&sin->sin_port, ni + ni_id.size() + 4, 2);
addr.second = sizeof(sockaddr_in);
if (isMartian(addr) || isNodeBlacklisted(addr))
sockaddr_in sin;
std::fill_n((uint8_t*)&sin, sizeof(sockaddr_in), 0);
sin.sin_family = AF_INET;
memcpy(&sin.sin_addr, ni + ni_id.size(), 4);
memcpy(&sin.sin_port, ni + ni_id.size() + 4, 2);
if (isMartian((sockaddr*)&sin, sizeof(sin)) || isNodeBlacklisted((sockaddr*)&sin, sizeof(sin)))
continue;
msg.nodes4.emplace_back(cache.getNode(ni_id, addr, now, 0));
msg.nodes4.emplace_back(cache.getNode(ni_id, (sockaddr*)&sin, sizeof(sin), now, 0));
onNewNode(msg.nodes4.back(), 0);
}
for (unsigned i = 0; i < msg.nodes6_raw.size() / NODE6_INFO_BUF_LEN; i++) {
@@ -655,31 +646,30 @@ NetworkEngine::deserializeNodesValues(ParsedMessage& msg) {
const InfoHash& ni_id = *reinterpret_cast<InfoHash*>(ni);
if (ni_id == myid)
continue;
auto sin6 = (sockaddr_in6*)&addr.first;
std::fill_n((uint8_t*)sin6, sizeof(sockaddr_in6), 0);
sin6->sin6_family = AF_INET6;
memcpy(&sin6->sin6_addr, ni + HASH_LEN, 16);
memcpy(&sin6->sin6_port, ni + HASH_LEN + 16, 2);
addr.second = sizeof(sockaddr_in6);
if (isMartian(addr) || isNodeBlacklisted(addr))
sockaddr_in6 sin6;
std::fill_n((uint8_t*)&sin6, sizeof(sockaddr_in6), 0);
sin6.sin6_family = AF_INET6;
memcpy(&sin6.sin6_addr, ni + HASH_LEN, 16);
memcpy(&sin6.sin6_port, ni + HASH_LEN + 16, 2);
if (isMartian((sockaddr*)&sin6, sizeof(sin6)) || isNodeBlacklisted((sockaddr*)&sin6, sizeof(sin6)))
continue;
msg.nodes6.emplace_back(cache.getNode(ni_id, addr, now, 0));
msg.nodes6.emplace_back(cache.getNode(ni_id, (sockaddr*)&sin6, sizeof(sin6), now, 0));
onNewNode(msg.nodes6.back(), 0);
}
}
}
void
NetworkEngine::sendNodesValues(const SockAddr& addr, TransId tid, const Blob& nodes, const Blob& nodes6,
NetworkEngine::sendNodesValues(const sockaddr* sa, socklen_t salen, TransId tid, const Blob& nodes, const Blob& nodes6,
const std::vector<std::shared_ptr<Value>>& st, const Blob& token) {
msgpack::sbuffer buffer;
msgpack::packer<msgpack::sbuffer> pk(&buffer);
pk.pack_map(4+(network?1:0));
pk.pack_map(4);
pk.pack(std::string("r"));
pk.pack_map(2 + (not st.empty()?1:0) + (nodes.size()>0?1:0) + (nodes6.size()>0?1:0) + (not token.empty()?1:0));
pk.pack(std::string("id")); pk.pack(myid);
insertAddr(pk, addr);
insertAddr(pk, sa, salen);
if (nodes.size() > 0) {
pk.pack(std::string("n4"));
pk.pack_bin(nodes.size());
@@ -725,11 +715,8 @@ NetworkEngine::sendNodesValues(const SockAddr& addr, TransId tid, const Blob& no
pk.pack_bin_body((const char*)tid.data(), tid.size());
pk.pack(std::string("y")); pk.pack(std::string("r"));
pk.pack(std::string("v")); pk.pack(my_v);
if (network) {
pk.pack(std::string("n")); pk.pack(network);
}
send(buffer.data(), buffer.size(), 0, addr);
send(buffer.data(), buffer.size(), 0, sa, salen);
}
Blob
@@ -738,14 +725,14 @@ NetworkEngine::bufferNodes(sa_family_t af, const InfoHash& id, std::vector<std::
std::sort(nodes.begin(), nodes.end(), [&](const std::shared_ptr<Node>& a, const std::shared_ptr<Node>& b){
return id.xorCmp(a->id, b->id) < 0;
});
size_t nnode = std::min<size_t>(SEND_NODES, nodes.size());
size_t nnode = std::min<size_t>(TARGET_NODES, nodes.size());
Blob bnodes;
if (af == AF_INET) {
bnodes.resize(NODE4_INFO_BUF_LEN * nnode);
const constexpr size_t size = HASH_LEN + sizeof(in_addr) + sizeof(in_port_t); // 26
for (size_t i=0; i<nnode; i++) {
const Node& n = *nodes[i];
sockaddr_in *sin = (sockaddr_in*)&n.addr.first;
sockaddr_in *sin = (sockaddr_in*)&n.ss;
auto dest = bnodes.data() + size * i;
memcpy(dest, n.id.data(), HASH_LEN);
memcpy(dest + HASH_LEN, &sin->sin_addr, sizeof(in_addr));
@@ -756,7 +743,7 @@ NetworkEngine::bufferNodes(sa_family_t af, const InfoHash& id, std::vector<std::
const constexpr size_t size = HASH_LEN + sizeof(in6_addr) + sizeof(in_port_t); // 38
for (size_t i=0; i<nnode; i++) {
const Node& n = *nodes[i];
sockaddr_in6 *sin6 = (sockaddr_in6*)&n.addr.first;
sockaddr_in6 *sin6 = (sockaddr_in6*)&n.ss;
auto dest = bnodes.data() + size * i;
memcpy(dest, n.id.data(), HASH_LEN);
memcpy(dest + HASH_LEN, &sin6->sin6_addr, sizeof(in6_addr));
@@ -790,7 +777,7 @@ NetworkEngine::sendListen(std::shared_ptr<Node> n, const InfoHash& infohash, con
auto tid = TransId {TransPrefix::LISTEN, getNewTid()};
msgpack::sbuffer buffer;
msgpack::packer<msgpack::sbuffer> pk(&buffer);
pk.pack_map(5+(network?1:0));
pk.pack_map(5);
pk.pack(std::string("a")); pk.pack_map(3);
pk.pack(std::string("id")); pk.pack(myid);
@@ -802,9 +789,7 @@ NetworkEngine::sendListen(std::shared_ptr<Node> n, const InfoHash& infohash, con
pk.pack_bin_body((const char*)tid.data(), tid.size());
pk.pack(std::string("y")); pk.pack(std::string("q"));
pk.pack(std::string("v")); pk.pack(my_v);
if (network) {
pk.pack(std::string("n")); pk.pack(network);
}
Blob b {buffer.data(), buffer.data() + buffer.size()};
std::shared_ptr<Request> req(new Request {tid.getTid(), n, std::move(b),
@@ -824,24 +809,21 @@ NetworkEngine::sendListen(std::shared_ptr<Node> n, const InfoHash& infohash, con
}
void
NetworkEngine::sendListenConfirmation(const SockAddr& addr, TransId tid) {
NetworkEngine::sendListenConfirmation(const sockaddr* sa, socklen_t salen, TransId tid) {
msgpack::sbuffer buffer;
msgpack::packer<msgpack::sbuffer> pk(&buffer);
pk.pack_map(4+(network?1:0));
pk.pack_map(4);
pk.pack(std::string("r")); pk.pack_map(2);
pk.pack(std::string("id")); pk.pack(myid);
insertAddr(pk, addr);
insertAddr(pk, sa, salen);
pk.pack(std::string("t")); pk.pack_bin(tid.size());
pk.pack_bin_body((const char*)tid.data(), tid.size());
pk.pack(std::string("y")); pk.pack(std::string("r"));
pk.pack(std::string("v")); pk.pack(my_v);
if (network) {
pk.pack(std::string("n")); pk.pack(network);
}
send(buffer.data(), buffer.size(), 0, addr);
send(buffer.data(), buffer.size(), 0, sa, salen);
}
std::shared_ptr<Request>
@@ -850,7 +832,7 @@ NetworkEngine::sendAnnounceValue(std::shared_ptr<Node> n, const InfoHash& infoha
auto tid = TransId {TransPrefix::ANNOUNCE_VALUES, getNewTid()};
msgpack::sbuffer buffer;
msgpack::packer<msgpack::sbuffer> pk(&buffer);
pk.pack_map(5+(network?1:0));
pk.pack_map(5);
pk.pack(std::string("a")); pk.pack_map((created < scheduler.time() ? 5 : 4));
pk.pack(std::string("id")); pk.pack(myid);
@@ -867,9 +849,6 @@ NetworkEngine::sendAnnounceValue(std::shared_ptr<Node> n, const InfoHash& infoha
pk.pack_bin_body((const char*)tid.data(), tid.size());
pk.pack(std::string("y")); pk.pack(std::string("q"));
pk.pack(std::string("v")); pk.pack(my_v);
if (network) {
pk.pack(std::string("n")); pk.pack(network);
}
Blob b {buffer.data(), buffer.data() + buffer.size()};
std::shared_ptr<Request> req(new Request {tid.getTid(), n, std::move(b),
@@ -896,29 +875,27 @@ NetworkEngine::sendAnnounceValue(std::shared_ptr<Node> n, const InfoHash& infoha
}
void
NetworkEngine::sendValueAnnounced(const SockAddr& addr, TransId tid, Value::Id vid) {
NetworkEngine::sendValueAnnounced(const sockaddr* sa, socklen_t salen, TransId tid, Value::Id vid) {
msgpack::sbuffer buffer;
msgpack::packer<msgpack::sbuffer> pk(&buffer);
pk.pack_map(4+(network?1:0));
pk.pack_map(4);
pk.pack(std::string("r")); pk.pack_map(3);
pk.pack(std::string("id")); pk.pack(myid);
pk.pack(std::string("vid")); pk.pack(vid);
insertAddr(pk, addr);
insertAddr(pk, sa, salen);
pk.pack(std::string("t")); pk.pack_bin(tid.size());
pk.pack_bin_body((const char*)tid.data(), tid.size());
pk.pack(std::string("y")); pk.pack(std::string("r"));
pk.pack(std::string("v")); pk.pack(my_v);
if (network) {
pk.pack(std::string("n")); pk.pack(network);
}
send(buffer.data(), buffer.size(), 0, addr);
send(buffer.data(), buffer.size(), 0, sa, salen);
}
void
NetworkEngine::sendError(const SockAddr& addr,
NetworkEngine::sendError(const sockaddr* sa,
socklen_t salen,
TransId tid,
uint16_t code,
const std::string& message,
@@ -940,11 +917,8 @@ NetworkEngine::sendError(const SockAddr& addr,
pk.pack_bin_body((const char*)tid.data(), tid.size());
pk.pack(std::string("y")); pk.pack(std::string("e"));
pk.pack(std::string("v")); pk.pack(my_v);
if (network) {
pk.pack(std::string("n")); pk.pack(network);
}
send(buffer.data(), buffer.size(), 0, addr);
send(buffer.data(), buffer.size(), 0, sa, salen);
}
msgpack::object*
@@ -986,9 +960,6 @@ ParsedMessage::msgpack_unpack(msgpack::object msg)
error_code = e->via.array.ptr[0].as<uint16_t>();
}
if (auto netid = findMapValue(msg, "n"))
network = netid->as<NetId>();
if (auto rid = findMapValue(req, "id"))
id = {*rid};
+4 -3
View File
@@ -65,9 +65,10 @@ Node::getPendingMessageCount() const
}
void
Node::update(const SockAddr& new_addr)
Node::update(const sockaddr* sa, socklen_t salen)
{
addr = new_addr;
std::copy_n((const uint8_t*)sa, salen, (uint8_t*)&ss);
sslen = salen;
}
/** To be called when a message was sent to the node */
@@ -118,7 +119,7 @@ Node::toString() const
std::ostream& operator<< (std::ostream& s, const Node& h)
{
s << h.id << " " << h.addr.toString();
s << h.id << " " << print_addr(h.ss, h.sslen);
return s;
}
+6 -6
View File
@@ -27,10 +27,10 @@ NodeCache::getNode(const InfoHash& id, sa_family_t family) {
}
std::shared_ptr<Node>
NodeCache::getNode(const InfoHash& id, const SockAddr& addr, time_point now, int confirm) {
NodeCache::getNode(const InfoHash& id, const sockaddr* sa, socklen_t sa_len, time_point now, int confirm) {
if (id == zeroes)
return std::make_shared<Node>(id, addr);
return (addr.getFamily() == AF_INET ? cache_4 : cache_6).getNode(id, addr, now, confirm);
return std::make_shared<Node>(id, sa, sa_len);
return (sa->sa_family == AF_INET ? cache_4 : cache_6).getNode(id, sa, sa_len, now, confirm);
}
void
@@ -57,15 +57,15 @@ NodeCache::NodeMap::getNode(const InfoHash& id)
}
std::shared_ptr<Node>
NodeCache::NodeMap::getNode(const InfoHash& id, const SockAddr& addr, time_point now, int confirm)
NodeCache::NodeMap::getNode(const InfoHash& id, const sockaddr* sa, socklen_t sa_len, time_point now, int confirm)
{
auto it = emplace(id, std::weak_ptr<Node>{});
auto node = it.first->second.lock();
if (not node) {
node = std::make_shared<Node>(id, addr);
node = std::make_shared<Node>(id, sa, sa_len);
it.first->second = node;
} else if (confirm || node->time < now - Node::NODE_EXPIRE_TIME) {
node->update(addr);
node->update(sa, sa_len);
}
return node;
}
+2 -2
View File
@@ -333,7 +333,7 @@ SecureDht::putSigned(const InfoHash& hash, std::shared_ptr<Value> val, DoneCallb
},
[hash,val,this,callback,permanent] (bool /* ok */) {
sign(*val);
put(hash, val, callback, time_point::max(), permanent);
put(hash, val, callback, {}, permanent);
},
Value::IdFilter(val->id)
);
@@ -350,7 +350,7 @@ SecureDht::putEncrypted(const InfoHash& hash, const InfoHash& to, std::shared_pt
}
DHT_LOG.WARN("Encrypting data for PK: %s", pk->getId().toString().c_str());
try {
put(hash, encrypt(*val, *pk), callback, time_point::max(), permanent);
put(hash, encrypt(*val, *pk), callback, {}, permanent);
} catch (const std::exception& e) {
DHT_LOG.ERR("Error putting encrypted data: %s", e.what());
if (callback)
+1 -1
View File
@@ -47,7 +47,7 @@ print_addr(const sockaddr_storage& ss, socklen_t sslen)
}
std::string
printAddr(const SockAddr& addr) {
printAddr(const Address& addr) {
return print_addr((const sockaddr*)&addr.first, addr.second);
}
+1 -1
View File
@@ -53,7 +53,7 @@ main(int argc, char **argv)
throw std::runtime_error(std::string("Error initializing GnuTLS: ")+gnutls_strerror(rc));
DhtRunner dht;
dht.run(params.port, dht::crypto::generateIdentity("DHT Chat Node"), true, params.network);
dht.run(params.port, dht::crypto::generateIdentity("DHT Chat Node"), true);
if (not params.bootstrap.first.empty())
dht.bootstrap(params.bootstrap.first.c_str(), params.bootstrap.second.c_str());
+4 -2
View File
@@ -29,7 +29,7 @@ extern "C" {
using namespace dht;
void print_usage() {
std::cout << "Usage: dhtnode [-v [-l logfile]] [-i] [-d] [-n network_id] [-p local_port] [-b bootstrap_host[:port]]" << std::endl << std::endl;
std::cout << "Usage: dhtnode [-p local_port] [-b bootstrap_host:port]" << std::endl << std::endl;
std::cout << "dhtnode, a simple OpenDHT command line node runner." << std::endl;
std::cout << "Report bugs to: http://opendht.net" << std::endl;
}
@@ -40,6 +40,8 @@ void print_id_req() {
void print_node_info(const DhtRunner& dht, const dht_params& params) {
std::cout << "OpenDht node " << dht.getNodeId() << " running on port " << dht.getBoundPort() << std::endl;
if (params.is_bootstrap_node)
std::cout << "Running in bootstrap mode (discouraged)." << std::endl;
if (params.generate_identity)
std::cout << "Public key ID " << dht.getId() << std::endl;
}
@@ -254,7 +256,7 @@ main(int argc, char **argv)
crt = dht::crypto::generateIdentity("DHT Node", ca_tmp);
}
dht.run(params.port, crt, true, params.network);
dht.run(params.port, crt, true, params.is_bootstrap_node);
if (params.log) {
if (not params.logfile.empty())
+1 -1
View File
@@ -80,7 +80,7 @@ main(int argc, char **argv)
auto crt_tmp = dht::crypto::generateIdentity("Scanner node", ca_tmp);
DhtRunner dht;
dht.run(params.port, crt_tmp, true, params.network);
dht.run(params.port, crt_tmp, true, [](dht::NodeStatus /* ipv4 */, dht::NodeStatus /* ipv6 */) {});
if (not params.bootstrap.first.empty())
dht.bootstrap(params.bootstrap.first.c_str(), params.bootstrap.second.c_str());
+29 -18
View File
@@ -142,29 +142,27 @@ struct dht_params {
bool log {false};
std::string logfile {};
in_port_t port {0};
dht::NetId network {0};
bool is_bootstrap_node {false};
bool generate_identity {false};
bool daemonize {false};
std::pair<std::string, std::string> bootstrap {};
};
static const constexpr struct option long_options[] = {
{"help", no_argument , nullptr, 'h'},
{"help", no_argument, nullptr, 'h'},
{"port", required_argument, nullptr, 'p'},
{"net", required_argument, nullptr, 'n'},
{"bootstrap", required_argument, nullptr, 'b'},
{"bootstrap", optional_argument, nullptr, 'b'},
{"identity", no_argument , nullptr, 'i'},
{"verbose", no_argument , nullptr, 'v'},
{"daemonize", no_argument , nullptr, 'd'},
{"logfile", required_argument, nullptr, 'l'},
{nullptr, 0 , nullptr, 0}
{nullptr, 0, nullptr, 0}
};
dht_params
parseArgs(int argc, char **argv) {
dht_params params;
int opt;
while ((opt = getopt_long(argc, argv, "hidvp:n:b:l:", long_options, nullptr)) != -1) {
while ((opt = getopt_long(argc, argv, ":hidv:p:b:", long_options, nullptr)) != -1) {
switch (opt) {
case 'p': {
int port_arg = atoi(optarg);
@@ -174,24 +172,24 @@ parseArgs(int argc, char **argv) {
std::cout << "Invalid port: " << port_arg << std::endl;
}
break;
case 'n':
params.network = strtoul(optarg, nullptr, 0);
break;
case 'b':
params.bootstrap = splitPort((optarg[0] == '=') ? optarg+1 : optarg);
if (not params.bootstrap.first.empty() and params.bootstrap.second.empty()) {
std::stringstream ss;
ss << DHT_DEFAULT_PORT;
params.bootstrap.second = ss.str();
if (optarg) {
params.bootstrap = splitPort((optarg[0] == '=') ? optarg+1 : optarg);
if (not params.bootstrap.first.empty() and params.bootstrap.second.empty()) {
std::stringstream ss;
ss << DHT_DEFAULT_PORT;
params.bootstrap.second = ss.str();
}
}
else
params.is_bootstrap_node = true;
break;
case 'h':
params.help = true;
break;
case 'l':
params.logfile = optarg;
break;
case 'v':
if (optarg)
params.logfile = optarg;
params.log = true;
break;
case 'i':
@@ -200,6 +198,19 @@ parseArgs(int argc, char **argv) {
case 'd':
params.daemonize = true;
break;
case ':':
switch (optopt) {
case 'b':
params.is_bootstrap_node = true;
break;
case 'v':
params.log = true;
break;
default:
std::cout << "option requires an argument -- '" << optopt << '\'' << std::endl;
break;
}
break;
default:
break;
}