Server.hpp

#pragma once
 
#include "data/LedgerCacheInterface.hpp"
#include "util/Taggable.hpp"
#include "util/log/Logger.hpp"
#include "web/AdminVerificationStrategy.hpp"
#include "web/HttpSession.hpp"
#include "web/ProxyIpResolver.hpp"
#include "web/SslHttpSession.hpp"
#include "web/dosguard/DOSGuardInterface.hpp"
#include "web/interface/Concepts.hpp"
#include "web/ng/impl/ServerSslContext.hpp"
 
#include <boost/asio/io_context.hpp>
#include <boost/asio/ip/address.hpp>
#include <boost/asio/ip/tcp.hpp>
#include <boost/asio/socket_base.hpp>
#include <boost/asio/spawn.hpp>
#include <boost/asio/ssl/context.hpp>
#include <boost/asio/ssl/error.hpp>
#include <boost/asio/strand.hpp>
#include <boost/beast/core/error.hpp>
#include <boost/beast/core/flat_buffer.hpp>
#include <boost/beast/core/stream_traits.hpp>
#include <boost/beast/core/tcp_stream.hpp>
#include <fmt/format.h>
 
#include <atomic>
#include <chrono>
#include <cstdint>
#include <exception>
#include <functional>
#include <memory>
#include <optional>
#include <stdexcept>
#include <string>
#include <utility>

namespace web {

template <
    template <typename> class PlainSessionType,
    template <typename> class SslSessionType,
    SomeServerHandler HandlerType>
class Detector
    : public std::enable_shared_from_this<Detector<PlainSessionType, SslSessionType, HandlerType>> {
    using std::enable_shared_from_this<
        Detector<PlainSessionType, SslSessionType, HandlerType>>::shared_from_this;
 
    util::Logger log_{"WebServer"};
    boost::beast::tcp_stream stream_;
    std::optional<std::reference_wrapper<boost::asio::ssl::context>> ctx_;
    std::reference_wrapper<util::TagDecoratorFactory const> tagFactory_;
    std::reference_wrapper<dosguard::DOSGuardInterface> const dosGuard_;
    std::shared_ptr<HandlerType> const handler_;
    std::reference_wrapper<data::LedgerCacheInterface const> cache_;
    boost::beast::flat_buffer buffer_;
    std::shared_ptr<AdminVerificationStrategy> const adminVerification_;
    std::uint32_t maxWsSendingQueueSize_;
    std::shared_ptr<ProxyIpResolver> proxyIpResolver_;
 
public:
    Detector(
        tcp::socket&& socket,
        std::optional<std::reference_wrapper<boost::asio::ssl::context>> ctx,
        std::reference_wrapper<util::TagDecoratorFactory const> tagFactory,
        std::reference_wrapper<dosguard::DOSGuardInterface> dosGuard,
        std::shared_ptr<HandlerType> handler,
        std::reference_wrapper<data::LedgerCacheInterface const> cache,
        std::shared_ptr<AdminVerificationStrategy> adminVerification,
        std::uint32_t maxWsSendingQueueSize,
        std::shared_ptr<ProxyIpResolver> proxyIpResolver
    )
        : stream_(std::move(socket))
        , ctx_(ctx)
        , tagFactory_(std::cref(tagFactory))
        , dosGuard_(dosGuard)
        , handler_(std::move(handler))
        , cache_(cache)
        , adminVerification_(std::move(adminVerification))
        , maxWsSendingQueueSize_(maxWsSendingQueueSize)
        , proxyIpResolver_(std::move(proxyIpResolver))
    {
    }

    void
    fail(boost::system::error_code ec, char const* message)
    {
        if (ec == boost::asio::ssl::error::stream_truncated)
            return;
 
        LOG(log_.info()) << "Detector failed (" << message << "): " << ec.message();
    }

    void
    run()
    {
        boost::beast::get_lowest_layer(stream_).expires_after(std::chrono::seconds(30));
        async_detect_ssl(
            stream_,
            buffer_,
            boost::beast::bind_front_handler(&Detector::onDetect, shared_from_this())
        );
    }

    void
    onDetect(boost::beast::error_code ec, bool result)
    {
        if (ec)
            return fail(ec, "detect");
 
        std::string ip;
        try {
            ip = stream_.socket().remote_endpoint().address().to_string();
        } catch (std::exception const&) {
            return fail(ec, "cannot get remote endpoint");
        }
 
        if (result) {
            if (!ctx_)
                return fail(ec, "SSL is not supported by this server");
 
            std::make_shared<SslSessionType<HandlerType>>(
                stream_.release_socket(),
                ip,
                adminVerification_,
                proxyIpResolver_,
                *ctx_,
                tagFactory_,
                dosGuard_,
                handler_,
                cache_,
                std::move(buffer_),
                maxWsSendingQueueSize_
            )
                ->run();
            return;
        }
 
        std::make_shared<PlainSessionType<HandlerType>>(
            stream_.release_socket(),
            ip,
            adminVerification_,
            proxyIpResolver_,
            tagFactory_,
            dosGuard_,
            handler_,
            cache_,
            std::move(buffer_),
            maxWsSendingQueueSize_
        )
            ->run();
    }
};

template <
    template <typename> class PlainSessionType,
    template <typename> class SslSessionType,
    SomeServerHandler HandlerType>
class Server
    : public ServerTag,
      public std::enable_shared_from_this<Server<PlainSessionType, SslSessionType, HandlerType>> {
    using std::enable_shared_from_this<
        Server<PlainSessionType, SslSessionType, HandlerType>>::shared_from_this;
 
    util::Logger log_{"WebServer"};
    std::reference_wrapper<boost::asio::io_context> ioc_;
    std::optional<boost::asio::ssl::context> ctx_;
    util::TagDecoratorFactory tagFactory_;
    std::reference_wrapper<dosguard::DOSGuardInterface> dosGuard_;
    std::shared_ptr<HandlerType> handler_;
    std::reference_wrapper<data::LedgerCacheInterface const> cache_;
    tcp::acceptor acceptor_;
    std::shared_ptr<AdminVerificationStrategy> adminVerification_;
    std::uint32_t maxWsSendingQueueSize_;
    std::shared_ptr<ProxyIpResolver> proxyIpResolver_;
    std::atomic_bool isStopped_{false};
 
public:
    Server(
        boost::asio::io_context& ioc,
        std::optional<boost::asio::ssl::context> ctx,
        tcp::endpoint endpoint,
        util::TagDecoratorFactory tagFactory,
        dosguard::DOSGuardInterface& dosGuard,
        std::shared_ptr<HandlerType> handler,
        std::reference_wrapper<data::LedgerCacheInterface const> cache,
        std::shared_ptr<AdminVerificationStrategy> adminVerification,
        std::uint32_t maxWsSendingQueueSize,
        ProxyIpResolver proxyIpResolver
    )
        : ioc_(std::ref(ioc))
        , ctx_(std::move(ctx))
        , tagFactory_(tagFactory)
        , dosGuard_(std::ref(dosGuard))
        , handler_(std::move(handler))
        , cache_(cache)
        , acceptor_(boost::asio::make_strand(ioc))
        , adminVerification_(std::move(adminVerification))
        , maxWsSendingQueueSize_(maxWsSendingQueueSize)
        , proxyIpResolver_(std::make_shared<ProxyIpResolver>(std::move(proxyIpResolver)))
    {
        boost::beast::error_code ec;
 
        acceptor_.open(endpoint.protocol(), ec);
        if (ec)
            return;
 
        acceptor_.set_option(boost::asio::socket_base::reuse_address(true), ec);
        if (ec)
            return;
 
        acceptor_.bind(endpoint, ec);
        if (ec) {
            LOG(log_.error()) << "Failed to bind to endpoint: " << endpoint
                              << ". message: " << ec.message();
            throw std::runtime_error(
                fmt::format(
                    "Failed to bind to endpoint: {}:{}",
                    endpoint.address().to_string(),
                    endpoint.port()
                )
            );
        }
 
        acceptor_.listen(boost::asio::socket_base::max_listen_connections, ec);
        if (ec) {
            LOG(log_.error()) << "Failed to listen at endpoint: " << endpoint
                              << ". message: " << ec.message();
            throw std::runtime_error(
                fmt::format(
                    "Failed to listen at endpoint: {}:{}",
                    endpoint.address().to_string(),
                    endpoint.port()
                )
            );
        }
    }

    void
    run()
    {
        doAccept();
    }

    void
    stop(boost::asio::yield_context)
    {
        isStopped_ = true;
    }
 
private:
    void
    doAccept()
    {
        acceptor_.async_accept(
            boost::asio::make_strand(ioc_.get()),
            boost::beast::bind_front_handler(&Server::onAccept, shared_from_this())
        );
    }
 
    void
    onAccept(boost::beast::error_code ec, tcp::socket socket)
    {
        if (isStopped_) {
            return;
        }
 
        if (!ec) {
            auto ctxRef = ctx_
                ? std::optional<std::reference_wrapper<boost::asio::ssl::context>>{*ctx_}
                : std::nullopt;
 
            std::make_shared<Detector<PlainSessionType, SslSessionType, HandlerType>>(
                std::move(socket),
                ctxRef,
                std::cref(tagFactory_),
                dosGuard_,
                handler_,
                cache_,
                adminVerification_,
                maxWsSendingQueueSize_,
                proxyIpResolver_
            )
                ->run();
        }
 
        doAccept();
    }
};

template <typename HandlerType>
using HttpServer = Server<HttpSession, SslHttpSession, HandlerType>;

template <typename HandlerType>
static std::expected<std::shared_ptr<HttpServer<HandlerType>>, std::string>
makeHttpServer(
    util::config::ClioConfigDefinition const& config,
    boost::asio::io_context& ioc,
    dosguard::DOSGuardInterface& dosGuard,
    std::shared_ptr<HandlerType> const& handler,
    std::reference_wrapper<data::LedgerCacheInterface const> cache
)
{
    static util::Logger const log{"WebServer"};  // NOLINT(readability-identifier-naming)
 
    auto expectedSslContext = ng::impl::makeServerSslContext(config);
    if (not expectedSslContext) {
        return std::unexpected(
            fmt::format("Failed to create SSL context: {}", expectedSslContext.error())
        );
    }
 
    auto const serverConfig = config.getObject("server");
 
    auto const ipFromConfig = serverConfig.get<std::string>("ip");
    boost::system::error_code ec;
    auto const address = boost::asio::ip::make_address(ipFromConfig, ec);
    if (ec.failed())
        return std::unexpected(fmt::format("Invalid 'server.ip' config value: {}", ipFromConfig));
 
    auto const port = serverConfig.get<unsigned short>("port");
 
    auto expectedAdminVerification = makeAdminVerificationStrategy(config);
    if (not expectedAdminVerification.has_value())
        return std::unexpected(expectedAdminVerification.error());
 
    // If the transactions number is 200 per ledger, A client which subscribes everything will send
    // 400+ feeds for each ledger. we allow user delay 3 ledgers by default
    auto const maxWsSendingQueueSize = serverConfig.get<uint32_t>("ws_max_sending_queue_size");
 
    auto proxyIpResolver = ProxyIpResolver::fromConfig(config);
 
    auto server = std::make_shared<HttpServer<HandlerType>>(
        ioc,
        std::move(expectedSslContext).value(),
        boost::asio::ip::tcp::endpoint{address, port},
        util::TagDecoratorFactory(config),
        dosGuard,
        handler,
        cache,
        std::move(expectedAdminVerification).value(),
        maxWsSendingQueueSize,
        std::move(proxyIpResolver)
    );
 
    server->run();
    return server;
}
 
}  // namespace web