rebase server

CMakeLists.txt

@@ -67,7 +67,6 @@ include(Postgres)
 
 
 target_sources(reporting PRIVATE
-<<<<<<< HEAD
     backend/CassandraBackend.cpp
     backend/PostgresBackend.cpp
     backend/BackendIndexer.cpp
@@ -78,17 +77,6 @@ target_sources(reporting PRIVATE
     etl/ReportingETL.cpp
     server/session.cpp
     server/SubscriptionManager.cpp
-=======
-    reporting/ETLSource.cpp
-    reporting/CassandraBackend.cpp
-    reporting/PostgresBackend.cpp
-    reporting/BackendIndexer.cpp
-    reporting/Pg.cpp
-    reporting/DBHelpers.cpp
-    reporting/ReportingETL.cpp
-    reporting/server/Handlers.cpp
-    reporting/server/SubscriptionManager.cpp
->>>>>>> 27506bc (rebase handlers)
     handlers/AccountInfo.cpp
     handlers/Tx.cpp
     handlers/RPCHelpers.cpp

handlers/Subscribe.cpp

@@ -1,9 +1,4 @@
 #include <boost/json.hpp>
-<<<<<<< HEAD
-=======
-#include <reporting/server/WsSession.h>
-#include <reporting/server/SubscriptionManager.h>
->>>>>>> 27506bc (rebase handlers)
 #include <handlers/RPCHelpers.h>
 #include <server/session.h>
 

reporting/server/HttpSession.h

@@ -118,13 +118,24 @@ handle_request(
     {
         auto request = boost::json::parse(req.body()).as_object();
 
-        auto response = buildResponse(request, etl, nullptr);
+        boost::json::object builtResponse;
+        try 
+        {
+            builtResponse = buildResponse(request, etl, nullptr);
+        }
+        catch (std::exception const& e)
+        {
+            return send(response(
+                http::status::internal_server_error,
+                "text/html",
+                "Internal Error"
+            ));
+        }
 
         return send(response(
             http::status::ok,
             "application/json",
-            boost::json::serialize(response)
-        ));
+            boost::json::serialize(builtResponse)));
     }
     catch (std::exception const& e)
     {
@@ -140,8 +151,6 @@ handle_request(
 // Handles an HTTP server connection
 class HttpSession : public std::enable_shared_from_this<HttpSession>
 {
-    // This is the C++11 equivalent of a generic lambda.
-    // The function object is used to send an HTTP message.
     struct send_lambda
     {
         HttpSession& self_;

reporting/server/SslCerts.h

@@ -0,0 +1,124 @@
+//
+// Copyright (c) 2016-2019 Vinnie Falco (vinnie dot falco at gmail dot com)
+//
+// Distributed under the Boost Software License, Version 1.0. (See accompanying
+// file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)
+//
+// Official repository: https://github.com/boostorg/beast
+//
+
+#ifndef BOOST_BEAST_EXAMPLE_COMMON_SERVER_CERTIFICATE_HPP
+#define BOOST_BEAST_EXAMPLE_COMMON_SERVER_CERTIFICATE_HPP
+
+#include <boost/asio/buffer.hpp>
+#include <boost/asio/ssl/context.hpp>
+#include <cstddef>
+#include <memory>
+
+/*  Load a signed certificate into the ssl context, and configure
+    the context for use with a server.
+
+    For this to work with the browser or operating system, it is
+    necessary to import the "Beast Test CA" certificate into
+    the local certificate store, browser, or operating system
+    depending on your environment Please see the documentation
+    accompanying the Beast certificate for more details.
+*/
+inline
+void
+load_server_certificate(boost::asio::ssl::context& ctx)
+{
+    /*
+        The certificate was generated from CMD.EXE on Windows 10 using:
+
+        winpty openssl dhparam -out dh.pem 2048
+        winpty openssl req -newkey rsa:2048 -nodes -keyout key.pem -x509 -days 10000 -out cert.pem -subj "//C=US\ST=CA\L=Los Angeles\O=Beast\CN=www.example.com"
+    */
+
+    std::string const cert =
+        "-----BEGIN CERTIFICATE-----\n"
+        "MIIDaDCCAlCgAwIBAgIJAO8vBu8i8exWMA0GCSqGSIb3DQEBCwUAMEkxCzAJBgNV\n"
+        "BAYTAlVTMQswCQYDVQQIDAJDQTEtMCsGA1UEBwwkTG9zIEFuZ2VsZXNPPUJlYXN0\n"
+        "Q049d3d3LmV4YW1wbGUuY29tMB4XDTE3MDUwMzE4MzkxMloXDTQ0MDkxODE4Mzkx\n"
+        "MlowSTELMAkGA1UEBhMCVVMxCzAJBgNVBAgMAkNBMS0wKwYDVQQHDCRMb3MgQW5n\n"
+        "ZWxlc089QmVhc3RDTj13d3cuZXhhbXBsZS5jb20wggEiMA0GCSqGSIb3DQEBAQUA\n"
+        "A4IBDwAwggEKAoIBAQDJ7BRKFO8fqmsEXw8v9YOVXyrQVsVbjSSGEs4Vzs4cJgcF\n"
+        "xqGitbnLIrOgiJpRAPLy5MNcAXE1strVGfdEf7xMYSZ/4wOrxUyVw/Ltgsft8m7b\n"
+        "Fu8TsCzO6XrxpnVtWk506YZ7ToTa5UjHfBi2+pWTxbpN12UhiZNUcrRsqTFW+6fO\n"
+        "9d7xm5wlaZG8cMdg0cO1bhkz45JSl3wWKIES7t3EfKePZbNlQ5hPy7Pd5JTmdGBp\n"
+        "yY8anC8u4LPbmgW0/U31PH0rRVfGcBbZsAoQw5Tc5dnb6N2GEIbq3ehSfdDHGnrv\n"
+        "enu2tOK9Qx6GEzXh3sekZkxcgh+NlIxCNxu//Dk9AgMBAAGjUzBRMB0GA1UdDgQW\n"
+        "BBTZh0N9Ne1OD7GBGJYz4PNESHuXezAfBgNVHSMEGDAWgBTZh0N9Ne1OD7GBGJYz\n"
+        "4PNESHuXezAPBgNVHRMBAf8EBTADAQH/MA0GCSqGSIb3DQEBCwUAA4IBAQCmTJVT\n"
+        "LH5Cru1vXtzb3N9dyolcVH82xFVwPewArchgq+CEkajOU9bnzCqvhM4CryBb4cUs\n"
+        "gqXWp85hAh55uBOqXb2yyESEleMCJEiVTwm/m26FdONvEGptsiCmF5Gxi0YRtn8N\n"
+        "V+KhrQaAyLrLdPYI7TrwAOisq2I1cD0mt+xgwuv/654Rl3IhOMx+fKWKJ9qLAiaE\n"
+        "fQyshjlPP9mYVxWOxqctUdQ8UnsUKKGEUcVrA08i1OAnVKlPFjKBvk+r7jpsTPcr\n"
+        "9pWXTO9JrYMML7d+XRSZA1n3856OqZDX4403+9FnXCvfcLZLLKTBvwwFgEFGpzjK\n"
+        "UEVbkhd5qstF6qWK\n"
+        "-----END CERTIFICATE-----\n";
+
+    std::string const key =
+        "-----BEGIN PRIVATE KEY-----\n"
+        "MIIEvgIBADANBgkqhkiG9w0BAQEFAASCBKgwggSkAgEAAoIBAQDJ7BRKFO8fqmsE\n"
+        "Xw8v9YOVXyrQVsVbjSSGEs4Vzs4cJgcFxqGitbnLIrOgiJpRAPLy5MNcAXE1strV\n"
+        "GfdEf7xMYSZ/4wOrxUyVw/Ltgsft8m7bFu8TsCzO6XrxpnVtWk506YZ7ToTa5UjH\n"
+        "fBi2+pWTxbpN12UhiZNUcrRsqTFW+6fO9d7xm5wlaZG8cMdg0cO1bhkz45JSl3wW\n"
+        "KIES7t3EfKePZbNlQ5hPy7Pd5JTmdGBpyY8anC8u4LPbmgW0/U31PH0rRVfGcBbZ\n"
+        "sAoQw5Tc5dnb6N2GEIbq3ehSfdDHGnrvenu2tOK9Qx6GEzXh3sekZkxcgh+NlIxC\n"
+        "Nxu//Dk9AgMBAAECggEBAK1gV8uETg4SdfE67f9v/5uyK0DYQH1ro4C7hNiUycTB\n"
+        "oiYDd6YOA4m4MiQVJuuGtRR5+IR3eI1zFRMFSJs4UqYChNwqQGys7CVsKpplQOW+\n"
+        "1BCqkH2HN/Ix5662Dv3mHJemLCKUON77IJKoq0/xuZ04mc9csykox6grFWB3pjXY\n"
+        "OEn9U8pt5KNldWfpfAZ7xu9WfyvthGXlhfwKEetOuHfAQv7FF6s25UIEU6Hmnwp9\n"
+        "VmYp2twfMGdztz/gfFjKOGxf92RG+FMSkyAPq/vhyB7oQWxa+vdBn6BSdsfn27Qs\n"
+        "bTvXrGe4FYcbuw4WkAKTljZX7TUegkXiwFoSps0jegECgYEA7o5AcRTZVUmmSs8W\n"
+        "PUHn89UEuDAMFVk7grG1bg8exLQSpugCykcqXt1WNrqB7x6nB+dbVANWNhSmhgCg\n"
+        "VrV941vbx8ketqZ9YInSbGPWIU/tss3r8Yx2Ct3mQpvpGC6iGHzEc/NHJP8Efvh/\n"
+        "CcUWmLjLGJYYeP5oNu5cncC3fXUCgYEA2LANATm0A6sFVGe3sSLO9un1brA4zlZE\n"
+        "Hjd3KOZnMPt73B426qUOcw5B2wIS8GJsUES0P94pKg83oyzmoUV9vJpJLjHA4qmL\n"
+        "CDAd6CjAmE5ea4dFdZwDDS8F9FntJMdPQJA9vq+JaeS+k7ds3+7oiNe+RUIHR1Sz\n"
+        "VEAKh3Xw66kCgYB7KO/2Mchesu5qku2tZJhHF4QfP5cNcos511uO3bmJ3ln+16uR\n"
+        "GRqz7Vu0V6f7dvzPJM/O2QYqV5D9f9dHzN2YgvU9+QSlUeFK9PyxPv3vJt/WP1//\n"
+        "zf+nbpaRbwLxnCnNsKSQJFpnrE166/pSZfFbmZQpNlyeIuJU8czZGQTifQKBgHXe\n"
+        "/pQGEZhVNab+bHwdFTxXdDzr+1qyrodJYLaM7uFES9InVXQ6qSuJO+WosSi2QXlA\n"
+        "hlSfwwCwGnHXAPYFWSp5Owm34tbpp0mi8wHQ+UNgjhgsE2qwnTBUvgZ3zHpPORtD\n"
+        "23KZBkTmO40bIEyIJ1IZGdWO32q79nkEBTY+v/lRAoGBAI1rbouFYPBrTYQ9kcjt\n"
+        "1yfu4JF5MvO9JrHQ9tOwkqDmNCWx9xWXbgydsn/eFtuUMULWsG3lNjfst/Esb8ch\n"
+        "k5cZd6pdJZa4/vhEwrYYSuEjMCnRb0lUsm7TsHxQrUd6Fi/mUuFU/haC0o0chLq7\n"
+        "pVOUFq5mW8p0zbtfHbjkgxyF\n"
+        "-----END PRIVATE KEY-----\n";
+
+    std::string const dh =
+        "-----BEGIN DH PARAMETERS-----\n"
+        "MIIBCAKCAQEArzQc5mpm0Fs8yahDeySj31JZlwEphUdZ9StM2D8+Fo7TMduGtSi+\n"
+        "/HRWVwHcTFAgrxVdm+dl474mOUqqaz4MpzIb6+6OVfWHbQJmXPepZKyu4LgUPvY/\n"
+        "4q3/iDMjIS0fLOu/bLuObwU5ccZmDgfhmz1GanRlTQOiYRty3FiOATWZBRh6uv4u\n"
+        "tff4A9Bm3V9tLx9S6djq31w31Gl7OQhryodW28kc16t9TvO1BzcV3HjRPwpe701X\n"
+        "oEEZdnZWANkkpR/m/pfgdmGPU66S2sXMHgsliViQWpDCYeehrvFRHEdR9NV+XJfC\n"
+        "QMUk26jPTIVTLfXmmwU0u8vUkpR7LQKkwwIBAg==\n"
+        "-----END DH PARAMETERS-----\n";
+    
+    ctx.set_password_callback(
+        [](std::size_t,
+            boost::asio::ssl::context_base::password_purpose)
+        {
+            return "test";
+        });
+
+    ctx.set_options(
+        boost::asio::ssl::context::default_workarounds |
+        boost::asio::ssl::context::no_sslv2 |
+        boost::asio::ssl::context::single_dh_use);
+
+    ctx.use_certificate_chain(
+        boost::asio::buffer(cert.data(), cert.size()));
+
+    ctx.use_private_key(
+        boost::asio::buffer(key.data(), key.size()),
+        boost::asio::ssl::context::file_format::pem);
+
+    ctx.use_tmp_dh(
+        boost::asio::buffer(dh.data(), dh.size()));
+}
+
+#endif

reporting/server/WsSession.h

@@ -23,6 +23,8 @@
 #include <boost/asio/dispatch.hpp>
 #include <boost/beast/core.hpp>
 #include <boost/beast/websocket.hpp>
+#include <boost/beast/ssl.hpp>
+#include <boost/beast/websocket/ssl.hpp>
 
 #include <backend/BackendInterface.h>
 #include <etl/ETLSource.h>
@@ -169,7 +171,8 @@ fail(boost::beast::error_code ec, char const* what);
 // Echoes back all received WebSocket messages
 class WsSession : public std::enable_shared_from_this<WsSession>
 {
-    boost::beast::websocket::stream<boost::beast::tcp_stream> ws_;
+    boost::beast::websocket::stream<
+        boost::beast::ssl_stream<boost::beast::tcp_stream>> ws_;
     boost::beast::flat_buffer buffer_;
     std::string response_;
 
@@ -252,21 +255,45 @@ private:
     void
     on_run()
     {
+        boost::beast::get_lowest_layer(ws_).expires_after(std::chrono::seconds(30));
+
+         // Perform the SSL handshake
+        ws_.next_layer().async_handshake(
+            ssl::stream_base::server,
+            boost::beast::bind_front_handler(
+                &WsSession::on_handshake,
+                shared_from_this()));
+    }
+
+    void
+    on_handshake(boost::beast::error_code ec)
+    {
+        if(ec)
+            return wsFail(ec, "handshake");
+
+        // Turn off the timeout on the tcp_stream, because
+        // the websocket stream has its own timeout system.
+        boost::beast::get_lowest_layer(ws_).expires_never();
+
         // Set suggested timeout settings for the websocket
-        ws_.set_option(boost::beast::websocket::stream_base::timeout::suggested(
+        ws_.set_option(
+            boost::beast::websocket::stream_base::timeout::suggested(
                 boost::beast::role_type::server));
 
         // Set a decorator to change the Server of the handshake
         ws_.set_option(boost::beast::websocket::stream_base::decorator(
-            [](boost::beast::websocket::response_type& res) {
-                res.set(
-                    boost::beast::http::field::server,
+            [](boost::beast::websocket::response_type& res)
+            {
+                res.set(http::field::server,
                     std::string(BOOST_BEAST_VERSION_STRING) +
-                        " websocket-server-async");
+                        " websocket-server-async-ssl");
             }));
+
         // Accept the websocket handshake
-        ws_.async_accept(boost::beast::bind_front_handler(
-            &WsSession::on_accept, shared_from_this()));
+        ws_.async_accept(
+            boost::beast::bind_front_handler(
+                &WsSession::on_accept,
+                shared_from_this()));
     }
 
     void

server/SubscriptionManager.h

@@ -20,16 +20,10 @@
 #ifndef SUBSCRIPTION_MANAGER_H
 #define SUBSCRIPTION_MANAGER_H
 
-<<<<<<< HEAD:server/SubscriptionManager.h
 #include <server/session.h>
 
 #include <memory>
 #include <set>
-=======
-#include <set>
-#include <memory>
-#include <reporting/server/WsSession.h>
->>>>>>> 27506bc (rebase handlers):reporting/server/SubscriptionManager.h
 
 class WsSession;
 
@@ -105,17 +99,10 @@ public:
         std::shared_ptr<session>& session);
 
     void
-<<<<<<< HEAD:server/SubscriptionManager.h
     subProposedTransactions(std::shared_ptr<session>& session);
 
     void
     unsubProposedTransactions(std::shared_ptr<session>& session);
-=======
-    subAccount(ripple::AccountID const& account, std::shared_ptr<WsSession>& session);
-
-    void
-    unsubAccount(ripple::AccountID const& account, std::shared_ptr<WsSession>& session);
->>>>>>> 27506bc (rebase handlers):reporting/server/SubscriptionManager.h
 };
 
 #endif  // SUBSCRIPTION_MANAGER_H

server/listener.h

@@ -34,16 +34,15 @@ class SubscriptionManager;
 template <class Session>
 class listener : public std::enable_shared_from_this<listener<Session>>
 {
+    using std::enable_shared_from_this<listener<Session>>::shared_from_this;
+
     boost::asio::io_context& ioc_;
     boost::asio::ip::tcp::acceptor acceptor_;
-<<<<<<< HEAD:server/listener.h
     std::shared_ptr<BackendInterface> backend_;
     std::shared_ptr<SubscriptionManager> subscriptions_;
     std::shared_ptr<ETLLoadBalancer> balancer_;
     DOSGuard& dosGuard_;
-=======
-    ReportingETL& etl_;
->>>>>>> 27506bc (rebase handlers):reporting/server/listener.h
+    ssl::context& ctx_;
 
 public:
     static void
@@ -63,7 +62,6 @@ public:
     listener(
         boost::asio::io_context& ioc,
         boost::asio::ip::tcp::endpoint endpoint,
-<<<<<<< HEAD:server/listener.h
         std::shared_ptr<BackendInterface> backend,
         std::shared_ptr<SubscriptionManager> subscriptions,
         std::shared_ptr<ETLLoadBalancer> balancer,
@@ -74,12 +72,6 @@ public:
         , subscriptions_(subscriptions)
         , balancer_(balancer)
         , dosGuard_(dosGuard)
-=======
-        ReportingETL& etl)
-        : ioc_(ioc)
-        , acceptor_(ioc)
-        , etl_(etl)
->>>>>>> 27506bc (rebase handlers):reporting/server/listener.h
     {
         boost::beast::error_code ec;
 
@@ -149,17 +141,12 @@ private:
         }
         else
         {
-<<<<<<< HEAD:server/listener.h
             session::make_session(
                 std::move(socket),
                 backend_,
                 subscriptions_,
                 balancer_,
                 dosGuard_);
-=======
-            // Create the session and run it
-            std::make_shared<Session>(std::move(socket), etl_)->run();
->>>>>>> 27506bc (rebase handlers):reporting/server/listener.h
         }
 
         // Accept another connection

server/websocket_server_async.cpp

@@ -30,15 +30,12 @@
 #include <functional>
 #include <iostream>
 #include <memory>
-<<<<<<< HEAD:server/websocket_server_async.cpp
 #include <server/listener.h>
 #include <server/session.h>
-=======
 #include <reporting/ReportingETL.h>
 #include <reporting/server/listener.h>
 #include <reporting/server/WsSession.h>
 #include <reporting/server/HttpSession.h>
->>>>>>> 27506bc (rebase handlers):websocket_server_async.cpp
 #include <sstream>
 #include <string>
 #include <thread>
@@ -106,45 +103,12 @@ initLogLevel(int level)
 void
 start(boost::asio::io_context& ioc, std::uint32_t numThreads)
 {
-<<<<<<< HEAD:server/websocket_server_async.cpp
     std::vector<std::thread> v;
     v.reserve(numThreads - 1);
     for (auto i = numThreads - 1; i > 0; --i)
         v.emplace_back([&ioc] { ioc.run(); });
 
     ioc.run();
-=======
-    auto const address = 
-        boost::asio::ip::make_address(wsConfig.at("ip").as_string().c_str());
-    auto const port = 
-        static_cast<unsigned short>(wsConfig.at("port").as_uint64());
-
-    // Create and launch a listening port
-    std::make_shared<listener<WsSession>>(
-        ioc,
-        boost::asio::ip::tcp::endpoint{address, port},
-        etl)
-        ->run();
-}
-
-void
-openHttpServer(
-    boost::json::object const& httpConfig,
-    boost::asio::io_context& ioc,
-    ReportingETL& etl)
-{
-    auto const address = 
-        boost::asio::ip::make_address(httpConfig.at("ip").as_string().c_str());
-    auto const port = 
-        static_cast<unsigned short>(httpConfig.at("port").as_uint64());
-
-    // Create and launch a listening port
-    std::make_shared<listener<HttpSession>>(
-        ioc,
-        boost::asio::ip::tcp::endpoint{address, port},
-        etl)
-        ->run();
->>>>>>> 27506bc (rebase handlers):websocket_server_async.cpp
 }
 
 int
@@ -163,6 +127,7 @@ main(int argc, char* argv[])
 
     auto const threads = std::max<int>(1, std::atoi(argv[1]));
     auto const config = parse_config(argv[2]);
+
     if (argc > 5)
     {
         initLogLevel(std::atoi(argv[5]));