RPCServerHandler.hpp

#pragma once
 
#include "data/BackendInterface.hpp"
#include "etl/ETLServiceInterface.hpp"
#include "rpc/Errors.hpp"
#include "rpc/Factories.hpp"
#include "rpc/JS.hpp"
#include "rpc/RPCHelpers.hpp"
#include "rpc/common/impl/APIVersionParser.hpp"
#include "util/Assert.hpp"
#include "util/CoroutineGroup.hpp"
#include "util/JsonUtils.hpp"
#include "util/Profiler.hpp"
#include "util/Taggable.hpp"
#include "util/log/Logger.hpp"
#include "web/SubscriptionContextInterface.hpp"
#include "web/dosguard/DOSGuardInterface.hpp"
#include "web/ng/Connection.hpp"
#include "web/ng/Request.hpp"
#include "web/ng/Response.hpp"
#include "web/ng/impl/ErrorHandling.hpp"
 
#include <boost/asio/spawn.hpp>
#include <boost/asio/steady_timer.hpp>
#include <boost/beast/core/error.hpp>
#include <boost/beast/http/status.hpp>
#include <boost/json/array.hpp>
#include <boost/json/object.hpp>
#include <boost/json/parse.hpp>
#include <boost/json/serialize.hpp>
#include <boost/system/system_error.hpp>
#include <xrpl/protocol/jss.h>
 
#include <chrono>
#include <exception>
#include <functional>
#include <memory>
#include <optional>
#include <ratio>
#include <string>
#include <utility>
 
namespace web::ng {

template <typename RPCEngineType>
class RPCServerHandler {
    std::shared_ptr<BackendInterface const> const backend_;
    std::shared_ptr<RPCEngineType> const rpcEngine_;
    std::shared_ptr<etl::ETLServiceInterface const> const etl_;
    std::reference_wrapper<dosguard::DOSGuardInterface> dosguard_;
    util::TagDecoratorFactory const tagFactory_;
    rpc::impl::ProductionAPIVersionParser apiVersionParser_;  // can be injected if needed
 
    util::Logger log_{"RPC"};
    util::Logger perfLog_{"Performance"};
 
public:
    RPCServerHandler(
        util::config::ClioConfigDefinition const& config,
        std::shared_ptr<BackendInterface const> const& backend,
        std::shared_ptr<RPCEngineType> const& rpcEngine,
        std::shared_ptr<etl::ETLServiceInterface const> const& etl,
        dosguard::DOSGuardInterface& dosguard
    )
        : backend_(backend)
        , rpcEngine_(rpcEngine)
        , etl_(etl)
        , dosguard_(dosguard)
        , tagFactory_(config)
        , apiVersionParser_(config.getObject("api_version"))
    {
    }

    [[nodiscard]] Response
    operator()(
        Request const& request,
        ConnectionMetadata const& connectionMetadata,
        SubscriptionContextPtr subscriptionContext,
        boost::asio::yield_context yield
    )
    {
        if (not dosguard_.get().isOk(connectionMetadata.ip())) {
            return makeSlowDownResponse(request, std::nullopt);
        }
 
        std::optional<Response> response;
        util::CoroutineGroup coroutineGroup{yield, 1};
        auto const onTaskComplete = coroutineGroup.registerForeign(yield);
        ASSERT(onTaskComplete.has_value(), "Coroutine group can't be full");
 
        bool const postSuccessful = rpcEngine_->post(
            [this,
             &request,
             &response,
             &onTaskComplete = *onTaskComplete,  // NOLINT(bugprone-unchecked-optional-access)
             &connectionMetadata,
             subscriptionContext =
                 std::move(subscriptionContext)](boost::asio::yield_context innerYield) mutable {
                try {
                    boost::system::error_code ec;
                    auto parsedRequest = boost::json::parse(request.message(), ec);
                    if (ec.failed() or not parsedRequest.is_object()) {
                        rpcEngine_->notifyBadSyntax();
                        response = impl::ErrorHelper{request}.makeJsonParsingError();
                        if (ec.failed()) {
                            LOG(log_.warn()) << "Error parsing JSON: " << ec.message()
                                             << ". For request: " << request.message();
                        } else {
                            LOG(log_.warn())
                                << "Received not a JSON object. For request: " << request.message();
                        }
                    } else {
                        auto parsedObject = std::move(parsedRequest).as_object();
 
                        if (not dosguard_.get().request(connectionMetadata.ip(), parsedObject)) {
                            response = makeSlowDownResponse(request, parsedObject);
                        } else {
                            LOG(perfLog_.debug())
                                << connectionMetadata.tag() << "Adding to work queue";
 
                            if (not connectionMetadata.wasUpgraded() and
                                shouldReplaceParams(parsedObject)) {
                                parsedObject[JS(params)] =
                                    boost::json::array({boost::json::object{}});
                            }
 
                            response = handleRequest(
                                innerYield,
                                request,
                                std::move(parsedObject),
                                connectionMetadata,
                                std::move(subscriptionContext)
                            );
                        }
                    }
                } catch (std::exception const& ex) {
                    LOG(perfLog_.error())
                        << connectionMetadata.tag() << "Caught exception: " << ex.what();
                    rpcEngine_->notifyInternalError();
                    response = impl::ErrorHelper{request}.makeInternalError();
                }
 
                // notify the coroutine group that the foreign task is done
                onTaskComplete();
            },
            connectionMetadata.ip()
        );
 
        if (not postSuccessful) {
            // onTaskComplete must be called to notify coroutineGroup that the foreign task is done
            (*onTaskComplete)();  // NOLINT(bugprone-unchecked-optional-access)
            rpcEngine_->notifyTooBusy();
            return impl::ErrorHelper{request}.makeTooBusyError();
        }
 
        // Put the coroutine to sleep until the foreign task is done
        coroutineGroup.asyncWait(yield);
        ASSERT(response.has_value(), "Woke up coroutine without setting response");
 
        // NOLINTBEGIN(bugprone-unchecked-optional-access)
        if (not dosguard_.get().add(connectionMetadata.ip(), response->message().size())) {
            response->setMessage(makeLoadWarning(*response));
        }
 
        return *std::move(response);
        // NOLINTEND(bugprone-unchecked-optional-access)
    }
 
private:
    Response
    handleRequest(
        boost::asio::yield_context yield,
        Request const& rawRequest,
        boost::json::object&& request,
        ConnectionMetadata const& connectionMetadata,
        SubscriptionContextPtr subscriptionContext
    )
    {
        LOG(log_.info()) << connectionMetadata.tag()
                         << (connectionMetadata.wasUpgraded() ? "ws" : "http")
                         << " received request from work queue: " << util::removeSecret(request)
                         << " ip = " << connectionMetadata.ip();
 
        try {
            auto const range = backend_->fetchLedgerRange();
            if (!range) {
                // for error that happened before the handler, we don't attach any warnings
                rpcEngine_->notifyNotReady();
                return impl::ErrorHelper{rawRequest, std::move(request)}.makeNotReadyError();
            }
 
            auto const context = [&] {
                if (connectionMetadata.wasUpgraded()) {
                    ASSERT(
                        subscriptionContext != nullptr,
                        "Subscription context must exist for a WS connection"
                    );
                    return rpc::makeWsContext(
                        yield,
                        request,
                        std::move(subscriptionContext),
                        tagFactory_.with(connectionMetadata.tag()),
                        *range,
                        connectionMetadata.ip(),
                        std::cref(apiVersionParser_),
                        connectionMetadata.isAdmin()
                    );
                }
                return rpc::makeHttpContext(
                    yield,
                    request,
                    tagFactory_.with(connectionMetadata.tag()),
                    *range,
                    connectionMetadata.ip(),
                    std::cref(apiVersionParser_),
                    connectionMetadata.isAdmin()
                );
            }();
 
            if (!context) {
                auto const err = context.error();
                LOG(perfLog_.warn())
                    << connectionMetadata.tag() << "Could not create Web context: " << err;
                LOG(log_.warn()) << connectionMetadata.tag()
                                 << "Could not create Web context: " << err;
 
                // we count all those as BadSyntax - as the WS path would.
                // Although over HTTP these will yield a 400 status with a plain text response (for
                // most).
                rpcEngine_->notifyBadSyntax();
                return impl::ErrorHelper(rawRequest, std::move(request)).makeError(err);
            }
 
            auto [result, timeDiff] =
                util::timed([&]() { return rpcEngine_->buildResponse(*context); });
 
            auto us = std::chrono::duration<int, std::milli>(timeDiff);
            rpc::logDuration(request, context->tag(), us);
 
            boost::json::object response;
 
            if (!result.response.has_value()) {
                // note: error statuses are counted/notified in buildResponse itself
                response =
                    impl::ErrorHelper(rawRequest, request).composeError(result.response.error());
                auto const responseStr = boost::json::serialize(response);
 
                LOG(perfLog_.debug()) << context->tag() << "Encountered error: " << responseStr;
                LOG(log_.debug()) << context->tag() << "Encountered error: " << responseStr;
            } else {
                auto& json = result.response.value();
                auto const isForwarded = json.contains("forwarded") &&
                    json.at("forwarded").is_bool() && json.at("forwarded").as_bool();
 
                // This can still technically be an error. Clio counts forwarded requests
                // as successful.
                rpcEngine_->notifyComplete(*context, us, isForwarded);
 
                if (isForwarded)
                    json.erase("forwarded");
 
                // if the result is forwarded - just use it as is
                // if forwarded request has error, for http, error should be in "result"; for ws,
                // error should be at top
                if (isForwarded &&
                    (json.contains(JS(result)) || connectionMetadata.wasUpgraded())) {
                    for (auto const& [k, v] : json)
                        response.insert_or_assign(k, v);
                } else {
                    response[JS(result)] = json;
                }
 
                if (isForwarded)
                    response["forwarded"] = true;
 
                // for ws there is an additional field "status" in the response,
                // otherwise the "status" is in the "result" field
                if (connectionMetadata.wasUpgraded()) {
                    auto const appendFieldIfExist = [&](auto const& field) {
                        if (request.contains(field) and not request.at(field).is_null())
                            response[field] = request.at(field);
                    };
 
                    appendFieldIfExist(JS(id));
                    appendFieldIfExist(JS(api_version));
 
                    if (!response.contains(JS(error)))
                        response[JS(status)] = JS(success);
 
                    response[JS(type)] = JS(response);
                } else {
                    if (response.contains(JS(result)) &&
                        !response[JS(result)].as_object().contains(JS(error)))
                        response[JS(result)].as_object()[JS(status)] = JS(success);
                }
            }
 
            boost::json::array warnings = std::move(result.warnings);
            warnings.emplace_back(rpc::makeWarning(rpc::WarningCode::WarnRpcClio));
 
            if (etl_->lastCloseAgeSeconds() >= 60)
                warnings.emplace_back(rpc::makeWarning(rpc::WarningCode::WarnRpcOutdated));
 
            response["warnings"] = warnings;
            return Response{boost::beast::http::status::ok, response, rawRequest};
        } catch (std::exception const& ex) {
            // note: while we are catching this in buildResponse too, this is here to make sure
            // that any other code that may throw is outside of buildResponse is also worked around.
            LOG(perfLog_.error()) << connectionMetadata.tag() << "Caught exception: " << ex.what();
            LOG(log_.error()) << connectionMetadata.tag() << "Caught exception: " << ex.what();
 
            rpcEngine_->notifyInternalError();
            return impl::ErrorHelper(rawRequest, std::move(request)).makeInternalError();
        }
    }
 
    static Response
    makeSlowDownResponse(Request const& request, std::optional<boost::json::value> requestJson)
    {
        auto error = rpc::makeError(rpc::RippledError::rpcSLOW_DOWN);
 
        if (not request.isHttp()) {
            try {
                if (not requestJson.has_value()) {
                    requestJson = boost::json::parse(request.message());
                }
                if (requestJson->is_object() && requestJson->as_object().contains("id"))
                    error["id"] = requestJson->as_object().at("id");
                error["request"] = request.message();
            } catch (std::exception const&) {
                error["request"] = request.message();
            }
        }
        return web::ng::Response{boost::beast::http::status::service_unavailable, error, request};
    }
 
    static boost::json::object
    makeLoadWarning(Response const& response)
    {
        auto jsonResponse = boost::json::parse(response.message()).as_object();
        jsonResponse["warning"] = "load";
        if (jsonResponse.contains("warnings") && jsonResponse["warnings"].is_array()) {
            jsonResponse["warnings"].as_array().push_back(
                rpc::makeWarning(rpc::WarningCode::WarnRpcRateLimit)
            );
        } else {
            jsonResponse["warnings"] =
                boost::json::array{rpc::makeWarning(rpc::WarningCode::WarnRpcRateLimit)};
        }
        return jsonResponse;
    }
 
    [[nodiscard]] bool
    shouldReplaceParams(boost::json::object const& req) const
    {
        auto const hasParams = req.contains(JS(params));
        auto const paramsIsArray = hasParams and req.at(JS(params)).is_array();
        auto const paramsIsEmptyString =
            hasParams and req.at(JS(params)).is_string() and req.at(JS(params)).as_string().empty();
        auto const paramsIsEmptyObject =
            hasParams and req.at(JS(params)).is_object() and req.at(JS(params)).as_object().empty();
        auto const paramsIsNull = hasParams and req.at(JS(params)).is_null();
        auto const arrayIsEmpty = paramsIsArray and req.at(JS(params)).as_array().empty();
        auto const arrayIsNotEmpty = paramsIsArray and not req.at(JS(params)).as_array().empty();
        auto const firstArgIsNull =
            arrayIsNotEmpty and req.at(JS(params)).as_array().at(0).is_null();
        auto const firstArgIsEmptyString = arrayIsNotEmpty and
            req.at(JS(params)).as_array().at(0).is_string() and
            req.at(JS(params)).as_array().at(0).as_string().empty();
 
        // Note: all this compatibility dance is to match `rippled` as close as possible
        return not hasParams or paramsIsEmptyString or paramsIsNull or paramsIsEmptyObject or
            arrayIsEmpty or firstArgIsEmptyString or firstArgIsNull;
    }
};
 
}  // namespace web::ng