mirror of
https://github.com/savoirfairelinux/opendht.git
synced 2026-11-08 20:37:23 +00:00
Compare commits
6
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
694f421a4f | ||
|
|
a178a8f820 | ||
|
|
ec359d50fb | ||
|
|
3c5aeccd53 | ||
|
|
f99fe16a46 | ||
|
|
43dc7cf222 |
No files matched your search
@@ -78,7 +78,6 @@ struct OPENDHT_PUBLIC NodeStats {
|
||||
|
||||
struct OPENDHT_PUBLIC NodeInfo {
|
||||
InfoHash id;
|
||||
PkId pkid;
|
||||
InfoHash node_id;
|
||||
NodeStats ipv4 {};
|
||||
NodeStats ipv6 {};
|
||||
@@ -95,7 +94,7 @@ struct OPENDHT_PUBLIC NodeInfo {
|
||||
explicit NodeInfo(const Json::Value& v);
|
||||
#endif
|
||||
|
||||
MSGPACK_DEFINE_MAP(id, pkid, node_id, ipv4, ipv6)
|
||||
MSGPACK_DEFINE_MAP(id, node_id, ipv4, ipv6)
|
||||
};
|
||||
|
||||
/**
|
||||
@@ -168,7 +167,6 @@ 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);
|
||||
|
||||
@@ -66,11 +66,8 @@ private:
|
||||
|
||||
public:
|
||||
virtual void unpackValue(const Value& v) override {
|
||||
if (v.owner) {
|
||||
owner = v.owner;
|
||||
if (v.owner)
|
||||
from = v.owner->getId();
|
||||
fromId = v.owner->getLongId();
|
||||
}
|
||||
BaseClass::unpackValue(v);
|
||||
}
|
||||
|
||||
@@ -78,9 +75,7 @@ public:
|
||||
return [](const Value& v){ return v.isSigned(); };
|
||||
}
|
||||
|
||||
Sp<crypto::PublicKey> owner;
|
||||
dht::InfoHash from;
|
||||
dht::PkId fromId;
|
||||
};
|
||||
|
||||
template <typename T>
|
||||
|
||||
@@ -95,7 +95,7 @@ public:
|
||||
/**
|
||||
* Performs final operations before quitting.
|
||||
*/
|
||||
void shutdown(ShutdownCallback cb) override;
|
||||
void shutdown(ShutdownCallback cb, bool stop = false) override;
|
||||
|
||||
/**
|
||||
* Returns true if the node is running (have access to an open socket).
|
||||
|
||||
@@ -61,8 +61,10 @@ public:
|
||||
|
||||
/**
|
||||
* Performs final operations before quitting.
|
||||
* stop: if true, cancel ongoing operations and call their 'done'
|
||||
* callbacks synchronously.
|
||||
*/
|
||||
virtual void shutdown(ShutdownCallback cb) = 0;
|
||||
virtual void shutdown(ShutdownCallback cb, bool stop = false) = 0;
|
||||
|
||||
/**
|
||||
* Returns true if the node is running (have access to an open socket).
|
||||
|
||||
@@ -85,7 +85,7 @@ public:
|
||||
/**
|
||||
* Performs final operations before quitting.
|
||||
*/
|
||||
void shutdown(ShutdownCallback cb) override;
|
||||
void shutdown(ShutdownCallback cb, bool) override;
|
||||
|
||||
/**
|
||||
* Returns true if the node is running (have access to an open socket).
|
||||
|
||||
@@ -74,7 +74,6 @@ public:
|
||||
std::shared_ptr<PeerDiscovery> peerDiscovery {};
|
||||
StatusCallback statusChangedCallback {};
|
||||
CertificateStoreQuery certificateStore {};
|
||||
CertificateStoreLongQuery certificateStoreSha256 {};
|
||||
IdentityAnnouncedCb identityAnnouncedCb {};
|
||||
Context() {}
|
||||
};
|
||||
@@ -258,27 +257,6 @@ 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.
|
||||
@@ -325,7 +303,6 @@ 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.
|
||||
@@ -390,7 +367,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, CertificateStoreLongQuery&& query_sha256_method = {});
|
||||
void setLocalCertificateStore(CertificateStoreQuery&& query_method);
|
||||
|
||||
/**
|
||||
* @param port: Local port to bind. Both IPv4 and IPv6 will be tried (ANY).
|
||||
@@ -431,7 +408,7 @@ public:
|
||||
/**
|
||||
* Gracefuly disconnect from network.
|
||||
*/
|
||||
void shutdown(ShutdownCallback cb = {});
|
||||
void shutdown(ShutdownCallback cb = {}, bool stop = false);
|
||||
|
||||
/**
|
||||
* Quit and wait for all threads to terminate.
|
||||
|
||||
@@ -115,14 +115,6 @@ 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.
|
||||
@@ -134,35 +126,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<crypto::PublicKey>)>& cb);
|
||||
void findPublicKey(const InfoHash& node, const std::function<void(const Sp<const crypto::PublicKey>)>& cb);
|
||||
|
||||
Sp<crypto::Certificate> registerCertificate(const InfoHash& node, const Blob& cert);
|
||||
const Sp<crypto::Certificate> registerCertificate(const InfoHash& node, const Blob& cert);
|
||||
void registerCertificate(Sp<crypto::Certificate>& cert);
|
||||
|
||||
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;
|
||||
const Sp<crypto::Certificate> getCertificate(const InfoHash& node) const;
|
||||
const Sp<const crypto::PublicKey> getPublicKey(const InfoHash& 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, CertificateStoreLongQuery&& query_sha256_method) {
|
||||
void setLocalCertificateStore(CertificateStoreQuery&& query_method) {
|
||||
localQueryMethod_ = std::move(query_method);
|
||||
localQuerySha256Method_ = std::move(query_sha256_method);
|
||||
}
|
||||
|
||||
/**
|
||||
* SecureDht to Dht proxy
|
||||
*/
|
||||
void shutdown(ShutdownCallback cb) override {
|
||||
dht_->shutdown(cb);
|
||||
void shutdown(ShutdownCallback cb, bool stop = false) override {
|
||||
dht_->shutdown(cb, stop);
|
||||
}
|
||||
void dumpTables() const override {
|
||||
dht_->dumpTables();
|
||||
@@ -366,13 +351,10 @@ 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<crypto::PublicKey>> nodesPubKeys_ {};
|
||||
std::map<PkId, Sp<crypto::Certificate>> nodesCertificatesLong_ {};
|
||||
std::map<PkId, Sp<crypto::PublicKey>> nodesPubKeysLong_ {};
|
||||
std::map<InfoHash, Sp<const crypto::PublicKey>> nodesPubKeys_ {};
|
||||
|
||||
std::atomic_bool forward_all_ {false};
|
||||
bool enableCache_ {false};
|
||||
|
||||
@@ -68,7 +68,7 @@ public:
|
||||
* Build from existing address.
|
||||
*/
|
||||
SockAddr(const sockaddr* sa, socklen_t length) {
|
||||
if (length > (socklen_t)sizeof(sockaddr_storage))
|
||||
if (length > static_cast<socklen_t>(sizeof(sockaddr_storage)))
|
||||
throw std::runtime_error("Socket address length is too large");
|
||||
set(sa, length);
|
||||
}
|
||||
|
||||
@@ -93,4 +93,71 @@ private:
|
||||
void schedule();
|
||||
};
|
||||
|
||||
class OPENDHT_PUBLIC ExecutionContext {
|
||||
public:
|
||||
ExecutionContext(ThreadPool& pool)
|
||||
: threadPool_(pool), state_(std::make_shared<SharedState>())
|
||||
{}
|
||||
|
||||
~ExecutionContext() {
|
||||
state_->destroy();
|
||||
}
|
||||
|
||||
/** Wait for ongoing tasks to complete execution and drop other pending tasks */
|
||||
void stop() {
|
||||
state_->destroy(false);
|
||||
}
|
||||
|
||||
void run(std::function<void()>&& task) {
|
||||
std::lock_guard<std::mutex> lock(state_->mtx);
|
||||
if (state_->shutdown_) return;
|
||||
state_->pendingTasks++;
|
||||
threadPool_.get().run([task = std::move(task), state = state_] {
|
||||
state->run(task);
|
||||
});
|
||||
}
|
||||
|
||||
private:
|
||||
struct SharedState {
|
||||
std::mutex mtx {};
|
||||
std::condition_variable cv {};
|
||||
unsigned pendingTasks {0};
|
||||
unsigned ongoingTasks {0};
|
||||
/** When true, prevents new tasks to be scheduled */
|
||||
bool shutdown_ {false};
|
||||
/** When true, prevents scheduled tasks to be executed */
|
||||
std::atomic_bool destroyed {false};
|
||||
|
||||
void destroy(bool wait = true) {
|
||||
std::unique_lock<std::mutex> lock(mtx);
|
||||
if (destroyed) return;
|
||||
if (wait) {
|
||||
cv.wait(lock, [this] { return pendingTasks == 0 && ongoingTasks == 0; });
|
||||
}
|
||||
shutdown_ = true;
|
||||
if (not wait) {
|
||||
cv.wait(lock, [this] { return ongoingTasks == 0; });
|
||||
}
|
||||
destroyed = true;
|
||||
}
|
||||
|
||||
void run(const std::function<void()>& task) {
|
||||
{
|
||||
std::lock_guard<std::mutex> lock(mtx);
|
||||
pendingTasks--;
|
||||
ongoingTasks++;
|
||||
}
|
||||
if (destroyed) return;
|
||||
task();
|
||||
{
|
||||
std::lock_guard<std::mutex> lock(mtx);
|
||||
ongoingTasks--;
|
||||
cv.notify_all();
|
||||
}
|
||||
}
|
||||
};
|
||||
std::reference_wrapper<ThreadPool> threadPool_;
|
||||
std::shared_ptr<SharedState> state_;
|
||||
};
|
||||
|
||||
}
|
||||
@@ -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<crypto::PublicKey>(key.getPublicKey());
|
||||
owner = std::make_shared<const 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<crypto::PublicKey> getOwner() const {
|
||||
return std::static_pointer_cast<crypto::PublicKey>(owner);
|
||||
std::shared_ptr<const crypto::PublicKey> getOwner() const {
|
||||
return std::static_pointer_cast<const crypto::PublicKey>(owner);
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -594,7 +594,7 @@ struct OPENDHT_PUBLIC Value
|
||||
/**
|
||||
* Public key of the signer.
|
||||
*/
|
||||
std::shared_ptr<crypto::PublicKey> owner {};
|
||||
std::shared_ptr<const crypto::PublicKey> owner {};
|
||||
|
||||
/**
|
||||
* Hash of the recipient (optional).
|
||||
|
||||
@@ -141,8 +141,6 @@ 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();
|
||||
@@ -154,8 +152,6 @@ 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"]);
|
||||
|
||||
+17
-1
@@ -77,11 +77,27 @@ Dht::Kad::getStatus(time_point now) const
|
||||
}
|
||||
|
||||
void
|
||||
Dht::shutdown(ShutdownCallback cb)
|
||||
Dht::shutdown(ShutdownCallback cb, bool stop)
|
||||
{
|
||||
if (not persistPath.empty())
|
||||
saveState(persistPath);
|
||||
|
||||
if (stop) {
|
||||
for (auto dht : {&dht4, &dht6}) {
|
||||
for (auto& sr : dht->searches) {
|
||||
for (const auto& r : sr.second->callbacks)
|
||||
r.second.done_cb(false, {});
|
||||
sr.second->callbacks.clear();
|
||||
for (const auto& a : sr.second->announce) {
|
||||
if (a.callback) a.callback(false, {});
|
||||
}
|
||||
sr.second->announce.clear();
|
||||
sr.second->listeners.clear();
|
||||
}
|
||||
}
|
||||
network_engine.clear();
|
||||
}
|
||||
|
||||
if (not maintain_storage) {
|
||||
if (cb) cb();
|
||||
return;
|
||||
|
||||
@@ -242,7 +242,7 @@ DhtProxyClient::cancelAllListeners()
|
||||
}
|
||||
|
||||
void
|
||||
DhtProxyClient::shutdown(ShutdownCallback cb)
|
||||
DhtProxyClient::shutdown(ShutdownCallback cb, bool)
|
||||
{
|
||||
stop();
|
||||
if (cb)
|
||||
|
||||
+50
-93
@@ -119,22 +119,24 @@ DhtRunner::run(const Config& config, Context&& context)
|
||||
local6.setFamily(AF_INET6);
|
||||
}
|
||||
auto state_path = config.dht_config.node_config.persist_path;
|
||||
if (not state_path.empty()) {
|
||||
if (not state_path.empty() && (local4.getPort() == 0 || local6.getPort() == 0)) {
|
||||
state_path += "_port.txt";
|
||||
std::ifstream inConfig(state_path);
|
||||
if (inConfig.is_open()) {
|
||||
in_port_t port;
|
||||
if (inConfig >> port) {
|
||||
if (context.logger)
|
||||
context.logger->d("[runner %p] Using IPv4 port %hu from saved configuration", this, port);
|
||||
if (local4.getPort() == 0)
|
||||
if (local4.getPort() == 0) {
|
||||
if (context.logger)
|
||||
context.logger->d("[runner %p] Using IPv4 port %hu from saved configuration", this, port);
|
||||
local4.setPort(port);
|
||||
}
|
||||
}
|
||||
if (inConfig >> port) {
|
||||
if (context.logger)
|
||||
context.logger->d("[runner %p] Using IPv6 port %hu from saved configuration", this, port);
|
||||
if (local6.getPort() == 0)
|
||||
if (local6.getPort() == 0) {
|
||||
if (context.logger)
|
||||
context.logger->d("[runner %p] Using IPv6 port %hu from saved configuration", this, port);
|
||||
local6.setPort(port);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -185,10 +187,10 @@ DhtRunner::run(const Config& config, Context&& context)
|
||||
if (context.statusChangedCallback) {
|
||||
statusCb = std::move(context.statusChangedCallback);
|
||||
}
|
||||
if (context.certificateStore || context.certificateStoreSha256) {
|
||||
dht_->setLocalCertificateStore(std::move(context.certificateStore), std::move(context.certificateStoreSha256));
|
||||
if (context.certificateStore) {
|
||||
dht_->setLocalCertificateStore(std::move(context.certificateStore));
|
||||
if (dht_via_proxy_)
|
||||
dht_via_proxy_->setLocalCertificateStore(std::move(context.certificateStore), std::move(context.certificateStoreSha256));
|
||||
dht_via_proxy_->setLocalCertificateStore(std::move(context.certificateStore));
|
||||
}
|
||||
|
||||
if (not config.threaded)
|
||||
@@ -266,29 +268,31 @@ DhtRunner::run(const Config& config, Context&& context)
|
||||
}
|
||||
|
||||
void
|
||||
DhtRunner::shutdown(ShutdownCallback cb) {
|
||||
DhtRunner::shutdown(ShutdownCallback cb, bool stop) {
|
||||
std::unique_lock<std::mutex> lck(storage_mtx);
|
||||
auto expected = State::Running;
|
||||
if (not running.compare_exchange_strong(expected, State::Stopping)) {
|
||||
if (expected == State::Stopping and ongoing_ops) {
|
||||
std::lock_guard<std::mutex> lck(storage_mtx);
|
||||
shutdownCallbacks_.emplace_back(std::move(cb));
|
||||
}
|
||||
else if (cb) cb();
|
||||
else if (cb) {
|
||||
lck.unlock();
|
||||
cb();
|
||||
}
|
||||
return;
|
||||
}
|
||||
if (logger_)
|
||||
logger_->d("[runner %p] state changed to Stopping, %zu ongoing ops", this, ongoing_ops.load());
|
||||
std::lock_guard<std::mutex> lck(storage_mtx);
|
||||
ongoing_ops++;
|
||||
shutdownCallbacks_.emplace_back(std::move(cb));
|
||||
pending_ops_prio.emplace([=](SecureDht&) mutable {
|
||||
pending_ops.emplace([=](SecureDht&) mutable {
|
||||
auto onShutdown = [this]{ opEnded(); };
|
||||
#ifdef OPENDHT_PROXY_CLIENT
|
||||
if (dht_via_proxy_)
|
||||
dht_via_proxy_->shutdown(onShutdown);
|
||||
dht_via_proxy_->shutdown(onShutdown, stop);
|
||||
#endif
|
||||
if (dht_)
|
||||
dht_->shutdown(onShutdown);
|
||||
dht_->shutdown(onShutdown, stop);
|
||||
});
|
||||
cv.notify_all();
|
||||
}
|
||||
@@ -317,11 +321,11 @@ DhtRunner::bindOpDoneCallback(DoneCallbackSimple&& cb) {
|
||||
|
||||
bool
|
||||
DhtRunner::checkShutdown() {
|
||||
if (running != State::Stopping or ongoing_ops)
|
||||
return false;
|
||||
decltype(shutdownCallbacks_) cbs;
|
||||
{
|
||||
std::lock_guard<std::mutex> lck(storage_mtx);
|
||||
if (running != State::Stopping or ongoing_ops)
|
||||
return false;
|
||||
cbs = std::move(shutdownCallbacks_);
|
||||
}
|
||||
for (auto& cb : cbs)
|
||||
@@ -353,9 +357,13 @@ DhtRunner::join()
|
||||
|
||||
{
|
||||
std::lock_guard<std::mutex> lck(storage_mtx);
|
||||
if (ongoing_ops and logger_) {
|
||||
logger_->w("[runner %p] stopping with %zu remaining ops", this, ongoing_ops.load());
|
||||
}
|
||||
pending_ops = decltype(pending_ops)();
|
||||
pending_ops_prio = decltype(pending_ops_prio)();
|
||||
ongoing_ops = 0;
|
||||
shutdownCallbacks_.clear();
|
||||
}
|
||||
{
|
||||
std::lock_guard<std::mutex> lck(dht_mtx);
|
||||
@@ -398,15 +406,6 @@ DhtRunner::getId() const
|
||||
return {};
|
||||
}
|
||||
|
||||
PkId
|
||||
DhtRunner::getLongId() const
|
||||
{
|
||||
if (auto dht = activeDht())
|
||||
return dht->getLongId();
|
||||
return {};
|
||||
}
|
||||
|
||||
|
||||
InfoHash
|
||||
DhtRunner::getNodeId() const
|
||||
{
|
||||
@@ -518,7 +517,6 @@ 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);
|
||||
@@ -540,7 +538,6 @@ 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);
|
||||
@@ -627,14 +624,14 @@ DhtRunner::registerCertificate(std::shared_ptr<crypto::Certificate> cert) {
|
||||
activeDht()->registerCertificate(cert);
|
||||
}
|
||||
void
|
||||
DhtRunner::setLocalCertificateStore(CertificateStoreQuery&& query_method, CertificateStoreLongQuery&& query_sha256_method) {
|
||||
DhtRunner::setLocalCertificateStore(CertificateStoreQuery&& query_method) {
|
||||
std::lock_guard<std::mutex> lck(dht_mtx);
|
||||
#ifdef OPENDHT_PROXY_CLIENT
|
||||
if (dht_via_proxy_)
|
||||
dht_via_proxy_->setLocalCertificateStore(std::move(query_method), std::move(query_sha256_method));
|
||||
dht_via_proxy_->setLocalCertificateStore(std::forward<CertificateStoreQuery>(query_method));
|
||||
#endif
|
||||
if (dht_)
|
||||
dht_->setLocalCertificateStore(std::move(query_method), std::move(query_sha256_method));
|
||||
dht_->setLocalCertificateStore(std::forward<CertificateStoreQuery>(query_method));
|
||||
}
|
||||
|
||||
time_point
|
||||
@@ -718,11 +715,12 @@ DhtRunner::loop_()
|
||||
void
|
||||
DhtRunner::get(InfoHash hash, GetCallback vcb, DoneCallback dcb, Value::Filter f, Where w)
|
||||
{
|
||||
std::unique_lock<std::mutex> lck(storage_mtx);
|
||||
if (running != State::Running) {
|
||||
lck.unlock();
|
||||
if (dcb) dcb(false, {});
|
||||
return;
|
||||
}
|
||||
std::lock_guard<std::mutex> lck(storage_mtx);
|
||||
ongoing_ops++;
|
||||
pending_ops.emplace([=](SecureDht& dht) mutable {
|
||||
dht.get(hash, std::move(vcb), bindOpDoneCallback(std::move(dcb)), std::move(f), std::move(w));
|
||||
@@ -737,11 +735,12 @@ DhtRunner::get(const std::string& key, GetCallback vcb, DoneCallbackSimple dcb,
|
||||
}
|
||||
void
|
||||
DhtRunner::query(const InfoHash& hash, QueryCallback cb, DoneCallback done_cb, Query q) {
|
||||
std::unique_lock<std::mutex> lck(storage_mtx);
|
||||
if (running != State::Running) {
|
||||
lck.unlock();
|
||||
if (done_cb) done_cb(false, {});
|
||||
return;
|
||||
}
|
||||
std::lock_guard<std::mutex> lck(storage_mtx);
|
||||
ongoing_ops++;
|
||||
pending_ops.emplace([=](SecureDht& dht) mutable {
|
||||
dht.query(hash, std::move(cb), bindOpDoneCallback(std::move(done_cb)), std::move(q));
|
||||
@@ -753,11 +752,12 @@ std::future<size_t>
|
||||
DhtRunner::listen(InfoHash hash, ValueCallback vcb, Value::Filter f, Where w)
|
||||
{
|
||||
auto ret_token = std::make_shared<std::promise<size_t>>();
|
||||
std::unique_lock<std::mutex> lck(storage_mtx);
|
||||
if (running != State::Running) {
|
||||
lck.unlock();
|
||||
ret_token->set_value(0);
|
||||
return ret_token->get_future();
|
||||
}
|
||||
std::lock_guard<std::mutex> lck(storage_mtx);
|
||||
pending_ops.emplace([=](SecureDht& dht) mutable {
|
||||
#ifdef OPENDHT_PROXY_CLIENT
|
||||
auto tokenbGlobal = listener_token_++;
|
||||
@@ -838,11 +838,12 @@ DhtRunner::cancelListen(InfoHash h, std::shared_future<size_t> ftoken)
|
||||
void
|
||||
DhtRunner::put(InfoHash hash, Value&& value, DoneCallback cb, time_point created, bool permanent)
|
||||
{
|
||||
std::unique_lock<std::mutex> lck(storage_mtx);
|
||||
if (running != State::Running) {
|
||||
lck.unlock();
|
||||
if (cb) cb(false, {});
|
||||
return;
|
||||
}
|
||||
std::lock_guard<std::mutex> lck(storage_mtx);
|
||||
ongoing_ops++;
|
||||
pending_ops.emplace([=,
|
||||
cb = std::move(cb),
|
||||
@@ -856,11 +857,12 @@ DhtRunner::put(InfoHash hash, Value&& value, DoneCallback cb, time_point created
|
||||
void
|
||||
DhtRunner::put(InfoHash hash, std::shared_ptr<Value> value, DoneCallback cb, time_point created, bool permanent)
|
||||
{
|
||||
std::unique_lock<std::mutex> lck(storage_mtx);
|
||||
if (running != State::Running) {
|
||||
lck.unlock();
|
||||
if (cb) cb(false, {});
|
||||
return;
|
||||
}
|
||||
std::lock_guard<std::mutex> lck(storage_mtx);
|
||||
ongoing_ops++;
|
||||
pending_ops.emplace([=, cb = std::move(cb)](SecureDht& dht) mutable {
|
||||
dht.put(hash, value, bindOpDoneCallback(std::move(cb)), created, permanent);
|
||||
@@ -897,11 +899,12 @@ DhtRunner::cancelPut(const InfoHash& h, const std::shared_ptr<Value>& value)
|
||||
void
|
||||
DhtRunner::putSigned(InfoHash hash, std::shared_ptr<Value> value, DoneCallback cb, bool permanent)
|
||||
{
|
||||
std::unique_lock<std::mutex> lck(storage_mtx);
|
||||
if (running != State::Running) {
|
||||
lck.unlock();
|
||||
if (cb) cb(false, {});
|
||||
return;
|
||||
}
|
||||
std::lock_guard<std::mutex> lck(storage_mtx);
|
||||
ongoing_ops++;
|
||||
pending_ops.emplace([=,
|
||||
cb = std::move(cb),
|
||||
@@ -927,11 +930,12 @@ DhtRunner::putSigned(const std::string& key, Value&& value, DoneCallbackSimple c
|
||||
void
|
||||
DhtRunner::putEncrypted(InfoHash hash, InfoHash to, std::shared_ptr<Value> value, DoneCallback cb, bool permanent)
|
||||
{
|
||||
std::unique_lock<std::mutex> lck(storage_mtx);
|
||||
if (running != State::Running) {
|
||||
lck.unlock();
|
||||
if (cb) cb(false, {});
|
||||
return;
|
||||
}
|
||||
std::lock_guard<std::mutex> lck(storage_mtx);
|
||||
ongoing_ops++;
|
||||
pending_ops.emplace([=,
|
||||
cb = std::move(cb),
|
||||
@@ -954,55 +958,6 @@ 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)
|
||||
{
|
||||
@@ -1066,11 +1021,12 @@ DhtRunner::bootstrap(std::vector<SockAddr> nodes, DoneCallbackSimple&& cb)
|
||||
void
|
||||
DhtRunner::bootstrap(const SockAddr& addr, DoneCallbackSimple&& cb)
|
||||
{
|
||||
std::unique_lock<std::mutex> lck(storage_mtx);
|
||||
if (running != State::Running) {
|
||||
lck.unlock();
|
||||
if (cb) cb(false);
|
||||
return;
|
||||
}
|
||||
std::lock_guard<std::mutex> lck(storage_mtx);
|
||||
ongoing_ops++;
|
||||
pending_ops_prio.emplace([addr, cb = bindOpDoneCallback(std::move(cb))](SecureDht& dht) mutable {
|
||||
dht.pingNode(std::move(addr), std::move(cb));
|
||||
@@ -1081,9 +1037,9 @@ DhtRunner::bootstrap(const SockAddr& addr, DoneCallbackSimple&& cb)
|
||||
void
|
||||
DhtRunner::bootstrap(const InfoHash& id, const SockAddr& address)
|
||||
{
|
||||
std::lock_guard<std::mutex> lck(storage_mtx);
|
||||
if (running != State::Running)
|
||||
return;
|
||||
std::unique_lock<std::mutex> lck(storage_mtx);
|
||||
pending_ops_prio.emplace([id, address](SecureDht& dht) mutable {
|
||||
dht.insertNode(id, address);
|
||||
});
|
||||
@@ -1093,9 +1049,9 @@ DhtRunner::bootstrap(const InfoHash& id, const SockAddr& address)
|
||||
void
|
||||
DhtRunner::bootstrap(const std::vector<NodeExport>& nodes)
|
||||
{
|
||||
std::lock_guard<std::mutex> lck(storage_mtx);
|
||||
if (running != State::Running)
|
||||
return;
|
||||
std::lock_guard<std::mutex> lck(storage_mtx);
|
||||
pending_ops_prio.emplace([=](SecureDht& dht) {
|
||||
for (auto& node : nodes)
|
||||
dht.insertNode(node);
|
||||
@@ -1119,11 +1075,12 @@ DhtRunner::connectivityChanged()
|
||||
|
||||
void
|
||||
DhtRunner::findCertificate(InfoHash hash, std::function<void(const Sp<crypto::Certificate>&)> cb) {
|
||||
std::unique_lock<std::mutex> lck(storage_mtx);
|
||||
if (running != State::Running) {
|
||||
lck.unlock();
|
||||
cb({});
|
||||
return;
|
||||
}
|
||||
std::lock_guard<std::mutex> lck(storage_mtx);
|
||||
ongoing_ops++;
|
||||
pending_ops.emplace([this, hash, cb = std::move(cb)] (SecureDht& dht) {
|
||||
dht.findCertificate(hash, [this, cb = std::move(cb)](const Sp<crypto::Certificate>& crt){
|
||||
|
||||
+17
-167
@@ -118,7 +118,7 @@ SecureDht::secureType(ValueType&& type)
|
||||
return type;
|
||||
}
|
||||
|
||||
Sp<crypto::Certificate>
|
||||
const 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;
|
||||
}
|
||||
|
||||
Sp<crypto::PublicKey>
|
||||
const Sp<const crypto::PublicKey>
|
||||
SecureDht::getPublicKey(const InfoHash& node) const
|
||||
{
|
||||
if (node == getId())
|
||||
@@ -142,7 +142,7 @@ SecureDht::getPublicKey(const InfoHash& node) const
|
||||
return it->second;
|
||||
}
|
||||
|
||||
Sp<crypto::Certificate>
|
||||
const Sp<crypto::Certificate>
|
||||
SecureDht::registerCertificate(const InfoHash& node, const Blob& data)
|
||||
{
|
||||
Sp<crypto::Certificate> crt;
|
||||
@@ -155,9 +155,12 @@ SecureDht::registerCertificate(const InfoHash& node, const Blob& data)
|
||||
if (node == h) {
|
||||
if (logger_)
|
||||
logger_->d("Registering certificate for %s", h.toString().c_str());
|
||||
nodesCertificates_[h] = crt;
|
||||
nodesCertificatesLong_[crt->getLongId()] = crt;
|
||||
return crt;
|
||||
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;
|
||||
} else {
|
||||
if (logger_)
|
||||
logger_->w("Certificate %s for node %s does not match node id !", h.toString().c_str(), node.toString().c_str());
|
||||
@@ -168,10 +171,8 @@ 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
|
||||
@@ -191,7 +192,6 @@ 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,6 +200,8 @@ 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;
|
||||
@@ -210,7 +212,7 @@ SecureDht::findCertificate(const InfoHash& node, const std::function<void(const
|
||||
return false;
|
||||
}
|
||||
}
|
||||
return !*found;
|
||||
return true;
|
||||
}, [cb,found](bool) {
|
||||
if (!*found and cb)
|
||||
cb(nullptr);
|
||||
@@ -218,7 +220,7 @@ SecureDht::findCertificate(const InfoHash& node, const std::function<void(const
|
||||
}
|
||||
|
||||
void
|
||||
SecureDht::findPublicKey(const InfoHash& node, const std::function<void(const Sp<crypto::PublicKey>)>& cb)
|
||||
SecureDht::findPublicKey(const InfoHash& node, const std::function<void(const Sp<const crypto::PublicKey>)>& cb)
|
||||
{
|
||||
auto pk = getPublicKey(node);
|
||||
if (pk && *pk) {
|
||||
@@ -233,7 +235,6 @@ 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;
|
||||
}
|
||||
@@ -242,106 +243,6 @@ 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)
|
||||
{
|
||||
@@ -361,10 +262,8 @@ 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;
|
||||
}
|
||||
@@ -382,10 +281,8 @@ 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_)
|
||||
@@ -508,7 +405,7 @@ SecureDht::putEncrypted(const InfoHash& hash, const InfoHash& to, Sp<Value> val,
|
||||
callback(false, {});
|
||||
return;
|
||||
}
|
||||
findPublicKey(to, [this, hash, val = std::move(val), callback = std::move(callback), permanent](const Sp<crypto::PublicKey>& pk) {
|
||||
findPublicKey(to, [=](const Sp<const crypto::PublicKey>& pk) {
|
||||
if(!pk || !*pk) {
|
||||
if (callback)
|
||||
callback(false, {});
|
||||
@@ -527,53 +424,6 @@ 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
|
||||
{
|
||||
|
||||
+5
-5
@@ -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<crypto::PublicKey>(std::move(new_owner));
|
||||
owner = std::make_shared<const 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<crypto::PublicKey>(ownerBlob);
|
||||
owner = std::make_shared<const crypto::PublicKey>(ownerBlob);
|
||||
}
|
||||
const auto& jto = json[VALUE_KEY_TO];
|
||||
if (jto.isString())
|
||||
@@ -496,10 +496,10 @@ Query::msgpack_unpack(const msgpack::object& o)
|
||||
}
|
||||
|
||||
template <typename T>
|
||||
bool subset(std::vector<T> fds, std::vector<T> qfds)
|
||||
bool subset(const std::vector<T>& fds, const std::vector<T>& qfds)
|
||||
{
|
||||
for (auto& fd : fds) {
|
||||
if (std::find_if(qfds.begin(), qfds.end(), [&fd](T& _vfd) { return fd == _vfd; }) == qfds.end())
|
||||
for (const auto& fd : fds) {
|
||||
if (std::find_if(qfds.begin(), qfds.end(), [&fd](const T& _vfd) { return fd == _vfd; }) == qfds.end())
|
||||
return false;
|
||||
}
|
||||
return true;
|
||||
|
||||
@@ -71,7 +71,7 @@ DhtProxyTester::tearDown() {
|
||||
cv.notify_all();
|
||||
});
|
||||
std::unique_lock<std::mutex> lk(cv_m);
|
||||
CPPUNIT_ASSERT(cv.wait_for(lk, 5s, [&]{ return done; }));
|
||||
CPPUNIT_ASSERT(cv.wait_for(lk, 15s, [&]{ return done; }));
|
||||
serverProxy.reset();
|
||||
nodeProxy.reset();
|
||||
}
|
||||
@@ -289,4 +289,47 @@ DhtProxyTester::testFuzzy()
|
||||
CPPUNIT_ASSERT(value->data == mtu);
|
||||
}
|
||||
|
||||
void
|
||||
DhtProxyTester::testShutdownStop()
|
||||
{
|
||||
constexpr size_t N = 40000;
|
||||
constexpr unsigned C = 100;
|
||||
|
||||
// Arrange
|
||||
auto key = dht::InfoHash::get("testShutdownStop");
|
||||
std::vector<std::shared_ptr<dht::Value>> values;
|
||||
std::vector<uint8_t> mtu;
|
||||
mtu.reserve(N);
|
||||
for (size_t i = 0; i < N; i++)
|
||||
mtu.emplace_back((i % 2) ? 'T' : 'M');
|
||||
|
||||
std::atomic_uint callback_count {0};
|
||||
|
||||
// Act
|
||||
for (size_t i = 0; i < C; i++) {
|
||||
auto nodeTest = std::make_shared<dht::DhtRunner>();
|
||||
nodeTest->run(0, clientConfig);
|
||||
nodeTest->put(key, dht::Value(mtu), [&](bool ok) {
|
||||
callback_count++;
|
||||
});
|
||||
nodeTest->get(key, [&](const std::vector<std::shared_ptr<dht::Value>>& vals){
|
||||
values.insert(values.end(), vals.begin(), vals.end());
|
||||
return true;
|
||||
},[&](bool ok){
|
||||
callback_count++;
|
||||
});
|
||||
bool done = false;
|
||||
std::condition_variable cv;
|
||||
std::mutex cv_m;
|
||||
nodeTest->shutdown([&]{
|
||||
std::lock_guard<std::mutex> lk(cv_m);
|
||||
done = true;
|
||||
cv.notify_all();
|
||||
}, true);
|
||||
std::unique_lock<std::mutex> lk(cv_m);
|
||||
CPPUNIT_ASSERT(cv.wait_for(lk, 10s, [&]{ return done; }));
|
||||
}
|
||||
CPPUNIT_ASSERT_EQUAL(2*C, callback_count.load());
|
||||
}
|
||||
|
||||
} // namespace test
|
||||
@@ -36,6 +36,7 @@ class DhtProxyTester : public CppUnit::TestFixture {
|
||||
CPPUNIT_TEST(testResubscribeGetValues);
|
||||
CPPUNIT_TEST(testPutGet40KChars);
|
||||
CPPUNIT_TEST(testFuzzy);
|
||||
CPPUNIT_TEST(testShutdownStop);
|
||||
CPPUNIT_TEST_SUITE_END();
|
||||
|
||||
public:
|
||||
@@ -68,6 +69,8 @@ class DhtProxyTester : public CppUnit::TestFixture {
|
||||
|
||||
void testFuzzy();
|
||||
|
||||
void testShutdownStop();
|
||||
|
||||
private:
|
||||
dht::DhtRunner::Config clientConfig {};
|
||||
dht::DhtRunner nodePeer;
|
||||
|
||||
@@ -59,7 +59,7 @@ ThreadPoolTester::testExecutor()
|
||||
auto executor8 = std::make_shared<dht::Executor>(pool, 8);
|
||||
|
||||
constexpr unsigned N = 64 * 1024;
|
||||
std::atomic_uint count1 {0};
|
||||
unsigned count1 {0};
|
||||
std::atomic_uint count4 {0};
|
||||
std::atomic_uint count8 {0};
|
||||
for (unsigned i=0; i<N; i++) {
|
||||
@@ -69,7 +69,7 @@ ThreadPoolTester::testExecutor()
|
||||
}
|
||||
|
||||
auto start = clock::now();
|
||||
while ((count1.load() != N ||
|
||||
while ((count1 != N ||
|
||||
count4.load() != N ||
|
||||
count8.load() != N) && clock::now() - start < std::chrono::seconds(20))
|
||||
{
|
||||
@@ -78,11 +78,28 @@ ThreadPoolTester::testExecutor()
|
||||
executor1.reset();
|
||||
executor4.reset();
|
||||
executor8.reset();
|
||||
CPPUNIT_ASSERT_EQUAL(N, count1.load());
|
||||
CPPUNIT_ASSERT_EQUAL(N, count1);
|
||||
CPPUNIT_ASSERT_EQUAL(N, count4.load());
|
||||
CPPUNIT_ASSERT_EQUAL(N, count8.load());
|
||||
}
|
||||
|
||||
void
|
||||
ThreadPoolTester::testContext()
|
||||
{
|
||||
std::atomic_uint count {0};
|
||||
constexpr unsigned N = 64 * 1024;
|
||||
|
||||
{
|
||||
dht::ExecutionContext ctx(dht::ThreadPool::computation());
|
||||
for (unsigned i=0; i<N; i++) {
|
||||
ctx.run([&] { count++; });
|
||||
}
|
||||
}
|
||||
|
||||
CPPUNIT_ASSERT_EQUAL(N, count.load());
|
||||
|
||||
}
|
||||
|
||||
void
|
||||
ThreadPoolTester::tearDown() {
|
||||
}
|
||||
|
||||
@@ -29,6 +29,7 @@ class ThreadPoolTester : public CppUnit::TestFixture {
|
||||
CPPUNIT_TEST_SUITE(ThreadPoolTester);
|
||||
CPPUNIT_TEST(testThreadPool);
|
||||
CPPUNIT_TEST(testExecutor);
|
||||
CPPUNIT_TEST(testContext);
|
||||
CPPUNIT_TEST_SUITE_END();
|
||||
|
||||
public:
|
||||
@@ -43,6 +44,7 @@ class ThreadPoolTester : public CppUnit::TestFixture {
|
||||
|
||||
void testThreadPool();
|
||||
void testExecutor();
|
||||
void testContext();
|
||||
};
|
||||
|
||||
} // namespace test
|
||||
@@ -200,8 +200,6 @@ 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[] = {
|
||||
|
||||
Reference in new issue
Block a user