diff --git a/include/opendht/dht_proxy_server.h b/include/opendht/dht_proxy_server.h index 900f50f5..23bc19b5 100644 --- a/include/opendht/dht_proxy_server.h +++ b/include/opendht/dht_proxy_server.h @@ -206,7 +206,6 @@ private: * HTTP 400, body: {"err":"xxxx"} if bad json * @param session */ - //void putSigned(const std::shared_ptr& session) const; RequestStatus putSigned(restinio::request_handle_t request, restinio::router::route_params_t params) const; @@ -219,9 +218,8 @@ private: * HTTP 400, body: {"err":"xxxx"} if bad json * @param session */ - //void putEncrypted(const std::shared_ptr& session) const; - //RequestStatus putEncrypted(restinio::request_handle_t request, - // restinio::router::route_params_t params); + RequestStatus putEncrypted(restinio::request_handle_t request, + restinio::router::route_params_t params); #endif // OPENDHT_PROXY_SERVER_IDENTITY @@ -341,6 +339,7 @@ private: proxy::ListenToken tokenPushNotif_ {0}; #endif //OPENDHT_PUSH_NOTIFICATIONS + const std::string RESP_MSG_DESTINATION_NOT_FOUND = "{\"err\":\"No destination found\"}"; const std::string RESP_MSG_NO_TOKEN = "{\"err\":\"No token\"}"; const std::string RESP_MSG_JSON_NOT_ENABLED = "{\"err\":\"JSON not enabled on this instance\"}"; const std::string RESP_MSG_JSON_INCORRECT = "{\"err:\":\"Incorrect JSON\"}"; diff --git a/src/dht_proxy_server.cpp b/src/dht_proxy_server.cpp index 32813a5f..ceb363ce 100644 --- a/src/dht_proxy_server.cpp +++ b/src/dht_proxy_server.cpp @@ -60,7 +60,7 @@ DhtProxyServer::DhtProxyServer(std::shared_ptr dht, in_port_t port , //service_ = std::unique_ptr(new restbed::Service()); std::cout << "Running DHT proxy server on port " << port << std::endl; - if (not pushServer.empty()) { + if (not pushServer.empty()){ #ifdef OPENDHT_PUSH_NOTIFICATIONS std::cout << "Using push notification server: " << pushServer << std::endl; #else @@ -71,7 +71,7 @@ DhtProxyServer::DhtProxyServer(std::shared_ptr dht, in_port_t port , jsonBuilder_["commentStyle"] = "None"; jsonBuilder_["indentation"] = ""; - server_thread = std::thread([this, port]() { + server_thread = std::thread([this, port](){ using namespace std::chrono; auto maxThreads = std::thread::hardware_concurrency() - 1; auto restThreads = maxThreads > 1 ? maxThreads : 1; @@ -91,26 +91,26 @@ DhtProxyServer::DhtProxyServer(std::shared_ptr dht, in_port_t port , try { restinio::run(std::move(settings)); } - catch(const std::exception &ex) { + catch(const std::exception &ex){ std::cerr << "Error starting RESTinio: " << ex.what() << std::endl; } }); - listenThread_ = std::thread([this]() { - while (not server_thread.joinable() and not stopListeners) { + listenThread_ = std::thread([this](){ + while (not server_thread.joinable() and not stopListeners){ std::this_thread::sleep_for(std::chrono::seconds(1)); } - while (server_thread.joinable() and not stopListeners) { + while (server_thread.joinable() and not stopListeners){ removeClosedListeners(); std::this_thread::sleep_for(std::chrono::seconds(1)); } // Remove last listeners removeClosedListeners(false); }); - schedulerThread_ = std::thread([this]() { - while (not server_thread.joinable() and not stopListeners) { + schedulerThread_ = std::thread([this](){ + while (not server_thread.joinable() and not stopListeners){ std::this_thread::sleep_for(std::chrono::seconds(1)); } - while (server_thread.joinable() and not stopListeners) { + while (server_thread.joinable() and not stopListeners){ std::unique_lock lock(schedulerLock_); auto next = scheduler_.run(); if (next == time_point::max()) @@ -148,7 +148,7 @@ DhtProxyServer::stop() std::lock_guard lock(lockListener_); auto listener = currentListeners_.begin(); - while (listener != currentListeners_.end()) { + while (listener != currentListeners_.end()){ ++listener; } stopListeners = true; @@ -206,6 +206,7 @@ DhtProxyServer::createRestRouter() #endif //OPENDHT_PUSH_NOTIFICATIONS #ifdef OPENDHT_PROXY_SERVER_IDENTITY router->http_post("/:hash/sign", std::bind(&DhtProxyServer::putSigned, this, _1, _2)); + router->http_post("/:hash/encrypt", std::bind(&DhtProxyServer::putEncrypted, this, _1, _2)); #endif // OPENDHT_PROXY_SERVER_IDENTITY router->add_handler(restinio::http_method_t::http_options, "/:hash", std::bind(&DhtProxyServer::options, this, _1, _2)); @@ -239,7 +240,7 @@ DhtProxyServer::getStats(restinio::request_handle_t request, { requestNum_++; try { - if (dht_) { + if (dht_){ #ifdef OPENDHT_JSONCPP auto output = Json::writeString(jsonBuilder_, stats_.toJson()) + "\n"; auto response = this->initHttpResponse(request->create_response()); @@ -257,7 +258,7 @@ DhtProxyServer::getStats(restinio::request_handle_t request, response.set_body(this->RESP_MSG_SERVICE_UNAVAILABLE); return response.done(); } - } catch (...) { + } catch (...){ auto response = this->initHttpResponse( request->create_response(restinio::status_internal_server_error())); response.set_body(this->RESP_MSG_INTERNAL_SERVER_ERRROR); @@ -295,7 +296,7 @@ DhtProxyServer::get(restinio::request_handle_t request, [response] (bool /*ok*/){ response->done(); }); - } catch (const std::exception& e) { + } catch (const std::exception& e){ auto response = this->initHttpResponse( request->create_response(restinio::status_internal_server_error())); response.set_body(this->RESP_MSG_INTERNAL_SERVER_ERRROR); @@ -342,14 +343,14 @@ DhtProxyServer::subscribe(restinio::request_handle_t request, Json::CharReaderBuilder rbuilder; auto* char_data = reinterpret_cast(b.data()); auto reader = std::unique_ptr(rbuilder.newCharReader()); - if (!reader->parse(char_data, char_data + b.size(), &root, &err)) { + if (!reader->parse(char_data, char_data + b.size(), &root, &err)){ auto response = this->initHttpResponse( request->create_response(restinio::status_bad_request())); response.set_body(this->RESP_MSG_JSON_INCORRECT); return response.done(); } auto pushToken = root["key"].asString(); - if (pushToken.empty()) { + if (pushToken.empty()){ auto response = this->initHttpResponse( request->create_response(restinio::status_bad_request())); response.set_body(this->RESP_MSG_NO_TOKEN); @@ -370,8 +371,8 @@ DhtProxyServer::subscribe(restinio::request_handle_t request, // One push listener per pushToken.infoHash.clientId auto pushListener = pushListeners_.emplace(pushToken, PushListener{}).first; auto listeners = pushListener->second.listeners.emplace(infoHash, std::vector{}).first; - for (auto& listener: listeners->second) { - if (listener.clientId == clientId) { + for (auto& listener: listeners->second){ + if (listener.clientId == clientId){ scheduler_.edit(listener.expireJob, timeout); scheduler_.edit(listener.expireNotifyJob, timeout - proxy::OP_MARGIN); @@ -379,7 +380,7 @@ DhtProxyServer::subscribe(restinio::request_handle_t request, this->initHttpResponse(request->create_response())); response->flush(); - if (!root.isMember("refresh") or !root["refresh"].asBool()) { + if (!root.isMember("refresh") or !root["refresh"].asBool()){ dht_->get(infoHash, [this, response](const dht::Sp& value){ auto output = Json::writeString(jsonBuilder_, value->toJson()) + "\n"; response->append_chunk(output); @@ -406,15 +407,15 @@ DhtProxyServer::subscribe(restinio::request_handle_t request, // The listener is not found, so add it. listener.internalToken = dht_->listen(infoHash, - [this, infoHash, pushToken, isAndroid, clientId](const std::vector>& values, bool expired) { - threadPool_->run([this, infoHash, pushToken, isAndroid, clientId, values, expired]() { + [this, infoHash, pushToken, isAndroid, clientId](const std::vector>& values, bool expired){ + threadPool_->run([this, infoHash, pushToken, isAndroid, clientId, values, expired](){ // Build message content Json::Value json; json["key"] = infoHash.toString(); json["to"] = clientId; - if (expired and values.size() < 3) { + if (expired and values.size() < 3){ std::stringstream ss; - for(size_t i = 0; i < values.size(); ++i) { + for(size_t i = 0; i < values.size(); ++i){ if(i != 0) ss << ","; ss << values[i]->id; } @@ -444,7 +445,7 @@ DhtProxyServer::subscribe(restinio::request_handle_t request, auto response = this->initHttpResponse(request->create_response()); response.set_body("{}\n"); return response.done(); - } catch (...) { + } catch (...){ auto response = this->initHttpResponse( request->create_response(restinio::status_internal_server_error())); response.set_body(this->RESP_MSG_INTERNAL_SERVER_ERRROR); @@ -477,7 +478,7 @@ DhtProxyServer::unsubscribe(restinio::request_handle_t request, auto* char_data = reinterpret_cast(b.data()); auto reader = std::unique_ptr(rbuilder.newCharReader()); - if (!reader->parse(char_data, char_data + b.size(), &root, &err)) { + if (!reader->parse(char_data, char_data + b.size(), &root, &err)){ auto response = this->initHttpResponse( request->create_response(restinio::status_bad_request())); response.set_body(this->RESP_MSG_JSON_INCORRECT); @@ -490,7 +491,7 @@ DhtProxyServer::unsubscribe(restinio::request_handle_t request, cancelPushListen(pushToken, infoHash, clientId); auto response = this->initHttpResponse(request->create_response()); return response.done(); - } catch (...) { + } catch (...){ auto response = this->initHttpResponse( request->create_response(restinio::status_internal_server_error())); response.set_body(this->RESP_MSG_INTERNAL_SERVER_ERRROR); @@ -509,8 +510,8 @@ DhtProxyServer::cancelPushListen(const std::string& pushToken, const dht::InfoHa 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) { + 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); @@ -518,10 +519,10 @@ DhtProxyServer::cancelPushListen(const std::string& pushToken, const dht::InfoHa ++listener; } } - if (listeners->second.empty()) { + if (listeners->second.empty()){ pushListener->second.listeners.erase(listeners); } - if (pushListener->second.listeners.empty()) { + if (pushListener->second.listeners.empty()){ pushListeners_.erase(pushListener); } } @@ -563,7 +564,7 @@ DhtProxyServer::put(restinio::request_handle_t request, response.set_body(this->RESP_MSG_SERVICE_UNAVAILABLE); return response.done(); } - else if (request->body().empty()) { + else if (request->body().empty()){ auto response = this->initHttpResponse( request->create_response(restinio::status_bad_request())); response.set_body(this->RESP_MSG_MISSING_PARAMS); @@ -582,10 +583,10 @@ DhtProxyServer::put(restinio::request_handle_t request, bool permanent = root.isMember("permanent"); std::cout << "Got put " << infoHash << " " << *value << " " << (permanent ? "permanent" : "") << std::endl; - if (permanent) { + if (permanent){ std::string pushToken, clientId, platform; auto& pVal = root["permanent"]; - if (pVal.isObject()) { + if (pVal.isObject()){ pushToken = pVal["key"].asString(); clientId = pVal["client_id"].asString(); platform = pVal["platform"].asString(); @@ -597,13 +598,13 @@ DhtProxyServer::put(restinio::request_handle_t request, auto sPuts = puts_.emplace(infoHash, SearchPuts{}).first; auto r = sPuts->second.puts.emplace(vid, PermanentPut{}); auto& pput = r.first->second; - if (r.second) { + if (r.second){ pput.expireJob = scheduler_.add(timeout, [this, infoHash, vid]{ std::cout << "Permanent put expired: " << infoHash << " " << vid << std::endl; cancelPut(infoHash, vid); }); #ifdef OPENDHT_PUSH_NOTIFICATIONS - if (not pushToken.empty()) { + if (not pushToken.empty()){ bool isAndroid = platform == "android"; pput.expireNotifyJob = scheduler_.add(timeout - proxy::OP_MARGIN, [this, infoHash, vid, pushToken, clientId, isAndroid] @@ -644,7 +645,7 @@ DhtProxyServer::put(restinio::request_handle_t request, response.set_body(this->RESP_MSG_JSON_INCORRECT); return response.done(); } - } catch (const std::exception& e) { + } catch (const std::exception& e){ std::cout << "Error performing put: " << e.what() << std::endl; auto response = this->initHttpResponse( request->create_response(restinio::status_internal_server_error())); @@ -670,7 +671,7 @@ RequestStatus DhtProxyServer::putSigned(restinio::request_handle_t request, response.set_body(this->RESP_MSG_SERVICE_UNAVAILABLE); return response.done(); } - else if (request->body().empty()) { + else if (request->body().empty()){ auto response = this->initHttpResponse( request->create_response(restinio::status_bad_request())); response.set_body(this->RESP_MSG_MISSING_PARAMS); @@ -707,7 +708,7 @@ RequestStatus DhtProxyServer::putSigned(restinio::request_handle_t request, response.set_body(this->RESP_MSG_JSON_INCORRECT); return response.done(); } - } catch (const std::exception& e) { + } catch (const std::exception& e){ std::cout << "Error performing put: " << e.what() << std::endl; auto response = this->initHttpResponse( request->create_response(restinio::status_internal_server_error())); @@ -717,7 +718,72 @@ RequestStatus DhtProxyServer::putSigned(restinio::request_handle_t request, return restinio::request_handling_status_t::accepted; } -//DhtProxyServer::putEncrypted() +RequestStatus +DhtProxyServer::putEncrypted(restinio::request_handle_t request, + restinio::router::route_params_t params) +{ + requestNum_++; + dht::InfoHash infoHash(params["hash"].to_string()); + if (!infoHash) + infoHash = dht::InfoHash::get(params["hash"].to_string()); + + if (!dht_){ + auto response = this->initHttpResponse( + request->create_response(restinio::status_service_unavailable())); + response.set_body(this->RESP_MSG_SERVICE_UNAVAILABLE); + return response.done(); + } + else if (request->body().empty()){ + auto response = this->initHttpResponse( + request->create_response(restinio::status_bad_request())); + response.set_body(this->RESP_MSG_MISSING_PARAMS); + return response.done(); + } + + try { + std::string err; + Json::Value root; + Json::CharReaderBuilder rbuilder; + auto* char_data = reinterpret_cast(request->body().data()); + auto reader = std::unique_ptr(rbuilder.newCharReader()); + + if (reader->parse(char_data, char_data + request->body().size(), &root, &err)){ + InfoHash to(root["to"].asString()); + if (!to){ + auto response = this->initHttpResponse( + request->create_response(restinio::status_bad_request())); + response.set_body(this->RESP_MSG_DESTINATION_NOT_FOUND); + return response.done(); + } + auto value = std::make_shared(root); + dht_->putEncrypted(infoHash, to, value, [this, request, value](bool ok){ + if (ok){ + auto output = Json::writeString(jsonBuilder_, value->toJson()) + "\n"; + auto response = this->initHttpResponse(request->create_response()); + response.append_body(output); + response.done(); + } else { + auto response = this->initHttpResponse(request->create_response( + restinio::status_bad_gateway())); + response.set_body(this->RESP_MSG_PUT_FAILED); + response.done(); + } + }); + } else { + auto response = this->initHttpResponse( + request->create_response(restinio::status_bad_request())); + response.set_body(this->RESP_MSG_JSON_INCORRECT); + return response.done(); + } + } catch (const std::exception& e){ + std::cout << "Error performing put: " << e.what() << std::endl; + auto response = this->initHttpResponse( + request->create_response(restinio::status_internal_server_error())); + response.set_body(this->RESP_MSG_INTERNAL_SERVER_ERRROR); + return response.done(); + } + return restinio::request_handling_status_t::accepted; +} #endif // OPENDHT_PROXY_SERVER_IDENTITY @@ -770,7 +836,7 @@ DhtProxyServer::getFiltered(restinio::request_handle_t request, }, {}, value ); - } catch (const std::exception& e) { + } catch (const std::exception& e){ auto response = this->initHttpResponse( request->create_response(restinio::status_internal_server_error())); response.set_body(this->RESP_MSG_INTERNAL_SERVER_ERRROR); @@ -785,9 +851,9 @@ DhtProxyServer::removeClosedListeners(bool testSession) // clean useless listeners std::lock_guard lock(lockListener_); auto listener = currentListeners_.begin(); - while (listener != currentListeners_.end()) { + while (listener != currentListeners_.end()){ auto cancel = dht_ and (not testSession or not server_thread.joinable()); - if (cancel) { + if (cancel){ dht_->cancelListen(listener->hash, std::move(listener->token)); // Remove listener if unused listener = currentListeners_.erase(listener);