Compare commits

...
19 changed files with 406 additions and 59 deletions

No files matched your search

+4 -1
View File
@@ -71,8 +71,11 @@ if (NOT MSVC)
)
endif()
if (OPENDHT_HTTP)
if (OPENDHT_HTTP OR OPENDHT_PEER_DISCOVERY)
find_path(ASIO_INCLUDE_DIR asio.hpp REQUIRED)
endif ()
if (OPENDHT_HTTP)
find_package(Restinio REQUIRED)
find_library(FMT_LIBRARY fmt)
add_library(fmt SHARED IMPORTED)
+3 -1
View File
@@ -78,6 +78,7 @@ struct OPENDHT_PUBLIC NodeStats {
struct OPENDHT_PUBLIC NodeInfo {
InfoHash id;
PkId pkid;
InfoHash node_id;
NodeStats ipv4 {};
NodeStats ipv6 {};
@@ -94,7 +95,7 @@ struct OPENDHT_PUBLIC NodeInfo {
explicit NodeInfo(const Json::Value& v);
#endif
MSGPACK_DEFINE_MAP(id, node_id, ipv4, ipv6)
MSGPACK_DEFINE_MAP(id, pkid, node_id, ipv4, ipv6)
};
/**
@@ -167,6 +168,7 @@ using ShutdownCallback = std::function<void()>;
using IdentityAnnouncedCb = std::function<void(bool)>;
using CertificateStoreQuery = std::function<std::vector<std::shared_ptr<crypto::Certificate>>(const InfoHash& pk_id)>;
using CertificateStoreLongQuery = std::function<std::vector<std::shared_ptr<crypto::Certificate>>(const PkId& pk_id)>;
typedef bool (*GetCallbackRaw)(std::shared_ptr<Value>, void *user_data);
typedef bool (*ValueCallbackRaw)(std::shared_ptr<Value>, bool expired, void *user_data);
+2
View File
@@ -80,6 +80,8 @@ struct OPENDHT_PUBLIC PublicKey
* Takes ownership of an existing gnutls_pubkey.
*/
PublicKey(gnutls_pubkey_t k) : pk(k) {}
/** Import public key from serialized data */
PublicKey(const uint8_t* dat, size_t dat_size);
PublicKey(const Blob& pk) : PublicKey(pk.data(), pk.size()) {}
PublicKey(PublicKey&& o) noexcept : pk(o.pk) { o.pk = nullptr; };
+6 -1
View File
@@ -66,8 +66,11 @@ private:
public:
virtual void unpackValue(const Value& v) override {
if (v.owner)
if (v.owner) {
owner = v.owner;
from = v.owner->getId();
fromId = v.owner->getLongId();
}
BaseClass::unpackValue(v);
}
@@ -75,7 +78,9 @@ public:
return [](const Value& v){ return v.isSigned(); };
}
Sp<crypto::PublicKey> owner;
dht::InfoHash from;
dht::PkId fromId;
};
template <typename T>
-1
View File
@@ -445,7 +445,6 @@ private:
std::map<InfoHash, std::vector<Listener>> listeners;
MSGPACK_DEFINE_ARRAY(listeners)
};
std::mutex lockPushListeners_;
std::map<std::string, PushListener> pushListeners_;
proxy::ListenToken tokenPushNotif_ {0};
#endif //OPENDHT_PUSH_NOTIFICATIONS
+24 -1
View File
@@ -74,6 +74,7 @@ public:
std::shared_ptr<PeerDiscovery> peerDiscovery {};
StatusCallback statusChangedCallback {};
CertificateStoreQuery certificateStore {};
CertificateStoreLongQuery certificateStoreSha256 {};
IdentityAnnouncedCb identityAnnouncedCb {};
Context() {}
};
@@ -257,6 +258,27 @@ public:
}
void putEncrypted(const std::string& key, InfoHash to, Value&& value, DoneCallback cb={}, bool permanent = false);
void putEncrypted(InfoHash hash, PkId to, std::shared_ptr<Value> value, DoneCallback cb={}, bool permanent = false);
void putEncrypted(InfoHash hash, PkId to, std::shared_ptr<Value> value, DoneCallbackSimple cb, bool permanent = false) {
putEncrypted(hash, to, value, bindDoneCb(cb), permanent);
}
void putEncrypted(InfoHash hash, PkId to, Value&& value, DoneCallback cb={}, bool permanent = false);
void putEncrypted(InfoHash hash, PkId to, Value&& value, DoneCallbackSimple cb, bool permanent = false) {
putEncrypted(hash, to, std::forward<Value>(value), bindDoneCb(cb), permanent);
}
void putEncrypted(InfoHash hash, const std::shared_ptr<crypto::PublicKey>& to, std::shared_ptr<Value> value, DoneCallback cb={}, bool permanent = false);
void putEncrypted(InfoHash hash, const std::shared_ptr<crypto::PublicKey>& to, std::shared_ptr<Value> value, DoneCallbackSimple cb, bool permanent = false) {
putEncrypted(hash, to, value, bindDoneCb(cb), permanent);
}
void putEncrypted(InfoHash hash, const std::shared_ptr<crypto::PublicKey>& to, Value&& value, DoneCallback cb={}, bool permanent = false);
void putEncrypted(InfoHash hash, const std::shared_ptr<crypto::PublicKey>& to, Value&& value, DoneCallbackSimple cb, bool permanent = false) {
putEncrypted(hash, to, std::forward<Value>(value), bindDoneCb(cb), permanent);
}
/**
* 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.
@@ -303,6 +325,7 @@ public:
* Get the public key fingerprint if an identity is used with this node, 0 otherwise.
*/
InfoHash getId() const;
PkId getLongId() const;
/**
* Get the ID of the DHT node.
@@ -367,7 +390,7 @@ public:
void findCertificate(InfoHash hash, std::function<void(const std::shared_ptr<crypto::Certificate>&)>);
void registerCertificate(std::shared_ptr<crypto::Certificate> cert);
void setLocalCertificateStore(CertificateStoreQuery&& query_method);
void setLocalCertificateStore(CertificateStoreQuery&& query_method, CertificateStoreLongQuery&& query_sha256_method = {});
/**
* @param port: Local port to bind. Both IPv4 and IPv6 will be tried (ANY).
+24 -6
View File
@@ -115,6 +115,14 @@ public:
void putEncrypted(const InfoHash& hash, const InfoHash& to, Value&& v, DoneCallback callback, bool permanent = false) {
putEncrypted(hash, to, std::make_shared<Value>(std::move(v)), callback, permanent);
}
void putEncrypted(const InfoHash& hash, const PkId& to, Sp<Value> val, DoneCallback callback, bool permanent = false);
void putEncrypted(const InfoHash& hash, const PkId& to, Value&& v, DoneCallback callback, bool permanent = false) {
putEncrypted(hash, to, std::make_shared<Value>(std::move(v)), callback, permanent);
}
void putEncrypted(const InfoHash& hash, const crypto::PublicKey& to, Sp<Value> val, DoneCallback callback, bool permanent = false);
void putEncrypted(const InfoHash& hash, const crypto::PublicKey& to, Value&& v, DoneCallback callback, bool permanent = false) {
putEncrypted(hash, to, std::make_shared<Value>(std::move(v)), callback, permanent);
}
/**
* Take ownership of the value and sign it using our private key.
@@ -126,21 +134,28 @@ public:
Value decrypt(const Value& v);
void findCertificate(const InfoHash& node, const std::function<void(const Sp<crypto::Certificate>)>& cb);
void findPublicKey(const InfoHash& node, const std::function<void(const Sp<const crypto::PublicKey>)>& cb);
void findPublicKey(const InfoHash& node, const std::function<void(const Sp<crypto::PublicKey>)>& cb);
const Sp<crypto::Certificate> registerCertificate(const InfoHash& node, const Blob& cert);
Sp<crypto::Certificate> registerCertificate(const InfoHash& node, const Blob& cert);
void registerCertificate(Sp<crypto::Certificate>& cert);
const Sp<crypto::Certificate> getCertificate(const InfoHash& node) const;
const Sp<const crypto::PublicKey> getPublicKey(const InfoHash& node) const;
Sp<crypto::Certificate> getCertificate(const InfoHash& node) const;
Sp<crypto::PublicKey> getPublicKey(const InfoHash& node) const;
void findCertificate(const PkId& node, const std::function<void(const Sp<crypto::Certificate>)>& cb);
void findPublicKey(const PkId& node, const std::function<void(const Sp<crypto::PublicKey>)>& cb);
Sp<crypto::Certificate> registerCertificate(const PkId& node, const Blob& cert);
Sp<crypto::Certificate> getCertificate(const PkId& node) const;
Sp<crypto::PublicKey> getPublicKey(const PkId& node) const;
/**
* Allows to set a custom callback called by the library to find a locally-stored certificate.
* The search key used is the public key ID, so there may be multiple certificates retured, signed with
* the same private key.
*/
void setLocalCertificateStore(CertificateStoreQuery&& query_method) {
void setLocalCertificateStore(CertificateStoreQuery&& query_method, CertificateStoreLongQuery&& query_sha256_method) {
localQueryMethod_ = std::move(query_method);
localQuerySha256Method_ = std::move(query_sha256_method);
}
/**
@@ -351,10 +366,13 @@ private:
// method to query the local certificate store
CertificateStoreQuery localQueryMethod_ {};
CertificateStoreLongQuery localQuerySha256Method_ {};
// our certificate cache
std::map<InfoHash, Sp<crypto::Certificate>> nodesCertificates_ {};
std::map<InfoHash, Sp<const crypto::PublicKey>> nodesPubKeys_ {};
std::map<InfoHash, Sp<crypto::PublicKey>> nodesPubKeys_ {};
std::map<PkId, Sp<crypto::Certificate>> nodesCertificatesLong_ {};
std::map<PkId, Sp<crypto::PublicKey>> nodesPubKeysLong_ {};
std::atomic_bool forward_all_ {false};
bool enableCache_ {false};
+1 -1
View File
@@ -68,7 +68,7 @@ public:
* Build from existing address.
*/
SockAddr(const sockaddr* sa, socklen_t length) {
if (length > sizeof(sockaddr_storage))
if (length > (socklen_t)sizeof(sockaddr_storage))
throw std::runtime_error("Socket address length is too large");
set(sa, length);
}
+4 -4
View File
@@ -356,7 +356,7 @@ struct OPENDHT_PUBLIC Value
void sign(const crypto::PrivateKey& key) {
if (isEncrypted())
throw DhtException("Can't sign encrypted data.");
owner = std::make_shared<const crypto::PublicKey>(key.getPublicKey());
owner = std::make_shared<crypto::PublicKey>(key.getPublicKey());
signature = key.sign(getToSign());
}
@@ -368,8 +368,8 @@ struct OPENDHT_PUBLIC Value
return isSigned() and owner->checkSignature(getToSign(), signature);
}
std::shared_ptr<const crypto::PublicKey> getOwner() const {
return std::static_pointer_cast<const crypto::PublicKey>(owner);
std::shared_ptr<crypto::PublicKey> getOwner() const {
return std::static_pointer_cast<crypto::PublicKey>(owner);
}
/**
@@ -594,7 +594,7 @@ struct OPENDHT_PUBLIC Value
/**
* Public key of the signer.
*/
std::shared_ptr<const crypto::PublicKey> owner {};
std::shared_ptr<crypto::PublicKey> owner {};
/**
* Hash of the recipient (optional).
+14 -2
View File
@@ -36,6 +36,7 @@ from libcpp.memory cimport shared_ptr
from cython.parallel import parallel, prange
from cython.operator cimport dereference as deref, preincrement as inc, predecrement as dec
from cpython cimport ref
from datetime import timedelta
cimport opendht_cpp as cpp
@@ -235,8 +236,9 @@ cdef class Where(object):
cdef class Value(object):
cdef shared_ptr[cpp.Value] _value
def __init__(self, bytes val=b''):
self._value.reset(new cpp.Value(val, len(val)))
def __init__(self, bytes val=b'', cpp.uint16_t id=0):
self._value.reset(new cpp.Value(id, val, len(val)))
def __str__(self):
return self._value.get().toString().decode()
property owner:
@@ -270,6 +272,14 @@ cdef class Value(object):
def __get__(self):
return self._value.get().size()
cdef class ValueType(object):
cdef cpp.ValueType * _value
def __init__(self, cpp.uint16_t id, str name, expiration: timedelta):
if not isinstance(expiration, timedelta):
raise TypeError("expiration argument must be of type timedelta")
cdef cpp.seconds duration = cpp.seconds(int(expiration.total_seconds()))
self._value = new cpp.ValueType(id, name.encode(), duration)
cdef class NodeSetIter(object):
cdef map[cpp.InfoHash, shared_ptr[cpp.Node]]* _nodes
cdef map[cpp.InfoHash, shared_ptr[cpp.Node]].iterator _curIter
@@ -513,6 +523,8 @@ cdef class DhtRunner(_WithID):
cb_obj = {'shutdown':shutdown_cb}
ref.Py_INCREF(cb_obj)
self.thisptr.get().shutdown(cpp.bindShutdownCb(shutdown_callback, <void*>cb_obj))
def registerType(self, ValueType type):
self.thisptr.get().registerType(deref(type._value))
def enableLogging(self):
cpp.enableLogging(self.thisptr.get()[0])
def disableLogging(self):
+16 -2
View File
@@ -26,6 +26,14 @@ from libc.string cimport const_char, const_uchar
ctypedef uint16_t in_port_t
ctypedef unsigned short int sa_family_t;
cdef extern from "<iostream>" namespace "std::chrono" nogil:
cdef cppclass duration[ulong]:
duration() except +
cdef cppclass seconds:
duration seconds(uint64_t) except +
duration seconds()
cdef extern from "<memory>" namespace "std" nogil:
cdef cppclass shared_ptr[T]:
shared_ptr() except +
@@ -131,7 +139,6 @@ cdef extern from "opendht/value.h" namespace "dht::Value":
cdef extern from "opendht/value.h" namespace "dht::Value::Field":
cdef Field None
cdef Field Id
cdef Field ValueType
cdef Field OwnerPk
cdef Field SeqNum
cdef Field UserType
@@ -141,7 +148,7 @@ cdef extern from "opendht/value.h" namespace "dht":
cdef cppclass Value:
Value() except +
Value(vector[uint8_t]) except +
Value(const uint8_t* dat_ptr, size_t dat_len) except +
Value(const uint16_t t, const uint8_t* dat_ptr, size_t dat_len) except +
string toString() const
size_t size() const
uint64_t id
@@ -150,6 +157,12 @@ cdef extern from "opendht/value.h" namespace "dht":
vector[uint8_t] data
string user_type
cdef cppclass ValueType:
ValueType(uint16_t id, string name, seconds expiration) except +
uint16_t id
string name
seconds expiration
cdef cppclass Query:
Query() except +
Query(Select s, Where w) except +
@@ -234,6 +247,7 @@ cdef extern from "opendht/dhtrunner.h" namespace "dht":
void run(const_char*, const_char*, const_char*, Config config)
void join()
void shutdown(ShutdownCallback)
void registerType(ValueType& value);
bool isRunning()
SockAddr getBound(sa_family_t af) const
string getStorageLog() const
+4
View File
@@ -141,6 +141,8 @@ NodeInfo::toJson() const
Json::Value val;
if (id)
val["id"] = id.toString();
if (pkid)
val["pkid"] = pkid.toString();
val["node_id"] = node_id.toString();
val["ipv4"] = ipv4.toJson();
val["ipv6"] = ipv6.toJson();
@@ -152,6 +154,8 @@ NodeInfo::NodeInfo(const Json::Value& v)
{
if (v.isMember("id"))
id = InfoHash(v["id"].asString());
if (v.isMember("pkid"))
pkid = PkId(v["pkid"].asString());
node_id = InfoHash(v["node_id"].asString());
ipv4 = NodeStats(v["ipv4"]);
ipv6 = NodeStats(v["ipv6"]);
+6 -6
View File
@@ -710,13 +710,13 @@ CertificateRequest::sign(const PrivateKey& key, const std::string& password)
}
bool
CertificateRequest::verify() const
CertificateRequest::verify() const
{
return gnutls_x509_crq_verify(request, 0) >= 0;
}
Blob
CertificateRequest::pack() const
Blob
CertificateRequest::pack() const
{
gnutls_datum_t dat {nullptr, 0};
if (auto err = gnutls_x509_crq_export2(request, GNUTLS_X509_FMT_PEM, &dat))
@@ -727,7 +727,7 @@ CertificateRequest::pack() const
}
std::string
CertificateRequest::toString() const
CertificateRequest::toString() const
{
gnutls_datum_t dat {nullptr, 0};
if (auto err = gnutls_x509_crq_export2(request, GNUTLS_X509_FMT_PEM, &dat))
@@ -1147,8 +1147,8 @@ void
setRandomSerial(gnutls_x509_crt_t cert)
{
random_device rdev;
std::uniform_int_distribution<uint64_t> dist{};
uint64_t cert_serial = dist(rdev);
std::uniform_int_distribution<int64_t> dist{1};
int64_t cert_serial = dist(rdev);
gnutls_x509_crt_set_serial(cert, &cert_serial, sizeof(cert_serial));
}
+1 -1
View File
@@ -822,7 +822,7 @@ DhtProxyServer::subscribe(restinio::request_handle_t request,
logger_->d("[proxy:server] [subscribe %s] [client %s] [session %s]", infoHash.toString().c_str(), clientId.c_str(), sessionId.c_str());
// Insert new or return existing push listeners of a token
std::lock_guard<std::mutex> lock(lockPushListeners_);
std::lock_guard<std::mutex> lock(lockListener_);
auto& pushListener = pushListeners_[pushToken];
auto& pushListeners = pushListener.listeners[infoHash];
+66 -6
View File
@@ -185,10 +185,10 @@ DhtRunner::run(const Config& config, Context&& context)
if (context.statusChangedCallback) {
statusCb = std::move(context.statusChangedCallback);
}
if (context.certificateStore) {
dht_->setLocalCertificateStore(std::move(context.certificateStore));
if (context.certificateStore || context.certificateStoreSha256) {
dht_->setLocalCertificateStore(std::move(context.certificateStore), std::move(context.certificateStoreSha256));
if (dht_via_proxy_)
dht_via_proxy_->setLocalCertificateStore(std::move(context.certificateStore));
dht_via_proxy_->setLocalCertificateStore(std::move(context.certificateStore), std::move(context.certificateStoreSha256));
}
if (not config.threaded)
@@ -398,6 +398,15 @@ DhtRunner::getId() const
return {};
}
PkId
DhtRunner::getLongId() const
{
if (auto dht = activeDht())
return dht->getLongId();
return {};
}
InfoHash
DhtRunner::getNodeId() const
{
@@ -509,6 +518,7 @@ DhtRunner::getNodeInfo() const {
NodeInfo info {};
if (auto dht = activeDht()) {
info.id = dht->getId();
info.pkid = dht->getLongId();
info.node_id = dht->getNodeId();
info.ipv4 = dht->getNodesStats(AF_INET);
info.ipv6 = dht->getNodesStats(AF_INET6);
@@ -530,6 +540,7 @@ DhtRunner::getNodeInfo(std::function<void(std::shared_ptr<NodeInfo>)> cb)
auto sinfo = std::make_shared<NodeInfo>();
auto& info = *sinfo;
info.id = dht.getId();
info.pkid = dht.getLongId();
info.node_id = dht.getNodeId();
info.ipv4 = dht.getNodesStats(AF_INET);
info.ipv6 = dht.getNodesStats(AF_INET6);
@@ -616,14 +627,14 @@ DhtRunner::registerCertificate(std::shared_ptr<crypto::Certificate> cert) {
activeDht()->registerCertificate(cert);
}
void
DhtRunner::setLocalCertificateStore(CertificateStoreQuery&& query_method) {
DhtRunner::setLocalCertificateStore(CertificateStoreQuery&& query_method, CertificateStoreLongQuery&& query_sha256_method) {
std::lock_guard<std::mutex> lck(dht_mtx);
#ifdef OPENDHT_PROXY_CLIENT
if (dht_via_proxy_)
dht_via_proxy_->setLocalCertificateStore(std::forward<CertificateStoreQuery>(query_method));
dht_via_proxy_->setLocalCertificateStore(std::move(query_method), std::move(query_sha256_method));
#endif
if (dht_)
dht_->setLocalCertificateStore(std::forward<CertificateStoreQuery>(query_method));
dht_->setLocalCertificateStore(std::move(query_method), std::move(query_sha256_method));
}
time_point
@@ -943,6 +954,55 @@ DhtRunner::putEncrypted(const std::string& key, InfoHash to, Value&& value, Done
putEncrypted(InfoHash::get(key), to, std::forward<Value>(value), std::move(cb), permanent);
}
void
DhtRunner::putEncrypted(InfoHash hash, PkId to, std::shared_ptr<Value> value, DoneCallback cb, bool permanent)
{
if (running != State::Running) {
if (cb) cb(false, {});
return;
}
std::lock_guard<std::mutex> lck(storage_mtx);
ongoing_ops++;
pending_ops.emplace([=,
cb = std::move(cb),
value = std::move(value)
] (SecureDht& dht) mutable {
dht.putEncrypted(hash, to, value, bindOpDoneCallback(std::move(cb)), permanent);
});
cv.notify_all();
}
void
DhtRunner::putEncrypted(InfoHash hash, PkId to, Value&& value, DoneCallback cb, bool permanent)
{
putEncrypted(hash, to, std::make_shared<Value>(std::move(value)), std::move(cb), permanent);
}
void
DhtRunner::putEncrypted(InfoHash hash, const std::shared_ptr<crypto::PublicKey>& to, std::shared_ptr<Value> value, DoneCallback cb, bool permanent)
{
if (running != State::Running) {
if (cb) cb(false, {});
return;
}
std::lock_guard<std::mutex> lck(storage_mtx);
ongoing_ops++;
pending_ops.emplace([=,
cb = std::move(cb),
value = std::move(value)
] (SecureDht& dht) mutable {
dht.putEncrypted(hash, *to, value, bindOpDoneCallback(std::move(cb)), permanent);
});
cv.notify_all();
}
void
DhtRunner::putEncrypted(InfoHash hash, const std::shared_ptr<crypto::PublicKey>& to, Value&& value, DoneCallback cb, bool permanent)
{
putEncrypted(hash, to, std::make_shared<Value>(std::move(value)), std::move(cb), permanent);
}
void
DhtRunner::bootstrap(const std::string& host, const std::string& service)
{
+60 -7
View File
@@ -561,12 +561,63 @@ void
Connection::async_connect(std::vector<asio::ip::tcp::endpoint>&& endpoints, ConnectHandlerCb cb)
{
std::lock_guard<std::mutex> lock(mutex_);
if (ssl_socket_)
asio::async_connect(ssl_socket_->lowest_layer(), std::move(endpoints), wrapCallabck(std::move(cb)));
else if (socket_)
asio::async_connect(*socket_, std::move(endpoints), wrapCallabck(std::move(cb)));
else if (cb)
if (!ssl_socket_ && !socket_) {
cb(asio::error::operation_aborted, {});
return;
}
auto& base = ssl_socket_? ssl_socket_->lowest_layer() : *socket_;
ConnectHandlerCb wcb = [&base, cb=std::move(cb)](const asio::error_code& ec, const asio::ip::tcp::endpoint& endpoint) {
if (!ec) {
auto socket = base.native_handle();
// Once connected, set a keep alive on the TCP socket with 30 seconds delay
// This will generate broken pipes as soon as possible.
// Note this needs to be done once connected to have a valid native_handle()
uint32_t start = 30;
uint32_t interval = 30;
uint32_t cnt = 1;
#ifdef _WIN32
std::string val = "1";
setsockopt(socket, SOL_SOCKET, SO_KEEPALIVE, val.c_str(), sizeof(val));
// TCP_KEEPIDLE and TCP_KEEPINTVL are available since Win 10 version 1709
// TCP_KEEPCNT since Win 10 version 1703
#ifdef TCP_KEEPIDLE
std::string start_str = std::to_string(start);
setsockopt(socket, IPPROTO_TCP, TCP_KEEPIDLE,
start_str.c_str(), sizeof(start_str));
#endif
#ifdef TCP_KEEPINTVL
std::string interval_str = std::to_string(interval);
setsockopt(socket, IPPROTO_TCP, TCP_KEEPINTVL,
interval_str.c_str(), sizeof(interval_str));
#endif
#ifdef TCP_KEEPCNT
std::string cnt_str = std::to_string(cnt);
setsockopt(socket, IPPROTO_TCP, TCP_KEEPCNT,
cnt_str.c_str(), sizeof(cnt_str));
#endif
#else
uint32_t val = 1;
setsockopt(socket, SOL_SOCKET, SO_KEEPALIVE, &val, sizeof(uint32_t));
#ifdef __APPLE__
// Apple devices only have one parameter
setsockopt(socket, IPPROTO_TCP, TCP_KEEPALIVE, &start, sizeof(uint32_t));
#else
// Linux based systems
setsockopt(socket, SOL_TCP, TCP_KEEPIDLE, &start, sizeof(uint32_t));
setsockopt(socket, SOL_TCP, TCP_KEEPINTVL, &interval, sizeof(uint32_t));
setsockopt(socket, SOL_TCP, TCP_KEEPCNT, &cnt, sizeof(uint32_t));
#endif
#endif
}
if (cb)
cb(ec, endpoint);
};
if (ssl_socket_)
asio::async_connect(ssl_socket_->lowest_layer(), std::move(endpoints), wrapCallabck(std::move(wcb)));
else
asio::async_connect(*socket_, std::move(endpoints), wrapCallabck(std::move(wcb)));
}
void
@@ -1254,9 +1305,11 @@ Request::terminate(const asio::error_code& ec)
return;
response_.aborted = ec == asio::error::operation_aborted;
if (ec == asio::error::basic_errors::broken_pipe)
response_.status_code = 0U; // Avoid to give a successful answer (happen with a broken pipe, takes the last status)
if (logger_) {
if (ec and ec != asio::error::eof and ec != asio::error::operation_aborted)
if (ec and ec != asio::error::eof and !response_.aborted)
logger_->e("[http:request:%i] end with error: %s", id_, ec.message().c_str());
else
logger_->d("[http:request:%i] done with status code %u", id_, response_.status_code);
@@ -1415,7 +1468,7 @@ Request::await()
std::unique_lock<std::mutex> lock(mtx);
std::condition_variable cv;
bool ok {false};
add_on_done_callback([&](const Response& resp){
add_on_done_callback([&](const Response&){
std::lock_guard<std::mutex> lk(mtx);
ok = true;
cv.notify_all();
+167 -17
View File
@@ -118,7 +118,7 @@ SecureDht::secureType(ValueType&& type)
return type;
}
const Sp<crypto::Certificate>
Sp<crypto::Certificate>
SecureDht::getCertificate(const InfoHash& node) const
{
if (node == getId())
@@ -130,7 +130,7 @@ SecureDht::getCertificate(const InfoHash& node) const
return it->second;
}
const Sp<const crypto::PublicKey>
Sp<crypto::PublicKey>
SecureDht::getPublicKey(const InfoHash& node) const
{
if (node == getId())
@@ -142,7 +142,7 @@ SecureDht::getPublicKey(const InfoHash& node) const
return it->second;
}
const Sp<crypto::Certificate>
Sp<crypto::Certificate>
SecureDht::registerCertificate(const InfoHash& node, const Blob& data)
{
Sp<crypto::Certificate> crt;
@@ -155,12 +155,9 @@ SecureDht::registerCertificate(const InfoHash& node, const Blob& data)
if (node == h) {
if (logger_)
logger_->d("Registering certificate for %s", h.toString().c_str());
auto it = nodesCertificates_.find(h);
if (it == nodesCertificates_.end())
std::tie(it, std::ignore) = nodesCertificates_.emplace(h, std::move(crt));
else
it->second = std::move(crt);
return it->second;
nodesCertificates_[h] = crt;
nodesCertificatesLong_[crt->getLongId()] = crt;
return crt;
} else {
if (logger_)
logger_->w("Certificate %s for node %s does not match node id !", h.toString().c_str(), node.toString().c_str());
@@ -171,8 +168,10 @@ SecureDht::registerCertificate(const InfoHash& node, const Blob& data)
void
SecureDht::registerCertificate(Sp<crypto::Certificate>& cert)
{
if (cert)
if (cert) {
nodesCertificates_[cert->getId()] = cert;
nodesCertificatesLong_[cert->getLongId()] = cert;
}
}
void
@@ -192,6 +191,7 @@ SecureDht::findCertificate(const InfoHash& node, const std::function<void(const
if (logger_)
logger_->d("Registering certificate from local store for %s", node.to_c_str());
nodesCertificates_.emplace(node, res.front());
nodesCertificatesLong_.emplace(res.front()->getLongId(), res.front());
if (cb)
cb(res.front());
return;
@@ -200,8 +200,6 @@ SecureDht::findCertificate(const InfoHash& node, const std::function<void(const
auto found = std::make_shared<bool>(false);
dht_->get(node, [cb,node,found,this](const std::vector<Sp<Value>>& vals) {
if (*found)
return false;
for (const auto& v : vals) {
if (auto cert = registerCertificate(node, v->data)) {
*found = true;
@@ -212,7 +210,7 @@ SecureDht::findCertificate(const InfoHash& node, const std::function<void(const
return false;
}
}
return true;
return !*found;
}, [cb,found](bool) {
if (!*found and cb)
cb(nullptr);
@@ -220,7 +218,7 @@ SecureDht::findCertificate(const InfoHash& node, const std::function<void(const
}
void
SecureDht::findPublicKey(const InfoHash& node, const std::function<void(const Sp<const crypto::PublicKey>)>& cb)
SecureDht::findPublicKey(const InfoHash& node, const std::function<void(const Sp<crypto::PublicKey>)>& cb)
{
auto pk = getPublicKey(node);
if (pk && *pk) {
@@ -235,6 +233,7 @@ SecureDht::findPublicKey(const InfoHash& node, const std::function<void(const Sp
auto pk = std::make_shared<crypto::PublicKey>(crt->getPublicKey());
if (*pk) {
nodesPubKeys_[pk->getId()] = pk;
nodesPubKeysLong_[pk->getLongId()] = pk;
if (cb) cb(pk);
return;
}
@@ -243,6 +242,106 @@ SecureDht::findPublicKey(const InfoHash& node, const std::function<void(const Sp
});
}
void
SecureDht::findCertificate(const PkId& node, const std::function<void(const Sp<crypto::Certificate>)>& cb)
{
Sp<crypto::Certificate> b = getCertificate(node);
if (b && *b) {
if (logger_)
logger_->d("Using certificate from cache for %s", node.to_c_str());
if (cb)
cb(b);
return;
}
if (localQueryMethod_) {
auto res = localQuerySha256Method_(node);
if (not res.empty()) {
if (logger_)
logger_->d("Registering certificate from local store for %s", node.to_c_str());
nodesCertificates_.emplace(res.front()->getId(), res.front());
nodesCertificatesLong_.emplace(node, res.front());
if (cb)
cb(res.front());
return;
}
}
if (cb)
cb(nullptr);
}
void
SecureDht::findPublicKey(const PkId& node, const std::function<void(const Sp<crypto::PublicKey>)>& cb)
{
auto pk = getPublicKey(node);
if (pk && *pk) {
if (logger_)
logger_->d("Found public key from cache for %s", node.to_c_str());
if (cb)
cb(pk);
return;
}
findCertificate(node, [=](const Sp<crypto::Certificate> crt) {
if (crt && *crt) {
auto pk = std::make_shared<crypto::PublicKey>(crt->getPublicKey());
if (*pk) {
nodesPubKeys_[pk->getId()] = pk;
nodesPubKeysLong_[pk->getLongId()] = pk;
if (cb) cb(pk);
return;
}
}
if (cb) cb(nullptr);
});
}
Sp<crypto::Certificate>
SecureDht::registerCertificate(const PkId& pkId, const Blob& data)
{
Sp<crypto::Certificate> crt;
try {
crt = std::make_shared<crypto::Certificate>(data);
} catch (const std::exception& e) {
return nullptr;
}
PkId id = crt->getPublicKey().getLongId();
if (pkId == id) {
if (logger_)
logger_->d("Registering certificate for %s", id.toString().c_str());
nodesCertificates_[crt->getId()] = crt;
nodesCertificatesLong_[id] = crt;
return crt;
} else {
if (logger_)
logger_->w("Certificate %s for node %s does not match node id !", id.toString().c_str(), pkId.toString().c_str());
return nullptr;
}
}
Sp<crypto::Certificate>
SecureDht::getCertificate(const PkId& pkId) const
{
if (pkId == getLongId())
return certificate_;
auto it = nodesCertificatesLong_.find(pkId);
if (it == nodesCertificatesLong_.end())
return nullptr;
else
return it->second;
}
Sp<crypto::PublicKey>
SecureDht::getPublicKey(const PkId& pkId) const
{
if (pkId == getLongId())
return std::make_shared<crypto::PublicKey>(certificate_->getPublicKey());
auto it = nodesPubKeysLong_.find(pkId);
if (it == nodesPubKeysLong_.end())
return nullptr;
else
return it->second;
}
Sp<Value>
SecureDht::checkValue(const Sp<Value>& v)
{
@@ -262,8 +361,10 @@ SecureDht::checkValue(const Sp<Value>& v)
try {
Value decrypted_val (decrypt(*v));
if (decrypted_val.recipient == getId()) {
if (decrypted_val.owner)
if (decrypted_val.owner) {
nodesPubKeys_[decrypted_val.owner->getId()] = decrypted_val.owner;
nodesPubKeysLong_[decrypted_val.owner->getLongId()] = decrypted_val.owner;
}
v->decryptedValue = std::make_shared<Value>(std::move(decrypted_val));
return v->decryptedValue;
}
@@ -281,8 +382,10 @@ SecureDht::checkValue(const Sp<Value>& v)
v->signatureChecked = true;
if (v->owner and v->owner->checkSignature(v->getToSign(), v->signature)) {
v->signatureValid = true;
if (enableCache_)
if (enableCache_) {
nodesPubKeys_[v->owner->getId()] = v->owner;
nodesPubKeysLong_[v->owner->getLongId()] = v->owner;
}
return v;
}
else if (logger_)
@@ -405,7 +508,7 @@ SecureDht::putEncrypted(const InfoHash& hash, const InfoHash& to, Sp<Value> val,
callback(false, {});
return;
}
findPublicKey(to, [=](const Sp<const crypto::PublicKey>& pk) {
findPublicKey(to, [this, hash, val = std::move(val), callback = std::move(callback), permanent](const Sp<crypto::PublicKey>& pk) {
if(!pk || !*pk) {
if (callback)
callback(false, {});
@@ -424,6 +527,53 @@ SecureDht::putEncrypted(const InfoHash& hash, const InfoHash& to, Sp<Value> val,
});
}
void
SecureDht::putEncrypted(const InfoHash& hash, const PkId& to, Sp<Value> val, DoneCallback callback, bool permanent)
{
if (not key_) {
if (callback)
callback(false, {});
return;
}
findPublicKey(to, [this, hash, val = std::move(val), callback = std::move(callback), permanent](const Sp<crypto::PublicKey>& pk) {
if(!pk || !*pk) {
if (callback)
callback(false, {});
return;
}
if (logger_)
logger_->w("Encrypting data for PK: %s", pk->getLongId().toString().c_str());
try {
dht_->put(hash, encrypt(*val, *pk), callback, time_point::max(), permanent);
} catch (const std::exception& e) {
if (logger_)
logger_->e("Error putting encrypted data: %s", e.what());
if (callback)
callback(false, {});
}
});
}
void
SecureDht::putEncrypted(const InfoHash& hash, const crypto::PublicKey& pk, Sp<Value> val, DoneCallback callback, bool permanent)
{
if (not key_) {
if (callback)
callback(false, {});
return;
}
if (logger_)
logger_->w("Encrypting data for PK: %s", pk.getLongId().to_c_str());
try {
dht_->put(hash, encrypt(*val, pk), callback, time_point::max(), permanent);
} catch (const std::exception& e) {
if (logger_)
logger_->e("Error putting encrypted data: %s", e.what());
if (callback)
callback(false, {});
}
}
void
SecureDht::sign(Value& v) const
{
+2 -2
View File
@@ -163,7 +163,7 @@ Value::msgpack_unpack_body(const msgpack::object& o)
throw msgpack::type_error();
crypto::PublicKey new_owner;
new_owner.msgpack_unpack(*rowner);
owner = std::make_shared<const crypto::PublicKey>(std::move(new_owner));
owner = std::make_shared<crypto::PublicKey>(std::move(new_owner));
if (auto rrecipient = findMapValue(*rbody, VALUE_KEY_TO)) {
recipient = rrecipient->as<InfoHash>();
}
@@ -193,7 +193,7 @@ Value::Value(const Json::Value& json)
if (jowner.isString()) {
auto ownerStr = jowner.asString();
auto ownerBlob = std::vector<unsigned char>(ownerStr.begin(), ownerStr.end());
owner = std::make_shared<const crypto::PublicKey>(ownerBlob);
owner = std::make_shared<crypto::PublicKey>(ownerBlob);
}
const auto& jto = json[VALUE_KEY_TO];
if (jto.isString())
+2
View File
@@ -200,6 +200,8 @@ void print_node_info(const dht::NodeInfo& info) {
std::cout << "IPv4 port " << info.bound4 << ", IPv6 port " << info.bound6 << std::endl;
if (info.id)
std::cout << "Public key ID " << info.id << std::endl;
if (info.pkid)
std::cout << "Public key long ID " << info.pkid << std::endl;
}
static const constexpr struct option long_options[] = {