diff --git a/src/httpserver.cpp b/src/httpserver.cpp index c155544dc9a..1e785a5b3f7 100644 --- a/src/httpserver.cpp +++ b/src/httpserver.cpp @@ -1072,6 +1072,65 @@ void HTTPServer::NewSockAccepted(std::unique_ptr&& sock, const CService& a addr.ToStringAddrPort(), id); } +void HTTPServer::SocketHandlerConnected(const IOReadiness& io_readiness) const +{ + for (const auto& [sock, events] : io_readiness.events_per_sock) { + if (m_interrupt_net) { + return; + } + + auto it{io_readiness.httpclients_per_sock.find(sock)}; + if (it == io_readiness.httpclients_per_sock.end()) { + continue; + } + const std::shared_ptr& client{it->second}; + + bool send_ready = events.occurred & Sock::SEND; + bool recv_ready = events.occurred & Sock::RECV; + bool err_ready = events.occurred & Sock::ERR; + + if (send_ready) { + // TODO: send data + } + + if (recv_ready || err_ready) { + std::byte buf[0x10000]; // typical socket buffer is 8K-64K + + const ssize_t nrecv{WITH_LOCK( + client->m_sock_mutex, + return client->m_sock->Recv(buf, sizeof(buf), MSG_DONTWAIT);)}; + + if (nrecv < 0) { + const int err = WSAGetLastError(); + if (IOErrorIsPermanent(err)) { + LogDebug( + BCLog::HTTP, + "Permanent read error from %s (id=%llu): %s", + client->m_origin, + client->m_id, + NetworkErrorString(err)); + // TODO: Disconnect + } + } else if (nrecv == 0) { + LogDebug( + BCLog::HTTP, + "Received EOF from %s (id=%llu)", + client->m_origin, + client->m_id); + // TODO: Disconnect + } else { + // Copy data from socket buffer to client receive buffer + client->m_recv_buffer.insert( + client->m_recv_buffer.end(), + buf, + buf + nrecv); + // Process as much received data as we can + MaybeDispatchRequestsFromClient(client); + } + } + } +} + void HTTPServer::SocketHandlerListening(const Sock::EventsPerSock& events_per_sock) { for (const auto& sock : m_listen) { @@ -1131,8 +1190,66 @@ void HTTPServer::ThreadSocketHandler() m_interrupt_net.sleep_for(SELECT_TIMEOUT); } + // Service (send/receive) each of the already connected sockets. + SocketHandlerConnected(io_readiness); + // Accept new connections from listening sockets. SocketHandlerListening(io_readiness.events_per_sock); } } + +void HTTPServer::MaybeDispatchRequestsFromClient(const std::shared_ptr& client) const +{ + // Try reading (potentially multiple) HTTP requests from the buffer + while (!client->m_recv_buffer.empty()) { + // Create a new request object and try to fill it with data from the receive buffer + auto req = std::make_unique(client); + try { + // Stop reading if we need more data from the client to parse a complete request + if (!client->ReadRequest(*req)) break; + } catch (const std::runtime_error& e) { + LogDebug( + BCLog::HTTP, + "Error reading HTTP request from client %s (id=%llu): %s", + client->m_origin, + client->m_id, + e.what()); + + // We failed to read a complete request from the buffer + // TODO: respond with HTTP_BAD_REQUEST and disconnect + + return; + } + + // We read a complete request from the buffer into the queue + LogDebug( + BCLog::HTTP, + "Received a %s request for %s from %s (id=%llu)", + req->m_method, + req->m_target, + client->m_origin, + client->m_id); + + // handle request + m_request_dispatcher(std::move(req)); + } +} + +bool HTTPRemoteClient::ReadRequest(HTTPRequest& req) +{ + LineReader reader(m_recv_buffer, MAX_HEADERS_SIZE); + + if (!req.LoadControlData(reader)) return false; + if (!req.LoadHeaders(reader)) return false; + if (!req.LoadBody(reader)) return false; + + // Remove the bytes read out of the buffer. + // If one of the above calls throws an error, the caller must + // catch it and disconnect the client. + m_recv_buffer.erase( + m_recv_buffer.begin(), + m_recv_buffer.begin() + reader.Consumed()); + + return true; +} } // namespace http_bitcoin diff --git a/src/httpserver.h b/src/httpserver.h index 73a06441a80..092c593533b 100644 --- a/src/httpserver.h +++ b/src/httpserver.h @@ -281,6 +281,8 @@ public: std::string StringifyHeaders() const; }; +class HTTPRemoteClient; + class HTTPRequest { public: @@ -290,6 +292,13 @@ public: HTTPHeaders m_headers; std::string m_body; + //! Pointer to the client that made the request so we know who to respond to. + std::shared_ptr m_client; + + explicit HTTPRequest(std::shared_ptr client) : m_client{std::move(client)} {} + //! Construct with a null client for unit tests + explicit HTTPRequest() : m_client{} {} + /** * Methods that attempt to parse HTTP request fields line-by-line * from a receive buffer. @@ -305,8 +314,6 @@ public: /// @} }; -class HTTPRemoteClient; - class HTTPServer { public: @@ -315,6 +322,8 @@ public: */ using Id = uint64_t; + explicit HTTPServer(std::function&&)> func) : m_request_dispatcher{std::move(func)} {} + virtual ~HTTPServer() { Assume(!m_thread_socket_handler.joinable()); // Missing call to JoinSocketsThreads() @@ -344,13 +353,6 @@ public: */ size_t GetConnectionsCount() const { return m_connected_size.load(std::memory_order_acquire); } - /** - * This is a temporary method used to get a pointer to an HTTPRemoteClient - * so its connection can be inspected in a unit test. - * It will be removed in a future commit. - */ - const HTTPRemoteClient& GetFirstConnection() { return *m_connected.front(); } - /** * Start the necessary threads for sockets IO. */ @@ -436,6 +438,11 @@ private: */ std::thread m_thread_socket_handler; + /* + * What to do with HTTP requests once received, validated and parsed + */ + std::function&&)> m_request_dispatcher; + /** * Accept a connection. * @param[in] listen_sock Socket on which to accept the connection. @@ -457,6 +464,12 @@ private: */ void NewSockAccepted(std::unique_ptr&& sock, const CService& addr); + /** + * Do the read/write for connected sockets that are ready for IO. + * @param[in] io_readiness Which sockets are ready and their corresponding HTTPRemoteClients. + */ + void SocketHandlerConnected(const IOReadiness& io_readiness) const; + /** * Accept incoming connections, one from each read-ready listening socket. * @param[in] events_per_sock Sockets that are ready for IO. @@ -475,6 +488,15 @@ private: * This is the main I/O loop of the server. */ void ThreadSocketHandler(); + + /** + * Try to read HTTPRequests from a client's receive buffer. + * Complete requests are dispatched, incomplete requests are + * left in the buffer to wait for more data. Some read errors + * will mark this client for disconnection. + * @param[in] client The HTTPRemoteClient to read requests from + */ + void MaybeDispatchRequestsFromClient(const std::shared_ptr& client) const; }; class HTTPRemoteClient @@ -489,6 +511,13 @@ public: //! IP:port of connected client, cached for logging purposes const std::string m_origin; + /** + * In lieu of an intermediate transport class like p2p uses, + * we copy data from the socket buffer to the client object + * and attempt to read HTTP requests from here. + */ + std::vector m_recv_buffer{}; + /** * Mutex that serializes the Send() and Recv() calls on `m_sock`. Reading * from the client occurs in the I/O thread but writing back to a client @@ -511,6 +540,13 @@ public: // Disable copies (should only be used as shared pointers) HTTPRemoteClient(const HTTPRemoteClient&) = delete; HTTPRemoteClient& operator=(const HTTPRemoteClient&) = delete; + + /** + * Try to read an HTTP request from the receive buffer. + * @param[in] req A HTTPRequest to read into + * @returns true upon reading a complete request, otherwise false (may throw). + */ + bool ReadRequest(HTTPRequest& req); }; } // namespace http_bitcoin diff --git a/src/test/httpserver_tests.cpp b/src/test/httpserver_tests.cpp index b30a76f0458..bb9da546d74 100644 --- a/src/test/httpserver_tests.cpp +++ b/src/test/httpserver_tests.cpp @@ -19,6 +19,16 @@ using http_bitcoin::MAX_BODY_SIZE; using http_bitcoin::MAX_HEADERS_SIZE; using util::LineReader; +// HTTP request captured from bitcoin-cli +constexpr std::string_view full_request = "POST / HTTP/1.1\r\n" + "Host: 127.0.0.1\r\n" + "Connection: close\r\n" + "Content-Type: application/json\r\n" + "Authorization: Basic X19jb29raWVfXzo5OGQ5ODQ3MWNmNjg0NzAzYTkzN2EzNzk0ZDFlODQ1NjZmYTRkZjJiMzFkYjhhODI4ZGY4MjVjOTg5ZGI4OTVl\r\n" + "Content-Length: 46\r\n" + "\r\n" + R"({"method":"getblockcount","params":[],"id":1})""\n"; + BOOST_FIXTURE_TEST_SUITE(httpserver_tests, SocketTestingSetup) BOOST_AUTO_TEST_CASE(test_query_parameters) @@ -174,15 +184,6 @@ BOOST_AUTO_TEST_CASE(http_response_tests) BOOST_AUTO_TEST_CASE(http_request_tests) { { - // Reading request captured from bitcoin-cli - constexpr std::string_view full_request = "POST / HTTP/1.1\r\n" - "Host: 127.0.0.1\r\n" - "Connection: close\r\n" - "Content-Type: application/json\r\n" - "Authorization: Basic X19jb29raWVfXzo5OGQ5ODQ3MWNmNjg0NzAzYTkzN2EzNzk0ZDFlODQ1NjZmYTRkZjJiMzFkYjhhODI4ZGY4MjVjOTg5ZGI4OTVl\r\n" - "Content-Length: 46\r\n" - "\r\n" - R"({"method":"getblockcount","params":[],"id":1})""\n"; HTTPRequest req; LineReader reader(full_request, MAX_HEADERS_SIZE); BOOST_CHECK(req.LoadControlData(reader)); @@ -377,7 +378,16 @@ BOOST_AUTO_TEST_CASE(http_request_tests) BOOST_AUTO_TEST_CASE(http_server_socket_tests) { - HTTPServer server; + // Prepare a request handler that just stores received requests so we can examine them. + // Mutex is required to prevent a race between this test's main thread and the server's I/O loop. + Mutex requests_mutex; + std::deque> requests; + auto StoreRequest = [&](std::unique_ptr&& req) { + LOCK(requests_mutex); + requests.push_back(std::move(req)); + }; + + HTTPServer server{StoreRequest}; { // We can only bind to NET_IPV4 and NET_IPV6 @@ -403,8 +413,8 @@ BOOST_AUTO_TEST_CASE(http_server_socket_tests) // No connections yet BOOST_CHECK_EQUAL(server.GetConnectionsCount(), 0); - // Create a mock client and add it to the local CreateSock queue - ConnectClient(); + // Create a mock client with pre-loaded request data and add it to the local CreateSock queue + ConnectClient(std::as_bytes(std::span(full_request))); // Wait up to a minute to find and connect the client in the I/O loop int attempts{6000}; @@ -413,9 +423,31 @@ BOOST_AUTO_TEST_CASE(http_server_socket_tests) BOOST_REQUIRE(--attempts > 0); } - // Inspect the connection - const HTTPRemoteClient& client = server.GetFirstConnection(); - BOOST_CHECK_EQUAL(client.m_origin, "5.5.5.5:6789"); + // Prepare a pointer to the client, we'll assign it from the request itself. + std::shared_ptr client; + + // Wait up to a minute to read the request from the client. + // Given that the mock client is itself a mock socket + // with hard-coded data it should only take a fraction of that. + attempts = 6000; + while (true) { + { + LOCK(requests_mutex); + // Connected client should have one request already from the static content. + if (requests.size() == 1) { + // Check the received request + BOOST_CHECK_EQUAL(requests.front()->m_body, R"({"method":"getblockcount","params":[],"id":1})""\n"); + + // Inspect the connection pointed to from the request + client = requests.front()->m_client; + BOOST_CHECK_EQUAL(client->m_origin, "5.5.5.5:6789"); + + break; + } + } + std::this_thread::sleep_for(10ms); + BOOST_REQUIRE(--attempts > 0); + } // Stop the I/O thread before modifying m_connected. CloseConnection() is a // temporary test-only API that is not thread-safe against GenerateWaitSockets(), @@ -423,7 +455,7 @@ BOOST_AUTO_TEST_CASE(http_server_socket_tests) server.InterruptNet(); server.JoinSocketsThreads(); // Close connection - BOOST_REQUIRE(server.CloseConnection(client)); + BOOST_REQUIRE(server.CloseConnection(*client)); // Close all listening sockets server.StopListening(); } diff --git a/src/test/util/setup_common.cpp b/src/test/util/setup_common.cpp index 846f47032aa..6a4f8ff178b 100644 --- a/src/test/util/setup_common.cpp +++ b/src/test/util/setup_common.cpp @@ -658,12 +658,13 @@ SocketTestingSetup::~SocketTestingSetup() CreateSock = m_create_sock_orig; } -void SocketTestingSetup::ConnectClient() +void SocketTestingSetup::ConnectClient(std::span data) { // I/O pipes for a mock Connected Socket we can read and write to. auto connected_socket_pipes(std::make_shared()); - // TODO: Insert a payload + // Insert the payload + connected_socket_pipes->recv.PushBytes(data.data(), data.size()); // Create the Mock Connected Socket that represents a client. // It needs I/O pipes but its queue can remain empty diff --git a/src/test/util/setup_common.h b/src/test/util/setup_common.h index bf9ec42ebf2..6773a689b84 100644 --- a/src/test/util/setup_common.h +++ b/src/test/util/setup_common.h @@ -256,8 +256,8 @@ public: explicit SocketTestingSetup(); ~SocketTestingSetup(); - //! Connect to the socket with a mock client (a DynSock) - void ConnectClient(); + //! Connect to the socket with a mock client (a DynSock) and send pre-loaded data. + void ConnectClient(std::span data); private: //! Save the original value of CreateSock here and restore it when the test ends.