Compare commits

...
74 Commits
Author SHA1 Message Date
Adrien Béraud 5fde50be4a dhtrunner: use const char* for setopt value
fix build on Win32
2017-11-17 19:34:31 +01:00
Adrien Béraud 1c16e6f38d update to version 1.5.0
* added REST server implementing OpenDHT proxy API
* tools: added simple Web client using the proxy server
* tools: allow dhtnode to run a proxy server
* tools: allow dhtnode to use any string as key
* InfoHash: added operator bool()
* minor performance improvement
* SockAddr: fix endianness issue
* net: fix incorrect reported address parsing
* minor memory leak fixes
2017-11-17 18:51:45 +01:00
Adrien Béraud 6790d1ec5c SockAddr: set service to null if string is empty 2017-11-17 18:31:28 +01:00
Adrien Béraud 4dc1847978 autotools: fix build without proxy server 2017-11-17 16:42:04 +01:00
Adrien Béraud 3a61692618 net: set MSG_NOSIGNAL if possible 2017-11-17 02:20:31 +01:00
Adrien Béraud 6f9afac0a5 dhtrunner: cleanup 2017-11-17 02:12:03 +01:00
Adrien Béraud b17c7c2965 dht: reserve address vector size 2017-11-17 02:06:32 +01:00
Adrien Béraud 6d238836ad dht: return empty search on error 2017-11-17 01:01:47 +01:00
Adrien Béraud 5115fab9e6 dht: increase max searches limit 2017-11-16 16:49:59 +01:00
Adrien Béraud 93498ecf0b tools: JS node: cleanup 2017-11-14 21:55:03 +01:00
Adrien Béraud 72d790953a tools: JS node: allow to set the proxy address 2017-11-14 20:58:18 +01:00
Adrien Béraud 72b556a09b tools: JS node: cleanup 2017-11-14 19:38:55 +01:00
Adrien Béraud 747f86f8f4 dhtrunner: use SockAddr::resolve 2017-11-10 17:59:05 -05:00
Adrien Béraud 161769f4ec SockAddr: use host endianess for bitwise-shift operations 2017-11-10 17:48:36 -05:00
Adrien Béraud 9948079c97 SockAddr: add resolve() 2017-11-10 17:48:36 -05:00
Adrien Béraud 607e5eeccf tools: use https url 2017-11-10 17:29:24 -05:00
Adrien Béraud 111ac57b6f tools: JS node: add listen support 2017-11-09 12:23:30 -05:00
Adrien Béraud 153641dde5 tools: add JavaScript proxy node 2017-11-08 15:20:06 -05:00
Adrien Béraud 113bebabd4 DhtRunner: remove log 2017-11-08 15:14:01 -05:00
Adrien Béraud b549055034 DhtRunner: use SockAddr 2017-11-08 15:12:43 -05:00
Adrien Béraud 7d7410b472 dhtnode: allow to use arbitrary string as key
Key will be h(key).
2017-11-08 15:12:43 -05:00
Adrien Béraud d7aaa4ae56 proxy server: add Access-Control-Allow-Origin 2017-11-08 15:12:43 -05:00
Adrien Béraud 5a049f89f3 cmake: find Jsoncpp using pkgconfig 2017-11-08 15:12:43 -05:00
Sébastien Blin 65e88f2fa6 build: toInfoHash->to 2017-11-08 15:07:04 -05:00
Adrien Béraud 4430b1e616 Merge pull request #217 from AmarOk1412/dhtproxy
Communicate with OpenDHT through a REST API
2017-11-08 14:35:31 -05:00
Sébastien Blin e2c534df5a add std::bind, fix put, avoid overflow on base64 2017-11-08 14:33:07 -05:00
Sébastien Blin 2ca2835c9b add options method handler 2017-11-08 12:14:44 -05:00
Sébastien Blin bbe922dd7d remove headers, fix HTTP status, fix get / 2017-11-07 16:08:55 -05:00
Sébastien Blin 38ad5b9ef2 add --proxyserver option for dhtnode 2017-11-06 16:49:46 -05:00
Sébastien Blin 67ba435ba6 base64: take the old implementation from ring-daemon 2017-11-06 16:24:00 -05:00
Sébastien Blin 6413571301 clean code 2017-11-06 12:53:35 -05:00
Adrien Beraud 0dccae68e6 InfoHash: make false in case of parsing error 2017-11-03 16:03:11 -04:00
Adrien Beraud 92320cef59 InfoHash: throw in case of invalid input string 2017-11-03 14:03:34 -04:00
Adrien Beraud bf4e838e20 InfoHash: return stream in operator>> 2017-11-03 14:00:49 -04:00
Adrien Beraud 88b4233bda python: update InfoHash bool operator 2017-11-03 14:00:49 -04:00
Adrien Beraud e9eb97ea9a InfoHash: improve hex parsing performance 2017-11-02 17:42:45 -04:00
Adrien Beraud 0e865976f9 InfoHash: unify with Hash<N> 2017-11-02 14:28:09 -04:00
Adrien Beraud 87dc080bbf crypto: add raw hash function 2017-11-02 14:25:46 -04:00
Sébastien Blin 9abc88f378 add --enable-proxy-server option for autotools 2017-11-01 17:11:14 -04:00
Adrien Beraud 834d62e6e9 InfoHash: use bool operator where relevent 2017-11-01 16:28:17 -04:00
Sébastien Blin 61a148e83f update README.md 2017-11-01 14:52:29 -04:00
Sébastien Blin 8e43775394 update dhtnode + clean code 2017-11-01 14:52:29 -04:00
Sébastien Blin 8344138bc8 add optional + clean start/stop 2017-11-01 14:52:29 -04:00
Sébastien Blin dc66b3847a make code for DHT Proxy Server optionnal 2017-11-01 14:52:29 -04:00
Sébastien Blin d608da4307 cmake: add optionnal dependencies for proxy 2017-11-01 14:52:29 -04:00
Sébastien Blin 7ddd6629e3 add getFiltered 2017-11-01 14:52:29 -04:00
Sébastien Blin 20abc3e495 add LISTEN endpoint 2017-11-01 14:52:29 -04:00
Sébastien Blin c4ba8670e4 add Putxxx value endpoints 2017-11-01 14:52:29 -04:00
Sébastien Blin a3348c85d6 value to JSON and begin to clean endpoints 2017-11-01 14:52:29 -04:00
Sébastien Blin ca465d7159 set timeout to a high value + add jsoncpp 2017-11-01 14:52:29 -04:00
Sébastien Blin 1413d32ee6 add listen endpoint 2017-11-01 14:52:29 -04:00
Sébastien Blin 58fdfa9452 add get hash endpoint 2017-11-01 14:51:36 -04:00
Sébastien Blin 6f641073ae proxy put and putEncrypted 2017-11-01 14:51:36 -04:00
Sébastien Blin 32153b11ab proxy putSigned 2017-11-01 14:51:36 -04:00
Sébastien Blin e33ac02351 test restbed 2017-11-01 14:51:36 -04:00
Adrien Beraud dc99919338 InfoHash: add explicit operator bool() 2017-11-01 11:59:11 -04:00
Adrien Béraud fcaaa01b2d network engine: add "value update" message type 2017-10-30 13:12:16 -04:00
Adrien Béraud 8cd3ff9fc4 network engine: remove unused constant 2017-10-30 13:12:16 -04:00
Adrien Béraud 0a130c7045 dht: update permanent state on announce 2017-10-30 13:12:16 -04:00
Adrien Béraud eb9f314356 dhtnode: allow to perform and cancel persistent put 2017-10-30 13:12:16 -04:00
Adrien Béraud a401aea132 dht: simplify searchSendAnnounceValue 2017-10-30 13:12:16 -04:00
Adrien Béraud 41ebbe21d4 node cache: expire nodes on destructions
Expiring nodes on cache destruction allow to cancel
any remaining request before the dht instance is destructed,
since Node instances might remain alive after the dht instance
is destroyed.
2017-10-30 13:12:16 -04:00
Adrien Béraud 6378840d33 dhtnode: allow to cancel a listen operation 2017-10-30 13:12:16 -04:00
Adrien Béraud 64f0aa1532 parsed message: don't override the reported address family 2017-10-30 13:12:16 -04:00
Adrien Béraud 63366722a9 dht: reduce verbosity of storage log 2017-10-21 18:01:32 +02:00
Adrien Béraud cd17858ae1 dht: periodic cleanup of unused quota entries 2017-10-21 18:00:46 +02:00
Adrien Béraud f4ebbc6651 update to version 1.4.1
* updated MSVC project
* bug fixes
2017-10-21 01:30:12 +02:00
Adrien Béraud 464339822b Merge pull request #215 from AmarOk1412/typoPeriod
rate_limiter: typo fix, 'Perdiod'->'Period'
2017-10-20 23:56:02 +02:00
Sébastien Blin fe09146609 rate_limiter: typo fix, 'Perdiod'->'Period' 2017-10-20 14:51:35 -04:00
Adrien Béraud a789ffe7ae dht: prevent use-after-free of socket callback 2017-10-20 17:46:27 +02:00
Adrien Béraud 476fbb1160 network: use full 32-bits of transaction IDs
We used to use the first 16 bits for the message type.
2017-10-20 17:42:28 +02:00
Adrien Béraud 4501217c21 Merge pull request #214 from atraczyk/master
remove vs2015 based constexpr error, update vs project files
2017-10-19 01:19:43 +02:00
atraczyk 207675a181 remove vs2015 based constexpr error, update vs project files 2017-10-18 17:50:19 -04:00
Adrien Béraud cf6447dc80 NetworkEngine: use proper type for constant 2017-10-17 19:22:15 +02:00
48 changed files with 1836 additions and 659 deletions

No files matched your search

+5
View File
@@ -48,6 +48,11 @@ Makefile.in
# Python
*.pyc
python/opendht.cpp
python/setup.py
# Doxygen
doc/Doxyfile
# git backup files
*.orig
+37 -4
View File
@@ -1,7 +1,7 @@
cmake_minimum_required (VERSION 3.1)
project (opendht)
set (opendht_VERSION_MAJOR 1)
set (opendht_VERSION_MINOR 4.0)
set (opendht_VERSION_MINOR 5.0)
set (opendht_VERSION ${opendht_VERSION_MAJOR}.${opendht_VERSION_MINOR})
set (PACKAGE_VERSION ${opendht_VERSION})
set (VERSION "${opendht_VERSION}")
@@ -16,6 +16,9 @@ option (OPENDHT_SYSTEMD "Install systemd module" OFF)
option (OPENDHT_ARGON2 "Use included argon2 sources" OFF)
option (OPENDHT_LTO "Build with LTO" OFF)
option (OPENDHT_SANITIZE "Build with address sanitizer and stack protector" OFF)
option (OPENDHT_PROXY_SERVER "Enable DHT proxy server, use Restbed and jsoncpp" OFF)
option (OPENDHT_PROXY_SERVER_IDENTITY "Allow clients to use the node identity" OFF)
option (OPENDHT_PROXY_CLIENT "Enable DHT proxy client, use Restbed and jsoncpp" OFF)
find_package(Doxygen)
option (OPENDHT_DOCUMENTATION "Create and install the HTML based API documentation (requires Doxygen)" ${DOXYGEN_FOUND})
@@ -41,6 +44,10 @@ if (NOT OPENDHT_ARGON2)
set(OPENDHT_ARGON2 ON)
endif()
endif ()
if (OPENDHT_PROXY_SERVER OR OPENDHT_PROXY_CLIENT)
find_package(Restbed REQUIRED)
pkg_search_module(Jsoncpp jsoncpp)
endif()
# Build flags
set (CMAKE_CXX_STANDARD 11)
@@ -77,7 +84,14 @@ endif ()
if (Nettle_INCLUDE_DIRS)
include_directories (SYSTEM "${Nettle_INCLUDE_DIRS}")
endif ()
if (RESTBED_INCLUDE)
include_directories (SYSTEM "${RESTBED_INCLUDE}")
endif ()
if (Jsoncpp_INCLUDE_DIRS)
include_directories (SYSTEM "${Jsoncpp_INCLUDE_DIRS}")
endif ()
link_directories (${Nettle_LIBRARY_DIRS})
link_directories (${Jsoncpp_LIBRARY_DIRS})
include_directories (
./
include/
@@ -107,6 +121,7 @@ list (APPEND opendht_SOURCES
src/storage.h
src/listener.h
src/search.h
src/net.h
src/parsed_message.h
src/request.h
src/callbacks.cpp
@@ -134,7 +149,6 @@ list (APPEND opendht_HEADERS
include/opendht/routing_table.h
include/opendht/node_cache.h
include/opendht/network_engine.h
include/opendht/net.h
include/opendht/scheduler.h
include/opendht/rate_limiter.h
include/opendht/securedht.h
@@ -144,6 +158,25 @@ list (APPEND opendht_HEADERS
include/opendht/indexation/pht.h
)
if (OPENDHT_PROXY_SERVER)
add_definitions(-DOPENDHT_PROXY_SERVER=true)
if (OPENDHT_PROXY_SERVER_IDENTITY)
add_definitions(-DOPENDHT_PROXY_SERVER_IDENTITY=true)
else ()
add_definitions(-DOPENDHT_PROXY_SERVER_IDENTITY=false)
endif()
list (APPEND opendht_HEADERS
include/opendht/dht_proxy_server.h
)
list (APPEND opendht_SOURCES
src/dht_proxy_server.cpp
src/base64.h
src/base64.cpp
)
else ()
add_definitions(-DENABLE_PROXY_SERVER=false)
endif ()
if(OPENDHT_ARGON2)
# make sure argon2 submodule is up to date and initialized
message("Initializing Argon2 submodule")
@@ -177,7 +210,7 @@ if (OPENDHT_STATIC)
target_link_libraries(opendht-static ${argon2_LIBRARIES})
target_include_directories(opendht-static SYSTEM PRIVATE ${argon2_INCLUDE_DIRS})
endif ()
target_link_libraries(opendht-static ${CMAKE_THREAD_LIBS_INIT} ${GNUTLS_LIBRARIES} ${Nettle_LIBRARIES})
target_link_libraries(opendht-static ${CMAKE_THREAD_LIBS_INIT} ${GNUTLS_LIBRARIES} ${Nettle_LIBRARIES} ${Restbed_LIBRARIES} ${Jsoncpp_LIBRARIES})
install (TARGETS opendht-static DESTINATION ${CMAKE_INSTALL_LIBDIR} EXPORT opendht)
endif ()
@@ -198,7 +231,7 @@ if (OPENDHT_SHARED)
target_link_libraries(opendht PRIVATE ${argon2_LIBRARIES})
target_include_directories(opendht SYSTEM PRIVATE ${argon2_INCLUDE_DIRS})
endif ()
target_link_libraries(opendht PRIVATE ${CMAKE_THREAD_LIBS_INIT} ${GNUTLS_LIBRARIES} ${Nettle_LIBRARIES})
target_link_libraries(opendht PRIVATE ${CMAKE_THREAD_LIBS_INIT} ${GNUTLS_LIBRARIES} ${Nettle_LIBRARIES} ${Restbed_LIBRARIES} ${Jsoncpp_LIBRARIES})
install (TARGETS opendht DESTINATION ${CMAKE_INSTALL_LIBDIR} EXPORT opendht)
endif ()
+4
View File
@@ -39,6 +39,7 @@
<ClInclude Include="..\include\opendht.h" />
<ClInclude Include="..\include\opendht\callbacks.h" />
<ClInclude Include="..\include\opendht\crypto.h" />
<ClInclude Include="..\include\opendht\def.h" />
<ClInclude Include="..\include\opendht\default_types.h" />
<ClInclude Include="..\include\opendht\dht.h" />
<ClInclude Include="..\include\opendht\dhtrunner.h" />
@@ -55,9 +56,12 @@
<ClInclude Include="..\include\opendht\routing_table.h" />
<ClInclude Include="..\include\opendht\scheduler.h" />
<ClInclude Include="..\include\opendht\securedht.h" />
<ClInclude Include="..\include\opendht\sockaddr.h" />
<ClInclude Include="..\include\opendht\utils.h" />
<ClInclude Include="..\include\opendht\value.h" />
<ClInclude Include="..\src\net.h" />
<ClInclude Include="..\src\listener.h" />
<ClInclude Include="..\src\parsed_message.h" />
<ClInclude Include="..\src\search.h" />
<ClInclude Include="..\src\storage.h" />
</ItemGroup>
+4 -2
View File
@@ -1,7 +1,7 @@
<img src="https://raw.githubusercontent.com/savoirfairelinux/opendht/master/resources/opendht_logo_512.png" width="100" align="right">
<br>
<h1 style="margin-top:10px">
<a id="user-content-opendht-" class="anchor" href="/savoirfairelinux/opendht/blob/master/README.md#opendht-" aria-hidden="true"></a>OpenDHT
<a id="user-content-opendht-" class="anchor" href="/savoirfairelinux/opendht/blob/master/README.md#opendht-" aria-hidden="true"></a>OpenDHT
</h1>
A lightweight C++11 Distributed Hash Table implementation originally based on https://github.com/jech/dht by Juliusz Chroboczek.
@@ -18,7 +18,7 @@ See the wiki: <https://github.com/savoirfairelinux/opendht/wiki>
#### How-to build and install
Build instructions : <https://github.com/savoirfairelinux/opendht/wiki/Build-the-library>
Build instructions: <https://github.com/savoirfairelinux/opendht/wiki/Build-the-library>
#### How-to build a simple client app
```bash
@@ -91,6 +91,8 @@ for r in results:
- msgpack-c 1.2+, used for data serialization.
- GnuTLS 3.3+, used for cryptographic operations.
- Nettle 2.4+, a GnuTLS dependency for crypto.
- (optional) restbed 4.0+, used for the REST API.
- (optional) jsoncpp 1.7.4-3+, used for the REST API.
- Build tested with GCC 4.8+ (GNU/Linux, Android, Windows with MinGW), Clang/LLVM (Linux, macOS).
- Build tested with Microsoft Visual Studio 2015
+17
View File
@@ -0,0 +1,17 @@
find_path (RESTBED_INCLUDE restbed
HINTS
"/usr/include"
"/usr/local/include"
"/opt/local/include"
)
if (RESTBED_INCLUDE)
message(STATUS "${green}Found Restbed: ${RESTBED_INCLUDE}")
else()
message(FATAL_ERROR "${red}Failed to locate Restbed.}")
endif()
if (RESTBED_INCLUDE)
set(RESTBED_FOUND TRUE)
set(Restbed_LIBRARIES restbed)
endif()
+20 -1
View File
@@ -1,6 +1,6 @@
dnl define macros
m4_define([opendht_major_version], 1)
m4_define([opendht_minor_version], 4)
m4_define([opendht_minor_version], 5)
m4_define([opendht_patch_version], 0)
m4_define([opendht_version],
[opendht_major_version.opendht_minor_version.opendht_patch_version])
@@ -128,6 +128,25 @@ AM_COND_IF([ENABLE_TOOLS], [
])
])
AC_ARG_ENABLE([proxy_server], AS_HELP_STRING([--enable-proxy-server], [Enable proxy server ability]), proxy_server=yes, proxy_server=no)
AM_CONDITIONAL(ENABLE_PROXY_SERVER, test x$proxy_server == xyes)
AM_COND_IF([ENABLE_PROXY_SERVER], [
AC_CHECK_LIB(restbed, exit,, AC_MSG_ERROR([Missing restbed files]))
PKG_CHECK_MODULES([Jsoncpp], [jsoncpp >= 1.7.4])
CPPFLAGS+=" -DOPENDHT_PROXY_SERVER=true -ljsoncpp -lrestbed"
], [
CPPFLAGS+=" -DOPENDHT_PROXY_SERVER=false"
])
AC_ARG_ENABLE([proxy_server_identity], AS_HELP_STRING([--enable-proxy-server-identity],
[Enable proxy server ability]), proxy_server_identity=yes, proxy_server_identity=no)
AM_CONDITIONAL(ENABLE_PROXY_SERVER_IDENTITY, test x$proxy_server_identity == xyes -a x$proxy_server == xyes)
AM_COND_IF([ENABLE_PROXY_SERVER_IDENTITY], [
CPPFLAGS+=" -DOPENDHT_PROXY_SERVER_IDENTITY=true"
], [
CPPFLAGS+=" -DOPENDHT_PROXY_SERVER_IDENTITY=false"
])
AC_CONFIG_FILES([doc/Doxyfile doc/Makefile])
dnl Configure setup.py if we build the python module
+3
View File
@@ -19,6 +19,9 @@
#pragma once
#include "opendht/dhtrunner.h"
#if OPENDHT_PROXY_SERVER
#include "opendht/dht_proxy_server.h"
#endif
#include "opendht/log.h"
#include "opendht/default_types.h"
#include "opendht/indexation/pht.h"
+10
View File
@@ -26,6 +26,10 @@
#include <memory>
#include <functional>
#if OPENDHT_PROXY_SERVER
#include <json/json.h>
#endif //OPENDHT_PROXY_SERVER
namespace dht {
struct Node;
@@ -47,6 +51,12 @@ struct OPENDHT_PUBLIC NodeStats {
unsigned table_depth;
unsigned getKnownNodes() const { return good_nodes + dubious_nodes; }
std::string toString() const;
#if OPENDHT_PROXY_SERVER
/**
* Build a json object from a NodeStats
*/
Json::Value toJson() const;
#endif //OPENDHT_PROXY_SERVER
};
/**
+2
View File
@@ -601,6 +601,8 @@ OPENDHT_PUBLIC Identity generateEcIdentity(const std::string& name = "dhtnode",
*/
OPENDHT_PUBLIC Blob hash(const Blob& data, size_t hash_length = 512/8);
OPENDHT_PUBLIC void hash(const uint8_t* data, size_t data_length, uint8_t* hash, size_t hash_length);
/**
* Generates an encryption key from a text password,
* making the key longer to bruteforce.
+1 -1
View File
@@ -93,7 +93,7 @@ public:
static Value::Filter getFilter() {
return Value::Filter::chain(
BaseClass::getFilter(),
[](const Value& v){ return v.recipient != InfoHash(); }
[](const Value& v) { return static_cast<bool>(v.recipient); }
);
}
+1 -2
View File
@@ -339,7 +339,7 @@ private:
static constexpr unsigned MAX_HASHES {16384};
/* The maximum number of searches we keep data about. */
static constexpr unsigned MAX_SEARCHES {2048};
static constexpr unsigned MAX_SEARCHES {16384};
static constexpr std::chrono::minutes MAX_STORAGE_MAINTENANCE_EXPIRE_TIME {10};
@@ -428,7 +428,6 @@ private:
void storageChanged(const InfoHash& id, Storage& st, ValueStorage&);
std::string printStorageLog(const decltype(store)::value_type&) const;
void printStorageQuota(std::ostream& out, const decltype(store_quota)::value_type& ip) const;
/**
* For a given storage, if values don't belong there anymore because this
+168
View File
@@ -0,0 +1,168 @@
/*
* Copyright (C) 2017 Savoir-faire Linux Inc.
* Author : Sébastien Blin <sebastien.blin@savoirfairelinux.com>
*
* This program 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.
*
* You should have received a copy of the GNU General Public License
* along with this program. If not, see <https://www.gnu.org/licenses/>.
*/
#if OPENDHT_PROXY_SERVER
#pragma once
#include "def.h"
#include "sockaddr.h"
#include "infohash.h"
#include <thread>
#include <memory>
#include <restbed>
namespace dht {
class DhtRunner;
/**
* Describes the REST API
*/
class OPENDHT_PUBLIC DhtProxyServer
{
public:
/**
* Start the Http server for OpenDHT
* @param dht the DhtRunner linked to this proxy server
* @param port to listen
* @note if the server fails to start (if port is already used or reserved),
* it will fails silently
*/
DhtProxyServer(std::shared_ptr<DhtRunner> dht, in_port_t port = 8000);
virtual ~DhtProxyServer();
DhtProxyServer(const DhtProxyServer& other) = delete;
DhtProxyServer(DhtProxyServer&& other) = delete;
DhtProxyServer& operator=(const DhtProxyServer& other) = delete;
DhtProxyServer& operator=(DhtProxyServer&& other) = delete;
/**
* Stop the DhtProxyServer
*/
void stop();
private:
/**
* Return the PublicKey id, the node id and node stats
* Method: GET "/"
* Result: HTTP 200, body: Node infos in JSON format
* On error: HTTP 503, body: {"err":"xxxx"}
* @param session
*/
void getNodeInfo(const std::shared_ptr<restbed::Session>& session) const;
/**
* Return Values of a InfoHash
* Method: GET "/{InfoHash: .*}"
* Return: Multiple JSON object in parts. Example:
* Value in JSON format\n
* Value in JSON format
*
* On error: HTTP 503, body: {"err":"xxxx"}
* @param session
*/
void get(const std::shared_ptr<restbed::Session>& session) const;
/**
* Listen incoming Values of a InfoHash.
* Method: LISTEN "/{InfoHash: .*}"
* Return: Multiple JSON object in parts. Example:
* Value in JSON format\n
* Value in JSON format
*
* On error: HTTP 503, body: {"err":"xxxx"}
* @param session
*/
void listen(const std::shared_ptr<restbed::Session>& session) const;
/**
* Put a value on the DHT
* Method: POST "/{InfoHash: .*}"
* body = Value to put in JSON
* Return: HTTP 200 if success and the value put in JSON
* On error: HTTP 503, body: {"err":"xxxx"} if no dht
* HTTP 400, body: {"err":"xxxx"} if bad json or HTTP 502 if put fails
* @param session
*/
void put(const std::shared_ptr<restbed::Session>& session) const;
#if OPENDHT_PROXY_SERVER_IDENTITY
/**
* Put a value to sign by the proxy on the DHT
* Method: SIGN "/{InfoHash: .*}"
* body = Value to put in JSON
* Return: HTTP 200 if success and the value put in JSON
* On error: HTTP 503, body: {"err":"xxxx"} if no dht
* HTTP 400, body: {"err":"xxxx"} if bad json
* @param session
*/
void putSigned(const std::shared_ptr<restbed::Session>& session) const;
/**
* Put a value to encrypt by the proxy on the DHT
* Method: ENCRYPT "/{hash: .*}"
* body = Value to put in JSON + "to":"infoHash"
* Return: HTTP 200 if success and the value put in JSON
* On error: HTTP 503, body: {"err":"xxxx"} if no dht
* HTTP 400, body: {"err":"xxxx"} if bad json
* @param session
*/
void putEncrypted(const std::shared_ptr<restbed::Session>& session) const;
#endif // OPENDHT_PROXY_SERVER_IDENTITY
/**
* Return Values of a InfoHash filtered by a value id
* Method: GET "/{InfoHash: .*}/{ValueId: .*}"
* Return: Multiple JSON object in parts. Example:
* Value in JSON format\n
* Value in JSON format
*
* On error: HTTP 503, body: {"err":"xxxx"}
* @param session
*/
void getFiltered(const std::shared_ptr<restbed::Session>& session) const;
/**
* Respond allowed Methods
* Method: OPTIONS "/{hash: .*}"
* Return: HTTP 200 + Allow: allowed methods
* See https://developer.mozilla.org/en-US/docs/Web/HTTP/Methods/OPTIONS
*/
void handleOptionsMethod(const std::shared_ptr<restbed::Session>& session) const;
std::thread server_thread {};
std::unique_ptr<restbed::Service> service_;
std::shared_ptr<DhtRunner> dht_;
// Handle client quit for listen.
// NOTE: can be simplified when we will supports restbed 5.0
std::thread listenThread_;
struct SessionToHashToken {
std::shared_ptr<restbed::Session> session;
InfoHash hash;
std::future<size_t> token;
};
mutable std::vector<SessionToHashToken> currentListeners_;
std::atomic_bool stopListeners {false};
};
}
#endif //OPENDHT_PROXY_SERVER
+3 -5
View File
@@ -202,7 +202,7 @@ public:
* Insert known nodes to the routing table, without necessarly ping them.
* Usefull to restart a node and get things running fast without putting load on the network.
*/
void bootstrap(const std::vector<std::pair<sockaddr_storage, socklen_t>>& nodes, DoneCallbackSimple&& cb={});
void bootstrap(const std::vector<SockAddr>& nodes, DoneCallbackSimple&& cb={});
void bootstrap(const SockAddr& addr, DoneCallbackSimple&& cb={});
/**
@@ -329,7 +329,7 @@ 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(const sockaddr_in* local4, const sockaddr_in6* local6, Config config);
void run(const SockAddr& local4, const SockAddr& local6, Config config);
/**
* Same as @run(sockaddr_in, sockaddr_in6, Identity, bool, StatusCallback), but with string IP addresses and service (port).
@@ -373,11 +373,9 @@ private:
*/
void tryBootstrapContinuously();
void doRun(const sockaddr_in* sin4, const sockaddr_in6* sin6, SecureDhtConfig config);
void doRun(const SockAddr& sin4, const SockAddr& sin6, SecureDhtConfig config);
time_point loop_();
static std::vector<std::pair<sockaddr_storage, socklen_t>> getAddrInfo(const std::string& host, const std::string& service);
NodeStatus getStatus() const {
return std::max(status4, status6);
}
+196 -142
View File
@@ -19,6 +19,7 @@
#pragma once
#include "def.h"
#include "rng.h"
#include <msgpack.hpp>
@@ -44,133 +45,99 @@ typedef uint16_t in_port_t;
#include <sstream>
#include <cstring>
// bytes
#define HASH_LEN 20u
namespace dht {
using byte = uint8_t;
template <size_t N>
class OPENDHT_PUBLIC Hash
{
public:
Hash() {
data_.fill(0);
}
Hash(const msgpack::object& o) {
msgpack_unpack(o);
}
byte* data() {
return data_.data();
}
const byte* data() const {
return data_.data();
}
constexpr size_t size() const { return data_.size(); }
explicit operator bool() const { return std::any_of(data_.begin(), data_.end(), [](byte _b) { return _b != 0; }); }
bool operator==(Hash const& _c) const { return data_ == _c.data_; }
bool operator!=(Hash const& _c) const { return data_ != _c.data_; }
bool operator<(Hash const& _c) const { for (unsigned i = 0; i < N; ++i) if (data_[i] < _c.data_[i]) return true; else if (data_[i] > _c.data_[i]) return false; return false; }
bool operator>=(Hash const& _c) const { return !operator<(_c); }
bool operator<=(Hash const& _c) const { return operator==(_c) || operator<(_c); }
bool operator>(Hash const& _c) const { return !operator<=(_c); }
template <typename Packer>
void msgpack_pack(Packer& pk) const
{
pk.pack_bin(N);
pk.pack_bin_body((char*)data_.data(), N);
}
void msgpack_unpack(msgpack::object o) {
if (o.type != msgpack::type::BIN or o.via.bin.size != N)
throw msgpack::type_error();
std::copy_n(o.via.bin.ptr, N, data_.data());
}
template <size_t M>
OPENDHT_PUBLIC friend std::ostream& operator<< (std::ostream& s, const Hash<M>& h);
template <size_t M>
OPENDHT_PUBLIC friend std::istream& operator>> (std::istream& s, Hash<M>& h);
std::string toString() const {
std::stringstream ss;
ss << *this;
return ss.str();
}
private:
std::array<byte, N> data_;
};
template <size_t N>
std::ostream& operator<< (std::ostream& s, const Hash<N>& h)
{
s << std::hex;
for (unsigned i=0; i<h.size(); i++)
s << std::setfill('0') << std::setw(2) << (unsigned)h.data_[i];
s << std::dec;
return s;
namespace crypto {
OPENDHT_PUBLIC void hash(const uint8_t* data, size_t data_length, uint8_t* hash, size_t hash_length);
}
template <size_t N>
std::istream& operator>> (std::istream& s, Hash<N>& h)
{
std::array<char, h.size()*2> dat;
s.exceptions(std::istream::eofbit | std::istream::failbit);
s.read(&(*dat.begin()), dat.size());
for (size_t i = 0; i < h.size(); i++) {
unsigned res = 0;
sscanf(dat.data() + 2*i, "%02x", &res);
h[i] = res;
}
return s;
}
using h256 = Hash<32>;
using PkId = h256;
/**
* Represents an InfoHash.
* An InfoHash is a byte array of HASH_LEN bytes.
* InfoHashes identify nodes and values in the Dht.
*/
class OPENDHT_PUBLIC InfoHash final : public std::array<uint8_t, HASH_LEN> {
template <size_t N>
class OPENDHT_PUBLIC Hash {
public:
constexpr InfoHash() : std::array<uint8_t, HASH_LEN>() {}
constexpr InfoHash(const std::array<uint8_t, HASH_LEN>& h) : std::array<uint8_t, HASH_LEN>(h) {}
InfoHash(const uint8_t* h, size_t h_len) : std::array<uint8_t, HASH_LEN>() {
if (h_len < HASH_LEN)
fill(0);
else
std::copy_n(h, HASH_LEN, begin());
}
using T = std::array<uint8_t, N>;
typedef typename T::iterator iterator;
typedef typename T::const_iterator const_iterator;
Hash () {
data_.fill(0);
}
Hash (const uint8_t* h, size_t data_len) {
if (data_len < N)
data_.fill(0);
else
std::copy_n(h, N, data_.begin());
}
/**
* Constructor from an hexadecimal string (without "0x").
* hex must be at least 2.HASH_LEN characters long.
* If too long, only the first 2.HASH_LEN characters are read.
*/
explicit InfoHash(const std::string& hex);
explicit Hash(const std::string& hex);
InfoHash(const msgpack::object& o) {
Hash(const msgpack::object& o) {
msgpack_unpack(o);
}
size_t size() const { return data_.size(); }
const uint8_t* data() const { return data_.data(); }
uint8_t* data() { return data_.data(); }
iterator begin() { return data_.begin(); }
const_iterator cbegin() const { return data_.cbegin(); }
iterator end() { return data_.end(); }
const_iterator cend() const { return data_.cend(); }
bool operator==(const Hash& h) const {
auto a = reinterpret_cast<const uint32_t*>(data_.data());
auto b = reinterpret_cast<const uint32_t*>(h.data_.data());
constexpr unsigned n = N / sizeof(uint32_t);
for (unsigned i=0; i < n; i++)
if (a[i] != b[i])
return false;
return true;
}
bool operator!=(const Hash& h) const { return !(*this == h); }
bool operator<(const Hash& o) const {
for(unsigned i = 0; i < N; i++) {
if(data_[i] != o.data_[i])
return data_[i] < o.data_[i];
}
return false;
}
explicit operator bool() const {
auto a = reinterpret_cast<const uint32_t*>(data_.data());
auto b = reinterpret_cast<const uint32_t*>(data_.data() + N);
for (; a != b; a++) {
if (*a)
return true;
}
return false;
}
uint8_t& operator[](size_t index) { return data_[index]; }
const uint8_t& operator[](size_t index) const { return data_[index]; }
/**
* Find the lowest 1 bit in an id.
* Result will allways be lower than 8*HASH_LEN
* Result will allways be lower than 8*N
*/
inline unsigned lowbit() const {
int i, j;
for(i = HASH_LEN-1; i >= 0; i--)
if((*this)[i] != 0)
for(i = N-1; i >= 0; i--)
if(data_[i] != 0)
break;
if(i < 0)
return -1;
for(j = 7; j >= 0; j--)
if(((*this)[i] & (0x80 >> j)) != 0)
if((data_[i] & (0x80 >> j)) != 0)
break;
return 8 * i + j;
}
@@ -179,25 +146,25 @@ public:
* Forget about the ``XOR-metric''. An id is just a path from the
* root of the tree, so bits are numbered from the start.
*/
static inline int cmp(const InfoHash& id1, const InfoHash& id2) {
return std::memcmp(id1.data(), id2.data(), HASH_LEN);
static inline int cmp(const Hash& id1, const Hash& id2) {
return std::memcmp(id1.data_.data(), id2.data_.data(), N);
}
/** Find how many bits two ids have in common. */
static inline unsigned
commonBits(const InfoHash& id1, const InfoHash& id2)
commonBits(const Hash& id1, const Hash& id2)
{
unsigned i, j;
uint8_t x;
for(i = 0; i < HASH_LEN; i++) {
if(id1[i] != id2[i])
for(i = 0; i < N; i++) {
if(id1.data_[i] != id2.data_[i])
break;
}
if(i == HASH_LEN)
return 8*HASH_LEN;
if(i == N)
return 8*N;
x = id1[i] ^ id2[i];
x = id1.data_[i] ^ id2.data_[i];
j = 0;
while((x & 0x80) == 0) {
@@ -210,14 +177,14 @@ public:
/** Determine whether id1 or id2 is closer to this */
int
xorCmp(const InfoHash& id1, const InfoHash& id2) const
xorCmp(const Hash& id1, const Hash& id2) const
{
for(unsigned i = 0; i < HASH_LEN; i++) {
for(unsigned i = 0; i < N; i++) {
uint8_t xor1, xor2;
if(id1[i] == id2[i])
if(id1.data_[i] == id2.data_[i])
continue;
xor1 = id1[i] ^ (*this)[i];
xor2 = id2[i] ^ (*this)[i];
xor1 = id1.data_[i] ^ data_[i];
xor2 = id2.data_[i] ^ data_[i];
if(xor1 < xor2)
return -1;
else
@@ -229,7 +196,7 @@ public:
bool
getBit(unsigned nbit) const
{
auto& num = *(cbegin()+(nbit/8));
auto& num = *(data_.cbegin()+(nbit/8));
unsigned bit = 7 - (nbit % 8);
return (num >> bit) & 1;
}
@@ -237,7 +204,7 @@ public:
void
setBit(unsigned nbit, bool b)
{
auto& num = (*this)[nbit/8];
auto& num = data_[nbit/8];
unsigned bit = 7 - (nbit % 8);
num ^= (-b ^ num) & (1 << bit);
}
@@ -245,45 +212,36 @@ public:
double toFloat() const {
using D = size_t;
double v = 0.;
for (size_t i = 0; i < std::min<size_t>(HASH_LEN, sizeof(D)-1); i++)
v += *(cbegin()+i) / (double)((D)1 << 8*(i+1));
for (size_t i = 0; i < std::min<size_t>(N, sizeof(D)-1); i++)
v += *(data_.cbegin()+i) / (double)((D)1 << 8*(i+1));
return v;
}
bool operator==(const InfoHash& h) const {
auto a = reinterpret_cast<const uint32_t*>(data());
auto b = reinterpret_cast<const uint32_t*>(h.data());
for (size_t i=0; i<(size()/sizeof(uint32_t)); i++)
if (a[i] != b[i])
return false;
return true;
}
bool operator<(const InfoHash& o) const {
for(unsigned i = 0; i < HASH_LEN; i++) {
if((*this)[i] != o[i])
return (*this)[i] < o[i];
}
return false;
}
static inline InfoHash get(const std::string& data) {
static inline Hash get(const std::string& data) {
return get((const uint8_t*)data.data(), data.size());
}
static inline InfoHash get(const std::vector<uint8_t>& data) {
static inline Hash get(const std::vector<uint8_t>& data) {
return get(data.data(), data.size());
}
/**
* Computes the hash from a given data buffer of size data_len.
*/
static InfoHash get(const uint8_t* data, size_t data_len);
static Hash get(const uint8_t* data, size_t data_len)
{
Hash ret;
crypto::hash(data, data_len, ret.data(), N);
return ret;
}
static InfoHash getRandom();
static Hash getRandom();
OPENDHT_PUBLIC friend std::ostream& operator<< (std::ostream& s, const InfoHash& h);
OPENDHT_PUBLIC friend std::istream& operator>> (std::istream& s, InfoHash& h);
template <size_t M>
OPENDHT_PUBLIC friend std::ostream& operator<< (std::ostream& s, const Hash<M>& h);
template <size_t M>
OPENDHT_PUBLIC friend std::istream& operator>> (std::istream& s, Hash<M>& h);
const char* to_c_str() const;
@@ -292,19 +250,115 @@ public:
template <typename Packer>
void msgpack_pack(Packer& pk) const
{
pk.pack_bin(HASH_LEN);
pk.pack_bin_body((char*)data(), HASH_LEN);
pk.pack_bin(N);
pk.pack_bin_body((char*)data_.data(), N);
}
void msgpack_unpack(msgpack::object o) {
if (o.type != msgpack::type::BIN or o.via.bin.size != HASH_LEN)
if (o.type != msgpack::type::BIN or o.via.bin.size != N)
throw msgpack::type_error();
std::copy_n(o.via.bin.ptr, HASH_LEN, data());
std::copy_n(o.via.bin.ptr, N, data_.data());
}
private:
T data_;
void fromString(const char*);
};
static constexpr const InfoHash zeroes {};
#define HASH_LEN 20u
using InfoHash = Hash<HASH_LEN>;
using h256 = Hash<32>;
using PkId = h256;
template <size_t N>
std::ostream& operator<< (std::ostream& s, const Hash<N>& h)
{
s.write(h.to_c_str(), N*2);
return s;
}
template <size_t N>
std::istream& operator>> (std::istream& s, Hash<N>& h)
{
std::array<char, h.size()*2> dat;
s.exceptions(std::istream::eofbit | std::istream::failbit);
s.read(&(*dat.begin()), dat.size());
fromString(dat.data());
return s;
}
template <size_t N>
Hash<N>::Hash(const std::string& hex) {
if (hex.size() < 2*N)
data_.fill(0);
else
fromString(hex.c_str());
}
template <size_t N>
void
Hash<N>::fromString(const char* in) {
auto hex2bin = [](char c) -> uint8_t {
if (c >= 'a' and c <= 'f') return 10 + c - 'a';
else if (c >= 'A' and c <= 'F') return 10 + c - 'A';
else if (c >= '0' and c <= '9') return c - '0';
else throw std::domain_error("not an hex character");
};
try {
for (size_t i=0; i<N; i++)
data_[i] = (hex2bin(in[2*i]) << 4) | hex2bin(in[2*i+1]);
} catch (const std::domain_error&) {
data_.fill(0);
}
}
template <size_t N>
Hash<N>
Hash<N>::getRandom()
{
Hash h;
crypto::random_device rdev;
std::uniform_int_distribution<uint32_t> rand_int;
auto a = reinterpret_cast<uint32_t*>(h.data());
auto b = reinterpret_cast<uint32_t*>(h.data() + h.size());
std::generate(a, b, std::bind(rand_int, std::ref(rdev)));
return h;
}
struct HexMap : public std::array<std::array<char, 2>, 256> {
HexMap() {
for (size_t i=0; i<size(); i++) {
auto& e = (*this)[i];
e[0] = hex_digits[(i >> 4) & 0x0F];
e[1] = hex_digits[i & 0x0F];
}
}
private:
static constexpr const char* hex_digits = "0123456789abcdef";
};
OPENDHT_PUBLIC extern const HexMap hex_map;
template <size_t N>
const char*
Hash<N>::to_c_str() const
{
thread_local std::array<char, N*2+1> buf;
for (size_t i=0; i<N; i++) {
auto b = buf.data()+i*2;
const auto& m = hex_map[data_[i]];
*((uint16_t*)b) = *((uint16_t*)&m);
}
return buf.data();
}
template <size_t N>
std::string
Hash<N>::toString() const
{
return std::string(to_c_str(), N*2);
}
const InfoHash zeroes {};
struct OPENDHT_PUBLIC NodeExport {
InfoHash id;
+1 -1
View File
@@ -75,7 +75,7 @@ struct Logger {
LogMethod ERR = NOLOG;
void setFilter(const InfoHash& f) {
filter_ = f;
filterEnable_ = filter_ != InfoHash{};
filterEnable_ = static_cast<bool>(filter_);
}
inline void log0(const LogMethod& logger, char const* format, va_list args) const {
#if OPENDHT_LOG
-67
View File
@@ -1,67 +0,0 @@
/*
* Copyright (C) 2014-2017 Savoir-faire Linux Inc.
* Author(s) : Adrien Béraud <adrien.beraud@savoirfairelinux.com>
* Simon Désaulniers <simon.desaulniers@savoirfairelinux.com>
*
* This program 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.
*
* You should have received a copy of the GNU General Public License
* along with this program. If not, see <https://www.gnu.org/licenses/>.
*/
#pragma once
#include <array>
#include <cstdint>
namespace dht {
namespace net {
struct TransPrefix : public std::array<uint8_t, 2> {
TransPrefix(const std::string& str) : std::array<uint8_t, 2>({{(uint8_t)str[0], (uint8_t)str[1]}}) {}
static const TransPrefix PING;
static const TransPrefix FIND_NODE;
static const TransPrefix GET_VALUES;
static const TransPrefix ANNOUNCE_VALUES;
static const TransPrefix REFRESH;
static const TransPrefix LISTEN;
};
/* Transaction-ids are 4-bytes long, with the first two bytes identifying
* the kind of request, and the remaining two a sequence number in
* host order.
*/
struct TransId final : public std::array<uint8_t, 4> {
static const constexpr uint16_t INVALID {0};
TransId() { std::fill(begin(), end(), 0); }
TransId(const std::array<char, 4>& o) { std::copy(o.begin(), o.end(), begin()); }
TransId(const TransPrefix prefix, uint16_t seqno = 0) {
std::copy_n(prefix.begin(), prefix.size(), begin());
*reinterpret_cast<uint16_t*>(data()+prefix.size()) = seqno;
}
TransId(uint32_t id) {
*reinterpret_cast<uint32_t*>(data()) = htonl(id);
}
uint32_t toInt() const {
return ntohl(*reinterpret_cast<const uint32_t*>(&(*this)[0]));
}
bool matches(const TransPrefix prefix) const {
return std::equal(begin(), begin()+2, prefix.begin());
}
};
TransId unpackTid(msgpack::object& o);
} /* namespace net */
} /* dht */
+12 -12
View File
@@ -27,7 +27,6 @@
#include "utils.h"
#include "rng.h"
#include "rate_limiter.h"
#include "net.h"
#include <vector>
#include <string>
@@ -41,6 +40,7 @@ namespace net {
struct Request;
struct Socket;
struct TransId;
#ifndef MSG_CONFIRM
#define MSG_CONFIRM 0
@@ -169,7 +169,7 @@ private:
std::function<RequestAnswer(Sp<Node>,
const InfoHash&,
const Blob&,
uint32_t,
Tid,
const Query&)> onListen {};
/**
* Called on announce request.
@@ -231,7 +231,7 @@ public:
* @param nodes6 The ipv6 closest nodes.
* @param values The values to send.
*/
void tellListener(Sp<Node> n, uint32_t socket_id, const InfoHash& hash, want_t want, const Blob& ntoken,
void tellListener(Sp<Node> n, Tid socket_id, const InfoHash& hash, want_t want, const Blob& ntoken,
std::vector<Sp<Node>>&& nodes, std::vector<Sp<Node>>&& nodes6,
std::vector<Sp<Value>>&& values, const Query& q);
@@ -465,18 +465,18 @@ private:
// basic wrapper for socket sendto function
int send(const char *buf, size_t len, int flags, const SockAddr& addr);
void sendValueParts(TransId tid, const std::vector<Blob>& svals, const SockAddr& addr);
void sendValueParts(const TransId& tid, const std::vector<Blob>& svals, const SockAddr& addr);
std::vector<Blob> packValueHeader(msgpack::sbuffer&, const std::vector<Sp<Value>>&);
void maintainRxBuffer(const TransId& tid);
void maintainRxBuffer(Tid tid);
/*************
* Answers *
*************/
/* answer to a ping request */
void sendPong(const SockAddr& addr, TransId tid);
void sendPong(const SockAddr& addr, Tid tid);
/* answer to findnodes/getvalues request */
void sendNodesValues(const SockAddr& addr,
TransId tid,
Tid tid,
const Blob& nodes,
const Blob& nodes6,
const std::vector<Sp<Value>>& st,
@@ -490,12 +490,12 @@ private:
std::vector<Sp<Node>>& nodes,
std::vector<Sp<Node>>& nodes6);
/* answer to a listen request */
void sendListenConfirmation(const SockAddr& addr, TransId tid);
void sendListenConfirmation(const SockAddr& addr, Tid tid);
/* answer to put request */
void sendValueAnnounced(const SockAddr& addr, TransId, Value::Id);
void sendValueAnnounced(const SockAddr& addr, Tid, Value::Id);
/* answer in case of error */
void sendError(const SockAddr& addr,
TransId tid,
Tid tid,
uint16_t code,
const std::string& message,
bool include_id=false);
@@ -519,8 +519,8 @@ private:
size_t limiter_maintenance {0};
// requests handling
std::map<TransId, Sp<Request>> requests {};
std::map<TransId, PartialMessage> partial_messages;
std::map<Tid, Sp<Request>> requests {};
std::map<Tid, PartialMessage> partial_messages;
MessageStats in_stats {}, out_stats {};
std::set<SockAddr> blacklist {};
+10 -11
View File
@@ -22,7 +22,6 @@
#include "infohash.h" // includes socket structures
#include "utils.h"
#include "sockaddr.h"
#include "net.h"
#include <list>
#include <map>
@@ -36,8 +35,8 @@ struct Socket;
struct RequestAnswer;
} /* namespace net */
using Tid = uint32_t;
using SocketCb = std::function<void(const Sp<Node>&, net::RequestAnswer&&)>;
using SocketId = uint32_t;
struct Socket {
Socket() {}
Socket(SocketCb&& on_receive) :
@@ -102,7 +101,7 @@ struct Node {
void requested(const Sp<net::Request>& req);
void received(time_point now, const Sp<net::Request>& req);
Sp<net::Request> getRequest(const net::TransId& tid);
Sp<net::Request> getRequest(Tid tid);
void cancelRequest(const Sp<net::Request>& req);
void setExpired();
@@ -116,16 +115,16 @@ struct Node {
*
* @return the socket.
*/
SocketId openSocket(SocketCb&& cb);
Tid openSocket(SocketCb&& cb);
Socket* getSocket(SocketId id);
Socket* getSocket(Tid id);
/**
* Closes a socket so that no further data will be red on that socket.
*
* @param socket The socket to close.
*/
void closeSocket(SocketId id);
void closeSocket(Tid id);
/**
* Resets the state of the node so it's not expired anymore.
@@ -137,9 +136,9 @@ struct Node {
*
* @return the new id.
*/
uint16_t getNewTid() {
Tid getNewTid() {
++transaction_id;
return transaction_id == net::TransId::INVALID ? ++transaction_id : transaction_id;
return transaction_id ? ++transaction_id : transaction_id;
}
std::string toString() const;
@@ -164,11 +163,11 @@ private:
time_point reply_time {time_point::min()}; /* time of last correct reply received */
unsigned auth_errors {0};
bool expired_ {false};
SocketId transaction_id;
Tid transaction_id;
using TransactionDist = std::uniform_int_distribution<decltype(transaction_id)>;
std::map<net::TransId, Sp<net::Request>> requests_ {};
std::map<SocketId, Socket> sockets_;
std::map<Tid, Sp<net::Request>> requests_ {};
std::map<Tid, Socket> sockets_;
};
}
+3
View File
@@ -37,12 +37,15 @@ struct NodeCache {
*/
void clearBadNodes(sa_family_t family = 0);
~NodeCache();
private:
class NodeMap : public std::map<InfoHash, std::weak_ptr<Node>> {
public:
Sp<Node> getNode(const InfoHash& id);
Sp<Node> getNode(const InfoHash& id, const SockAddr&, time_point now, bool confirmed, bool client);
void clearBadNodes();
void setExpired();
};
const NodeMap& cache(sa_family_t af) const { return af == AF_INET ? cache_4 : cache_6; }
+2 -2
View File
@@ -23,12 +23,12 @@
namespace dht {
template<size_t Quota, unsigned long Perdiod=1>
template<size_t Quota, unsigned long Period=1>
class RateLimiter {
public:
/** Clear outdated records and return current quota usage */
size_t maintain(const time_point& now) {
auto limit = now - std::chrono::seconds(Perdiod);
auto limit = now - std::chrono::seconds(Period);
while (not records.empty() and records.front() < limit)
records.pop();
return records.size();
+3
View File
@@ -36,6 +36,7 @@ typedef uint16_t in_port_t;
#include <string>
#include <memory>
#include <vector>
#include <cstring>
#include <cstddef>
@@ -71,6 +72,8 @@ public:
*/
SockAddr(const sockaddr_storage& ss, socklen_t len) : SockAddr((const sockaddr*)&ss, len) {}
static std::vector<SockAddr> resolve(const std::string& host, const std::string& service = {});
bool operator<(const SockAddr& o) const {
if (len != o.len)
return len < o.len;
+26 -2
View File
@@ -37,6 +37,10 @@
#include <chrono>
#include <set>
#if OPENDHT_PROXY_SERVER
#include <json/json.h>
#endif //OPENDHT_PROXY_SERVER
namespace dht {
struct Value;
@@ -346,6 +350,14 @@ struct OPENDHT_PUBLIC Value
Value(ValueType::Id t, const uint8_t* dat_ptr, size_t dat_len, Id id = INVALID_ID)
: id(id), type(t), data(dat_ptr, dat_ptr+dat_len) {}
#if OPENDHT_PROXY_SERVER
/**
* Build a value from a json object
* @param json
*/
Value(Json::Value& json);
#endif //OPENDHT_PROXY_SERVER
template <typename Type>
Value(ValueType::Id t, const Type& d, Id id = INVALID_ID)
: id(id), type(t), data(packMsg(d)) {}
@@ -417,6 +429,18 @@ struct OPENDHT_PUBLIC Value
return ss.str();
}
#if OPENDHT_PROXY_SERVER
/**
* Build a json object from a value
* Example:
* {
* "data":"base64ofdata",
* id":"0", "seq":0,"type":3
* }
*/
Json::Value toJson() const;
#endif //OPENDHT_PROXY_SERVER
/** Return the size in bytes used by this value in memory (minimum). */
size_t size() const;
@@ -424,11 +448,11 @@ struct OPENDHT_PUBLIC Value
void msgpack_pack_to_sign(Packer& pk) const
{
bool has_owner = owner && *owner;
pk.pack_map((user_type.empty()?0:1) + (has_owner?(recipient == InfoHash() ? 4 : 5):2));
pk.pack_map((user_type.empty()?0:1) + (has_owner?(recipient ? 5 : 4):2));
if (has_owner) { // isSigned
pk.pack(std::string("seq")); pk.pack(seq);
pk.pack(std::string("owner")); owner->msgpack_pack(pk);
if (recipient != InfoHash()) {
if (recipient) {
pk.pack(std::string("to")); pk.pack(recipient);
}
}
+2 -2
View File
@@ -92,9 +92,9 @@ cdef class _WithID(object):
cdef class InfoHash(_WithID):
cdef cpp.InfoHash _infohash
def __cinit__(self, bytes str=b''):
self._infohash = cpp.InfoHash(str)
self._infohash = cpp.InfoHash(str) if str else cpp.InfoHash()
def __bool__(InfoHash self):
return not (self._infohash == cpp.InfoHash())
return <bool>self._infohash
def __richcmp__(InfoHash self, InfoHash other, int op):
if op == 0:
return self._infohash < other._infohash
+3 -3
View File
@@ -65,8 +65,9 @@ cdef extern from "opendht/infohash.h" namespace "dht":
InfoHash get(string s)
@staticmethod
InfoHash getRandom()
bool operator==(InfoHash)
bool operator<(InfoHash)
bool operator==(InfoHash) const
bool operator<(InfoHash) const
bool operator bool() const
cdef extern from "opendht/sockaddr.h" namespace "dht":
cdef cppclass SockAddr:
@@ -259,4 +260,3 @@ cdef extern from "opendht/indexation/pht.h" namespace "dht::indexation":
void insert(IndexKey k, IndexValue v, DoneCallbackSimple cb)
@staticmethod
LookupCallback bindLookupCb(LookupCallbackRaw cb, void *user_data)
+6 -1
View File
@@ -9,6 +9,7 @@ libopendht_la_SOURCES = \
listener.h \
request.h \
search.h \
net.h \
parsed_message.h \
node_cache.cpp \
callbacks.cpp \
@@ -37,7 +38,6 @@ nobase_include_HEADERS = \
../include/opendht/node_cache.h \
../include/opendht/routing_table.h \
../include/opendht/network_engine.h \
../include/opendht/net.h \
../include/opendht/scheduler.h \
../include/opendht/rate_limiter.h \
../include/opendht/utils.h \
@@ -54,6 +54,11 @@ nobase_include_HEADERS = \
../include/opendht/rng.h \
../include/opendht/indexation/pht.h
if ENABLE_PROXY_SERVER
libopendht_la_SOURCES += base64.h base64.cpp dht_proxy_server.cpp
nobase_include_HEADERS += ../include/opendht/dht_proxy_server.h
endif
clean-local:
rm -rf libargon2.la
+143
View File
@@ -0,0 +1,143 @@
/*
* Copyright (C) 2004-2017 Savoir-faire Linux Inc.
*
* This program 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.
*
* You should have received a copy of the GNU General Public License
* along with this program. If not, see <https://www.gnu.org/licenses/>.
*/
#include "base64.h"
#include <stdint.h>
#include <stdlib.h>
/* Mainly based on the following stackoverflow question:
* http://stackoverflow.com/questions/342409/how-do-i-base64-encode-decode-in-c
*/
static const char encoding_table[] = {
'A', 'B', 'C', 'D', 'E', 'F', 'G', 'H', 'I', 'J', 'K',
'L', 'M', 'N', 'O', 'P', 'Q', 'R', 'S', 'T', 'U', 'V',
'W', 'X', 'Y', 'Z', 'a', 'b', 'c', 'd', 'e', 'f', 'g',
'h', 'i', 'j', 'k', 'l', 'm', 'n', 'o', 'p', 'q', 'r',
's', 't', 'u', 'v', 'w', 'x', 'y', 'z', '0', '1', '2',
'3', '4', '5', '6', '7', '8', '9', '+', '/'
};
static const size_t mod_table[] = { 0, 2, 1 };
char *base64_encode(const uint8_t *input, size_t input_length,
char *output, size_t *output_length)
{
size_t i, j;
size_t out_sz = *output_length;
*output_length = 4 * ((input_length + 2) / 3);
if (out_sz < *output_length || output == nullptr)
return nullptr;
for (i = 0, j = 0; i < input_length; ) {
uint8_t octet_a = i < input_length ? input[i++] : 0;
uint8_t octet_b = i < input_length ? input[i++] : 0;
uint8_t octet_c = i < input_length ? input[i++] : 0;
uint32_t triple = (octet_a << 0x10) + (octet_b << 0x08) + octet_c;
output[j++] = encoding_table[(triple >> 3 * 6) & 0x3F];
output[j++] = encoding_table[(triple >> 2 * 6) & 0x3F];
output[j++] = encoding_table[(triple >> 1 * 6) & 0x3F];
output[j++] = encoding_table[(triple >> 0 * 6) & 0x3F];
}
for (i = 0; i < mod_table[input_length % 3]; i++)
output[*output_length - 1 - i] = '=';
return output;
}
uint8_t *base64_decode(const char *input, size_t input_length,
uint8_t *output, size_t *output_length)
{
size_t i, j;
uint8_t decoding_table[256];
uint8_t c;
for (c = 0; c < 64; c++)
decoding_table[static_cast<int>(encoding_table[c])] = c;
if (input_length % 4 != 0 || input_length < 2)
return nullptr;
size_t out_sz = *output_length;
*output_length = input_length / 4 * 3;
if (input[input_length - 1] == '=')
(*output_length)--;
if (input[input_length - 2] == '=')
(*output_length)--;
if (out_sz < *output_length || output == nullptr)
return nullptr;
for (i = 0, j = 0; i < input_length;) {
uint8_t sextet_a = input[i] == '=' ? 0 & i++
: decoding_table[static_cast<int>(input[i++])];
uint8_t sextet_b = input[i] == '=' ? 0 & i++
: decoding_table[static_cast<int>(input[i++])];
uint8_t sextet_c = input[i] == '=' ? 0 & i++
: decoding_table[static_cast<int>(input[i++])];
uint8_t sextet_d = input[i] == '=' ? 0 & i++
: decoding_table[static_cast<int>(input[i++])];
uint32_t triple = (sextet_a << 3 * 6) +
(sextet_b << 2 * 6) +
(sextet_c << 1 * 6) +
(sextet_d << 0 * 6);
if (j < *output_length)
output[j++] = (triple >> 2 * 8) & 0xFF;
if (j < *output_length)
output[j++] = (triple >> 1 * 8) & 0xFF;
if (j < *output_length)
output[j++] = (triple >> 0 * 8) & 0xFF;
}
return output;
}
std::string
base64_encode(const std::vector<uint8_t>::const_iterator begin,
const std::vector<uint8_t>::const_iterator end)
{
size_t output_length = 4 * ((std::distance(begin, end) + 2) / 3);
std::string out;
out.resize(output_length);
base64_encode(&(*begin), std::distance(begin, end),
&(*out.begin()), &output_length);
out.resize(output_length);
return out;
}
std::string
base64_encode(const std::vector<unsigned char>& str)
{
return base64_encode(str.cbegin(), str.cend());
}
std::string
base64_decode(const std::string& str)
{
size_t output_length = str.length() / 4 * 3 + 2;
std::vector<uint8_t> output;
output.resize(output_length);
base64_decode(str.data(), str.size(), output.data(), &output_length);
output.resize(output_length);
return std::string(output.begin(), output.end());
}
+36
View File
@@ -0,0 +1,36 @@
/*
* Copyright (C) 2004-2017 Savoir-faire Linux Inc.
*
* This program 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.
*
* You should have received a copy of the GNU General Public License
* along with this program. If not, see <https://www.gnu.org/licenses/>.
*/
#pragma once
#include <string>
#include <vector>
/**
* Encode a buffer in base64.
*
* @param str the input buffer
* @return a base64-encoded buffer
*/
std::string base64_encode(const std::vector<unsigned char>& str);
/**
* Decode a buffer in base64.
*
* @param str the input buffer
* @return a base64-decoded buffer
*/
std::string base64_decode(const std::string& str);
+20
View File
@@ -68,4 +68,24 @@ NodeStats::toString() const
return ss.str();
}
#if OPENDHT_PROXY_SERVER
/**
* Build a json object from a NodeStats
*/
Json::Value
NodeStats::toJson() const
{
Json::Value val;
val["good"] = static_cast<Json::LargestUInt>(good_nodes);
val["dubious"] = static_cast<Json::LargestUInt>(dubious_nodes);
val["incoming"] = static_cast<Json::LargestUInt>(incoming_nodes);
if (table_depth > 1) {
val["table_depth"] = static_cast<Json::LargestUInt>(table_depth);
unsigned long tot_nodes = 8 * std::exp2(table_depth);
val["network_size_estimation"] = static_cast<Json::LargestUInt>(tot_nodes);
}
return val;
}
#endif //OPENDHT_PROXY_SERVER
}
+13 -4
View File
@@ -84,7 +84,7 @@ 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 : (
(min_res > 160/8) ? GNUTLS_DIG_SHA256 : (
GNUTLS_DIG_SHA1));
}
@@ -212,12 +212,21 @@ Blob hash(const Blob& data, size_t hash_len)
Blob res;
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))
throw CryptoException("Can't compute hash !");
if (auto err = gnutls_fingerprint(algo, &gdat, res.data(), &res_size))
throw CryptoException(std::string("Can't compute hash: ") + gnutls_strerror(err));
res.resize(std::min(hash_len, res_size));
return res;
}
void hash(const uint8_t* data, size_t data_length, uint8_t* hash, size_t hash_length)
{
auto algo = gnutlsHashAlgo(hash_length);
size_t res_size = hash_length;
const gnutls_datum_t gdat {(uint8_t*)data, (unsigned)data_length};
if (auto err = gnutls_fingerprint(algo, &gdat, hash, &res_size))
throw CryptoException(std::string("Can't compute hash: ") + gnutls_strerror(err));
}
PrivateKey::PrivateKey()
{}
@@ -529,7 +538,7 @@ PublicKey::getId() const
return {};
InfoHash id;
size_t sz = id.size();
constexpr int flags = (id.size() == 32) ? GNUTLS_KEYID_USE_SHA256 : 0;
const int flags = (id.size() == 32) ? GNUTLS_KEYID_USE_SHA256 : 0;
if (auto err = gnutls_pubkey_get_key_id(pk, flags, id.data(), &sz))
throw CryptoException(std::string("Can't get public key ID: ") + gnutls_strerror(err));
if (sz != id.size())
+95 -93
View File
@@ -115,6 +115,7 @@ Dht::getPublicAddress(sa_family_t family)
return a.first > b.first;
});
std::vector<SockAddr> ret;
ret.reserve(!family ? reported_addr.size() : reported_addr.size()/2);
for (const auto& addr : reported_addr)
if (!family || family == addr.second.getFamily())
ret.emplace_back(addr.second);
@@ -386,11 +387,12 @@ void Dht::searchSendAnnounceValue(const Sp<Search>& sr) {
std::weak_ptr<Search> ws = sr;
const auto& now = scheduler.time();
for (auto& n : sr->nodes) {
auto something_to_announce = std::find_if(sr->announce.cbegin(), sr->announce.cend(),
[this,&now,&sr,&n](const Announce& a) {
return n.isSynced(now) and n.getAnnounceTime(a.value->id) <= now;
}) != sr->announce.cend();
if (not something_to_announce)
if (not n.isSynced(now))
continue;
if (std::find_if(sr->announce.cbegin(), sr->announce.cend(),
[this,&now,&n](const Announce& a) {
return n.getAnnounceTime(a.value->id) <= now;
}) == sr->announce.cend())
continue;
auto onDone = [this,ws](const net::Request& req, net::RequestAnswer&& answer)
@@ -409,62 +411,63 @@ void Dht::searchSendAnnounceValue(const Sp<Search>& sr) {
auto onSelectDone =
[this,ws,onDone,onExpired](const net::Request& req, net::RequestAnswer&& answer) mutable
{ /* on probing done */
auto sr = ws.lock();
if (not sr) return;
auto sn = sr->getNode(req.node);
if (not sn) return;
const auto& now = scheduler.time();
if (auto sr = ws.lock()) {
if (auto sn = sr->getNode(req.node)) {
for (auto ait = sr->announce.begin(); ait != sr->announce.end();) {
auto& a = *ait;
if (sn->isSynced(now) and sn->getAnnounceTime(a.value->id) <= now) {
bool hasValue {false};
uint16_t seq_no = 0;
try {
const auto& f = std::find_if(answer.fields.cbegin(), answer.fields.cend(),
[&a](const Sp<FieldValueIndex>& i){
return i->index.at(Value::Field::Id).getInt() == a.value->id;
});
if (f != answer.fields.cend() and *f) {
hasValue = true;
seq_no = static_cast<uint16_t>((*f)->index.at(Value::Field::SeqNum).getInt());
}
} catch (std::out_of_range&) { }
auto next_refresh_time = now + getType(a.value->type).expiration;
/* only put the value if the node doesn't already have it */
if (not hasValue or seq_no < a.value->seq) {
DHT_LOG.w(sr->id, sn->node->id, "[search %s] [node %s] sending 'put' (vid: %d)",
sr->id.toString().c_str(), sn->node->toString().c_str(), a.value->id);
sn->acked[a.value->id] = std::make_pair(network_engine.sendAnnounceValue(sn->node,
sr->id,
a.value,
a.permanent ? time_point::max() : a.created,
sn->token,
onDone,
onExpired), next_refresh_time);
} else if (hasValue and a.permanent) {
DHT_LOG.w(sr->id, sn->node->id, "[search %s] [node %s] sending 'refresh' (vid: %d)",
sr->id.toString().c_str(), sn->node->toString().c_str(), a.value->id);
sn->acked[a.value->id] = std::make_pair(network_engine.sendRefreshValue(sn->node,
sr->id,
a.value->id,
sn->token,
onDone,
onExpired), next_refresh_time);
} else {
DHT_LOG.w(sr->id, sn->node->id, "[search %s] [node %s] already has value (vid: %d). Aborting.",
sr->id.toString().c_str(), sn->node->toString().c_str(), a.value->id);
auto ack_req = std::make_shared<net::Request>(net::Request::State::COMPLETED);
ack_req->reply_time = now;
sn->acked[a.value->id] = std::make_pair(std::move(ack_req), next_refresh_time);
/* step to clear announces */
scheduler.edit(sr->nextSearchStep, now);
}
} else {
/* Search is now unsynced. Let's call searchStep to sync again. */
scheduler.edit(sr->nextSearchStep, sr->getNextStepTime(now));
}
++ait;
if (not sn->isSynced(now)) {
/* Search is now unsynced. Let's call searchStep to sync again. */
scheduler.edit(sr->nextSearchStep, sr->getNextStepTime(now));
return;
}
for (auto& a : sr->announce) {
if (sn->getAnnounceTime(a.value->id) > now)
continue;
bool hasValue {false};
uint16_t seq_no = 0;
try {
const auto& f = std::find_if(answer.fields.cbegin(), answer.fields.cend(),
[&a](const Sp<FieldValueIndex>& i){
return i->index.at(Value::Field::Id).getInt() == a.value->id;
});
if (f != answer.fields.cend() and *f) {
hasValue = true;
seq_no = static_cast<uint16_t>((*f)->index.at(Value::Field::SeqNum).getInt());
}
} catch (std::out_of_range&) { }
auto next_refresh_time = now + getType(a.value->type).expiration;
/* only put the value if the node doesn't already have it */
if (not hasValue or seq_no < a.value->seq) {
DHT_LOG.w(sr->id, sn->node->id, "[search %s] [node %s] sending 'put' (vid: %d)",
sr->id.toString().c_str(), sn->node->toString().c_str(), a.value->id);
sn->acked[a.value->id] = std::make_pair(network_engine.sendAnnounceValue(sn->node,
sr->id,
a.value,
a.permanent ? time_point::max() : a.created,
sn->token,
onDone,
onExpired), next_refresh_time);
} else if (hasValue and a.permanent) {
DHT_LOG.w(sr->id, sn->node->id, "[search %s] [node %s] sending 'refresh' (vid: %d)",
sr->id.toString().c_str(), sn->node->toString().c_str(), a.value->id);
sn->acked[a.value->id] = std::make_pair(network_engine.sendRefreshValue(sn->node,
sr->id,
a.value->id,
sn->token,
onDone,
onExpired), next_refresh_time);
} else {
DHT_LOG.w(sr->id, sn->node->id, "[search %s] [node %s] already has value (vid: %d). Aborting.",
sr->id.toString().c_str(), sn->node->toString().c_str(), a.value->id);
auto ack_req = std::make_shared<net::Request>(net::Request::State::COMPLETED);
ack_req->reply_time = now;
sn->acked[a.value->id] = std::make_pair(std::move(ack_req), next_refresh_time);
/* step to clear announces */
scheduler.edit(sr->nextSearchStep, now);
}
}
};
@@ -553,8 +556,8 @@ Dht::searchStep(Sp<Search> sr)
[this,ws,query](const net::Request& req, net::RequestAnswer&& answer) mutable
{ /* on done */
if (auto sr = ws.lock()) {
onListenDone(req.node, answer, sr);
scheduler.edit(sr->nextSearchStep, scheduler.time());
onListenDone(req.node, answer, sr);
}
},
[this,ws,query](const net::Request& req, bool over) mutable
@@ -569,8 +572,8 @@ Dht::searchStep(Sp<Search> sr)
[this,ws,query](const Sp<Node>& node, net::RequestAnswer&& answer) mutable
{ /* on new values */
if (auto sr = ws.lock()) {
onGetValuesDone(node, answer, sr, query);
scheduler.edit(sr->nextSearchStep, scheduler.time());
onGetValuesDone(node, answer, sr, query);
}
}
);
@@ -670,8 +673,10 @@ Dht::search(const InfoHash& id, sa_family_t af, GetCallback gcb, QueryCallback q
break;
}
}
if (not sr)
throw DhtException("Can't create search");
if (not sr) {
DHT_LOG.e(id, "[search %s IPv%c] maximum number of searches reached !", id.toString().c_str(), (af == AF_INET) ? '4' : '6');
return {};
}
}
sr->af = af;
sr->tid = search_id++;
@@ -741,8 +746,9 @@ Dht::announce(const InfoHash& id,
n.probe_query.reset();
n.acked[value->id].first.reset();
}
}
else {
} else {
a_sr->permanent = permanent;
a_sr->created = created;
if (a_sr->value != value) {
a_sr->value = value;
for (auto& n : sr->nodes) {
@@ -754,9 +760,8 @@ Dht::announce(const InfoHash& id,
if (a_sr->callback)
a_sr->callback(true, {});
a_sr->callback = {};
if (callback) {
if (callback)
callback(true, {});
}
return;
} else {
if (a_sr->callback)
@@ -1256,8 +1261,7 @@ void
Dht::expireStore()
{
// removing expired values
auto i = store.begin();
while (i != store.end()) {
for (auto i = store.begin(); i != store.end();) {
expireStore(i);
if (i->second.empty() && i->second.listeners.empty() && i->second.local_listeners.empty()) {
@@ -1291,10 +1295,17 @@ Dht::expireStore()
DHT_LOG.w("Discarded %ld bytes, still %ld used", largest->first.toString().c_str(), total_store_size);
if (ret.values_diff)
break;
} else
std::cout << "exp_value not found " << exp_value.first << std::endl;
}
}
}
// remove unused quota entires
for (auto i = store_quota.begin(); i != store_quota.end();) {
if (i->second.size() == 0)
i = store_quota.erase(i);
else
++i;
}
}
void
@@ -1534,12 +1545,6 @@ Dht::dumpTables() const
DHT_LOG.d("%s", out.str().c_str());
}
void
Dht::printStorageQuota(std::ostream& out, const decltype(store_quota)::value_type& ip) const
{
out << "IP " << ip.first.toString() << " uses " << ip.second.size() << " bytes" << std::endl;
}
std::string
Dht::getStorageLog() const
{
@@ -1552,9 +1557,12 @@ Dht::getStorageLog() const
else
out << (total_store_size/1024) << " KB)";
out << std::endl << std::endl;
for (const auto& ip : store_quota) {
printStorageQuota(out, ip);
}
std::multimap<size_t, const SockAddr*> q_map;
for (const auto& ip : store_quota)
if (ip.second.size())
q_map.emplace(ip.second.size(), &ip.first);
for (auto ip = q_map.rbegin(); ip != q_map.rend(); ++ip)
out << "IP " << ip->second->toString() << " uses " << ip->first << " bytes" << std::endl;
out << std::endl;
return out.str();
}
@@ -1583,15 +1591,9 @@ Dht::printStorageLog(const decltype(store)::value_type& s) const
<< st.totalSize() << " bytes)" << std::endl;
if (not st.local_listeners.empty())
out << " " << st.local_listeners.size() << " local listeners" << std::endl;
const auto& now = scheduler.time();
for (const auto& node_listeners : st.listeners) {
const auto& node = node_listeners.first;
for (const auto& l : node_listeners.second) {
out << " " << "Listener " << node->toString();
auto since = duration_cast<seconds>(now - l.second.time);
auto expires = duration_cast<seconds>(l.second.time + Node::NODE_EXPIRE_TIME - now);
out << " (since " << since.count() << "s, exp in " << expires.count() << "s)" << std::endl;
}
out << " " << "Listener " << node->toString() << " : " << node_listeners.second.size() << " entries" << std::endl;
}
return out.str();
}
@@ -1656,7 +1658,7 @@ Dht::~Dht()
Dht::Dht() : store(), scheduler(DHT_LOG), network_engine(DHT_LOG, scheduler) {}
Dht::Dht(int s, int s6, Config config)
: myid(config.node_id != zeroes ? config.node_id : InfoHash::getRandom()),
: myid(config.node_id ? config.node_id : InfoHash::getRandom()),
is_bootstrap(config.is_bootstrap),
maintain_storage(config.maintain_storage), store(), store_quota(),
scheduler(DHT_LOG),
@@ -1968,7 +1970,7 @@ Dht::importValues(const std::vector<ValuesExport>& import)
throw msgpack::type_error();
for (unsigned i = 0; i < valarr.via.array.size; i++) {
auto& valel = valarr.via.array.ptr[i];
if (valel.via.array.size < 2)
if (valel.type != msgpack::type::ARRAY or valel.via.array.size < 2)
throw msgpack::type_error();
time_point val_time;
Value tmp_val;
@@ -2103,7 +2105,7 @@ Dht::onFindNode(Sp<Node> node, const InfoHash& target, want_t want)
net::RequestAnswer
Dht::onGetValues(Sp<Node> node, const InfoHash& hash, want_t, const Query& query)
{
if (hash == zeroes) {
if (not hash) {
DHT_LOG.w("[node %s] Eek! Got get_values with no info_hash", node->toString().c_str());
throw net::DhtProtocolException {
net::DhtProtocolException::NON_AUTHORITATIVE_INFORMATION,
@@ -2200,14 +2202,14 @@ void Dht::onGetValuesDone(const Sp<Node>& node,
net::RequestAnswer
Dht::onListen(Sp<Node> node, const InfoHash& hash, const Blob& token, size_t socket_id, const Query& query)
{
if (hash == zeroes) {
if (not hash) {
DHT_LOG.w(node->id, "[node %s] listen with no info_hash", node->toString().c_str());
throw net::DhtProtocolException {
net::DhtProtocolException::NON_AUTHORITATIVE_INFORMATION,
net::DhtProtocolException::LISTEN_NO_INFOHASH
};
}
if (!tokenMatch(token, node->getAddr())) {
if (not tokenMatch(token, node->getAddr())) {
DHT_LOG.w(hash, node->id, "[node %s] incorrect token %s for 'listen'", node->toString().c_str(), hash.toString().c_str());
throw net::DhtProtocolException {net::DhtProtocolException::UNAUTHORIZED, net::DhtProtocolException::LISTEN_WRONG_TOKEN};
}
@@ -2239,7 +2241,7 @@ Dht::onAnnounce(Sp<Node> n,
const time_point& creation_date)
{
auto& node = *n;
if (hash == zeroes) {
if (not hash) {
DHT_LOG.w(node.id, "put with no info_hash");
throw net::DhtProtocolException {
net::DhtProtocolException::NON_AUTHORITATIVE_INFORMATION,
+393
View File
@@ -0,0 +1,393 @@
/*
* Copyright (C) 2017 Savoir-faire Linux Inc.
* Author : Sébastien Blin <sebastien.blin@savoirfairelinux.com>
*
* This program 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.
*
* You should have received a copy of the GNU General Public License
* along with this program. If not, see <https://www.gnu.org/licenses/>.
*/
#if OPENDHT_PROXY_SERVER
#include "dht_proxy_server.h"
#include "default_types.h"
#include "dhtrunner.h"
#include "msgpack.hpp"
#include <chrono>
#include <functional>
#include <json/json.h>
#include <limits>
using namespace std::placeholders;
namespace dht {
DhtProxyServer::DhtProxyServer(std::shared_ptr<DhtRunner> dht, in_port_t port)
: dht_(dht)
{
// NOTE in c++14, use make_unique
service_ = std::unique_ptr<restbed::Service>(new restbed::Service());
server_thread = std::thread([this, port]() {
// Create endpoints
auto resource = std::make_shared<restbed::Resource>();
resource->set_path("/");
resource->set_method_handler("GET", std::bind(&DhtProxyServer::getNodeInfo, this, _1));
service_->publish(resource);
resource = std::make_shared<restbed::Resource>();
resource->set_path("/{hash: .*}");
resource->set_method_handler("GET", std::bind(&DhtProxyServer::get, this, _1));
resource->set_method_handler("LISTEN", std::bind(&DhtProxyServer::listen, this, _1));
resource->set_method_handler("POST", std::bind(&DhtProxyServer::put, this, _1));
#if OPENDHT_PROXY_SERVER_IDENTITY
resource->set_method_handler("SIGN", std::bind(&DhtProxyServer::putSigned, this, _1));
resource->set_method_handler("ENCRYPT", std::bind(&DhtProxyServer::putEncrypted, this, _1));
#endif // OPENDHT_PROXY_SERVER_IDENTITY
resource->set_method_handler("OPTIONS", std::bind(&DhtProxyServer::handleOptionsMethod, this, _1));
service_->publish(resource);
resource = std::make_shared<restbed::Resource>();
resource->set_path("/{hash: .*}/{value: .*}");
resource->set_method_handler("GET", std::bind(&DhtProxyServer::getFiltered, this, _1));
service_->publish(resource);
// Start server
auto settings = std::make_shared<restbed::Settings>();
settings->set_default_header("Content-Type", "application/json");
settings->set_default_header("Connection", "keep-alive");
settings->set_default_header("Access-Control-Allow-Origin", "*");
std::chrono::milliseconds timeout(std::numeric_limits<int>::max());
settings->set_connection_timeout(timeout); // there is a timeout, but really huge
settings->set_port(port);
try {
service_->start(settings);
} catch(std::system_error& e) {
std::cerr << "Error running server on port " << port << ": " << e.what() << std::endl;
}
});
listenThread_ = std::thread([this]() {
auto stop = false;
while (!stop) {
auto listener = currentListeners_.begin();
while (listener != currentListeners_.end()) {
if (listener->session->is_closed() && dht_) {
dht_->cancelListen(listener->hash, std::move(listener->token));
// Remove listener if unused
listener = currentListeners_.erase(listener);
} else {
++listener;
}
}
//NOTE: When supports restbed 5.0: service_->is_up() and remove stopListeners
stop = stopListeners;
if (!stop)
std::this_thread::sleep_for(std::chrono::seconds(1));
}
});
}
DhtProxyServer::~DhtProxyServer()
{
stop();
}
void
DhtProxyServer::stop()
{
service_->stop();
stopListeners = true;
// listenThreads_ will stop because there is no more sessions
if (listenThread_.joinable())
listenThread_.join();
if (server_thread.joinable())
server_thread.join();
}
void
DhtProxyServer::getNodeInfo(const std::shared_ptr<restbed::Session>& session) const
{
const auto request = session->get_request();
int content_length = std::stoi(request->get_header("Content-Length", "0"));
session->fetch(content_length,
[=](const std::shared_ptr<restbed::Session> s, const restbed::Bytes& /*b*/)
{
if (dht_) {
Json::Value result;
auto id = dht_->getId();
if (id)
result["id"] = id.toString();
result["node_id"] = dht_->getNodeId().toString();
result["ipv4"] = dht_->getNodesStats(AF_INET).toJson();
result["ipv6"] = dht_->getNodesStats(AF_INET6).toJson();
Json::FastWriter writer;
s->close(restbed::OK, writer.write(result));
}
else
s->close(restbed::SERVICE_UNAVAILABLE, "{\"err\":\"Incorrect DhtRunner\"}");
}
);
}
void
DhtProxyServer::get(const std::shared_ptr<restbed::Session>& session) const
{
const auto request = session->get_request();
int content_length = std::stoi(request->get_header("Content-Length", "0"));
auto hash = request->get_path_parameter("hash");
session->fetch(content_length,
[=](const std::shared_ptr<restbed::Session> s, const restbed::Bytes& /*b* */)
{
if (dht_) {
InfoHash infoHash(hash);
if (!infoHash) {
infoHash = InfoHash::get(hash);
}
s->yield(restbed::OK, "", [=]( const std::shared_ptr< restbed::Session > s) {
dht_->get(infoHash, [s](std::shared_ptr<Value> value) {
// Send values as soon as we get them
Json::FastWriter writer;
s->yield(writer.write(value->toJson()), [](const std::shared_ptr<restbed::Session> /*session*/){ });
return true;
}, [s](bool /*ok* */) {
// Communication is finished
s->close();
});
});
} else {
s->close(restbed::SERVICE_UNAVAILABLE, "{\"err\":\"Incorrect DhtRunner\"}");
}
}
);
}
void
DhtProxyServer::listen(const std::shared_ptr<restbed::Session>& session) const
{
const auto request = session->get_request();
int content_length = std::stoi(request->get_header("Content-Length", "0"));
auto hash = request->get_path_parameter("hash");
InfoHash infoHash(hash);
if (!infoHash)
infoHash = InfoHash::get(hash);
session->fetch(content_length,
[=](const std::shared_ptr<restbed::Session> s, const restbed::Bytes& /*b* */)
{
if (dht_) {
InfoHash infoHash(hash);
if (!infoHash) {
infoHash = InfoHash::get(hash);
}
s->yield(restbed::OK);
// Handle client deconnection
// NOTE: for now, there is no handler, so we test the session in a thread
// will be the case in restbed 5.0
SessionToHashToken listener;
listener.session = session;
listener.hash = infoHash;
listener.token = dht_->listen(infoHash, [s](std::shared_ptr<Value> value) {
// Send values as soon as we get them
if (!s->is_closed()) {
Json::FastWriter writer;
s->yield(writer.write(value->toJson()), [](const std::shared_ptr<restbed::Session> /*session*/){ });
}
return !s->is_closed();
});
currentListeners_.emplace_back(std::move(listener));
} else {
session->close(restbed::SERVICE_UNAVAILABLE, "{\"err\":\"Incorrect DhtRunner\"}");
}
}
);
}
void
DhtProxyServer::put(const std::shared_ptr<restbed::Session>& session) const
{
const auto request = session->get_request();
int content_length = std::stoi(request->get_header("Content-Length", "0"));
auto hash = request->get_path_parameter("hash");
InfoHash infoHash(hash);
if (!infoHash)
infoHash = InfoHash::get(hash);
session->fetch(content_length,
[=](const std::shared_ptr<restbed::Session> s, const restbed::Bytes& b)
{
if (dht_) {
if(b.empty()) {
std::string response("{\"err\":\"Missing parameters\"}");
s->close(restbed::BAD_REQUEST, response);
} else {
restbed::Bytes buf(b);
Json::Value root;
Json::Reader reader;
std::string strJson(buf.begin(), buf.end());
bool parsingSuccessful = reader.parse(strJson.c_str(), root);
if (parsingSuccessful) {
// Build the Value from json
auto value = std::make_shared<Value>(root);
dht_->put(infoHash, value, [s, value](bool ok) {
if (ok) {
Json::FastWriter writer;
s->close(restbed::OK, writer.write(value->toJson()));
} else {
s->close(restbed::BAD_GATEWAY, "{\"err\":\"put failed\"}");
}
});
} else {
s->close(restbed::BAD_REQUEST, "{\"err\":\"Incorrect JSON\"}");
}
}
} else {
s->close(restbed::SERVICE_UNAVAILABLE, "{\"err\":\"Incorrect DhtRunner\"}");
}
}
);
}
#if OPENDHT_PROXY_SERVER_IDENTITY
void
DhtProxyServer::putSigned(const std::shared_ptr<restbed::Session>& session) const
{
const auto request = session->get_request();
int content_length = std::stoi(request->get_header("Content-Length", "0"));
auto hash = request->get_path_parameter("hash");
InfoHash infoHash(hash);
if (!infoHash)
infoHash = InfoHash::get(hash);
session->fetch(content_length,
[=](const std::shared_ptr<restbed::Session> s, const restbed::Bytes& b)
{
if (dht_) {
if(b.empty()) {
std::string response("{\"err\":\"Missing parameters\"}");
s->close(restbed::BAD_REQUEST, response);
} else {
restbed::Bytes buf(b);
Json::Value root;
Json::Reader reader;
std::string strJson(buf.begin(), buf.end());
bool parsingSuccessful = reader.parse(strJson.c_str(), root);
if (parsingSuccessful) {
auto value = std::make_shared<Value>(root);
Json::FastWriter writer;
dht_->putSigned(infoHash, value);
s->close(restbed::OK, writer.write(value->toJson()));
} else {
s->close(restbed::BAD_REQUEST, "{\"err\":\"Incorrect JSON\"}");
}
}
} else {
s->close(restbed::SERVICE_UNAVAILABLE, "{\"err\":\"Incorrect DhtRunner\"}");
}
}
);
}
void
DhtProxyServer::putEncrypted(const std::shared_ptr<restbed::Session>& session) const
{
const auto request = session->get_request();
int content_length = std::stoi(request->get_header("Content-Length", "0"));
auto hash = request->get_path_parameter("hash");
InfoHash key(hash);
if (!key)
key = InfoHash::get(hash);
session->fetch(content_length,
[=](const std::shared_ptr<restbed::Session> s, const restbed::Bytes& b)
{
if (dht_) {
if(b.empty()) {
std::string response("{\"err\":\"Missing parameters\"}");
s->close(restbed::BAD_REQUEST, response);
} else {
restbed::Bytes buf(b);
Json::Value root;
Json::Reader reader;
std::string strJson(buf.begin(), buf.end());
bool parsingSuccessful = reader.parse(strJson.c_str(), root);
InfoHash to(root["to"].asString());
if (parsingSuccessful && to) {
auto value = std::make_shared<Value>(root);
Json::FastWriter writer;
dht_->putEncrypted(key, to, value);
s->close(restbed::OK, writer.write(value->toJson()));
} else {
if(!parsingSuccessful)
s->close(restbed::BAD_REQUEST, "{\"err\":\"Incorrect JSON\"}");
else
s->close(restbed::BAD_REQUEST, "{\"err\":\"No destination found\"}");
}
}
} else {
s->close(restbed::SERVICE_UNAVAILABLE, "{\"err\":\"Incorrect DhtRunner\"}");
}
}
);
}
#endif // OPENDHT_PROXY_SERVER_IDENTITY
void
DhtProxyServer::handleOptionsMethod(const std::shared_ptr<restbed::Session>& session) const
{
#if OPENDHT_PROXY_SERVER_IDENTITY
const auto allowed = "OPTIONS, GET, POST, LISTEN, SIGN, ENCRYPT";
#else
const auto allowed = "OPTIONS, GET, POST, LISTEN";
#endif //OPENDHT_PROXY_SERVER_IDENTITY
session->close(restbed::OK, {{"Access-Control-Allow-Methods", allowed},
{"Access-Control-Allow-Headers", "content-type"},
{"Access-Control-Max-Age", "86400"}});
}
void
DhtProxyServer::getFiltered(const std::shared_ptr<restbed::Session>& session) const
{
const auto request = session->get_request();
int content_length = std::stoi(request->get_header("Content-Length", "0"));
auto hash = request->get_path_parameter("hash");
auto value = request->get_path_parameter("value");
session->fetch(content_length,
[=](const std::shared_ptr<restbed::Session> s, const restbed::Bytes& /*b* */)
{
if (dht_) {
InfoHash infoHash(hash);
if (!infoHash) {
infoHash = InfoHash::get(hash);
}
s->yield(restbed::OK, "", [=]( const std::shared_ptr< restbed::Session > s) {
dht_->get(infoHash, [s](std::shared_ptr<Value> v) {
// Send values as soon as we get them
Json::FastWriter writer;
s->yield(writer.write(v->toJson()), [](const std::shared_ptr<restbed::Session> /*session*/){ });
return true;
}, [s](bool /*ok* */) {
// Communication is finished
s->close();
}, {}, value);
});
} else {
s->close(restbed::SERVICE_UNAVAILABLE, "{\"err\":\"Incorrect DhtRunner\"}");
}
}
);
}
}
#endif //OPENDHT_PROXY_SERVER
+46 -74
View File
@@ -58,28 +58,26 @@ DhtRunner::~DhtRunner()
void
DhtRunner::run(in_port_t port, DhtRunner::Config config)
{
sockaddr_in sin4;
std::fill_n((uint8_t*)&sin4, sizeof(sin4), 0);
sin4.sin_family = AF_INET;
sin4.sin_port = htons(port);
sockaddr_in6 sin6;
std::fill_n((uint8_t*)&sin6, sizeof(sin6), 0);
sin6.sin6_family = AF_INET6;
sin6.sin6_port = htons(port);
run(&sin4, &sin6, config);
SockAddr sin4;
sin4.setFamily(AF_INET);
sin4.setPort(port);
SockAddr sin6;
sin6.setFamily(AF_INET6);
sin6.setPort(port);
run(sin4, sin6, config);
}
void
DhtRunner::run(const char* ip4, const char* ip6, const char* service, DhtRunner::Config config)
{
auto res4 = getAddrInfo(ip4, service);
auto res6 = getAddrInfo(ip6, service);
run(res4.empty() ? nullptr : (sockaddr_in*) &res4.front().first,
res6.empty() ? nullptr : (sockaddr_in6*)&res6.front().first, config);
auto res4 = SockAddr::resolve(ip4, service);
auto res6 = SockAddr::resolve(ip6, service);
run(res4.empty() ? SockAddr() : res4.front(),
res6.empty() ? SockAddr() : res6.front(), config);
}
void
DhtRunner::run(const sockaddr_in* local4, const sockaddr_in6* local6, DhtRunner::Config config)
DhtRunner::run(const SockAddr& local4, const SockAddr& local6, DhtRunner::Config config)
{
if (running)
return;
@@ -375,8 +373,31 @@ DhtRunner::loop_()
return wakeup;
}
int bindSocket(const SockAddr& addr, SockAddr& bound)
{
bool is_ipv6 = addr.getFamily() == AF_INET6;
int sock = socket(is_ipv6 ? PF_INET6 : PF_INET, SOCK_DGRAM, 0);
if (sock < 0)
throw DhtException(std::string("Can't open socket: ") + strerror(sock));
int set = 1;
#ifdef SO_NOSIGPIPE
setsockopt(sock, SOL_SOCKET, SO_NOSIGPIPE, (const char*)&set, sizeof(set));
#endif
if (is_ipv6)
setsockopt(sock, IPPROTO_IPV6, IPV6_V6ONLY, (const char*)&set, sizeof(set));
int rc = bind(sock, addr.get(), addr.getLength());
if(rc < 0)
throw DhtException("Can't bind socket on " + addr.toString() + " " + strerror(rc));
sockaddr_storage ss;
socklen_t ss_len = sizeof(ss);
getsockname(sock, (sockaddr*)&ss, &ss_len);
bound = {ss, ss_len};
return sock;
}
void
DhtRunner::doRun(const sockaddr_in* sin4, const sockaddr_in6* sin6, SecureDht::Config config)
DhtRunner::doRun(const SockAddr& sin4, const SockAddr& sin6, SecureDht::Config config)
{
dht_.reset();
@@ -384,38 +405,13 @@ DhtRunner::doRun(const sockaddr_in* sin4, const sockaddr_in6* sin6, SecureDht::C
s6 = -1;
bound4 = {};
if (sin4) {
s4 = socket(PF_INET, SOCK_DGRAM, 0);
if(s4 >= 0) {
int rc = bind(s4, (sockaddr*)sin4, sizeof(sockaddr_in));
if(rc < 0)
throw DhtException("Can't bind IPv4 socket on " + dht::print_addr((sockaddr*)sin4, sizeof(sockaddr_in)));
sockaddr_storage ss;
socklen_t ss_len = sizeof(ss);
getsockname(s4, (sockaddr*)&ss, &ss_len);
bound4 = {ss, ss_len};
}
}
if (sin4)
s4 = bindSocket(sin4, bound4);
#if 1
bound6 = {};
if (sin6) {
s6 = socket(PF_INET6, SOCK_DGRAM, 0);
if(s6 >= 0) {
int val = 1;
int rc = setsockopt(s6, IPPROTO_IPV6, IPV6_V6ONLY, (char *)&val, sizeof(val));
if(rc < 0)
throw DhtException("Can't set IPV6_V6ONLY");
rc = bind(s6, (sockaddr*)sin6, sizeof(sockaddr_in6));
if(rc < 0)
throw DhtException("Can't bind IPv6 socket on " + dht::print_addr((sockaddr*)sin6, sizeof(sockaddr_in6)));
sockaddr_storage ss;
socklen_t ss_len = sizeof(ss);
getsockname(s6, (sockaddr*)&ss, &ss_len);
bound6 = {ss, ss_len};
}
}
if (sin6)
s6 = bindSocket(sin6, bound6);
#endif
dht_ = std::unique_ptr<SecureDht>(new SecureDht {s4, s6, config});
@@ -490,7 +486,8 @@ DhtRunner::get(const std::string& key, GetCallback vcb, DoneCallbackSimple dcb,
{
get(InfoHash::get(key), vcb, dcb, f, w);
}
void DhtRunner::query(const InfoHash& hash, QueryCallback cb, DoneCallback done_cb, Query q) {
void
DhtRunner::query(const InfoHash& hash, QueryCallback cb, DoneCallback done_cb, Query q) {
{
std::lock_guard<std::mutex> lck(storage_mtx);
pending_ops.emplace([=](SecureDht& dht) mutable {
@@ -664,7 +661,7 @@ DhtRunner::tryBootstrapContinuously()
for (auto it = nodes.rbegin(); it != nodes.rend(); it++) {
++ping_count;
try {
bootstrap(getAddrInfo(it->first, it->second), [&](bool) {
bootstrap(SockAddr::resolve(it->first, it->second), [&](bool) {
if (not running)
return;
{
@@ -695,31 +692,6 @@ DhtRunner::tryBootstrapContinuously()
});
}
std::vector<std::pair<sockaddr_storage, socklen_t>>
DhtRunner::getAddrInfo(const std::string& host, const std::string& service)
{
std::vector<std::pair<sockaddr_storage, socklen_t>> ips {};
if (host.empty())
return ips;
addrinfo hints;
memset(&hints, 0, sizeof(hints));
hints.ai_socktype = SOCK_DGRAM;
addrinfo* info = nullptr;
int rc = getaddrinfo(host.c_str(), service.c_str(), &hints, &info);
if(rc != 0)
throw std::invalid_argument(std::string("Error: `") + host + ":" + service + "`: " + gai_strerror(rc));
addrinfo* infop = info;
while (infop) {
ips.emplace_back(sockaddr_storage(), infop->ai_addrlen);
std::copy_n((uint8_t*)infop->ai_addr, infop->ai_addrlen, (uint8_t*)&ips.back().first);
infop = infop->ai_next;
}
freeaddrinfo(info);
return ips;
}
void
DhtRunner::bootstrap(const std::string& host, const std::string& service)
{
@@ -737,13 +709,13 @@ DhtRunner::clearBootstrap()
}
void
DhtRunner::bootstrap(const std::vector<std::pair<sockaddr_storage, socklen_t>>& nodes, DoneCallbackSimple&& cb)
DhtRunner::bootstrap(const std::vector<SockAddr>& nodes, DoneCallbackSimple&& cb)
{
std::lock_guard<std::mutex> lck(storage_mtx);
pending_ops_prio.emplace([=](SecureDht& dht) mutable {
auto rem = cb ? std::make_shared<std::pair<size_t, bool>>(nodes.size(), false) : nullptr;
for (auto& node : nodes)
dht.pingNode((sockaddr*)&node.first, node.second, cb ? [rem,cb](bool ok) {
for (const auto& node : nodes)
dht.pingNode(node.get(), node.getLength(), cb ? [rem,cb](bool ok) {
auto& r = *rem;
r.first--;
r.second |= ok;
+1 -96
View File
@@ -17,11 +17,6 @@
*/
#include "infohash.h"
#include "rng.h"
extern "C" {
#include <gnutls/gnutls.h>
}
#include <functional>
#include <sstream>
@@ -29,97 +24,7 @@ extern "C" {
namespace dht {
InfoHash::InfoHash(const std::string& hex) {
if (hex.size() < 2*HASH_LEN) {
fill(0);
return;
}
const auto p = (const char*)hex.data();
for (size_t i = 0; i < HASH_LEN; i++) {
unsigned res = 0;
sscanf(p + 2*i, "%02x", &res);
(*this)[i] = res;
}
}
InfoHash
InfoHash::get(const uint8_t* data, size_t data_len)
{
InfoHash h;
size_t s = h.size();
const gnutls_datum_t gnudata = {(uint8_t*)data, (unsigned)data_len};
const gnutls_digest_algorithm_t algo = (HASH_LEN == 64) ? GNUTLS_DIG_SHA512 : (
(HASH_LEN == 32) ? GNUTLS_DIG_SHA256 : (
(HASH_LEN == 20) ? GNUTLS_DIG_SHA1 :
GNUTLS_DIG_NULL ));
static_assert(algo != GNUTLS_DIG_NULL, "Can't find hash function to use.");
int rc = gnutls_fingerprint(algo, &gnudata, h.data(), &s);
if (rc == 0 && s == HASH_LEN)
return h;
throw std::runtime_error(std::string("Error hashing: ") + gnutls_strerror(rc));
}
InfoHash
InfoHash::getRandom()
{
InfoHash h;
crypto::random_device rdev;
std::uniform_int_distribution<uint32_t> rand_int;
auto a = reinterpret_cast<uint32_t*>(h.data());
auto b = reinterpret_cast<uint32_t*>(h.data() + h.size());
std::generate(a, b, std::bind(rand_int, std::ref(rdev)));
return h;
}
struct HexMap : public std::array<std::array<char, 2>, 256> {
HexMap() {
for (size_t i=0; i<size(); i++) {
auto& e = (*this)[i];
e[0] = hex_digits[(i >> 4) & 0x0F];
e[1] = hex_digits[i & 0x0F];
}
}
private:
static constexpr const char* hex_digits = "0123456789abcdef";
};
const char*
InfoHash::to_c_str() const
{
static const HexMap map;
thread_local std::array<char, HASH_LEN*2+1> buf;
for (size_t i=0; i<HASH_LEN; i++) {
auto b = buf.data()+i*2;
const auto& m = map[(*this)[i]];
*((uint16_t*)b) = *((uint16_t*)&m);
}
return buf.data();
}
std::string
InfoHash::toString() const
{
return std::string(to_c_str(), HASH_LEN*2);
}
std::ostream& operator<< (std::ostream& s, const InfoHash& h)
{
s.write(h.to_c_str(), HASH_LEN*2);
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;
}
const HexMap hex_map = {};
void
NodeExport::msgpack_unpack(msgpack::object o)
+38
View File
@@ -0,0 +1,38 @@
/*
* Copyright (C) 2017 Savoir-faire Linux Inc.
* Author(s) : Adrien Béraud <adrien.beraud@savoirfairelinux.com>
*
* This program 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.
*
* You should have received a copy of the GNU General Public License
* along with this program. If not, see <https://www.gnu.org/licenses/>.
*/
#pragma once
namespace dht {
namespace net {
enum class MessageType {
Error = 0,
Reply,
Ping,
FindNode,
GetValues,
AnnounceValue,
Refresh,
Listen,
ValueData,
ValueUpdate
};
} /* namespace net */
} /* dht */
+82 -54
View File
@@ -53,15 +53,7 @@ constexpr std::chrono::seconds NetworkEngine::RX_MAX_PACKET_TIME;
constexpr std::chrono::seconds NetworkEngine::RX_TIMEOUT;
const std::string NetworkEngine::my_v {"RNG1"};
const constexpr uint16_t TransId::INVALID;
const TransPrefix TransPrefix::PING = {"pn"};
const TransPrefix TransPrefix::FIND_NODE = {"fn"};
const TransPrefix TransPrefix::GET_VALUES = {"gt"};
const TransPrefix TransPrefix::ANNOUNCE_VALUES = {"pt"};
const TransPrefix TransPrefix::REFRESH = {"rf"};
const TransPrefix TransPrefix::LISTEN = {"lt"};
constexpr long unsigned NetworkEngine::MAX_REQUESTS_PER_SEC;
constexpr size_t NetworkEngine::MAX_REQUESTS_PER_SEC;
static const uint8_t v4prefix[16] = {
0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0xFF, 0xFF, 0, 0, 0, 0
@@ -95,6 +87,24 @@ set_nonblocking(int fd, int nonblocking)
#endif
/* Transaction-ids are 4-bytes long, with the first two bytes identifying
* the kind of request, and the remaining two a sequence number in
* host order.
*/
struct TransId final : public std::array<uint8_t, 4> {
TransId() { std::fill(begin(), end(), 0); }
TransId(const std::array<char, 4>& o) { std::copy(o.begin(), o.end(), begin()); }
TransId(uint32_t id) {
*reinterpret_cast<uint32_t*>(data()) = htonl(id);
}
uint32_t toInt() const {
return ntohl(*reinterpret_cast<const uint32_t*>(&(*this)[0]));
}
};
struct NetworkEngine::PartialMessage {
SockAddr from;
time_point start;
@@ -146,14 +156,14 @@ NetworkEngine::~NetworkEngine() {
}
void
NetworkEngine::tellListener(Sp<Node> node, uint32_t socket_id, const InfoHash& hash, want_t want,
NetworkEngine::tellListener(Sp<Node> node, Tid socket_id, const InfoHash& hash, want_t want,
const Blob& ntoken, std::vector<Sp<Node>>&& nodes,
std::vector<Sp<Node>>&& nodes6, std::vector<Sp<Value>>&& values,
const Query& query)
{
auto nnodes = bufferNodes(node->getFamily(), hash, want, nodes, nodes6);
try {
sendNodesValues(node->getAddr(), TransId(socket_id), nnodes.first, nnodes.second, values, query, ntoken);
sendNodesValues(node->getAddr(), socket_id, nnodes.first, nnodes.second, values, query, ntoken);
} catch (const std::overflow_error& e) {
DHT_LOG.e("Can't send value: buffer not large enough !");
}
@@ -196,7 +206,7 @@ NetworkEngine::requestStep(Sp<Request> sreq)
if (req.isExpired(now)) {
DHT_LOG.e(node.id, "[node %s] expired !", node.toString().c_str());
node.setExpired();
if (node.id == zeroes)
if (not node.id)
requests.erase(req.tid);
return;
} else if (req.attempt_count == 1) {
@@ -208,10 +218,11 @@ NetworkEngine::requestStep(Sp<Request> sreq)
node.getAddr());
if (err == ENETUNREACH ||
err == EHOSTUNREACH ||
err == EAFNOSUPPORT)
err == EAFNOSUPPORT ||
err == EPIPE)
{
node.setExpired();
if (node.id == zeroes)
if (not node.id)
requests.erase(req.tid);
} else {
if (err != EAGAIN) {
@@ -234,7 +245,7 @@ void
NetworkEngine::sendRequest(const Sp<Request>& request)
{
auto& node = request->node;
if (node->id == zeroes)
if (not node->id)
requests.emplace(request->tid, request);
request->start = scheduler.time();
node->requested(request);
@@ -363,7 +374,7 @@ NetworkEngine::processMessage(const uint8_t *buf, size_t buflen, const SockAddr&
return;
}
if (msg->id == myid || msg->id == zeroes) {
if (msg->id == myid or not msg->id) {
DHT_LOG.d("Received message from self");
return;
}
@@ -400,14 +411,23 @@ NetworkEngine::process(std::unique_ptr<ParsedMessage>&& msg, const SockAddr& fro
const auto& now = scheduler.time();
auto node = cache.getNode(msg->id, from, now, true, msg->is_client);
if (msg->type == MessageType::Error or msg->type == MessageType::Reply) {
auto rsocket = node->getSocket(msg->tid.toInt());
if (msg->type == MessageType::ValueUpdate) {
auto rsocket = node->getSocket(msg->tid);
if (not rsocket)
throw DhtProtocolException {DhtProtocolException::UNKNOWN_TID, "Can't find socket", msg->id};
node->received(now, {});
onNewNode(node, 2);
deserializeNodes(*msg, from);
rsocket->on_receive(node, std::move(*msg));
}
else if (msg->type == MessageType::Error or msg->type == MessageType::Reply) {
auto rsocket = node->getSocket(msg->tid);
auto req = node->getRequest(msg->tid);
/* either response for a request or data for an opened socket */
if (not req and not rsocket) {
auto req_it = requests.find(msg->tid);
if (req_it != requests.end() and req_it->second->node->id == zeroes) {
if (req_it != requests.end() and not req_it->second->node->id) {
req = req_it->second;
req->node = node;
requests.erase(req_it);
@@ -432,10 +452,10 @@ NetworkEngine::process(std::unique_ptr<ParsedMessage>&& msg, const SockAddr& fro
switch (msg->type) {
case MessageType::Error: {
if (msg->id != zeroes and (
(msg->error_code == DhtProtocolException::NOT_FOUND and msg->tid.matches(TransPrefix::REFRESH)) or
(msg->error_code == DhtProtocolException::UNAUTHORIZED and (msg->tid.matches(TransPrefix::ANNOUNCE_VALUES)
or msg->tid.matches(TransPrefix::LISTEN)))))
if (msg->id and req and (
(msg->error_code == DhtProtocolException::NOT_FOUND and req->getType() == MessageType::Refresh) or
(msg->error_code == DhtProtocolException::UNAUTHORIZED and (req->getType() == MessageType::AnnounceValue
or req->getType() == MessageType::Listen))))
{
req->last_try = time_point::min();
req->reply_time = time_point::min();
@@ -513,7 +533,7 @@ NetworkEngine::process(std::unique_ptr<ParsedMessage>&& msg, const SockAddr& fro
case MessageType::Listen: {
DHT_LOG.d(msg->info_hash, node->id, "[node %s] got 'listen' request for %s", node->toString().c_str(), msg->info_hash.toString().c_str());
++in_stats.listen;
RequestAnswer answer = onListen(node, msg->info_hash, msg->token, msg->socket_id.toInt(), std::move(msg->query));
RequestAnswer answer = onListen(node, msg->info_hash, msg->token, msg->socket_id, std::move(msg->query));
sendListenConfirmation(from, msg->tid);
break;
}
@@ -563,6 +583,9 @@ NetworkEngine::send(const char *buf, size_t len, int flags, const SockAddr& addr
if (s < 0)
return EAFNOSUPPORT;
#ifdef MSG_NOSIGNAL
flags &= MSG_NOSIGNAL;
#endif
if (sendto(s, buf, len, flags, addr.get(), addr.getLength()) == -1) {
int err = errno;
DHT_LOG.e("Can't send message to %s: %s", addr.toString().c_str(), strerror(err));
@@ -573,7 +596,7 @@ NetworkEngine::send(const char *buf, size_t len, int flags, const SockAddr& addr
Sp<Request>
NetworkEngine::sendPing(Sp<Node> node, RequestCb&& on_done, RequestExpiredCb&& on_expired) {
auto tid = TransId {TransPrefix::PING, node->getNewTid()};
TransId tid (node->getNewTid());
msgpack::sbuffer buffer;
msgpack::packer<msgpack::sbuffer> pk(&buffer);
pk.pack_map(5+(network?1:0));
@@ -590,7 +613,7 @@ NetworkEngine::sendPing(Sp<Node> node, RequestCb&& on_done, RequestExpiredCb&& o
pk.pack(std::string("n")); pk.pack(network);
}
auto req = std::make_shared<Request>(tid, node,
auto req = std::make_shared<Request>(MessageType::Ping, tid.toInt(), node,
Blob(buffer.data(), buffer.data() + buffer.size()),
[=](const Request& req_status, ParsedMessage&&) {
DHT_LOG.d(req_status.node->id, "[node %s] got pong !", req_status.node->toString().c_str());
@@ -610,7 +633,7 @@ NetworkEngine::sendPing(Sp<Node> node, RequestCb&& on_done, RequestExpiredCb&& o
}
void
NetworkEngine::sendPong(const SockAddr& addr, TransId tid) {
NetworkEngine::sendPong(const SockAddr& addr, Tid tid) {
msgpack::sbuffer buffer;
msgpack::packer<msgpack::sbuffer> pk(&buffer);
pk.pack_map(4+(network?1:0));
@@ -619,8 +642,9 @@ NetworkEngine::sendPong(const SockAddr& addr, TransId tid) {
pk.pack(std::string("id")); pk.pack(myid);
insertAddr(pk, addr);
pk.pack(std::string("t")); pk.pack_bin(tid.size());
pk.pack_bin_body((const char*)tid.data(), tid.size());
TransId t (tid);
pk.pack(std::string("t")); pk.pack_bin(t.size());
pk.pack_bin_body((const char*)t.data(), t.size());
pk.pack(std::string("y")); pk.pack(std::string("r"));
pk.pack(std::string("v")); pk.pack(my_v);
if (network) {
@@ -633,7 +657,7 @@ NetworkEngine::sendPong(const SockAddr& addr, TransId tid) {
Sp<Request>
NetworkEngine::sendFindNode(Sp<Node> n, const InfoHash& target, want_t want,
RequestCb&& on_done, RequestExpiredCb&& on_expired) {
auto tid = TransId {TransPrefix::FIND_NODE, n->getNewTid()};
TransId tid (n->getNewTid());
msgpack::sbuffer buffer;
msgpack::packer<msgpack::sbuffer> pk(&buffer);
pk.pack_map(5+(network?1:0));
@@ -657,7 +681,7 @@ NetworkEngine::sendFindNode(Sp<Node> n, const InfoHash& target, want_t want,
pk.pack(std::string("n")); pk.pack(network);
}
auto req = std::make_shared<Request>(tid, n,
auto req = std::make_shared<Request>(MessageType::FindNode, tid.toInt(), n,
Blob(buffer.data(), buffer.data() + buffer.size()),
[=](const Request& req_status, ParsedMessage&& msg) { /* on done */
if (on_done) {
@@ -679,7 +703,7 @@ NetworkEngine::sendFindNode(Sp<Node> n, const InfoHash& target, want_t want,
Sp<Request>
NetworkEngine::sendGetValues(Sp<Node> n, const InfoHash& info_hash, const Query& query, want_t want,
RequestCb&& on_done, RequestExpiredCb&& on_expired) {
auto tid = TransId {TransPrefix::GET_VALUES, n->getNewTid()};
TransId tid (n->getNewTid());
msgpack::sbuffer buffer;
msgpack::packer<msgpack::sbuffer> pk(&buffer);
pk.pack_map(5+(network?1:0));
@@ -706,7 +730,7 @@ NetworkEngine::sendGetValues(Sp<Node> n, const InfoHash& info_hash, const Query&
pk.pack(std::string("n")); pk.pack(network);
}
auto req = std::make_shared<Request>(tid, n,
auto req = std::make_shared<Request>(MessageType::GetValues, tid.toInt(), n,
Blob(buffer.data(), buffer.data() + buffer.size()),
[=](const Request& req_status, ParsedMessage&& msg) { /* on done */
if (on_done) {
@@ -807,7 +831,7 @@ NetworkEngine::packValueHeader(msgpack::sbuffer& buffer, const std::vector<Sp<Va
}
void
NetworkEngine::sendValueParts(TransId tid, const std::vector<Blob>& svals, const SockAddr& addr)
NetworkEngine::sendValueParts(const TransId& tid, const std::vector<Blob>& svals, const SockAddr& addr)
{
msgpack::sbuffer buffer;
unsigned i=0;
@@ -837,7 +861,7 @@ NetworkEngine::sendValueParts(TransId tid, const std::vector<Blob>& svals, const
}
void
NetworkEngine::sendNodesValues(const SockAddr& addr, TransId tid, const Blob& nodes, const Blob& nodes6,
NetworkEngine::sendNodesValues(const SockAddr& addr, Tid tid, const Blob& nodes, const Blob& nodes6,
const std::vector<Sp<Value>>& st, const Query& query, const Blob& token)
{
msgpack::sbuffer buffer;
@@ -878,8 +902,9 @@ NetworkEngine::sendNodesValues(const SockAddr& addr, TransId tid, const Blob& no
}
}
pk.pack(std::string("t")); pk.pack_bin(tid.size());
pk.pack_bin_body((const char*)tid.data(), tid.size());
TransId t (tid);
pk.pack(std::string("t")); pk.pack_bin(t.size());
pk.pack_bin_body((const char*)t.data(), t.size());
pk.pack(std::string("y")); pk.pack(std::string("r"));
pk.pack(std::string("v")); pk.pack(my_v);
if (network) {
@@ -954,8 +979,8 @@ NetworkEngine::sendListen(Sp<Node> n,
RequestExpiredCb&& on_expired,
SocketCb&& socket_cb)
{
uint32_t socket;
auto tid = TransId { TransPrefix::LISTEN, n->getNewTid() };
Tid socket;
TransId tid (n->getNewTid());
if (previous and previous->node == n) {
socket = previous->getSocket();
} else {
@@ -994,7 +1019,7 @@ NetworkEngine::sendListen(Sp<Node> n,
pk.pack(std::string("n")); pk.pack(network);
}
auto req = std::make_shared<Request>(tid, n,
auto req = std::make_shared<Request>(MessageType::Listen, tid.toInt(), n,
Blob(buffer.data(), buffer.data() + buffer.size()),
[=](const Request& req_status, ParsedMessage&& msg) { /* on done */
if (on_done)
@@ -1012,7 +1037,7 @@ NetworkEngine::sendListen(Sp<Node> n,
}
void
NetworkEngine::sendListenConfirmation(const SockAddr& addr, TransId tid) {
NetworkEngine::sendListenConfirmation(const SockAddr& addr, Tid tid) {
msgpack::sbuffer buffer;
msgpack::packer<msgpack::sbuffer> pk(&buffer);
pk.pack_map(4+(network?1:0));
@@ -1021,8 +1046,9 @@ NetworkEngine::sendListenConfirmation(const SockAddr& addr, TransId tid) {
pk.pack(std::string("id")); pk.pack(myid);
insertAddr(pk, addr);
pk.pack(std::string("t")); pk.pack_bin(tid.size());
pk.pack_bin_body((const char*)tid.data(), tid.size());
TransId t (tid);
pk.pack(std::string("t")); pk.pack_bin(t.size());
pk.pack_bin_body((const char*)t.data(), t.size());
pk.pack(std::string("y")); pk.pack(std::string("r"));
pk.pack(std::string("v")); pk.pack(my_v);
if (network) {
@@ -1041,7 +1067,7 @@ NetworkEngine::sendAnnounceValue(Sp<Node> n,
RequestCb&& on_done,
RequestExpiredCb&& on_expired)
{
auto tid = TransId {TransPrefix::ANNOUNCE_VALUES, n->getNewTid()};
TransId tid (n->getNewTid());
msgpack::sbuffer buffer;
msgpack::packer<msgpack::sbuffer> pk(&buffer);
pk.pack_map(5+(network?1:0));
@@ -1065,7 +1091,7 @@ NetworkEngine::sendAnnounceValue(Sp<Node> n,
pk.pack(std::string("n")); pk.pack(network);
}
auto req = std::make_shared<Request>(tid, n,
auto req = std::make_shared<Request>(MessageType::AnnounceValue, tid.toInt(), n,
Blob(buffer.data(), buffer.data() + buffer.size()),
[=](const Request& req_status, ParsedMessage&& msg) { /* on done */
if (msg.value_id == Value::INVALID_ID) {
@@ -1099,7 +1125,7 @@ NetworkEngine::sendRefreshValue(Sp<Node> n,
RequestCb&& on_done,
RequestExpiredCb&& on_expired)
{
auto tid = TransId {TransPrefix::REFRESH, n->getNewTid()};
TransId tid (n->getNewTid());
msgpack::sbuffer buffer;
msgpack::packer<msgpack::sbuffer> pk(&buffer);
pk.pack_map(5+(network?1:0));
@@ -1119,7 +1145,7 @@ NetworkEngine::sendRefreshValue(Sp<Node> n,
pk.pack(std::string("n")); pk.pack(network);
}
auto req = std::make_shared<Request>(tid, n,
auto req = std::make_shared<Request>(MessageType::Refresh, tid.toInt(), n,
Blob(buffer.data(), buffer.data() + buffer.size()),
[=](const Request& req_status, ParsedMessage&& msg) { /* on done */
if (msg.value_id == Value::INVALID_ID) {
@@ -1144,7 +1170,7 @@ NetworkEngine::sendRefreshValue(Sp<Node> n,
}
void
NetworkEngine::sendValueAnnounced(const SockAddr& addr, TransId tid, Value::Id vid) {
NetworkEngine::sendValueAnnounced(const SockAddr& addr, Tid tid, Value::Id vid) {
msgpack::sbuffer buffer;
msgpack::packer<msgpack::sbuffer> pk(&buffer);
pk.pack_map(4+(network?1:0));
@@ -1154,8 +1180,9 @@ NetworkEngine::sendValueAnnounced(const SockAddr& addr, TransId tid, Value::Id v
pk.pack(std::string("vid")); pk.pack(vid);
insertAddr(pk, addr);
pk.pack(std::string("t")); pk.pack_bin(tid.size());
pk.pack_bin_body((const char*)tid.data(), tid.size());
TransId t(tid);
pk.pack(std::string("t")); pk.pack_bin(t.size());
pk.pack_bin_body((const char*)t.data(), t.size());
pk.pack(std::string("y")); pk.pack(std::string("r"));
pk.pack(std::string("v")); pk.pack(my_v);
if (network) {
@@ -1167,7 +1194,7 @@ NetworkEngine::sendValueAnnounced(const SockAddr& addr, TransId tid, Value::Id v
void
NetworkEngine::sendError(const SockAddr& addr,
TransId tid,
Tid tid,
uint16_t code,
const std::string& message,
bool include_id)
@@ -1185,8 +1212,9 @@ NetworkEngine::sendError(const SockAddr& addr,
pk.pack(std::string("id")); pk.pack(myid);
}
pk.pack(std::string("t")); pk.pack_bin(tid.size());
pk.pack_bin_body((const char*)tid.data(), tid.size());
TransId t(tid);
pk.pack(std::string("t")); pk.pack_bin(t.size());
pk.pack_bin_body((const char*)t.data(), t.size());
pk.pack(std::string("y")); pk.pack(std::string("e"));
pk.pack(std::string("v")); pk.pack(my_v);
if (network) {
@@ -1197,7 +1225,7 @@ NetworkEngine::sendError(const SockAddr& addr,
}
void
NetworkEngine::maintainRxBuffer(const TransId& tid)
NetworkEngine::maintainRxBuffer(Tid tid)
{
auto msg = partial_messages.find(tid);
if (msg != partial_messages.end()) {
+5 -5
View File
@@ -35,7 +35,7 @@ Node::Node(const InfoHash& id, const SockAddr& addr, bool client)
: id(id), addr(addr), is_client(client), sockets_()
{
crypto::random_device rd;
transaction_id = std::uniform_int_distribution<SocketId>{1}(rd);
transaction_id = std::uniform_int_distribution<Tid>{1}(rd);
}
/* This is our definition of a known-good node. */
@@ -101,7 +101,7 @@ Node::received(time_point now, const Sp<net::Request>& req)
}
Sp<net::Request>
Node::getRequest(const net::TransId& tid)
Node::getRequest(Tid tid)
{
auto it = requests_.find(tid);
return it != requests_.end() ? it->second : nullptr;
@@ -128,7 +128,7 @@ Node::setExpired()
sockets_.clear();
}
SocketId
Tid
Node::openSocket(SocketCb&& cb)
{
if (++transaction_id == 0)
@@ -142,14 +142,14 @@ Node::openSocket(SocketCb&& cb)
}
Socket*
Node::getSocket(SocketId id)
Node::getSocket(Tid id)
{
auto it = sockets_.find(id);
return it == sockets_.end() ? nullptr : &it->second;
}
void
Node::closeSocket(SocketId id)
Node::closeSocket(Tid id)
{
if (id) {
sockets_.erase(id);
+15 -1
View File
@@ -20,6 +20,12 @@
namespace dht {
NodeCache::~NodeCache()
{
cache_4.setExpired();
cache_6.setExpired();
}
Sp<Node>
NodeCache::getNode(const InfoHash& id, sa_family_t family) {
return cache(family).getNode(id);
@@ -27,7 +33,7 @@ NodeCache::getNode(const InfoHash& id, sa_family_t family) {
Sp<Node>
NodeCache::getNode(const InfoHash& id, const SockAddr& addr, time_point now, bool confirm, bool client) {
if (id == zeroes)
if (not id)
return std::make_shared<Node>(id, addr);
return cache(addr.getFamily()).getNode(id, addr, now, confirm, client);
}
@@ -116,4 +122,12 @@ NodeCache::NodeMap::clearBadNodes() {
}
}
void
NodeCache::NodeMap::setExpired() {
for (auto& wn : *this)
if (auto n = wn.second.lock())
n->setExpired();
clear();
}
}
+15 -17
View File
@@ -26,17 +26,14 @@
namespace dht {
namespace net {
enum class MessageType {
Error = 0,
Reply,
Ping,
FindNode,
GetValues,
AnnounceValue,
Refresh,
Listen,
ValueData
};
Tid unpackTid(const msgpack::object& o) {
switch (o.type) {
case msgpack::type::POSITIVE_INTEGER:
return o.as<Tid>();
default:
return ntohl(*reinterpret_cast<const uint32_t*>(o.as<std::array<char, 4>>().data()));
}
}
struct ParsedMessage {
MessageType type;
@@ -51,9 +48,9 @@ struct ParsedMessage {
/* target id around which to find nodes */
InfoHash target;
/* transaction id */
TransId tid;
Tid tid;
/* tid for packets going through request socket */
TransId socket_id;
Tid socket_id;
/* security token */
Blob token;
/* the value id (announce confirmation) */
@@ -128,6 +125,7 @@ ParsedMessage::msgpack_unpack(msgpack::object msg)
{
auto y = findMapValue(msg, "y");
auto r = findMapValue(msg, "r");
auto u = findMapValue(msg, "u");
auto e = findMapValue(msg, "e");
auto v = findMapValue(msg, "p");
@@ -156,6 +154,8 @@ ParsedMessage::msgpack_unpack(msgpack::object msg)
type = MessageType::Reply;
else if (v)
type = MessageType::ValueData;
else if (u)
type = MessageType::ValueUpdate;
else if (y and y->as<std::string>() != "q")
throw msgpack::type_error();
else if (q == "ping")
@@ -188,9 +188,9 @@ ParsedMessage::msgpack_unpack(msgpack::object msg)
}
auto a = findMapValue(msg, "a");
if (!a && !r && !e)
if (!a && !r && !e && !u)
throw msgpack::type_error();
auto& req = a ? *a : (r ? *r : *e);
auto& req = a ? *a : (r ? *r : (u ? *u : *e));
if (e) {
if (e->type != msgpack::type::ARRAY)
@@ -232,13 +232,11 @@ ParsedMessage::msgpack_unpack(msgpack::object msg)
if (l == sizeof(in_addr)) {
addr.setFamily(AF_INET);
auto& a = addr.getIPv4();
std::fill_n((uint8_t*)&a, sizeof(a), 0);
a.sin_port = 0;
std::copy_n(sa->via.bin.ptr, l, (char*)&a.sin_addr);
} else if (l == sizeof(in6_addr)) {
addr.setFamily(AF_INET6);
auto& a = addr.getIPv6();
std::fill_n((uint8_t*)&a, sizeof(a), 0);
a.sin6_port = 0;
std::copy_n(sa->via.bin.ptr, l, (char*)&a.sin6_addr);
}
+10 -8
View File
@@ -59,18 +59,19 @@ struct Request {
State getState() const { return state_; }
Request(State state = State::PENDING) : state_(state) {}
Request(const TransId& tid,
Request(MessageType type, Tid tid,
Sp<Node> node,
Blob&& msg,
std::function<void(const Request&, ParsedMessage&&)> on_done,
std::function<void(const Request&, bool)> on_expired,
uint32_t socket = 0) :
node(node), on_done(on_done), on_expired(on_expired), tid(tid), msg(std::move(msg)), socket(socket) { }
Tid socket = 0) :
node(node), tid(tid), type(type), on_done(on_done), on_expired(on_expired), msg(std::move(msg)), socket(socket) { }
TransId getTid() const { return tid; }
Tid getTid() const { return tid; }
MessageType getType() const { return type; }
uint32_t getSocket() { return socket; }
uint32_t closeSocket() { auto ret = socket; socket = 0; return ret; }
Tid getSocket() { return socket; }
Tid closeSocket() { auto ret = socket; socket = 0; return ret; }
void setExpired() {
if (pending()) {
@@ -107,6 +108,8 @@ private:
msg.clear();
}
const Tid tid {0}; /* the request id. */
const MessageType type {};
State state_ {State::PENDING};
unsigned attempt_count {0}; /* number of attempt to process the request. */
@@ -116,9 +119,8 @@ private:
std::function<void(const Request&, ParsedMessage&&)> on_done {};
std::function<void(const Request&, bool)> on_expired {};
const TransId tid; /* the request id. */
Blob msg {}; /* the serialized message. */
uint32_t socket; /* the socket used for further reponses. */
Tid socket; /* the socket used for further reponses. */
};
} /* namespace net */
+1 -1
View File
@@ -76,7 +76,7 @@ RoutingTable::randomId(const RoutingTable::const_iterator& it) const
int b = bit/8;
InfoHash id_return;
std::copy_n(it->first.begin(), b, id_return.begin());
std::copy_n(it->first.cbegin(), b, id_return.begin());
id_return[b] = it->first[b] & (0xFF00 >> (bit % 8));
id_return[b] |= rand_byte(rd) >> (bit % 8);
for (unsigned i = b + 1; i < HASH_LEN; i++)
+1 -1
View File
@@ -35,7 +35,7 @@ namespace dht {
Config& getConfig(SecureDht::Config& conf)
{
auto& c = conf.node_config;
if (c.node_id == InfoHash() and conf.id.second)
if (not c.node_id and conf.id.second)
c.node_id = InfoHash::get("node:"+conf.id.second->getId().toString());
return c;
}
+32 -19
View File
@@ -19,7 +19,6 @@
#include "utils.h"
#include "sockaddr.h"
#include "default_types.h"
#include "net.h"
/* An IPv4 equivalent to IN6_IS_ADDR_UNSPECIFIED */
#ifndef IN_IS_ADDR_UNSPECIFIED
@@ -28,6 +27,31 @@
namespace dht {
std::vector<SockAddr>
SockAddr::resolve(const std::string& host, const std::string& service)
{
std::vector<SockAddr> ips {};
if (host.empty())
return ips;
addrinfo hints;
memset(&hints, 0, sizeof(hints));
hints.ai_socktype = SOCK_DGRAM;
addrinfo* info = nullptr;
int rc = getaddrinfo(host.c_str(), service.empty() ? nullptr : service.c_str(), &hints, &info);
if(rc != 0)
throw std::invalid_argument(std::string("Error: `") + host + ":" + service + "`: " + gai_strerror(rc));
addrinfo* infop = info;
while (infop) {
ips.emplace_back(infop->ai_addr, infop->ai_addrlen);
infop = infop->ai_next;
}
freeaddrinfo(info);
return ips;
}
std::string
print_addr(const sockaddr* sa, socklen_t slen)
{
@@ -70,7 +94,8 @@ SockAddr::isLoopback() const
{
switch (getFamily()) {
case AF_INET: {
uint8_t b1 = (uint8_t)(getIPv4().sin_addr.s_addr >> 24);
auto addr_host = ntohl(getIPv4().sin_addr.s_addr);
uint8_t b1 = (uint8_t)(addr_host >> 24);
return b1 == 127;
}
case AF_INET6:
@@ -87,10 +112,11 @@ SockAddr::isPrivate() const
return true;
}
switch (getFamily()) {
case AF_INET:
case AF_INET: {
auto addr_host = ntohl(getIPv4().sin_addr.s_addr);
uint8_t b1, b2;
b1 = (uint8_t)(getIPv4().sin_addr.s_addr >> 24);
b2 = (uint8_t)((getIPv4().sin_addr.s_addr >> 16) & 0x0ff);
b1 = (uint8_t)(addr_host >> 24);
b2 = (uint8_t)((addr_host >> 16) & 0x0ff);
// 10.x.y.z
if (b1 == 10)
return true;
@@ -101,6 +127,7 @@ SockAddr::isPrivate() const
if ((b1 == 192) && (b2 == 168))
return true;
return false;
}
case AF_INET6: {
const uint8_t* addr6 = reinterpret_cast<const uint8_t*>(&getIPv6().sin6_addr);
if (addr6[0] == 0xfc)
@@ -161,18 +188,4 @@ findMapValue(msgpack::object& map, const std::string& key) {
return nullptr;
}
namespace net {
TransId
unpackTid(msgpack::object& o) {
switch (o.type) {
case msgpack::type::POSITIVE_INTEGER:
return o.as<uint32_t>();
default:
return o.as<std::array<char, 4>>();
}
}
}
}
+71 -1
View File
@@ -22,6 +22,11 @@
#include "default_types.h"
#include "securedht.h" // print certificate ID
#if OPENDHT_PROXY_SERVER
#include "base64.h"
#endif //OPENDHT_PROXY_SERVER
namespace dht {
const std::string Query::QUERY_PARSE_ERROR {"Error parsing query."};
@@ -41,7 +46,7 @@ std::ostream& operator<< (std::ostream& s, const Value& v)
s << "encrypted ";
else if (v.isSigned()) {
s << "signed (v" << v.seq << ") ";
if (v.recipient != InfoHash())
if (v.recipient)
s << "decrypted ";
}
if (not v.isEncrypted()) {
@@ -167,6 +172,71 @@ Value::msgpack_unpack_body(const msgpack::object& o)
}
}
#if OPENDHT_PROXY_SERVER
Value::Value(Json::Value& json)
{
try {
if (json.isMember("id"))
id = ValueType::Id(json["id"].asInt());
} catch (...) { }
if (json.isMember("cypher")) {
auto cypherStr = json["cypher"].asString();
cypherStr = base64_decode(cypherStr);
cypher = std::vector<unsigned char>(cypherStr.begin(), cypherStr.end());
}
if (json.isMember("sig")) {
auto sigStr = json["sig"].asString();
sigStr = base64_decode(sigStr);
signature = std::vector<unsigned char>(sigStr.begin(), sigStr.end());
}
if (json.isMember("seq"))
seq = json["seq"].asInt();
if (json.isMember("owner")) {
auto ownerStr = json["owner"].asString();
auto ownerBlob = std::vector<unsigned char>(ownerStr.begin(), ownerStr.end());
owner = std::make_shared<const crypto::PublicKey>(ownerBlob);
}
if (json.isMember("to")) {
auto toStr = json["to"].asString();
recipient = InfoHash(toStr);
}
if (json.isMember("type"))
type = json["type"].asInt();
if (json.isMember("data")){
auto dataStr = json["data"].asString();
dataStr = base64_decode(dataStr);
data = std::vector<unsigned char>(dataStr.begin(), dataStr.end());
}
if (json.isMember("utype"))
user_type = json["utype"].asString();
}
Json::Value
Value::toJson() const
{
Json::Value val;
val["id"] = std::to_string(id);
if (isEncrypted()) {
val["cypher"] = base64_encode(cypher);
} else {
if (isSigned())
val["sig"] = base64_encode(signature);
bool has_owner = owner && *owner;
if (has_owner) { // isSigned
val["seq"] = seq;
val["owner"] = owner->toString();
if (recipient != InfoHash())
val["to"] = recipient.toString();
}
val["type"] = type;
val["data"] = base64_encode(data);
if (not user_type.empty())
val["utype"] = user_type;
}
return val;
}
#endif //OPENDHT_PROXY_SERVER
bool
FieldValue::operator==(const FieldValue& vfd) const
{
+2 -4
View File
@@ -45,7 +45,7 @@ void print_node_info(const DhtRunner& dht, const dht_params&) {
void print_usage() {
std::cout << "Usage: dhtchat [-n network_id] [-p local_port] [-b bootstrap_host[:port]]" << std::endl << std::endl;
std::cout << "dhtchat, a simple OpenDHT command line chat client." << std::endl;
std::cout << "Report bugs to: http://opendht.net" << std::endl;
std::cout << "Report bugs to: https://opendht.net" << std::endl;
}
int
@@ -88,8 +88,6 @@ main(int argc, char **argv)
if (line.empty())
continue;
static constexpr dht::InfoHash INVALID_ID {};
std::istringstream iss(line);
std::string op, idstr;
iss >> op;
@@ -99,7 +97,7 @@ main(int argc, char **argv)
else if (op == "c") {
iss >> idstr;
room = InfoHash(idstr);
if (room == INVALID_ID) {
if (not room) {
room = InfoHash::get(idstr);
std::cout << "Joining h(" << idstr << ") = " << room << std::endl;
}
+79 -18
View File
@@ -29,9 +29,9 @@ 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 [-v [-l logfile]] [-i] [-d] [-n network_id] [-p local_port] [-b bootstrap_host[:port]] [--proxyserver local_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;
std::cout << "Report bugs to: https://opendht.net" << std::endl;
}
void print_id_req() {
@@ -55,15 +55,23 @@ void print_help() {
<< " ll Print basic information and stats about the current node." << std::endl
<< " ls [key] Print basic information about current search(es)." << std::endl
<< " ld [key] Print basic information about currenty stored values on this node (or key)." << std::endl
<< " lr Print the full current routing table of this node" << std::endl;
<< " lr Print the full current routing table of this node." << std::endl;
#if OPENDHT_PROXY_SERVER
std::cout << std::endl << "Operations with the proxy:" << std::endl
<< " pst [port] Start the proxy interface on port." << std::endl
<< " psp [port] Stop the proxy interface on port." << std::endl;
#endif //OPENDHT_PROXY_SERVER
std::cout << std::endl << "Operations on the DHT:" << std::endl
<< " b <ip:port> Ping potential node at given IP address/port." << std::endl
<< " b <ip:port> Ping potential node at given IP address/port." << std::endl
<< " g <key> Get values at <key>." << std::endl
<< " l <key> Listen for value changes at <key>." << std::endl
<< " p <key> <str> Put string value at <key>." << std::endl
<< " pp <key> <str> Put string value at <key> (persistent version)." << std::endl
<< " s <key> <str> Put string value at <key>, signed with our generated private key." << std::endl
<< " e <key> <dest> <str> Put string value at <key>, encrypted for <dest> with its public key (if found)." << std::endl;
std::cout << std::endl << "Indexation operations on the DHT:" << std::endl
<< " il <name> <key> [exact match] Lookup the index named <name> with the key <key>." << std::endl
<< " Set [exact match] to 'false' for inexact match lookup." << std::endl
@@ -82,6 +90,12 @@ void cmd_loop(std::shared_ptr<DhtRunner>& dht, dht_params& params)
#endif
std::map<std::string, indexation::Pht> indexes;
#if OPENDHT_PROXY_SERVER
std::map<in_port_t, std::unique_ptr<DhtProxyServer>> proxies;
if (params.proxyserver != 0) {
proxies.emplace(params.proxyserver, new DhtProxyServer(dht, params.proxyserver));
}
#endif //OPENDHT_PROXY_SERVER
while (true)
{
@@ -115,20 +129,18 @@ void cmd_loop(std::shared_ptr<DhtRunner>& dht, dht_params& params)
} else if (op == "ld") {
iss >> idstr;
InfoHash filter(idstr);
if (filter == InfoHash{})
std::cout << dht->getStorageLog() << std::endl;
else
if (filter)
std::cout << dht->getStorageLog(filter) << std::endl;
else
std::cout << dht->getStorageLog() << std::endl;
continue;
} else if (op == "ls") {
iss >> idstr;
InfoHash filter(idstr);
if (filter == InfoHash{}) {
std::cout << "Searches:" << std::endl;
std::cout << dht->getSearchesLog() << std::endl;
} else {
if (filter)
std::cout << dht->getSearchLog(filter) << std::endl;
}
else
std::cout << dht->getSearchesLog() << std::endl;
continue;
} else if (op == "la") {
std::cout << "Reported public addresses:" << std::endl;
@@ -161,16 +173,35 @@ void cmd_loop(std::shared_ptr<DhtRunner>& dht, dht_params& params)
dht->setLogFilter(filter);
continue;
}
#if OPENDHT_PROXY_SERVER
else if (op == "pst") {
iss >> idstr;
try {
unsigned int port = std::stoi(idstr);
proxies.emplace(port, new DhtProxyServer(dht, port));
} catch (...) { }
continue;
} else if (op == "psp") {
iss >> idstr;
try {
auto it = proxies.find(std::stoi(idstr));
if (it != proxies.end())
proxies.erase(it);
} catch (...) { }
continue;
}
#endif //OPENDHT_PROXY_SERVER
if (op.empty())
continue;
static const std::set<std::string> VALID_OPS {"g", "l", "il", "ii", "p", "s", "e", "a"};
static const std::set<std::string> VALID_OPS {"g", "l", "cl", "il", "ii", "p", "pp", "cpp", "s", "e", "a", "q"};
if (VALID_OPS.find(op) == VALID_OPS.cend()) {
std::cout << "Unknown command: " << op << std::endl;
std::cout << " (type 'h' or 'help' for a list of possible commands)" << std::endl;
continue;
}
dht::InfoHash id;
if (op == "il" or op == "ii") {
// Pht syntax
@@ -197,12 +228,17 @@ void cmd_loop(std::shared_ptr<DhtRunner>& dht, dht_params& params)
else {
// Dht syntax
iss >> idstr;
InfoHash h {idstr};
if (not isInfoHash(h))
continue;
id = dht::InfoHash(idstr);
if (not id) {
if (idstr.empty()) {
std::cerr << "Syntax error: invalid InfoHash." << std::endl;
continue;
}
id = InfoHash::get(idstr);
std::cout << "Using h(" << idstr << ") = " << id << std::endl;
}
}
dht::InfoHash id {idstr};
// Dht
auto start = std::chrono::high_resolution_clock::now();
if (op == "g") {
@@ -236,11 +272,17 @@ void cmd_loop(std::shared_ptr<DhtRunner>& dht, dht_params& params)
else if (op == "l") {
std::string rem;
std::getline(iss, rem);
dht->listen(id, [](std::shared_ptr<Value> value) {
auto token = dht->listen(id, [](std::shared_ptr<Value> value) {
std::cout << "Listen: found value:" << std::endl;
std::cout << "\t" << *value << std::endl;
return true;
}, {}, dht::Where {std::move(rem)});
std::cout << "Listening, token: " << token.get() << std::endl;
}
else if (op == "cl") {
std::string rem;
iss >> rem;
dht->cancelListen(id, std::stoul(rem));
}
else if (op == "p") {
std::string v;
@@ -253,6 +295,25 @@ void cmd_loop(std::shared_ptr<DhtRunner>& dht, dht_params& params)
std::cout << "Put: " << (ok ? "success" : "failure") << " (took " << print_dt(end-start) << "s)" << std::endl;
});
}
else if (op == "pp") {
std::string v;
iss >> v;
auto value = std::make_shared<dht::Value>(
dht::ValueType::USER_DATA.id,
std::vector<uint8_t> {v.begin(), v.end()}
);
dht->put(id, value, [start,value](bool ok) {
auto end = std::chrono::high_resolution_clock::now();
auto flags(std::cout.flags());
std::cout << "Put: " << (ok ? "success" : "failure") << " (took " << print_dt(end-start) << "s). Value ID: " << std::hex << value->id << std::endl;
std::cout.flags(flags);
}, time_point::max(), true);
}
else if (op == "cpp") {
std::string rem;
iss >> rem;
dht->cancelPut(id, std::stoul(rem, nullptr, 16));
}
else if (op == "s") {
if (not params.generate_identity) {
print_id_req();
+1 -1
View File
@@ -32,7 +32,7 @@ using namespace dht;
void print_usage() {
std::cout << "Usage: dhtscanner [-n network_id] [-p local_port] [-b bootstrap_host[:port]]" << std::endl << std::endl;
std::cout << "dhtscanner, a simple OpenDHT command line utility generating scan result the network." << std::endl;
std::cout << "Report bugs to: http://opendht.net" << std::endl;
std::cout << "Report bugs to: https://opendht.net" << std::endl;
}
struct snode_compare {
+186
View File
@@ -0,0 +1,186 @@
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="utf-8" />
<meta name="viewport" content="width=device-width, initial-scale=1.0, maximum-scale=1.0, user-scalable=no">
<title>OpenDHT tester</title>
<link rel="stylesheet" href="https://cdnjs.cloudflare.com/ajax/libs/twitter-bootstrap/3.3.7/css/bootstrap.min.css" />
<link rel="stylesheet" href="https://cdnjs.cloudflare.com/ajax/libs/font-awesome/4.7.0/css/font-awesome.css" />
<script src="https://ajax.googleapis.com/ajax/libs/jquery/3.2.1/jquery.min.js"></script>
<script type="text/javascript" src="https://cdnjs.cloudflare.com/ajax/libs/twitter-bootstrap/3.3.7/js/bootstrap.min.js"></script>
<script type="text/javascript">"use strict";
let onGet;
let onPut;
let setServer;
const valueGetElement = function(o) {
const d = window.atob(o.data);
return d;
};
$(function() {
let request = undefined;
let server;
const getTools = $("#getTools");
const getBtn = $("#getBtn");
const getDropdown = $("#getDropdown");
const listenBtn = $("#listenBtn").click(function(){onGet('LISTEN');});
const setGetRequest = function() {
getBtn.button('loading');
getStopBtn.appendTo(getTools);
getDropdown.hide();
}
const clearGetRequest = function() {
if (request === undefined)
return;
request.abort();
request = undefined;
getStopBtn.detach();
getDropdown.show();
getBtn.button('reset');
}
const getStopBtn = $("#getStopBtn").detach().click(clearGetRequest);
const putBtn = $("#putBtn");
const result = $("#dhtResult");
const group = $('<ul class="list-group"/>').appendTo(result);
onGet = function (method) {
if (request !== undefined)
return false;
const input = $("#getKey").val();
group.empty();
let lastAppended = 0;
let start = new Date().getTime();
request = new XMLHttpRequest();
request.onreadystatechange = function(event) {
if (this.readyState >= XMLHttpRequest.LOADING) {
if (this.readyState == XMLHttpRequest.DONE) {
clearGetRequest();
}
if (this.status === 200) {
const elements = this.responseText.split("\n");
const elementsLength = elements.length;
const now = new Date().getTime();
for (let i = lastAppended; i < elementsLength; i++) {
const element = elements[i];
if (!element || element.length == 0)
return;
const o = JSON.parse(element);
const d = window.atob(o.data);
const delay = Math.max(0, start-now);
$('<li class="list-group-item"/>').append(valueGetElement(o)).appendTo(group).hide().delay(delay).slideDown(100);
lastAppended = i+1;
start = Math.max(start, now)+25;
}
} else if (this.status !== 0) {
group.empty().append($('<li class="list-group-item list-group-item-danger"/>').text("Error loading content: " + this.statusText));
}
}
};
request.onerror = function(event) {
clearGetRequest();
group.empty().append($('<li class="list-group-item list-group-item-danger"/>').text("Error loading content."));
};
request.open(method, server + input, true);
request.send(null);
setGetRequest();
return false;
};
onPut = function( ) {
const key = $("#getKey").val();
const value = $("#putValue").val();
$.ajax({
url: server + key,
type: 'POST',
data: JSON.stringify({
data:window.btoa(value)
}),
contentType: 'application/json; charset=utf-8',
dataType: 'json',
success: function( result ) {
putBtn.button('reset');
//$('<li class="list-group-item list-group-item-success"/>').append(valueGetElement(result)).appendTo(group.empty());
},
error: function(result) {
putBtn.button('reset');
group.empty().append($('<li class="list-group-item list-group-item-danger"/>').text(result.statusText));
}
});
putBtn.button('loading');
return false;
};
const serverValue = $("#serverValue");
const serverStatus = $("#serverStatus");
const serverBtn = $("#serverBtn");
setServer = function(event) {
server = 'http://'+serverValue.val() + '/';
serverStatus.empty();
serverBtn.button('loading');
$.getJSON(server, function(data){
$('<span><b>Node</b> '+data.node_id+'</span>').appendTo(serverStatus).hide().fadeIn();
}).fail(function(error) {
serverStatus.html("<div class='alert alert-danger' style='margin-bottom: 0px;'><span class='glyphicon glyphicon-remove' aria-hidden='true'></span> Can't access node</div>");
}).always(function(error) {
serverBtn.button('reset');
});
return false;
};
setServer();
});
</script>
</head>
<body>
<div class="container" style="max-width: 730px;">
<header class="page-header">
<div class="row">
<div class="col-sm-5">
<h1>OpenDHT tester</h1>
</div>
<div class="col-sm-7">
<div class="well well-sm" style="margin-top:10px; margin-bottom:0px;">
<form id="serverForm" class="form-inline" onsubmit="return setServer();" style="margin-bottom:4px;">
<div class="input-group">
<input type="text" class="form-control" id="serverValue" placeholder="Proxy server" value="127.0.0.1:8080"/>
<span class="input-group-btn">
<button id="serverBtn" type="submit" class="btn btn-default" data-loading-text="<i class='fa fa-circle-o-notch fa-spin'></i>"><span class="glyphicon glyphicon-refresh" aria-hidden="true"></span></button>
</span>
</div>
</form>
<div id="serverStatus"><i class='fa fa-circle-o-notch fa-spin'></i></div>
</div>
</div>
</div>
</header>
<div class="panel panel-default" id="dhtResult">
<div class="panel-heading">
<div class="row">
<div class="col-xs-6">
<form class="form-inline" onsubmit="return onGet('GET');">
<div class="input-group">
<input type="text" class="form-control" id="getKey" placeholder="Key" aria-label="Key" />
<span class="input-group-btn" id="getTools">
<button id="getBtn" class="btn btn-default" data-loading-text="<i class='fa fa-circle-o-notch fa-spin'></i>" type="submit">Get</button>
<button id="getDropdown"type="button" class="btn btn-default dropdown-toggle" data-toggle="dropdown" aria-haspopup="true" aria-expanded="false"> <span class="caret"></span> <span class="sr-only">Toggle Dropdown</span> </button>
<ul class="dropdown-menu">
<li><a id="listenBtn" href="#">Listen</a></li>
</ul>
<button id="getStopBtn" class="btn btn-default" type="submit"><span class="glyphicon glyphicon-remove" aria-hidden="true"></span></button>
</span>
</div>
</form>
</div>
<div class="col-xs-6">
<form class="form-inline" onsubmit="return onPut();">
<div class="input-group">
<input type="text" class="form-control input-group-input" id="putValue" placeholder="Value" />
<span class="input-group-btn">
<button id="putBtn" type="submit" class="btn btn-default" data-loading-text="<i class='fa fa-circle-o-notch fa-spin'></i> Loading">Put</button>
</span>
</div>
</form>
</div>
</div>
</div>
</div>
</div>
</body>
</html>
+12 -4
View File
@@ -98,10 +98,8 @@ dht::indexation::Pht::Key createPhtKey(std::map<std::string, std::string> pht_ke
return pht_key;
}
bool isInfoHash(dht::InfoHash& h) {
static constexpr dht::InfoHash INVALID_ID {};
if (h == INVALID_ID) {
bool isInfoHash(const dht::InfoHash& h) {
if (not h) {
std::cout << "Syntax error: invalid InfoHash." << std::endl;
return false;
}
@@ -121,6 +119,7 @@ struct dht_params {
bool daemonize {false};
bool service {false};
std::pair<std::string, std::string> bootstrap {};
in_port_t proxyserver {0};
};
static const constexpr struct option long_options[] = {
@@ -134,6 +133,7 @@ static const constexpr struct option long_options[] = {
{"service", no_argument , nullptr, 's'},
{"logfile", required_argument, nullptr, 'l'},
{"syslog", no_argument , nullptr, 'L'},
{"proxyserver",required_argument, nullptr, 'S'},
{nullptr, 0 , nullptr, 0}
};
@@ -151,6 +151,14 @@ parseArgs(int argc, char **argv) {
std::cout << "Invalid port: " << port_arg << std::endl;
}
break;
case 'S': {
int port_arg = atoi(optarg);
if (port_arg >= 0 && port_arg < 0x10000)
params.proxyserver = port_arg;
else
std::cout << "Invalid port: " << port_arg << std::endl;
}
break;
case 'n':
params.network = strtoul(optarg, nullptr, 0);
break;