mirrors/opendht
1
0 Fork 0
mirror of https://github.com/savoirfairelinux/opendht.git synced 2026-11-08 04:21:51 +00:00
Code Issues Packages Projects Releases Wiki Activity
Files
a5bfd3d9331b06015e12db7020751a4c5519a8ab
opendht/src/dht_proxy_server.cpp
T
Cyrille BéraudandCopilot 3ec03ccc0b proxy_server: keep push platform up to date 
The DHT listen callback of a push subscription kept the push type and
topic of the request that created it. Refreshing the subscription
with another platform updated the listener, but values kept being
pushed to the first platform: an Android device that once subscribed
without platform was pushed as an iOS device forever (BadDeviceToken)
and never received values such as connection requests.

Keep the push type and topic in the session context shared with the
callback, and update them on each subscription refresh and when the
state is restored.

Co-authored-by: Copilot (Claude Opus 5.5) <223556219+Copilot@users.noreply.github.com>
2026-10-26 02:55:59 +00:00

1690 lines
69 KiB
C++
Raw Blame History

// Copyright (c) 2014-2026 Savoir-faire Linux Inc.
// SPDX-License-Identifier: MIT
#ifdef HAVE_CONFIG_H
#include "config.h"
#endif
#include "dht_proxy_server.h"
#include "default_types.h"
#include "dhtrunner.h"
#include "base64.h"
#include <msgpack.hpp>
#include <json/json.h>
#include <chrono>
#include <functional>
#include <limits>
#include <iostream>
#include <fstream>
using namespace std::placeholders;
using namespace std::chrono_literals;
using namespace std::literals;
namespace dht {
constexpr char RESP_MSG_JSON_INCORRECT[] = "{\"err:\":\"Incorrect JSON\"}";
constexpr char RESP_MSG_SERVICE_UNAVAILABLE[] = "{\"err\":\"Incorrect DhtRunner\"}";
constexpr char RESP_MSG_INTERNAL_SERVER_ERRROR[] = "{\"err\":\"Internal server error\"}";
constexpr char RESP_MSG_MISSING_PARAMS[] = "{\"err\":\"Missing parameters\"}";
constexpr char RESP_MSG_PUT_FAILED[] = "{\"err\":\"Put failed\"}";
#ifdef OPENDHT_PROXY_SERVER_IDENTITY
constexpr char RESP_MSG_DESTINATION_NOT_FOUND[] = "{\"err\":\"No destination found\"}";
#endif
#ifdef OPENDHT_PUSH_NOTIFICATIONS
constexpr char RESP_MSG_NO_TOKEN[] = "{\"err\":\"No token\"}";
#endif
constexpr const std::chrono::minutes PRINT_STATS_PERIOD {2};
using ResponseByParts = restinio::chunked_output_t;
using ResponseByPartsBuilder = restinio::response_builder_t<ResponseByParts>;
namespace http {
/**
* Session value associated with a connection_id_t key.
*/
struct ListenerSession
{
ListenerSession() = default;
dht::InfoHash hash;
std::future<size_t> token;
std::shared_ptr<restinio::response_builder_t<restinio::chunked_output_t>> response;
};
} // namespace http
class opendht_logger_t
{
public:
opendht_logger_t(std::shared_ptr<Logger> logger = {})
: m_logger(std::move(logger))
{}
template<typename Builder>
void trace(Builder&& /* msg_builder */)
{
/* if (m_logger) m_logger->d("[proxy:server] %s", msg_builder().c_str()); */
}
template<typename Builder>
void info(Builder&& msg_builder)
{
if (m_logger)
m_logger->debug("[proxy:server] {}", msg_builder());
}
template<typename Builder>
void warn(Builder&& msg_builder)
{
if (m_logger)
m_logger->warn("[proxy:server] {}", msg_builder());
}
template<typename Builder>
void error(Builder&& msg_builder)
{
if (m_logger)
m_logger->error("[proxy:server] {}", msg_builder());
}
private:
std::shared_ptr<Logger> m_logger;
};
std::string
DhtProxyServer::ServerStats::toString() const
{
auto ret = fmt::format("Listens: {}, Puts: {}, PushListeners: {}\n"
"Push requests in the last {}: [Android: {}], [iOS: {}], [Unified: {}]\n"
"Requests: {} per second.",
listenCount,
putCount,
pushListenersCount,
print_duration(lastUpdated - serverStartTime),
androidPush.toString(),
iosPush.toString(),
unifiedPush.toString(),
requestRate);
if (nodeInfo) {
auto& ipv4 = nodeInfo->ipv4;
if (ipv4.table_depth > 1)
ret += fmt::format("IPv4 Network estimation: {}\n", ipv4.getNetworkSizeEstimation());
auto& ipv6 = nodeInfo->ipv6;
if (ipv6.table_depth > 1)
ret += fmt::format("IPv6 Network estimation: {}\n", ipv6.getNetworkSizeEstimation());
}
return ret;
}
Json::Value
DhtProxyServer::ServerStats::toJson() const
{
Json::Value result;
result["listenCount"] = static_cast<Json::UInt64>(listenCount);
result["putCount"] = static_cast<Json::UInt64>(putCount);
result["totalPermanentPuts"] = static_cast<Json::UInt64>(totalPermanentPuts);
result["pushListenersCount"] = static_cast<Json::UInt64>(pushListenersCount);
result["serverStartTime"] = static_cast<Json::LargestInt>(to_time_t(serverStartTime));
result["lastUpdated"] = static_cast<Json::LargestInt>(to_time_t(lastUpdated));
result["androidPush"] = androidPush.toJson();
result["iosPush"] = iosPush.toJson();
result["unifiedPush"] = unifiedPush.toJson();
result["requestRate"] = requestRate;
if (nodeInfo)
result["nodeInfo"] = nodeInfo->toJson();
return result;
}
restinio::request_handling_status_t
DhtProxyServer::serverError(restinio::request_t& request)
{
auto response = initHttpResponse(request.create_response(restinio::status_internal_server_error()));
response.set_body(RESP_MSG_INTERNAL_SERVER_ERRROR);
return response.done();
}
// connection listener
class DhtProxyServer::ConnectionListener
{
public:
ConnectionListener() {};
explicit ConnectionListener(std::function<void(restinio::connection_id_t)> onClosed)
: onClosed_(std::move(onClosed)) {};
~ConnectionListener() {};
/**
* Connection state change used to handle Listeners disconnects.
* RESTinio >= 0.5.1 https://github.com/Stiffstream/restinio/issues/28
*/
void state_changed(const restinio::connection_state::notice_t& notice) noexcept;
private:
std::function<void(restinio::connection_id_t)> onClosed_;
};
void
DhtProxyServer::ConnectionListener::state_changed(const restinio::connection_state::notice_t& notice) noexcept
{
if (std::holds_alternative<restinio::connection_state::closed_t>(notice.cause())) {
onClosed_(notice.connection_id());
}
}
void
DhtProxyServer::onConnectionClosed(restinio::connection_id_t id)
{
std::lock_guard lock(lockListener_);
auto it = listeners_.find(id);
if (it != listeners_.end()) {
dht_->cancelListen(it->second.hash, std::move(it->second.token));
listeners_.erase(it);
if (logger_)
logger_->debug("[proxy:server] [connection:{}] listener cancelled, {} still connected",
id,
listeners_.size());
}
}
struct DhtProxyServer::RestRouterTraitsTls : public restinio::default_tls_traits_t
{
using timer_manager_t = restinio::asio_timer_manager_t;
using logger_t = opendht_logger_t;
using request_handler_t = RestRouter;
using connection_state_listener_t = ConnectionListener;
};
struct DhtProxyServer::RestRouterTraits : public restinio::default_traits_t
{
using timer_manager_t = restinio::asio_timer_manager_t;
using logger_t = opendht_logger_t;
using request_handler_t = RestRouter;
using connection_state_listener_t = ConnectionListener;
};
void
DhtProxyServer::PermanentPut::msgpack_unpack(const msgpack::object& o)
{
if (auto cid = findMapValue(o, "cid"sv)) {
clientId = cid->as<std::string>();
}
if (auto exp = findMapValue(o, "exp"sv)) {
expiration = sys_clock::from_time_t(exp->as<time_t>());
}
if (auto token = findMapValue(o, "token"sv)) {
pushToken = token->as<std::string>();
}
if (auto sid = findMapValue(o, "sid"sv)) {
if (not sessionCtx)
sessionCtx = std::make_shared<PushSessionContext>(sid->as<std::string>());
else
sessionCtx->sessionId = sid->as<std::string>();
}
if (auto t = findMapValue(o, "t"sv)) {
type = t->as<PushType>();
}
if (auto val = findMapValue(o, "value"sv)) {
value = std::make_shared<dht::Value>(*val);
}
if (auto top = findMapValue(o, "top"sv)) {
topic = top->as<std::string>();
}
}
#ifdef OPENDHT_PUSH_NOTIFICATIONS
void
DhtProxyServer::Listener::msgpack_unpack(const msgpack::object& o)
{
if (auto cid = findMapValue(o, "cid"sv)) {
clientId = cid->as<std::string>();
}
if (auto exp = findMapValue(o, "exp"sv)) {
expiration = sys_clock::from_time_t(exp->as<time_t>());
}
if (auto sid = findMapValue(o, "sid"sv)) {
if (not sessionCtx)
sessionCtx = std::make_shared<PushSessionContext>(sid->as<std::string>());
else
sessionCtx->sessionId = sid->as<std::string>();
}
if (auto t = findMapValue(o, "t"sv)) {
type = t->as<PushType>();
}
if (auto top = findMapValue(o, "top"sv)) {
topic = top->as<std::string>();
}
}
#endif
DhtProxyServer::DhtProxyServer(const std::shared_ptr<DhtRunner>& dht,
const ProxyServerConfig& config,
const std::shared_ptr<dht::Logger>& logger)
: ioContext_(std::make_shared<asio::io_context>())
, dht_(dht)
, persistPath_(config.persistStatePath)
, logger_(logger)
, printStatsTimer_(std::make_unique<asio::steady_timer>(*ioContext_, 3s))
, serverStartTime_(clock::now())
, connListener_(std::make_shared<ConnectionListener>(
std::bind(&DhtProxyServer::onConnectionClosed, this, std::placeholders::_1)))
, pushServer_(config.pushServer)
, bundleId_(config.bundleId)
{
if (not dht_)
throw std::invalid_argument("A DHT instance must be provided");
if (logger_)
logger_->debug("[proxy:server] [init] running on {:d}", config.port);
if (not pushServer_.empty()) {
#ifdef OPENDHT_PUSH_NOTIFICATIONS
if (logger_)
logger_->debug("[proxy:server] [init] using push server {}", pushServer_);
#else
if (logger_)
logger_->error("[proxy:server] [init] opendht built without push notification support");
#endif
}
jsonBuilder_["commentStyle"] = "None";
jsonBuilder_["indentation"] = "";
if (!pushServer_.empty()) {
// no host delim, assume port only
if (pushServer_.find(":") == std::string::npos)
pushServer_ = "localhost:" + pushServer_;
// define http request destination for push notifications
pushHostPort_ = splitPort(pushServer_);
if (logger_)
logger_->debug("Using push server for notifications: {}:{}", pushHostPort_.first, pushHostPort_.second);
}
if (config.identity.first and config.identity.second) {
asio::error_code ec;
// define tls context
asio::ssl::context tls_context {asio::ssl::context::tlsv13_server};
tls_context.set_options(asio::ssl::context::default_workarounds | asio::ssl::context::no_sslv2
| asio::ssl::context::single_dh_use,
ec);
if (ec)
throw std::runtime_error("Error setting tls context options: " + ec.message());
// add more security options
#ifdef SSL_OP_NO_RENEGOTIATION
SSL_CTX_set_options(tls_context.native_handle(), SSL_OP_NO_RENEGOTIATION); // CVE-2009-3555
#endif
// node private key
auto key = config.identity.first->serialize();
tls_context.use_private_key(asio::const_buffer {key.data(), key.size()},
asio::ssl::context::file_format::pem,
ec);
if (ec)
throw std::runtime_error("Error setting node's private key: " + ec.message());
// certificate chain
auto certchain = config.identity.second->toString(true /*chain*/);
tls_context.use_certificate_chain(asio::const_buffer {certchain.data(), certchain.size()}, ec);
if (ec)
throw std::runtime_error("Error setting certificate chain: " + ec.message());
if (logger_)
logger_->debug("[proxy:server] using certificate chain for ssl:\n{}", certchain);
// build http server
auto settings = restinio::run_on_this_thread_settings_t<RestRouterTraitsTls>();
addServerSettings(settings);
if (not config.address.empty())
settings.address(config.address);
settings.port(config.port);
settings.tls_context(std::move(tls_context));
httpsServer_ = std::make_unique<restinio::http_server_t<RestRouterTraitsTls>>(
ioContext_, std::forward<restinio::run_on_this_thread_settings_t<RestRouterTraitsTls>>(std::move(settings)));
// run https server
serverThread_ = std::thread([this] {
httpsServer_->open_async([] { /*ok*/ }, [](std::exception_ptr ex) { std::rethrow_exception(ex); });
httpsServer_->io_context().run();
});
} else {
auto settings = restinio::run_on_this_thread_settings_t<RestRouterTraits>();
addServerSettings(settings);
if (not config.address.empty())
settings.address(config.address);
settings.port(config.port);
httpServer_ = std::make_unique<restinio::http_server_t<RestRouterTraits>>(
ioContext_, std::forward<restinio::run_on_this_thread_settings_t<RestRouterTraits>>(std::move(settings)));
// run http server
serverThread_ = std::thread([this]() {
httpServer_->open_async([] { /*ok*/ }, [](std::exception_ptr ex) { std::rethrow_exception(ex); });
httpServer_->io_context().run();
});
}
dht->forwardAllMessages(true);
updateStats();
printStatsTimer_->async_wait(std::bind(&DhtProxyServer::handlePrintStats, this, std::placeholders::_1));
if (not persistPath_.empty()) {
try {
std::ifstream stateFile(persistPath_, std::ios::binary | std::ios::ate);
if (stateFile) {
std::streamsize size = stateFile.tellg();
stateFile.seekg(0, std::ios::beg);
if (logger_)
logger_->debug("[proxy:server] Loading proxy state from {} ({} bytes)", persistPath_, size);
loadState(stateFile, size);
}
} catch (const std::exception& e) {
if (logger_)
logger_->error("[proxy:server] Error loading state from file: {}", e.what());
}
}
}
template<typename Os>
void
DhtProxyServer::saveState(Os& stream)
{
msgpack::packer<Os> pk(&stream);
pk.pack_map(2);
{
std::lock_guard lock(lockSearchPuts_);
pk.pack("puts");
pk.pack(puts_);
}
#ifdef OPENDHT_PUSH_NOTIFICATIONS
{
std::lock_guard lock(lockListener_);
pk.pack("pushListeners");
pk.pack(pushListeners_);
}
#endif
}
template<typename Is>
void
DhtProxyServer::loadState(Is& is, size_t size)
{
msgpack::unpacker pac;
pac.reserve_buffer(size);
if (is.read(pac.buffer(), size)) {
pac.buffer_consumed(size);
msgpack::object_handle oh;
while (pac.next(oh)) {
if (oh.get().type != msgpack::type::MAP)
continue;
if (auto puts = findMapValue(oh.get(), "puts"sv)) {
std::lock_guard lock(lockSearchPuts_);
puts_ = puts->as<decltype(puts_)>();
if (logger_)
logger_->debug("[proxy:server] Loading {} persistent puts", puts_.size());
for (auto& put : puts_) {
for (auto& pput : put.second.puts) {
pput.second.expireTimer = std::make_unique<asio::system_timer>(io_context(),
pput.second.expiration);
pput.second.expireTimer->async_wait(std::bind(&DhtProxyServer::handleCancelPermamentPut,
this,
std::placeholders::_1,
put.first,
pput.first));
#ifdef OPENDHT_PUSH_NOTIFICATIONS
if (not pput.second.pushToken.empty()) {
auto jsonProvider = [infoHash = put.first.toString(),
clientId = pput.second.clientId,
vid = pput.first,
sessionCtx = pput.second.sessionCtx]() {
Json::Value json;
json["timeout"] = infoHash;
json["to"] = clientId;
json["vid"] = std::to_string(vid);
if (sessionCtx) {
std::lock_guard l(sessionCtx->lock);
json["s"] = sessionCtx->sessionId;
}
return json;
};
pput.second.expireNotifyTimer = std::make_unique<asio::system_timer>(io_context(),
pput.second.expiration
- proxy::OP_MARGIN);
pput.second.expireNotifyTimer->async_wait(
std::bind(&DhtProxyServer::handleNotifyPushListenExpire,
this,
std::placeholders::_1,
pput.second.pushToken,
std::move(jsonProvider),
pput.second.type,
pput.second.topic));
}
#endif
dht_->put(put.first, pput.second.value, DoneCallbackSimple {}, dht::time_point::max(), true);
}
}
} else {
if (logger_)
logger_->debug("No persistent puts in state");
}
#ifdef OPENDHT_PUSH_NOTIFICATIONS
if (auto pushListeners = findMapValue(oh.get(), "pushListeners"sv)) {
std::lock_guard lock(lockListener_);
pushListeners_ = pushListeners->as<decltype(pushListeners_)>();
if (logger_)
logger_->debug("[proxy:server] Loading {} push listeners", pushListeners_.size());
for (auto& pushListener : pushListeners_) {
for (auto& listeners : pushListener.second.listeners) {
for (auto& listener : listeners.second) {
if (not listener.sessionCtx)
listener.sessionCtx = std::make_shared<PushSessionContext>("");
{
std::lock_guard l(listener.sessionCtx->lock);
listener.sessionCtx->type = listener.type;
listener.sessionCtx->topic = listener.topic;
}
// start listening
listener.internalToken
= dht_->listen(listeners.first,
[this,
infoHash = listeners.first,
pushToken = pushListener.first,
clientId = listener.clientId,
sessionCtx = listener.sessionCtx](const std::vector<Sp<Value>>& values,
bool expired) {
return this->handlePushListen(infoHash,
pushToken,
clientId,
sessionCtx,
values,
expired);
});
// expire notify
listener.expireNotifyTimer = std::make_unique<asio::system_timer>(io_context(),
listener.expiration
- proxy::OP_MARGIN);
auto jsonProvider = [infoHash = listeners.first.toString(),
clientId = listener.clientId,
sessionCtx = listener.sessionCtx]() {
Json::Value json;
json["timeout"] = infoHash;
json["to"] = clientId;
std::lock_guard l(sessionCtx->lock);
json["s"] = sessionCtx->sessionId;
return json;
};
listener.expireNotifyTimer->async_wait(
std::bind(&DhtProxyServer::handleNotifyPushListenExpire,
this,
std::placeholders::_1,
pushListener.first,
std::move(jsonProvider),
listener.type,
listener.topic));
// cancel push listen
listener.expireTimer = std::make_unique<asio::system_timer>(io_context(),
listener.expiration);
listener.expireTimer->async_wait(std::bind(&DhtProxyServer::handleCancelPushListen,
this,
std::placeholders::_1,
pushListener.first,
listeners.first,
listener.clientId));
}
}
}
} else {
if (logger_)
logger_->debug("[proxy:server] No push listeners in state");
}
#endif
}
if (logger_)
logger_->debug("[proxy:server] loading ended");
}
}
asio::io_context&
DhtProxyServer::io_context() const
{
return *ioContext_;
}
DhtProxyServer::~DhtProxyServer()
{
if (not persistPath_.empty()) {
if (logger_)
logger_->debug("[proxy:server] Saving proxy state to {}", persistPath_);
std::ofstream stateFile(persistPath_, std::ios::binary);
saveState(stateFile);
}
if (dht_) {
std::lock_guard lock(lockListener_);
for (auto& l : listeners_) {
dht_->cancelListen(l.second.hash, std::move(l.second.token));
if (l.second.response)
l.second.response->done();
}
#ifdef OPENDHT_PUSH_NOTIFICATIONS
for (auto& lm : pushListeners_) {
for (auto& ls : lm.second.listeners)
for (auto& l : ls.second) {
if (l.expireNotifyTimer)
l.expireNotifyTimer->cancel();
if (l.expireTimer)
l.expireTimer->cancel();
dht_->cancelListen(ls.first, std::move(l.internalToken));
}
}
pushListeners_.clear();
#endif
}
if (logger_)
logger_->debug("[proxy:server] closing http server");
ioContext_->stop();
if (serverThread_.joinable())
serverThread_.join();
{
// ~Request notifies DONE and its handler erases from requests_:
// detach pending push requests so nothing calls back into this object.
decltype(requests_) pending;
{
std::lock_guard l(requestLock_);
pending = std::move(requests_);
}
for (auto& r : pending)
r.second->add_on_state_change_callback({});
pending.clear();
}
if (logger_)
logger_->debug("[proxy:server] http server closed");
}
template<typename ServerSettings>
void
DhtProxyServer::addServerSettings(ServerSettings& settings, const unsigned int max_pipelined_requests)
{
using namespace std::chrono;
/**
* If max_pipelined_requests is greater than 1 then RESTinio will continue
* to read from the socket after parsing the first request.
* In that case, RESTinio can detect the disconnection
* and calls state listener as expected.
* https://github.com/Stiffstream/restinio/issues/28
*/
settings.max_pipelined_requests(max_pipelined_requests);
// one less to detect the listener disconnect
settings.concurrent_accepts_count(max_pipelined_requests - 1);
settings.separate_accept_and_create_connect(true);
settings.logger(logger_);
settings.protocol(restinio::asio_ns::ip::tcp::v6());
settings.request_handler(createRestRouter());
// time limits // ~ 0.8 month
std::chrono::milliseconds timeout_request(std::numeric_limits<int>::max());
settings.read_next_http_message_timelimit(timeout_request);
settings.write_http_response_timelimit(60s);
settings.handle_request_timeout(timeout_request);
// socket options
settings.socket_options_setter([](auto& options) {
options.set_option(asio::ip::tcp::no_delay {true});
options.set_option(asio::socket_base::keep_alive {true});
});
settings.connection_state_listener(connListener_);
}
std::shared_ptr<DhtProxyServer::ServerStats>
DhtProxyServer::updateStats(std::shared_ptr<NodeInfo> info) const
{
auto now = clock::now();
auto last = lastStatsReset_.exchange(now);
auto count = requestNum_.exchange(0);
auto dt = std::chrono::duration<double>(now - last);
auto sstats = std::make_shared<ServerStats>();
auto& stats = *sstats;
stats.requestRate = count / dt.count();
#ifdef OPENDHT_PUSH_NOTIFICATIONS
stats.pushListenersCount = pushListeners_.size();
{
std::lock_guard lk(pushStatsMutex_);
stats.androidPush = androidPush_;
stats.iosPush = iosPush_;
stats.unifiedPush = unifiedPush_;
}
#endif
stats.serverStartTime = serverStartTime_;
stats.lastUpdated = now;
stats.totalPermanentPuts = 0;
std::for_each(puts_.begin(), puts_.end(), [&stats](const auto& put) {
stats.totalPermanentPuts += put.second.puts.size();
});
stats.putCount = puts_.size();
stats.listenCount = listeners_.size();
stats.nodeInfo = std::move(info);
return sstats;
}
void
DhtProxyServer::updateStats()
{
dht_->getNodeInfo([this](std::shared_ptr<NodeInfo> newInfo) {
stats_ = updateStats(newInfo);
nodeInfo_ = newInfo;
if (logger_) {
auto str = Json::writeString(jsonBuilder_, newInfo->toJson());
logger_->debug("[proxy:server] [stats] {}", str);
}
});
}
void
DhtProxyServer::handlePrintStats(const asio::error_code& ec)
{
if (ec == asio::error::operation_aborted)
return;
updateStats();
printStatsTimer_->expires_at(printStatsTimer_->expiry() + PRINT_STATS_PERIOD);
printStatsTimer_->async_wait(std::bind(&DhtProxyServer::handlePrintStats, this, std::placeholders::_1));
}
template<typename HttpResponse>
HttpResponse
DhtProxyServer::initHttpResponse(HttpResponse response)
{
response.append_header("Server", "RESTinio");
response.append_header(restinio::http_field::content_type, "application/json");
response.append_header(restinio::http_field::access_control_allow_origin, "*");
return response;
}
std::unique_ptr<RestRouter>
DhtProxyServer::createRestRouter()
{
using namespace std::placeholders;
auto router = std::make_unique<RestRouter>();
// **************************** LEGACY ROUTES ****************************
// node.info
router->http_get("/", std::bind(&DhtProxyServer::getNodeInfo, this, _1, _2));
// key.options
router->add_handler(restinio::http_method_options(), "/:hash", std::bind(&DhtProxyServer::options, this, _1, _2));
// key.get
router->http_get("/:hash", std::bind(&DhtProxyServer::get, this, _1, _2));
// key.post
router->http_post("/:hash", std::bind(&DhtProxyServer::put, this, _1, _2));
#ifdef OPENDHT_PUSH_NOTIFICATIONS
// key.subscribe
router->add_handler(restinio::http_method_subscribe(),
"/:hash",
std::bind(&DhtProxyServer::subscribe, this, _1, _2));
// key.unsubscribe
router->add_handler(restinio::http_method_unsubscribe(),
"/:hash",
std::bind(&DhtProxyServer::unsubscribe, this, _1, _2));
#endif // OPENDHT_PUSH_NOTIFICATIONS
#ifdef OPENDHT_PROXY_SERVER_IDENTITY
#endif // OPENDHT_PROXY_SERVER_IDENTITY
// **************************** NEW ROUTES ****************************
// node.info
router->http_get("/node/info", std::bind(&DhtProxyServer::getNodeInfo, this, _1, _2));
// node.stats
router->http_get("/node/stats", std::bind(&DhtProxyServer::getStats, this, _1, _2));
// key.options
router->http_get("/key/:hash/options", std::bind(&DhtProxyServer::options, this, _1, _2));
// key.get
router->http_get("/key/:hash", std::bind(&DhtProxyServer::get, this, _1, _2));
// key.post
router->http_post("/key/:hash", std::bind(&DhtProxyServer::put, this, _1, _2));
router->add_handler(restinio::http_method_options(),
"/key/:hash/listen",
std::bind(&DhtProxyServer::options, this, _1, _2));
// key.listen
router->http_get("/key/:hash/listen", std::bind(&DhtProxyServer::listen, this, _1, _2));
#ifdef OPENDHT_PUSH_NOTIFICATIONS
// node.pingPush
router->http_post("/node/pingPush", std::bind(&DhtProxyServer::pingPush, this, _1, _2));
// key.subscribe
router->add_handler(restinio::http_method_subscribe(),
"/key/:hash",
std::bind(&DhtProxyServer::subscribe, this, _1, _2));
// key.unsubscribe
router->add_handler(restinio::http_method_unsubscribe(),
"/key/:hash",
std::bind(&DhtProxyServer::unsubscribe, this, _1, _2));
#endif // OPENDHT_PUSH_NOTIFICATIONS
#ifdef OPENDHT_PROXY_SERVER_IDENTITY
// key.sign
router->http_post("/key/:hash/sign", std::bind(&DhtProxyServer::putSigned, this, _1, _2));
// key.encrypt
router->http_post("/key/:hash/encrypt", std::bind(&DhtProxyServer::putEncrypted, this, _1, _2));
#endif // OPENDHT_PROXY_SERVER_IDENTITY
return router;
}
RequestStatus
DhtProxyServer::getNodeInfo(restinio::request_handle_t request, restinio::router::route_params_t /*params*/) const
{
try {
if (auto nodeInfo = nodeInfo_) {
auto result = nodeInfo->toJson();
// [ipv6:ipv4]:port or ipv4:port
result["public_ip"] = request->remote_endpoint().address().to_string();
auto response = initHttpResponse(request->create_response());
response.append_body(Json::writeString(jsonBuilder_, result) + "\n");
return response.done();
}
auto response = initHttpResponse(request->create_response(restinio::status_service_unavailable()));
response.set_body(RESP_MSG_SERVICE_UNAVAILABLE);
return response.done();
} catch (...) {
return serverError(*request);
}
}
RequestStatus
DhtProxyServer::getStats(restinio::request_handle_t request, restinio::router::route_params_t /*params*/)
{
requestNum_++;
try {
if (auto stats = stats_) {
auto response = initHttpResponse(request->create_response());
response.append_body(Json::writeString(jsonBuilder_, stats->toJson()) + "\n");
return response.done();
} else {
auto response = initHttpResponse(request->create_response(restinio::status_service_unavailable()));
response.set_body(RESP_MSG_SERVICE_UNAVAILABLE);
return response.done();
}
} catch (...) {
return serverError(*request);
}
}
RequestStatus
DhtProxyServer::get(restinio::request_handle_t request, restinio::router::route_params_t params)
{
requestNum_++;
try {
InfoHash infoHash(params["hash"]);
if (!infoHash)
infoHash = InfoHash::get(params["hash"]);
auto response = std::make_shared<ResponseByPartsBuilder>(
initHttpResponse(request->create_response<ResponseByParts>()));
response->flush();
dht_->get(
infoHash,
[this, response](const std::vector<Sp<Value>>& values) {
std::ostringstream output;
for (const auto& value : values) {
output << Json::writeString(jsonBuilder_, value->toJson()) << "\n";
}
response->append_chunk(output.str());
response->flush();
return true;
},
[response](bool /*ok*/) { response->done(); });
return restinio::request_handling_status_t::accepted;
} catch (const std::exception& e) {
return serverError(*request);
}
}
RequestStatus
DhtProxyServer::listen(restinio::request_handle_t request, restinio::router::route_params_t params)
{
requestNum_++;
try {
InfoHash infoHash(params["hash"]);
if (!infoHash)
infoHash = InfoHash::get(params["hash"]);
auto response = std::make_shared<ResponseByPartsBuilder>(
initHttpResponse(request->create_response<ResponseByParts>()));
response->flush();
std::lock_guard lock(lockListener_);
// save the listener to handle a disconnect
auto& session = listeners_[request->connection_id()];
session.hash = infoHash;
session.response = response;
session.token = dht_->listen(infoHash, [this, response](const std::vector<Sp<Value>>& values, bool expired) {
for (const auto& value : values) {
auto jsonVal = value->toJson();
if (expired)
jsonVal["expired"] = true;
response->append_chunk(Json::writeString(jsonBuilder_, jsonVal) + "\n");
}
response->flush();
return true;
});
return restinio::request_handling_status_t::accepted;
} catch (const std::exception& e) {
return serverError(*request);
}
}
#ifdef OPENDHT_PUSH_NOTIFICATIONS
PushType
DhtProxyServer::getTypeFromString(const std::string& type)
{
if (type == "android")
return PushType::Android;
else if (type == "ios")
return PushType::iOS;
else if (type == "unifiedpush")
return PushType::UnifiedPush;
return PushType::None;
}
std::string
DhtProxyServer::getDefaultTopic(PushType)
{
return bundleId_;
}
RequestStatus
DhtProxyServer::pingPush(restinio::request_handle_t request, restinio::router::route_params_t /*params*/)
{
requestNum_++;
try {
std::string err;
Json::Value r;
auto* char_data = reinterpret_cast<const char*>(request->body().data());
auto reader = std::unique_ptr<Json::CharReader>(jsonReaderBuilder_.newCharReader());
if (!reader->parse(char_data, char_data + request->body().size(), &r, &err)) {
auto response = initHttpResponse(request->create_response(restinio::status_bad_request()));
response.set_body(RESP_MSG_JSON_INCORRECT);
return response.done();
}
const Json::Value& root(r); // parse using const Json so [] never creates element
auto pushToken = root["key"].asString();
if (pushToken.empty()) {
auto response = initHttpResponse(request->create_response(restinio::status_bad_request()));
response.set_body(RESP_MSG_NO_TOKEN);
return response.done();
}
auto type = getTypeFromString(root["platform"].asString());
auto topic = root["topic"].asString();
if (topic.empty()) {
topic = getDefaultTopic(type);
}
Json::Value json;
json["to"] = root["client_id"];
json["pong"] = true;
sendPushNotification(pushToken, std::move(json), type, true, topic);
} catch (...) {
return serverError(*request);
}
return restinio::request_handling_status_t::accepted;
}
RequestStatus
DhtProxyServer::subscribe(restinio::request_handle_t request, restinio::router::route_params_t params)
{
requestNum_++;
try {
InfoHash infoHash(params["hash"]);
if (!infoHash)
infoHash = InfoHash::get(params["hash"]);
std::string err;
Json::Value r;
auto* char_data = reinterpret_cast<const char*>(request->body().data());
auto reader = std::unique_ptr<Json::CharReader>(jsonReaderBuilder_.newCharReader());
if (!reader->parse(char_data, char_data + request->body().size(), &r, &err)) {
auto response = initHttpResponse(request->create_response(restinio::status_bad_request()));
response.set_body(RESP_MSG_JSON_INCORRECT);
return response.done();
}
const Json::Value& root(r); // parse using const Json so [] never creates element
auto pushToken = root["key"].asString();
if (pushToken.empty()) {
auto response = initHttpResponse(request->create_response(restinio::status_bad_request()));
response.set_body(RESP_MSG_NO_TOKEN);
return response.done();
}
auto type = getTypeFromString(root["platform"].asString());
auto topic = root["topic"].asString();
if (topic.empty()) {
topic = getDefaultTopic(type);
}
auto clientId = root["client_id"].asString();
auto sessionId = root["session_id"].asString();
if (logger_)
logger_->debug("[proxy:server] [subscribe {}] [client {}] [session {}]", infoHash, clientId, sessionId);
// Insert new or return existing push listeners of a token
std::lock_guard lock(lockListener_);
auto& pushListener = pushListeners_[pushToken];
auto& pushListeners = pushListener.listeners[infoHash];
auto listIt = std::find_if(pushListeners.begin(), pushListeners.end(), [&](const Listener& l) {
return l.clientId == clientId;
});
bool newListener = listIt == pushListeners.end();
if (newListener) {
pushListeners.emplace_back(Listener {});
listIt = std::prev(pushListeners.end());
listIt->clientId = clientId;
listIt->sessionCtx = std::make_shared<PushSessionContext>(sessionId);
}
{
// A refresh may come with another push platform or topic (e.g. a device
// whose push configuration changed): values must then be pushed there.
std::lock_guard l(listIt->sessionCtx->lock);
listIt->sessionCtx->sessionId = sessionId;
listIt->sessionCtx->type = type;
listIt->sessionCtx->topic = topic;
}
auto& listener = *listIt;
// Expiration (wall-clock: persisted as epoch time, survives restarts)
auto timeout = sys_clock::now() + proxy::OP_TIMEOUT;
listener.expiration = timeout;
listener.type = type;
listener.topic = topic;
if (listener.expireNotifyTimer)
listener.expireNotifyTimer->expires_at(timeout - proxy::OP_MARGIN);
else
listener.expireNotifyTimer = std::make_unique<asio::system_timer>(io_context(), timeout - proxy::OP_MARGIN);
auto jsonProvider = [h = infoHash.toString(), clientId, sessionCtx = listener.sessionCtx]() {
Json::Value json;
json["timeout"] = h;
json["to"] = clientId;
std::lock_guard l(sessionCtx->lock);
json["s"] = sessionCtx->sessionId;
return json;
};
listener.expireNotifyTimer->async_wait(std::bind(&DhtProxyServer::handleNotifyPushListenExpire,
this,
std::placeholders::_1,
pushToken,
std::move(jsonProvider),
listener.type,
listener.topic));
if (!listener.expireTimer)
listener.expireTimer = std::make_unique<asio::system_timer>(io_context(), timeout);
else
listener.expireTimer->expires_at(timeout);
listener.expireTimer->async_wait(std::bind(&DhtProxyServer::handleCancelPushListen,
this,
std::placeholders::_1,
pushToken,
infoHash,
clientId));
// Send response
if (not newListener) {
if (logger_)
logger_->debug("[proxy:server] [subscribe {}] found [client {}]", infoHash, listener.clientId);
// Send response header
auto response = std::make_shared<ResponseByPartsBuilder>(
initHttpResponse(request->create_response<ResponseByParts>()));
response->flush();
if (!root["refresh"].asBool()) {
// No Refresh
dht_->get(
infoHash,
[this, response](const Sp<Value>& value) {
auto output = Json::writeString(jsonBuilder_, value->toJson()) + "\n";
response->append_chunk(output);
response->flush();
return true;
},
[response](bool) { response->done(); });
} else {
// Refresh
response->append_chunk("{}\n");
return response->done();
}
} else {
// =========== No existing listener for an infoHash ============
// Add listen on dht
if (logger_)
logger_->debug("[proxy:server] [subscribe {}] new", infoHash);
listener.internalToken = dht_->listen(
infoHash,
[this,
infoHash,
pushToken,
clientId,
sessionCtx = listener.sessionCtx](const std::vector<Sp<Value>>& values, bool expired) {
return this->handlePushListen(infoHash, pushToken, clientId, sessionCtx, values, expired);
});
// Send response header
auto response = initHttpResponse(request->create_response());
response.set_body("{}\n");
return response.done();
}
} catch (...) {
return serverError(*request);
}
return restinio::request_handling_status_t::accepted;
}
RequestStatus
DhtProxyServer::unsubscribe(restinio::request_handle_t request, restinio::router::route_params_t params)
{
requestNum_++;
InfoHash infoHash(params["hash"]);
if (!infoHash)
infoHash = InfoHash::get(params["hash"]);
if (logger_)
logger_->debug("[proxy:server] [unsubscribe {}]", infoHash);
try {
std::string err;
Json::Value root;
auto* char_data = reinterpret_cast<const char*>(request->body().data());
auto reader = std::unique_ptr<Json::CharReader>(jsonReaderBuilder_.newCharReader());
if (!reader->parse(char_data, char_data + request->body().size(), &root, &err)) {
auto response = initHttpResponse(request->create_response(restinio::status_bad_request()));
response.set_body(RESP_MSG_JSON_INCORRECT);
return response.done();
}
auto pushToken = root["key"].asString();
if (pushToken.empty())
return restinio::request_handling_status_t::rejected;
auto clientId = root["client_id"].asString();
handleCancelPushListen(asio::error_code() /*success*/, pushToken, infoHash, clientId);
auto response = initHttpResponse(request->create_response());
return response.done();
} catch (...) {
return serverError(*request);
}
}
void
DhtProxyServer::handleNotifyPushListenExpire(const asio::error_code& ec,
const std::string pushToken,
std::function<Json::Value()> jsonProvider,
PushType type,
const std::string& topic)
{
if (ec == asio::error::operation_aborted)
return;
else if (ec) {
if (logger_)
logger_->error("[proxy:server] [subscribe] error sending put refresh: {}", ec.message());
}
if (logger_)
logger_->debug("[proxy:server] [subscribe] sending refresh to {} token", pushToken);
sendPushNotification(pushToken, jsonProvider(), type, false, topic);
}
void
DhtProxyServer::handleCancelPushListen(const asio::error_code& ec,
const std::string pushToken,
const InfoHash key,
const std::string clientId)
{
if (ec == asio::error::operation_aborted)
return;
else if (ec) {
if (logger_)
logger_->error("[proxy:server] [listen:push {}] error cancel: {}", key, ec.message());
}
if (logger_)
logger_->debug("[proxy:server] [listen:push {}] cancelled for {}", key, clientId);
std::lock_guard lock(lockListener_);
auto pushListener = pushListeners_.find(pushToken);
if (pushListener == pushListeners_.end())
return;
auto listeners = pushListener->second.listeners.find(key);
if (listeners == pushListener->second.listeners.end())
return;
for (auto listener = listeners->second.begin(); listener != listeners->second.end();) {
if (listener->clientId == clientId) {
if (dht_)
dht_->cancelListen(key, std::move(listener->internalToken));
listener = listeners->second.erase(listener);
} else {
++listener;
}
}
if (listeners->second.empty())
pushListener->second.listeners.erase(listeners);
if (pushListener->second.listeners.empty())
pushListeners_.erase(pushListener);
}
bool
DhtProxyServer::handlePushListen(const InfoHash& infoHash,
const std::string& pushToken,
const std::string& clientId,
const std::shared_ptr<DhtProxyServer::PushSessionContext>& sessionCtx,
const std::vector<std::shared_ptr<Value>>& values,
bool expired)
{
Json::Value json;
json["key"] = infoHash.toString();
json["to"] = clientId;
using namespace std::chrono;
json["t"] = Json::Value::Int64(duration_cast<milliseconds>(system_clock::now().time_since_epoch()).count());
PushType type;
std::string topic;
{
std::lock_guard l(sessionCtx->lock);
json["s"] = sessionCtx->sessionId;
type = sessionCtx->type;
topic = sessionCtx->topic;
}
// Build a comma-separated list of ids from the values. This is intended to be used when
// streaming from the iOS notification extension in order to filter out unwanted values.
std::string ids;
std::string pushTypes;
for (const auto& value : values) {
if (!ids.empty()) {
ids += ",";
pushTypes += ",";
}
pushTypes += value->pushType;
ids += std::to_string(value->id);
}
json["ids"] = ids;
json["pt"] = pushTypes;
// If message is expired copy the value ID to the `exp` field.
// This is presumably used by the Android notification system.
if (expired && values.size() < 2) {
json["exp"] = json["ids"];
}
auto minPriority = 1000u;
for (const auto& v : values)
minPriority = std::min(minPriority, v->priority);
if (logger_)
logger_->debug("[proxy:server] [listen {}] [client {}] [session {}] [expired {}] [priority {}] [values {}]",
infoHash,
clientId,
sessionCtx->sessionId,
expired,
minPriority,
values.size());
sendPushNotification(pushToken, std::move(json), type, !expired and minPriority == 0, topic);
return true;
}
void
DhtProxyServer::sendPushNotification(
const std::string& token, Json::Value&& json, PushType type, bool highPriority, const std::string& topic)
{
if (pushServer_.empty() && type != PushType::UnifiedPush)
return;
unsigned reqid = 0;
try {
std::shared_ptr<http::Request> request;
if (type == PushType::UnifiedPush) {
request = std::make_shared<http::Request>(io_context(), token, logger_);
} else {
request = std::make_shared<http::Request>(io_context(),
pushHostPort_.first,
pushHostPort_.second,
pushHostPort_.first.find("https") == 0,
logger_);
}
reqid = request->id();
request->set_method(restinio::http_method_post());
request->set_header_field(restinio::http_field_t::user_agent, "RESTinio client");
request->set_header_field(restinio::http_field_t::accept, "*/*");
if (type == PushType::UnifiedPush) {
std::string_view topicView = topic;
Blob pubKey; ///< P-256 Public key
Blob auth; ///< Auth secret
auto delim = topicView.find('|');
if (delim != std::string_view::npos) {
pubKey = base64_default_or_url_decode(topicView.substr(0, delim));
auth = base64_default_or_url_decode(topicView.substr(delim + 1));
}
request->set_header_field(restinio::http_field_t::ttl, "86400");
request->set_header_field(restinio::http_field_t::urgency, highPriority ? "high" : "normal");
request->set_header_field(restinio::http_field_t::content_type, "application/json");
std::string payloadStr = Json::writeString(jsonBuilder_, std::move(json));
if (!pubKey.empty() && !auth.empty()) {
request->set_header_field(restinio::http_field_t::content_encoding, "aes128gcm");
try {
Blob encrypted = crypto::webPushEncrypt(pubKey,
auth,
reinterpret_cast<const uint8_t*>(payloadStr.data()),
payloadStr.size());
request->set_body(std::string(encrypted.begin(), encrypted.end()));
} catch (const std::exception& e) {
if (logger_)
logger_->error("[proxy:server] [notification] encryption failed: {}", e.what());
return;
}
} else {
request->set_body(payloadStr);
}
} else {
request->set_target("/api/push");
request->set_header_field(restinio::http_field_t::host, pushServer_);
request->set_header_field(restinio::http_field_t::content_type, "application/json");
// NOTE: see https://github.com/appleboy/gorush
auto isResubscribe = json.isMember("timeout");
Json::Value notification(Json::objectValue);
Json::Value tokens(Json::arrayValue);
tokens[0] = token;
notification["tokens"] = std::move(tokens);
notification["platform"] = type == PushType::Android ? 2 : 1;
notification["data"] = std::move(json);
auto priority = highPriority ? "high" : "normal";
if (type == PushType::Android) {
Json::Value androidConfig(Json::objectValue);
androidConfig["priority"] = priority;
androidConfig["ttl"] = "86400s"; // time to live = 24 hours
notification["android"] = std::move(androidConfig);
} else {
notification["priority"] = priority;
const auto expiration = std::chrono::system_clock::now() + std::chrono::hours(24);
uint32_t exp = std::chrono::duration_cast<std::chrono::seconds>(expiration.time_since_epoch()).count();
notification["expiration"] = exp;
if (!topic.empty())
notification["topic"] = topic;
if (highPriority || isResubscribe) {
Json::Value alert(Json::objectValue);
alert["title"] = "hello";
notification["push_type"] = "alert";
notification["alert"] = alert;
notification["mutable_content"] = true;
notification["priority"] = "high";
} else {
notification["push_type"] = "background";
notification["content_available"] = true;
}
}
Json::Value notifications(Json::arrayValue);
notifications[0] = std::move(notification);
Json::Value content;
content["notifications"] = std::move(notifications);
request->set_body(Json::writeString(jsonBuilder_, content));
}
request->add_on_state_change_callback([this, reqid](http::Request::State state, const http::Response& response) {
if (state == http::Request::State::DONE) {
if (logger_ and response.status_code != 200)
logger_->error("[proxy:server] [notification] push failed: {}", response.status_code);
std::lock_guard l(requestLock_);
requests_.erase(reqid);
}
});
{
std::lock_guard l(requestLock_);
requests_[reqid] = request;
}
request->send();
// For monitoring purposes
std::lock_guard lk(pushStatsMutex_);
switch (type) {
case PushType::Android:
androidPush_.increment(highPriority);
break;
case PushType::iOS:
iosPush_.increment(highPriority);
break;
case PushType::UnifiedPush:
unifiedPush_.increment(highPriority);
break;
default:
break;
}
} catch (const std::exception& e) {
if (logger_)
logger_->error("[proxy:server] [notification] error send push: {}", e.what());
if (reqid) {
std::lock_guard l(requestLock_);
requests_.erase(reqid);
}
}
}
#endif // OPENDHT_PUSH_NOTIFICATIONS
void
DhtProxyServer::handleCancelPermamentPut(const asio::error_code& ec, const InfoHash& key, Value::Id vid)
{
if (ec == asio::error::operation_aborted)
return;
else if (ec) {
if (logger_)
logger_->error("[proxy:server] [put:permament] error sending put refresh: {}", ec.message());
}
if (logger_)
logger_->debug("[proxy:server] [put {}] cancel permament put {}", key, vid);
std::lock_guard lock(lockSearchPuts_);
auto sPuts = puts_.find(key);
if (sPuts == puts_.end())
return;
auto& sPutsMap = sPuts->second.puts;
auto put = sPutsMap.find(vid);
if (put == sPutsMap.end())
return;
if (dht_)
dht_->cancelPut(key, vid);
if (put->second.expireTimer)
put->second.expireTimer->cancel();
if (put->second.expireNotifyTimer)
put->second.expireNotifyTimer->cancel();
sPutsMap.erase(put);
if (sPutsMap.empty())
puts_.erase(sPuts);
}
RequestStatus
DhtProxyServer::put(restinio::request_handle_t request, restinio::router::route_params_t params)
{
requestNum_++;
InfoHash infoHash(params["hash"]);
if (!infoHash)
infoHash = InfoHash::get(params["hash"]);
if (request->body().empty()) {
auto response = initHttpResponse(request->create_response(restinio::status_bad_request()));
response.set_body(RESP_MSG_MISSING_PARAMS);
return response.done();
}
try {
std::string err;
Json::Value root;
auto* char_data = reinterpret_cast<const char*>(request->body().data());
auto reader = std::unique_ptr<Json::CharReader>(jsonReaderBuilder_.newCharReader());
if (reader->parse(char_data, char_data + request->body().size(), &root, &err)) {
auto value = std::make_shared<Value>(root);
bool permanent = root.isMember("permanent");
if (logger_)
logger_->debug("[proxy:server] [put {}] {} {}",
infoHash,
value->toString(),
(permanent ? "permanent" : ""));
if (permanent) {
std::string pushToken, clientId, sessionId, platform, topic;
auto& pVal = root["permanent"];
if (pVal.isObject()) {
pushToken = pVal["key"].asString();
clientId = pVal["client_id"].asString();
platform = pVal["platform"].asString();
sessionId = pVal["session_id"].asString();
topic = pVal["topic"].asString();
}
std::lock_guard lock(lockSearchPuts_);
auto timeout = sys_clock::now() + proxy::OP_TIMEOUT;
auto& sPuts = puts_[infoHash];
if (value->id == Value::INVALID_ID) {
for (auto& pp : sPuts.puts) {
if (pp.second.pushToken == pushToken and pp.second.clientId == clientId
and pp.second.value->contentEquals(*value)) {
pp.second.expireTimer->expires_at(timeout);
pp.second.expireTimer->async_wait(std::bind(&DhtProxyServer::handleCancelPermamentPut,
this,
std::placeholders::_1,
infoHash,
pp.second.value->id));
if (not sessionId.empty()) {
if (not pp.second.sessionCtx)
pp.second.sessionCtx = std::make_shared<PushSessionContext>(sessionId);
else {
std::lock_guard l(pp.second.sessionCtx->lock);
pp.second.sessionCtx->sessionId = sessionId;
}
}
auto response = initHttpResponse(request->create_response());
response.append_body(Json::writeString(jsonBuilder_, pp.second.value->toJson()) + "\n");
return response.done();
}
}
value->id = std::uniform_int_distribution<Value::Id> {1}(rd);
}
auto vid = value->id;
auto& pput = sPuts.puts[vid];
pput.value = value;
pput.expiration = timeout;
if (not pput.expireTimer) {
// cancel permanent put
pput.expireTimer = std::make_unique<asio::system_timer>(io_context(), timeout);
#ifdef OPENDHT_PUSH_NOTIFICATIONS
if (not pushToken.empty()) {
pput.pushToken = pushToken;
pput.clientId = clientId;
pput.type = getTypeFromString(platform);
if (topic.empty())
topic = getDefaultTopic(pput.type);
pput.topic = topic;
pput.sessionCtx = std::make_shared<PushSessionContext>(sessionId);
}
#endif
} else {
if (not sessionId.empty()) {
if (not pput.sessionCtx)
pput.sessionCtx = std::make_shared<PushSessionContext>(sessionId);
else {
std::lock_guard l(pput.sessionCtx->lock);
pput.sessionCtx->sessionId = sessionId;
}
}
pput.expireTimer->expires_at(timeout);
}
pput.expireTimer->async_wait(
std::bind(&DhtProxyServer::handleCancelPermamentPut, this, std::placeholders::_1, infoHash, vid));
#ifdef OPENDHT_PUSH_NOTIFICATIONS
// notify put permanent expiration
if (pput.sessionCtx) {
auto jsonProvider = [infoHash, clientId, vid, sessionCtx = pput.sessionCtx]() {
Json::Value json;
json["timeout"] = infoHash.toString();
json["to"] = clientId;
json["vid"] = std::to_string(vid);
std::lock_guard l(sessionCtx->lock);
json["s"] = sessionCtx->sessionId;
return json;
};
if (!pput.expireNotifyTimer)
pput.expireNotifyTimer = std::make_unique<asio::system_timer>(io_context(),
timeout - proxy::OP_MARGIN);
else
pput.expireNotifyTimer->expires_at(timeout - proxy::OP_MARGIN);
pput.expireNotifyTimer->async_wait(std::bind(&DhtProxyServer::handleNotifyPushListenExpire,
this,
std::placeholders::_1,
pushToken,
std::move(jsonProvider),
pput.type,
pput.topic));
}
#endif
}
dht_->put(
infoHash,
value,
[this, request, value](bool ok) {
if (ok) {
auto response = initHttpResponse(request->create_response());
response.append_body(Json::writeString(jsonBuilder_, value->toJson()) + "\n");
response.done();
} else {
auto response = initHttpResponse(request->create_response(restinio::status_bad_gateway()));
response.set_body(RESP_MSG_PUT_FAILED);
response.done();
}
},
dht::time_point::max(),
permanent);
return restinio::request_handling_status_t::accepted;
} else {
auto response = initHttpResponse(request->create_response(restinio::status_bad_request()));
response.set_body(RESP_MSG_JSON_INCORRECT);
return response.done();
}
} catch (const std::exception& e) {
if (logger_)
logger_->error("[proxy:server] error in put: {}", e.what());
return serverError(*request);
}
}
#ifdef OPENDHT_PROXY_SERVER_IDENTITY
RequestStatus
DhtProxyServer::putSigned(restinio::request_handle_t request, restinio::router::route_params_t params) const
{
requestNum_++;
InfoHash infoHash(params["hash"]);
if (!infoHash)
infoHash = InfoHash::get(params["hash"]);
if (request->body().empty()) {
auto response = initHttpResponse(request->create_response(restinio::status_bad_request()));
response.set_body(RESP_MSG_MISSING_PARAMS);
return response.done();
}
try {
std::string err;
Json::Value root;
auto* char_data = reinterpret_cast<const char*>(request->body().data());
auto reader = std::unique_ptr<Json::CharReader>(jsonReaderBuilder_.newCharReader());
if (reader->parse(char_data, char_data + request->body().size(), &root, &err)) {
auto value = std::make_shared<Value>(root);
dht_->putSigned(infoHash, value, [this, request, value](bool ok) {
if (ok) {
auto output = Json::writeString(jsonBuilder_, value->toJson()) + "\n";
auto response = initHttpResponse(request->create_response());
response.append_body(output);
response.done();
} else {
auto response = initHttpResponse(request->create_response(restinio::status_bad_gateway()));
response.set_body(RESP_MSG_PUT_FAILED);
response.done();
}
});
return restinio::request_handling_status_t::accepted;
} else {
auto response = initHttpResponse(request->create_response(restinio::status_bad_request()));
response.set_body(RESP_MSG_JSON_INCORRECT);
return response.done();
}
} catch (const std::exception& e) {
if (logger_)
logger_->error("[proxy:server] error in putSigned: {}", e.what());
return serverError(*request);
}
}
RequestStatus
DhtProxyServer::putEncrypted(restinio::request_handle_t request, restinio::router::route_params_t params)
{
requestNum_++;
InfoHash infoHash(params["hash"]);
if (!infoHash)
infoHash = InfoHash::get(params["hash"]);
if (request->body().empty()) {
auto response = initHttpResponse(request->create_response(restinio::status_bad_request()));
response.set_body(RESP_MSG_MISSING_PARAMS);
return response.done();
}
try {
std::string err;
Json::Value root;
auto* char_data = reinterpret_cast<const char*>(request->body().data());
auto reader = std::unique_ptr<Json::CharReader>(jsonReaderBuilder_.newCharReader());
if (reader->parse(char_data, char_data + request->body().size(), &root, &err)) {
InfoHash to(root["to"].asString());
if (!to) {
auto response = initHttpResponse(request->create_response(restinio::status_bad_request()));
response.set_body(RESP_MSG_DESTINATION_NOT_FOUND);
return response.done();
}
auto value = std::make_shared<Value>(root);
dht_->putEncrypted(infoHash, to, value, [this, request, value](bool ok) {
if (ok) {
auto response = initHttpResponse(request->create_response());
response.append_body(Json::writeString(jsonBuilder_, value->toJson()) + "\n");
response.done();
} else {
auto response = initHttpResponse(request->create_response(restinio::status_bad_gateway()));
response.set_body(RESP_MSG_PUT_FAILED);
response.done();
}
});
return restinio::request_handling_status_t::accepted;
} else {
auto response = initHttpResponse(request->create_response(restinio::status_bad_request()));
response.set_body(RESP_MSG_JSON_INCORRECT);
return response.done();
}
} catch (const std::exception& e) {
if (logger_)
logger_->error("[proxy:server] error in put: {}", e.what());
return serverError(*request);
}
}
#endif // OPENDHT_PROXY_SERVER_IDENTITY
RequestStatus
DhtProxyServer::options(restinio::request_handle_t request, restinio::router::route_params_t /*params*/)
{
requestNum_++;
#ifdef OPENDHT_PROXY_SERVER_IDENTITY
const auto methods = "OPTIONS, GET, POST, LISTEN, SIGN, ENCRYPT";
#else
const auto methods = "OPTIONS, GET, POST, LISTEN";
#endif
auto response = initHttpResponse(request->create_response());
response.append_header(restinio::http_field::access_control_allow_methods, methods);
response.append_header(restinio::http_field::access_control_allow_headers, "content-type");
response.append_header(restinio::http_field::access_control_max_age, "86400");
return response.done();
}
RequestStatus
DhtProxyServer::getFiltered(restinio::request_handle_t request, restinio::router::route_params_t params)
{
requestNum_++;
auto query = params["value"];
InfoHash infoHash(params["hash"]);
if (!infoHash)
infoHash = InfoHash::get(params["hash"]);
try {
auto response = std::make_shared<ResponseByPartsBuilder>(
initHttpResponse(request->create_response<ResponseByParts>()));
response->flush();
dht_->get(
infoHash,
[this, response](const Sp<Value>& value) {
response->append_chunk(Json::writeString(jsonBuilder_, value->toJson()) + "\n");
response->flush();
return true;
},
[response](bool /*ok*/) { response->done(); },
{},
query);
return restinio::request_handling_status_t::accepted;
} catch (const std::exception& e) {
return serverError(*request);
}
}
} // namespace dht
Reference in new issue View Git Blame Copy Permalink