From 56d3a3990291810f71d3c4fd6ee99643c44c89c6 Mon Sep 17 00:00:00 2001 From: SSE4 Date: Tue, 18 Aug 2026 18:52:44 +0700 Subject: [PATCH 1/3] feat core: add TRACE as a first-class HTTP method TRACE is a standard HTTP method (RFC 9110 9.3.8) that llhttp parses fine, but server::http::HttpMethod had no value for it: an incoming TRACE was mapped to kUnknown and routing answered 405, and "TRACE" in a handler 'method:' list threw at service start. Add kTrace to the enum, to the string conversions, to the llhttp mapping and to kHandlerMethods, so a handler can register for TRACE and receive it like any other method. Answering with the RFC message/http loopback echo stays up to the handler - the framework only needs to be able to route the method. HTTP/2 takes the method from the ':method' pseudo-header rather than from llhttp, so it is covered by a separate functional test. --- .../http2server/static_config.yaml | 2 +- .../http2server/tests/conftest.py | 19 +++++++ .../http2server/tests/test_high_level.py | 13 +++++ .../userver/server/http/http_method.hpp | 1 + core/src/server/http/handler_methods.hpp | 1 + core/src/server/http/http_method.cpp | 8 +++ core/src/server/http/http_method_test.cpp | 51 +++++++++++++++++++ .../server/http/http_request_method_test.cpp | 1 + core/src/server/http/http_request_parser.cpp | 2 + 9 files changed, 97 insertions(+), 1 deletion(-) create mode 100644 core/src/server/http/http_method_test.cpp diff --git a/core/functional_tests/http2server/static_config.yaml b/core/functional_tests/http2server/static_config.yaml index 2593a497ae9d..c29f3a631589 100644 --- a/core/functional_tests/http2server/static_config.yaml +++ b/core/functional_tests/http2server/static_config.yaml @@ -74,7 +74,7 @@ components_manager: handler-http2: path: /http2server - method: GET,POST,PUT,DELETE,HEAD + method: GET,POST,PUT,DELETE,HEAD,TRACE task_processor: main-task-processor throttling_enabled: false max_request_size: 2097152 # 2Mib diff --git a/core/functional_tests/http2server/tests/conftest.py b/core/functional_tests/http2server/tests/conftest.py index 945444423478..f90760eac44f 100644 --- a/core/functional_tests/http2server/tests/conftest.py +++ b/core/functional_tests/http2server/tests/conftest.py @@ -161,6 +161,25 @@ async def head( timeout, ) + async def trace( + self, + path, + params={}, + headers={}, + data=None, + json={}, + timeout=DEFAULT_TIMEOUT, + ) -> httpx.Response: + return await self._request( + 'TRACE', + path, + params, + headers, + data, + json, + timeout, + ) + async def _request( self, method, diff --git a/core/functional_tests/http2server/tests/test_high_level.py b/core/functional_tests/http2server/tests/test_high_level.py index 37001fe46fa8..2ab919911e41 100644 --- a/core/functional_tests/http2server/tests/test_high_level.py +++ b/core/functional_tests/http2server/tests/test_high_level.py @@ -73,6 +73,19 @@ async def test_headers(http2_client): assert hval == r.text +async def test_trace_is_routed(http2_client): + # TRACE arrives as a ':method' pseudo-header, so unlike HTTP/1.1 it never + # goes through llhttp; the handler must be reached over HTTP/2 as well. + hval = 'traced' + r = await http2_client.trace( + DEFAULT_PATH, + params={'type': 'echo-header'}, + headers={'echo-header': hval}, + ) + assert 200 == r.status_code + assert hval == r.text + + async def test_head_response_has_no_body(http2_client): r = await http2_client.head( DEFAULT_PATH, diff --git a/core/include/userver/server/http/http_method.hpp b/core/include/userver/server/http/http_method.hpp index 5481511b6833..bac1c175f03e 100644 --- a/core/include/userver/server/http/http_method.hpp +++ b/core/include/userver/server/http/http_method.hpp @@ -27,6 +27,7 @@ enum class HttpMethod { kPatch, kConnect, kOptions, + kTrace, kUnknown, }; diff --git a/core/src/server/http/handler_methods.hpp b/core/src/server/http/handler_methods.hpp index ba0ead79a732..01080a01e86a 100644 --- a/core/src/server/http/handler_methods.hpp +++ b/core/src/server/http/handler_methods.hpp @@ -16,6 +16,7 @@ inline constexpr HttpMethod kHandlerMethods[] = { HttpMethod::kDelete, HttpMethod::kPatch, HttpMethod::kOptions, + HttpMethod::kTrace, HttpMethod::kUnknown }; diff --git a/core/src/server/http/http_method.cpp b/core/src/server/http/http_method.cpp index 3e9889d9a24a..e8783bcbd59e 100644 --- a/core/src/server/http/http_method.cpp +++ b/core/src/server/http/http_method.cpp @@ -19,6 +19,7 @@ struct HttpMethodStrings { const std::string kConnect = "CONNECT"; const std::string kPatch = "PATCH"; const std::string kOptions = "OPTIONS"; + const std::string kTrace = "TRACE"; const std::string kUnknown = "unknown"; }; // NOLINTEND(readability-identifier-naming) @@ -81,6 +82,11 @@ HttpMethod HttpMethodFromString(std::string_view method_str) { result = HttpMethod::kOptions; } break; + case 'T': + if (method_str == strings.kTrace) { + result = HttpMethod::kTrace; + } + break; } } @@ -116,6 +122,8 @@ const std::string& ToString(HttpMethod method) noexcept { return strings.kPatch; case HttpMethod::kOptions: return strings.kOptions; + case HttpMethod::kTrace: + return strings.kTrace; case HttpMethod::kUnknown: return strings.kUnknown; } diff --git a/core/src/server/http/http_method_test.cpp b/core/src/server/http/http_method_test.cpp new file mode 100644 index 000000000000..8cb490c5e04c --- /dev/null +++ b/core/src/server/http/http_method_test.cpp @@ -0,0 +1,51 @@ +#include + +#include + +#include + +USERVER_NAMESPACE_BEGIN + +namespace { +namespace sh = server::http; +} + +UTEST(ServerHttpMethodTest, ToString) { + EXPECT_EQ(sh::ToString(sh::HttpMethod::kDelete), "DELETE"); + EXPECT_EQ(sh::ToString(sh::HttpMethod::kGet), "GET"); + EXPECT_EQ(sh::ToString(sh::HttpMethod::kHead), "HEAD"); + EXPECT_EQ(sh::ToString(sh::HttpMethod::kPost), "POST"); + EXPECT_EQ(sh::ToString(sh::HttpMethod::kPut), "PUT"); + EXPECT_EQ(sh::ToString(sh::HttpMethod::kPatch), "PATCH"); + EXPECT_EQ(sh::ToString(sh::HttpMethod::kConnect), "CONNECT"); + EXPECT_EQ(sh::ToString(sh::HttpMethod::kOptions), "OPTIONS"); + EXPECT_EQ(sh::ToString(sh::HttpMethod::kTrace), "TRACE"); +} + +UTEST(ServerHttpMethodTest, FromString) { + EXPECT_EQ(sh::HttpMethodFromString("DELETE"), sh::HttpMethod::kDelete); + EXPECT_EQ(sh::HttpMethodFromString("GET"), sh::HttpMethod::kGet); + EXPECT_EQ(sh::HttpMethodFromString("HEAD"), sh::HttpMethod::kHead); + EXPECT_EQ(sh::HttpMethodFromString("POST"), sh::HttpMethod::kPost); + EXPECT_EQ(sh::HttpMethodFromString("PUT"), sh::HttpMethod::kPut); + EXPECT_EQ(sh::HttpMethodFromString("PATCH"), sh::HttpMethod::kPatch); + EXPECT_EQ(sh::HttpMethodFromString("CONNECT"), sh::HttpMethod::kConnect); + EXPECT_EQ(sh::HttpMethodFromString("OPTIONS"), sh::HttpMethod::kOptions); + EXPECT_EQ(sh::HttpMethodFromString("TRACE"), sh::HttpMethod::kTrace); + + UEXPECT_THROW(sh::HttpMethodFromString("TRAC"), std::runtime_error); + UEXPECT_THROW(sh::HttpMethodFromString("TRACER"), std::runtime_error); + UEXPECT_THROW(sh::HttpMethodFromString("trace"), std::runtime_error); +} + +// TRACE must be registrable in a handler `method:` list, unlike CONNECT, which +// userver never routes to a handler. +UTEST(ServerHttpMethodTest, IsHandlerMethod) { + EXPECT_TRUE(sh::IsHandlerMethod(sh::HttpMethod::kGet)); + EXPECT_TRUE(sh::IsHandlerMethod(sh::HttpMethod::kOptions)); + EXPECT_TRUE(sh::IsHandlerMethod(sh::HttpMethod::kTrace)); + + EXPECT_FALSE(sh::IsHandlerMethod(sh::HttpMethod::kConnect)); +} + +USERVER_NAMESPACE_END diff --git a/core/src/server/http/http_request_method_test.cpp b/core/src/server/http/http_request_method_test.cpp index 6eb9ff410ead..2bda89aa7b16 100644 --- a/core/src/server/http/http_request_method_test.cpp +++ b/core/src/server/http/http_request_method_test.cpp @@ -40,6 +40,7 @@ INSTANTIATE_UTEST_SUITE_P( MethodsData{"CONNECT", HttpMethod::kConnect}, MethodsData{"PATCH", HttpMethod::kPatch}, MethodsData{"OPTIONS", HttpMethod::kOptions}, + MethodsData{"TRACE", HttpMethod::kTrace}, MethodsData{"GE", HttpMethod::kUnknown}, MethodsData{"GETT", HttpMethod::kUnknown}, MethodsData{"get", HttpMethod::kUnknown}, diff --git a/core/src/server/http/http_request_parser.cpp b/core/src/server/http/http_request_parser.cpp index be2d6fab3ce1..913dcb76d6ae 100644 --- a/core/src/server/http/http_request_parser.cpp +++ b/core/src/server/http/http_request_parser.cpp @@ -58,6 +58,8 @@ HttpMethod ConvertHttpMethod(llhttp_method method) { return HttpMethod::kPatch; case HTTP_OPTIONS: return HttpMethod::kOptions; + case HTTP_TRACE: + return HttpMethod::kTrace; default: return HttpMethod::kUnknown; } From 18365f7cb491d97dab852e9d5465d04ab78fb19a Mon Sep 17 00:00:00 2001 From: SSE4 Date: Tue, 18 Aug 2026 18:52:45 +0700 Subject: [PATCH 2/3] feat core: allow sending TRACE requests with the HTTP client Mirrors the server-side TRACE support in clients::http::HttpMethod, so the enum and HttpMethodFromString() cover the same set of methods on both sides. TRACE carries no request body, so it is sent as a custom request just like DELETE and OPTIONS. --- core/include/userver/clients/http/request.hpp | 2 +- core/src/clients/http/http_method_test.cpp | 2 ++ core/src/clients/http/request.cpp | 3 ++- core/src/clients/http/request_state.cpp | 1 + core/utest/src/utest/http_server_mock.cpp | 2 ++ 5 files changed, 8 insertions(+), 2 deletions(-) diff --git a/core/include/userver/clients/http/request.hpp b/core/include/userver/clients/http/request.hpp index a5bc2d341fe8..9a2d49b65500 100644 --- a/core/include/userver/clients/http/request.hpp +++ b/core/include/userver/clients/http/request.hpp @@ -48,7 +48,7 @@ class EasyWrapper; } // namespace impl /// @brief HTTP request method -enum class HttpMethod { kGet, kPost, kHead, kPut, kDelete, kPatch, kOptions }; +enum class HttpMethod { kGet, kPost, kHead, kPut, kDelete, kPatch, kOptions, kTrace }; /// @brief Convert HTTP method enum value to string std::string_view ToStringView(HttpMethod method); diff --git a/core/src/clients/http/http_method_test.cpp b/core/src/clients/http/http_method_test.cpp index 3f6b3c3d2d20..47220051ed08 100644 --- a/core/src/clients/http/http_method_test.cpp +++ b/core/src/clients/http/http_method_test.cpp @@ -17,6 +17,7 @@ UTEST(ClientHttpMethodTest, Convert) { EXPECT_EQ(ch::ToStringView(ch::HttpMethod::kPut), "PUT"); EXPECT_EQ(ch::ToStringView(ch::HttpMethod::kDelete), "DELETE"); EXPECT_EQ(ch::ToStringView(ch::HttpMethod::kOptions), "OPTIONS"); + EXPECT_EQ(ch::ToStringView(ch::HttpMethod::kTrace), "TRACE"); EXPECT_EQ(ch::HttpMethodFromString("GET"), ch::HttpMethod::kGet); EXPECT_EQ(ch::HttpMethodFromString("HEAD"), ch::HttpMethod::kHead); @@ -25,6 +26,7 @@ UTEST(ClientHttpMethodTest, Convert) { EXPECT_EQ(ch::HttpMethodFromString("PUT"), ch::HttpMethod::kPut); EXPECT_EQ(ch::HttpMethodFromString("DELETE"), ch::HttpMethod::kDelete); EXPECT_EQ(ch::HttpMethodFromString("OPTIONS"), ch::HttpMethod::kOptions); + EXPECT_EQ(ch::HttpMethodFromString("TRACE"), ch::HttpMethod::kTrace); UEXPECT_THROW(ch::HttpMethodFromString("123"), std::runtime_error); } diff --git a/core/src/clients/http/request.cpp b/core/src/clients/http/request.cpp index f9f02d4cb868..43e3c7be2566 100644 --- a/core/src/clients/http/request.cpp +++ b/core/src/clients/http/request.cpp @@ -43,7 +43,8 @@ constexpr utils::TrivialBiMap kHttpMethodMap([](auto selector) { .Case(HttpMethod::kPut, "PUT") .Case(HttpMethod::kPatch, "PATCH") .Case(HttpMethod::kDelete, "DELETE") - .Case(HttpMethod::kOptions, "OPTIONS"); + .Case(HttpMethod::kOptions, "OPTIONS") + .Case(HttpMethod::kTrace, "TRACE"); }); curl::easy::http_version_t ToNative(HttpVersion version) { diff --git a/core/src/clients/http/request_state.cpp b/core/src/clients/http/request_state.cpp index a1b49e6bfd4a..600cc1e10db4 100644 --- a/core/src/clients/http/request_state.cpp +++ b/core/src/clients/http/request_state.cpp @@ -753,6 +753,7 @@ void RequestState::SetMethod(HttpMethod method) { switch (method) { case HttpMethod::kDelete: case HttpMethod::kOptions: + case HttpMethod::kTrace: easy().set_custom_request(ToString(method)); break; case HttpMethod::kGet: diff --git a/core/utest/src/utest/http_server_mock.cpp b/core/utest/src/utest/http_server_mock.cpp index ac80c24bfaee..d5c46721a12e 100644 --- a/core/utest/src/utest/http_server_mock.cpp +++ b/core/utest/src/utest/http_server_mock.cpp @@ -29,6 +29,8 @@ clients::http::HttpMethod ConvertHttpMethod(llhttp_method method) { return clients::http::HttpMethod::kPatch; case HTTP_OPTIONS: return clients::http::HttpMethod::kOptions; + case HTTP_TRACE: + return clients::http::HttpMethod::kTrace; default: ADD_FAILURE() << "Unknown HTTP method " << method; return clients::http::HttpMethod::kGet; From 6b12f5d635861a96acc496deb2390637183b8ca8 Mon Sep 17 00:00:00 2001 From: SSE4 Date: Tue, 18 Aug 2026 22:39:08 +0700 Subject: [PATCH 3/3] feat core: support websockets over HTTP/2.0 (RFC 8441) Accept websockets bootstrapped with the extended CONNECT method of RFC 8441, as Chrome, Firefox, .NET ClientWebSocket, HAProxy and Envoy send them. Enabled per listener with the new `http2-session.enable_connect_protocol` option (default off), which makes the server advertise SETTINGS_ENABLE_CONNECT_PROTOCOL. Such a request carries `:method: CONNECT` and `:protocol: websocket` instead of the Upgrade/Connection headers, and is answered with a plain 200 rather than a 101. It is routed as a GET so that path-registered handlers match, the way reverse proxies do when converting an HTTP/1.1 upgrade, so server::handlers::WebsocketHandlerBase subclasses need no changes at all: the same handler now serves both transports. HttpRequest::IsWebsocketExtendedConnect() tells them apart where it matters. Unlike the HTTP/1.1 upgrade, which hands the socket over and stops doing HTTP, the websocket lives inside a single stream, so the connection keeps multiplexing ordinary requests around it. The new Http2StreamRw presents that stream as an engine::io::RwBase: incoming DATA frames land in a lock-free pipe that outlives the stream (the peer may reset it while the handler unwinds), and writes are pushed as streaming events, since nghttp2_session may only be touched on the connection task. Notable details: * `:method` and `:protocol` may arrive in any order, so a CONNECT stream leaves its method unset -- which defers url parsing -- until the header block ends. * The read pipe is created when the request is finalized rather than on upgrade: a client may send websocket bytes before the handler accepted the stream. * Http2StreamRw::ReadNoblock yields. Over HTTP/1.1 the busy loop of WebSocketConnection::TryRecv is self-sufficient because the handler owns the socket; over HTTP/2.0 the pipe is filled by the connection task, which a non-yielding spinner starves. * A handler that does not upgrade must not answer 2xx to an extended CONNECT, as that claims the tunnel is established; Http2ResponseWriter rewrites it. * OnStreamClose now tolerates an already-removed stream. Rejecting a CONNECT (reset + remove) otherwise threw out of an nghttp2 callback and killed the whole connection -- the same latent bug the FinalizeRequest error path had. Also revives HTTP/2.0 streaming: Http2Session::HandleStreamingEvents() had lost its only caller in the connection refactor, so streamed bodies never reached the wire either. The connection loop awaits the streaming event again, which needs it to be non-auto-resetting to be usable with WaitAny. Tested by core/functional_tests/websocket_http2 (the same service as the HTTP/1.1 websocket test, driven by a hand-rolled h2 + wsproto client, as no mainstream Python websocket client speaks RFC 8441), including multiplexing an ordinary request onto a connection with a live websocket, and by unit tests for the settings, the routing and the negative cases. Green under asan/ubsan and TSAN. --- core/functional_tests/CMakeLists.txt | 3 + core/functional_tests/websocket/service.cpp | 14 ++ .../websocket/static_config.yaml | 4 + .../websocket_http2/CMakeLists.txt | 8 + .../websocket_http2/static_config.yaml | 72 ++++++ .../websocket_http2/tests/conftest.py | 196 ++++++++++++++++ .../tests/test_websocket_http2.py | 122 ++++++++++ .../userver/server/http/http_request.hpp | 7 + .../server/http/http_request_builder.hpp | 2 + core/src/server/component.yaml | 5 + .../src/server/handlers/websocket_handler.cpp | 57 +++-- core/src/server/http/http2_session.cpp | 159 +++++++++++-- core/src/server/http/http2_session.hpp | 12 +- core/src/server/http/http2_session_test.cpp | 214 ++++++++++++++++++ core/src/server/http/http2_stream.cpp | 17 ++ core/src/server/http/http2_stream.hpp | 24 ++ core/src/server/http/http2_stream_rw.cpp | 207 +++++++++++++++++ core/src/server/http/http2_stream_rw.hpp | 115 ++++++++++ core/src/server/http/http2_writer.cpp | 31 ++- core/src/server/http/http_request.cpp | 2 + core/src/server/http/http_request_builder.cpp | 5 + .../server/http/http_request_constructor.cpp | 4 + .../server/http/http_request_constructor.hpp | 1 + core/src/server/http/http_request_impl.hpp | 3 + core/src/server/net/connection_config.cpp | 1 + core/src/server/net/connection_config.hpp | 3 + core/src/server/net/http2_connection.cpp | 64 +++++- core/src/server/net/http2_connection.hpp | 7 +- scripts/docs/en/userver/http_server.md | 25 +- .../en/userver/tutorial/websocket_service.md | 5 + testsuite/requirements-internal-tests.txt | 3 + .../include/userver/http/common_headers.hpp | 2 + 32 files changed, 1344 insertions(+), 50 deletions(-) create mode 100644 core/functional_tests/websocket_http2/CMakeLists.txt create mode 100644 core/functional_tests/websocket_http2/static_config.yaml create mode 100644 core/functional_tests/websocket_http2/tests/conftest.py create mode 100644 core/functional_tests/websocket_http2/tests/test_websocket_http2.py create mode 100644 core/src/server/http/http2_stream_rw.cpp create mode 100644 core/src/server/http/http2_stream_rw.hpp diff --git a/core/functional_tests/CMakeLists.txt b/core/functional_tests/CMakeLists.txt index 85d2d9353eb7..42b39e3fb2d3 100644 --- a/core/functional_tests/CMakeLists.txt +++ b/core/functional_tests/CMakeLists.txt @@ -56,6 +56,9 @@ add_dependencies(${PROJECT_NAME} ${PROJECT_NAME}-cache-update) add_subdirectory(websocket) add_dependencies(${PROJECT_NAME} ${PROJECT_NAME}-websocket) +add_subdirectory(websocket_http2) +add_dependencies(${PROJECT_NAME} ${PROJECT_NAME}-websocket-http2) + # WebSocket client requires curl >= 7.86 with WebSocket support if(CURL_VERSION_STRING VERSION_GREATER_EQUAL "7.86") add_subdirectory(websocket_client) diff --git a/core/functional_tests/websocket/service.cpp b/core/functional_tests/websocket/service.cpp index 86a66daf434e..c2025fb21863 100644 --- a/core/functional_tests/websocket/service.cpp +++ b/core/functional_tests/websocket/service.cpp @@ -146,10 +146,24 @@ class WebsocketsPingPongHandler final : public server::handlers::WebsocketHandle } }; +/// Only configured by the HTTP/2.0 variant of this service, to check what an extended +/// CONNECT of RFC 8441 aimed at an ordinary handler answers. +class PlainHandler final : public server::handlers::HttpHandlerBase { +public: + static constexpr std::string_view kName = "plain-handler"; + + using HttpHandlerBase::HttpHandlerBase; + + std::string HandleRequestThrow(const server::http::HttpRequest&, server::request::RequestContext&) const override { + return "Not a websocket handler"; + } +}; + int main(int argc, char* argv[]) { const auto component_list = components::MinimalServerComponentList() .Append() + .Append() .Append() .Append() .Append() diff --git a/core/functional_tests/websocket/static_config.yaml b/core/functional_tests/websocket/static_config.yaml index 8730de66ec27..e848c2b65845 100644 --- a/core/functional_tests/websocket/static_config.yaml +++ b/core/functional_tests/websocket/static_config.yaml @@ -45,6 +45,10 @@ components_manager: task_processor: main-task-processor max-remote-payload: 100000 fragment-size: 10 + plain-handler: # Unused here; exercised by the HTTP/2.0 variant of this service. + path: /plain + method: GET + task_processor: main-task-processor testsuite-support: diff --git a/core/functional_tests/websocket_http2/CMakeLists.txt b/core/functional_tests/websocket_http2/CMakeLists.txt new file mode 100644 index 000000000000..138884adc364 --- /dev/null +++ b/core/functional_tests/websocket_http2/CMakeLists.txt @@ -0,0 +1,8 @@ +project(userver-core-tests-websocket-http2 CXX) + +# The very same handlers as the HTTP/1.1 websocket test: a websocket handler is not +# supposed to notice which transport bootstrapped it. +add_executable(${PROJECT_NAME} "${CMAKE_CURRENT_SOURCE_DIR}/../websocket/service.cpp") +target_link_libraries(${PROJECT_NAME} userver::core) + +userver_chaos_testsuite_add() diff --git a/core/functional_tests/websocket_http2/static_config.yaml b/core/functional_tests/websocket_http2/static_config.yaml new file mode 100644 index 000000000000..a27a001eb51f --- /dev/null +++ b/core/functional_tests/websocket_http2/static_config.yaml @@ -0,0 +1,72 @@ +components_manager: + + task_processors: # Task processor is an executor for coroutine tasks + main-task-processor: # Make a task processor for CPU-bound coroutine tasks. + worker_threads: 4 # Process tasks in 4 threads. + fs-task-processor: # Make a separate task processor for filesystem bound tasks. + worker_threads: 4 + + default_task_processor: main-task-processor + + components: # Configuring components that were registered via component_list + server: + listener: # configuring the main listening socket... + connection: + http-version: 2 + http2-session: + enable_connect_protocol: true + port: 8080 # ...to listen on this port and... + task_processor: main-task-processor # ...process incoming requests on this task processor. + logging: + fs-task-processor: fs-task-processor + loggers: + default: + file_path: '@stderr' + level: debug + overflow_behavior: discard # Drop logs if the system is too busy to write them down. + + websocket-handler: # Finally! Websocket handler. + path: /chat # Registering handlers '/*' find files. + method: GET # Handle only GET requests. + task_processor: main-task-processor # Run it on CPU bound task processor + max-remote-payload: 100000 + fragment-size: 10 + websocket-handler-alt: # Finally! Websocket handler. + path: /handler-alt # Registering handlers '/*' find files. + method: GET # Handle only GET requests. + task_processor: main-task-processor # Run it on CPU bound task processor + max-remote-payload: 100000 + fragment-size: 10 + websocket-duplex-handler: # Finally! Websocket handler. + path: /duplex # Registering handlers '/*' find files. + method: GET # Handle only GET requests. + task_processor: main-task-processor # Run it on CPU bound task processor + max-remote-payload: 100000 + fragment-size: 10 + websocket-ping-pong-handler: + path: /ping-pong + method: GET + task_processor: main-task-processor + max-remote-payload: 100000 + fragment-size: 10 + plain-handler: # An ordinary handler, to check that it never accepts a tunnel. + path: /plain + method: GET + task_processor: main-task-processor + + testsuite-support: + + http-client: + http-client-core: + fs-task-processor: main-task-processor + dns-client: + fs-task-processor: fs-task-processor + + tests-control: + method: POST + path: /tests/{action} + skip-unregistered-testpoints: true + task_processor: main-task-processor + testpoint-timeout: 10s + testpoint-url: $mockserver/testpoint + throttling_enabled: false diff --git a/core/functional_tests/websocket_http2/tests/conftest.py b/core/functional_tests/websocket_http2/tests/conftest.py new file mode 100644 index 000000000000..1daa1ba83329 --- /dev/null +++ b/core/functional_tests/websocket_http2/tests/conftest.py @@ -0,0 +1,196 @@ +import contextlib +import socket + +import h2.config +import h2.connection +import h2.events +import pytest +import wsproto.connection +import wsproto.events + +pytest_plugins = ['pytest_userver.plugins.core'] + + +class Rfc8441Error(Exception): + pass + + +class Rfc8441Client: + """A websocket bootstrapped over HTTP/2.0 with the extended CONNECT of RFC 8441. + + Hand-rolled because no mainstream Python websocket client speaks RFC 8441: the + handshake is done with `h2`, and `wsproto` provides plain RFC 6455 framing for + the bytes that then flow inside the DATA frames of the stream. + """ + + # Generous: a sanitized service on a loaded machine is slow, and a real hang still + # fails the test, only later. + def __init__(self, host: str, port: int, timeout: float = 60.0): + self._authority = f'{host}:{port}' + self._sock = socket.create_connection((host, port), timeout=timeout) + self._conn = h2.connection.H2Connection( + config=h2.config.H2Configuration(client_side=True), + ) + self._conn.initiate_connection() + self._flush() + self._ws = wsproto.connection.Connection( + wsproto.connection.ConnectionType.CLIENT, + ) + self._stream_id = None + self._pending = [] + self._got_settings = False + + # --- HTTP/2.0 plumbing --- + + def _flush(self): + self._sock.sendall(self._conn.data_to_send()) + + def _pump(self): + """Reads one batch of server bytes and returns the resulting h2 events.""" + data = self._sock.recv(65535) + if not data: + raise Rfc8441Error('the server closed the connection') + events = self._conn.receive_data(data) + self._flush() + return events + + @property + def enable_connect_protocol(self) -> int: + # The local default is 0, so wait for the server SETTINGS to actually arrive + # instead of reporting "not advertised" before it had a chance to. + while not self._got_settings: + for event in self._pump(): + if isinstance(event, h2.events.RemoteSettingsChanged): + self._got_settings = True + self._remember(event) + return self._conn.remote_settings.enable_connect_protocol + + def request(self, path: str, method: str = 'GET') -> int: + """Starts an ordinary request on its own stream and returns its id.""" + stream_id = self._conn.get_next_available_stream_id() + self._conn.send_headers( + stream_id, + [ + (':method', method), + (':scheme', 'http'), + (':path', path), + (':authority', self._authority), + ], + end_stream=True, + ) + self._flush() + return stream_id + + def status_of(self, stream_id: int) -> str: + status = None + while status is None: + # The whole batch has to be consumed even once the status is known: the + # server may well have put the response headers and the first bytes of the + # tunnelled protocol into one TCP segment. + for event in self._pump(): + if isinstance(event, h2.events.ResponseReceived) and event.stream_id == stream_id: + status = dict(event.headers)[b':status'].decode() + else: + self._remember(event) + return status + + # --- RFC 8441 --- + + def connect(self, path: str, extra_headers=()) -> str: + """Sends the extended CONNECT and returns the response `:status`.""" + assert self._stream_id is None, 'the client drives a single websocket' + self._stream_id = self._conn.get_next_available_stream_id() + self._conn.send_headers( + self._stream_id, + [ + (':method', 'CONNECT'), + (':protocol', 'websocket'), + (':scheme', 'http'), + (':path', path), + (':authority', self._authority), + ('sec-websocket-version', '13'), + *extra_headers, + ], + # The stream stays open for the lifetime of the websocket. + end_stream=False, + ) + self._flush() + + return self.status_of(self._stream_id) + + def _remember(self, event): + if isinstance(event, h2.events.DataReceived) and event.stream_id == self._stream_id: + if event.flow_controlled_length: + self._conn.acknowledge_received_data( + event.flow_controlled_length, + event.stream_id, + ) + self._flush() + if event.data: + self._ws.receive_data(event.data) + self._pending.extend(self._ws.events()) + elif isinstance(event, h2.events.StreamEnded) and event.stream_id == self._stream_id: + # Half-closing the stream is how the transport under the websocket goes away. + self._ws.receive_data(None) + self._pending.extend(self._ws.events()) + elif isinstance(event, h2.events.StreamReset) and event.stream_id == self._stream_id: + raise Rfc8441Error(f'the websocket stream was reset: {event.error_code}') + elif isinstance(event, h2.events.ConnectionTerminated): + raise Rfc8441Error(f'the connection was terminated: {event.error_code}') + + def send(self, event): + self._conn.send_data(self._stream_id, self._ws.send(event)) + self._flush() + + def send_text(self, payload: str): + self.send(wsproto.events.TextMessage(data=payload)) + + def send_bytes(self, payload: bytes): + self.send(wsproto.events.BytesMessage(data=payload)) + + def recv(self): + """Returns one websocket event, reassembling fragmented messages.""" + parts = None + while True: + while self._pending: + event = self._pending.pop(0) + if not isinstance(event, wsproto.events.Message): + return event + parts = event.data if parts is None else parts + event.data + if event.message_finished: + return type(event)(data=parts) + for event in self._pump(): + self._remember(event) + + def recv_message(self): + event = self.recv() + assert isinstance(event, wsproto.events.Message), event + return event.data + + def close(self, code: int = 1000): + self.send(wsproto.events.CloseConnection(code=code)) + return self.recv() + + def disconnect(self): + self._sock.close() + + +@pytest.fixture(name='rfc8441_client') +async def _rfc8441_client(service_client, service_port): + # `service_client` is required so that the daemon is up before we connect: the + # client speaks to the listener directly, bypassing the testsuite plumbing. + clients = [] + + @contextlib.contextmanager + def make_client(): + client = Rfc8441Client('localhost', service_port) + clients.append(client) + try: + yield client + finally: + client.disconnect() + + yield make_client + + for client in clients: + client.disconnect() diff --git a/core/functional_tests/websocket_http2/tests/test_websocket_http2.py b/core/functional_tests/websocket_http2/tests/test_websocket_http2.py new file mode 100644 index 000000000000..bdeb38717feb --- /dev/null +++ b/core/functional_tests/websocket_http2/tests/test_websocket_http2.py @@ -0,0 +1,122 @@ +"""Websockets bootstrapped over HTTP/2.0 with the extended CONNECT of RFC 8441. + +The service is the very same one the HTTP/1.1 websocket test drives; only the +listener speaks HTTP/2.0 with `enable_connect_protocol` on. +""" + +import pytest +import wsproto.events + +from conftest import Rfc8441Error + + +async def test_setting_is_advertised(rfc8441_client): + with rfc8441_client() as client: + assert client.enable_connect_protocol == 1 + + +async def test_echo(rfc8441_client): + with rfc8441_client() as client: + assert client.connect('/chat') == '200' + client.send_text('hello') + assert client.recv_message() == 'hello' + client.send_text('second message') + assert client.recv_message() == 'second message' + + +async def test_echo_bin(rfc8441_client): + with rfc8441_client() as client: + assert client.connect('/chat') == '200' + client.send_bytes(b'\x00\x01\x02\xff') + assert client.recv_message() == b'\x00\x01\x02\xff' + + +async def test_handshake_hook_sees_the_request(rfc8441_client): + with rfc8441_client() as client: + # The handler echoes the Origin back as its first message, which proves + # HandleHandshake() ran and saw the real headers. + assert client.connect('/chat', extra_headers=[('origin', 'localhost')]) == '200' + assert client.recv_message() == 'localhost' + + +async def test_close_handshake(rfc8441_client): + with rfc8441_client() as client: + assert client.connect('/chat') == '200' + client.send_text('hello') + assert client.recv_message() == 'hello' + assert isinstance(client.close(), wsproto.events.CloseConnection) + + +async def test_server_initiated_close(rfc8441_client): + with rfc8441_client() as client: + assert client.connect('/chat') == '200' + client.send_text('close') + closed = client.recv() + assert isinstance(closed, wsproto.events.CloseConnection) + assert closed.code == 1001 + + +async def test_multiplexing_with_a_plain_request(rfc8441_client, service_client): + """The property the HTTP/1.1 transport can never have.""" + with rfc8441_client() as client: + assert client.connect('/chat') == '200' + client.send_text('before') + assert client.recv_message() == 'before' + + # Another stream of the *same* connection, while the websocket is open. + assert client.status_of(client.request('/ping')) == '404' + + client.send_text('after') + assert client.recv_message() == 'after' + + +async def test_two_websockets_on_one_connection(rfc8441_client): + with rfc8441_client() as first, rfc8441_client() as second: + assert first.connect('/chat') == '200' + assert second.connect('/handler-alt') == '200' + + first.send_text('to first') + second.send_text('to second') + assert first.recv_message() == 'to first' + assert second.recv_message() == 'to second' + + +async def test_unknown_path_is_not_upgraded(rfc8441_client): + with rfc8441_client() as client: + assert client.connect('/no-such-handler') == '404' + + +async def test_non_websocket_handler_does_not_answer_2xx(rfc8441_client): + """A 2xx to an extended CONNECT would tell the client the tunnel is up.""" + with rfc8441_client() as client: + assert client.connect('/plain') == '502' + + +async def test_non_websocket_protocol_is_rejected(rfc8441_client): + with rfc8441_client() as client: + client._stream_id = client._conn.get_next_available_stream_id() # noqa: SLF001 + client._conn.send_headers( # noqa: SLF001 + client._stream_id, # noqa: SLF001 + [ + (':method', 'CONNECT'), + (':protocol', 'mqtt'), + (':scheme', 'http'), + (':path', '/chat'), + (':authority', client._authority), # noqa: SLF001 + ], + end_stream=False, + ) + client._flush() # noqa: SLF001 + with pytest.raises(Rfc8441Error): + client.recv() + + +async def test_service_survives_an_abandoned_websocket(rfc8441_client, service_client): + with rfc8441_client() as client: + assert client.connect('/chat') == '200' + client.send_text('hello') + assert client.recv_message() == 'hello' + client.disconnect() + + response = await service_client.get('/ping') + assert response.status in (404, 200) diff --git a/core/include/userver/server/http/http_request.hpp b/core/include/userver/server/http/http_request.hpp index 0343c6fad8d1..ee6aba2a6af2 100644 --- a/core/include/userver/server/http/http_request.hpp +++ b/core/include/userver/server/http/http_request.hpp @@ -258,6 +258,13 @@ class HttpRequest final { /// Get approximate time point of request handling start std::chrono::steady_clock::time_point GetStartTime() const; + /// @brief Whether the request is a websocket bootstrapped over HTTP/2.0 with the + /// extended CONNECT method of RFC 8441. + /// + /// Such a request is routed as a `GET` so that ordinary websocket handlers match + /// it, and it carries neither `Upgrade`/`Connection` nor `Sec-WebSocket-Key`. + bool IsWebsocketExtendedConnect() const; + /// @cond void MarkAsInternalServerError() const; diff --git a/core/include/userver/server/http/http_request_builder.hpp b/core/include/userver/server/http/http_request_builder.hpp index 201df9c8866b..b619a4e89705 100644 --- a/core/include/userver/server/http/http_request_builder.hpp +++ b/core/include/userver/server/http/http_request_builder.hpp @@ -47,6 +47,8 @@ class HttpRequestBuilder final { /// @cond HttpRequestBuilder& SetIsFinal(bool is_final); + HttpRequestBuilder& SetWebsocketExtendedConnect(bool is_websocket_extended_connect); + HttpRequestBuilder& SetFormDataArgs( utils::impl::TransparentMap, utils::StrCaseHash>&& form_data_args ); diff --git a/core/src/server/component.yaml b/core/src/server/component.yaml index dc85f4307afd..75c0f8a33394 100644 --- a/core/src/server/component.yaml +++ b/core/src/server/component.yaml @@ -198,6 +198,11 @@ properties: type: integer description: the initial window size of the server default: 65536 + enable_connect_protocol: + type: boolean + description: set to true to accept websockets bootstrapped over HTTP/2.0 with the extended + CONNECT method of RFC 8441 + default: false shards: type: integer description: how many concurrent tasks harvest data from a single socket; do not set if not sure what it is doing diff --git a/core/src/server/handlers/websocket_handler.cpp b/core/src/server/handlers/websocket_handler.cpp index fd5cb6d2fe08..a42ffe6770ac 100644 --- a/core/src/server/handlers/websocket_handler.cpp +++ b/core/src/server/handlers/websocket_handler.cpp @@ -21,6 +21,25 @@ USERVER_NAMESPACE_BEGIN namespace server::handlers { +namespace { + +/// @brief Checks the `Sec-WebSocket-Key` of the RFC 6455 handshake over HTTP/1.1. +/// The extended CONNECT of RFC 8441 has no counterpart: opening the stream is itself +/// the proof of intent. +const std::string& GetCheckedWebsocketKey(const server::http::HttpRequest& request) { + const std::string& sec_websocket_key = request.GetHeader(USERVER_NAMESPACE::http::headers::kWebsocketKey); + + // We are fine if `secWebsocketKey` is not properly base64-ecoded + static constexpr std::size_t kLengthOfBase64Encoded16Bytes = 24; + if (kLengthOfBase64Encoded16Bytes != sec_websocket_key.size()) { + LOG_WARNING() << "Empty or invalid Websocket Key"; + throw server::handlers::ClientError(); + } + return sec_websocket_key; +} + +} // namespace + WebsocketHandlerBase::WebsocketHandlerBase( const components::ComponentConfig& config, const components::ComponentContext& context @@ -34,6 +53,12 @@ WebsocketHandlerBase::WebsocketHandlerBase( } bool WebsocketHandlerBase::IsWebsocketRequest(const server::http::HttpRequest& request) const { + if (request.IsWebsocketExtendedConnect()) { + // RFC 8441 carries no Upgrade/Connection headers, and the extended CONNECT form + // has already been validated while parsing the HTTP/2.0 stream. + return true; + } + constexpr auto kIcaseEq = utils::StrIcaseEqual(); return request.GetMethod() == server::http::HttpMethod::kGet && @@ -45,14 +70,10 @@ void WebsocketHandlerBase::HandleWebsocketRequest( server::http::HttpRequest& request, server::request::RequestContext& context ) const { - const std::string& sec_websocket_key = request.GetHeader(USERVER_NAMESPACE::http::headers::kWebsocketKey); - - // We are fine if `secWebsocketKey` is not properly base64-ecoded - static constexpr std::size_t kLengthOfBase64Encoded16Bytes = 24; - if (kLengthOfBase64Encoded16Bytes != sec_websocket_key.size()) { - LOG_WARNING() << "Empty or invalid Websocket Key"; - throw server::handlers::ClientError(); - } + const bool is_extended_connect = request.IsWebsocketExtendedConnect(); + // Checked before anything else, as an invalid key is a bad request no matter what + // websocket version was asked for. + const std::string sec_websocket_key = is_extended_connect ? std::string{} : GetCheckedWebsocketKey(request); auto& response = request.GetHttpResponse(); @@ -68,13 +89,19 @@ void WebsocketHandlerBase::HandleWebsocketRequest( return; } - response.SetStatus(server::http::HttpStatus::kSwitchingProtocols); - response.SetHeader(USERVER_NAMESPACE::http::headers::kConnection, "Upgrade"); - response.SetHeader(USERVER_NAMESPACE::http::headers::kUpgrade, "websocket"); - response.SetHeader( - USERVER_NAMESPACE::http::headers::kWebsocketAccept, - websocket::impl::WebsocketSecAnswer(sec_websocket_key) - ); + if (is_extended_connect) { + // RFC 8441: the stream is accepted with a plain 200 and stays open. There is no + // protocol switch to announce, as the connection keeps speaking HTTP/2.0. + response.SetStatus(server::http::HttpStatus::kOk); + } else { + response.SetStatus(server::http::HttpStatus::kSwitchingProtocols); + response.SetHeader(USERVER_NAMESPACE::http::headers::kConnection, "Upgrade"); + response.SetHeader(USERVER_NAMESPACE::http::headers::kUpgrade, "websocket"); + response.SetHeader( + USERVER_NAMESPACE::http::headers::kWebsocketAccept, + websocket::impl::WebsocketSecAnswer(sec_websocket_key) + ); + } request.SetUpgradeWebsocket([context = std::make_shared(std::move(context)), this](std::unique_ptr socket, engine::io::Sockaddr&& peer_name) { diff --git a/core/src/server/http/http2_session.cpp b/core/src/server/http/http2_session.cpp index 1ed167b28280..ee979f093cf6 100644 --- a/core/src/server/http/http2_session.cpp +++ b/core/src/server/http/http2_session.cpp @@ -1,8 +1,11 @@ #include +#include #include #include +#include + #include #include #include @@ -16,6 +19,9 @@ namespace { constexpr std::size_t kFrameHeaderSize = 9; +// The only `:protocol` we bootstrap with the extended CONNECT method of RFC 8441. +constexpr std::string_view kWebsocketProtocol = "websocket"; + void ThrowIfErr(int error_code, std::string_view msg) { if (error_code != 0) { throw std::runtime_error{fmt::format("{}: {}", msg, nghttp2_strerror(error_code))}; @@ -58,7 +64,8 @@ Http2Session::Http2Session( streaming_consumer_(streaming_queue_->GetConsumer()) { UASSERT(streaming_queue_); - UASSERT(streaming_event_.IsAutoReset()); + // Not auto-resetting, so that the connection loop can await it directly. + UASSERT(!streaming_event_.IsAutoReset()); nghttp2_session_callbacks* callbacks{nullptr}; UINVARIANT(nghttp2_session_callbacks_new(&callbacks) == 0, "Failed to init callbacks for HTTP/2.0"); @@ -79,11 +86,15 @@ Http2Session::Http2Session( UASSERT(session); session_ = SessionPtr(session, nghttp2_session_del); - std::array settings{ + boost::container::small_vector settings{ nghttp2_settings_entry{NGHTTP2_SETTINGS_MAX_CONCURRENT_STREAMS, config.max_concurrent_streams}, nghttp2_settings_entry{NGHTTP2_SETTINGS_MAX_FRAME_SIZE, config.max_frame_size}, nghttp2_settings_entry{NGHTTP2_SETTINGS_INITIAL_WINDOW_SIZE, config.initial_window_size} }; + if (config.enable_connect_protocol) { + // Without this setting nghttp2 rejects any `:protocol` pseudo-header on our behalf. + settings.push_back(nghttp2_settings_entry{NGHTTP2_SETTINGS_ENABLE_CONNECT_PROTOCOL, 1}); + } auto rv = nghttp2_submit_settings(session_.get(), NGHTTP2_FLAG_NONE, settings.data(), settings.size()); ThrowIfErr(rv, "Error when submit settings"); @@ -97,17 +108,33 @@ int Http2Session::OnFrameRecv(nghttp2_session* session, const nghttp2_frame* fra auto& parser = GetParser(user_data); switch (frame->hd.type) { - case NGHTTP2_DATA: case NGHTTP2_HEADERS: { + // The stream may have been rejected and reset already, e.g. on a full stream pool. + auto* stream = parser.FindStream(Stream::Id{frame->hd.stream_id}); + if (stream == nullptr) { + break; + } + if (stream->GetReadPipe() != nullptr) { + // Trailers on an upgraded stream carry nothing we could act upon. + break; + } + if (stream->IsConnect()) { + // A CONNECT stream is half-closed only when the tunnelled protocol ends, + // so its request is finalized at the end of the header block instead. + parser.FinalizeConnectRequest(*stream); + } else if (frame->hd.flags & NGHTTP2_FLAG_END_STREAM) { + parser.FinalizeCompleteRequest(*stream); + } + } break; + case NGHTTP2_DATA: { if (frame->hd.flags & NGHTTP2_FLAG_END_STREAM) { auto& stream = parser.GetStreamChecked(Stream::Id{frame->hd.stream_id}); - try { - stream.RequestConstructor().AppendHeaderField(std::string_view{}); - } catch (const std::exception& e) { - IncStat(parser.stats_.http2_stats.streams_parse_error); - LOG_LIMITED_WARNING() << "can't append header field: " << e; + if (const auto& pipe = stream.GetReadPipe()) { + // A half-close of an upgraded stream is an EOF for the tunnelled protocol. + pipe->Close(); + } else { + parser.FinalizeCompleteRequest(stream); } - parser.FinalizeRequest(stream); } } break; case NGHTTP2_RST_STREAM: { @@ -171,8 +198,17 @@ int Http2Session::OnHeader( auto& stream = parser.GetStreamChecked(Stream::Id{frame->hd.stream_id}); auto& ctor = stream.RequestConstructor(); if (hname == USERVER_NAMESPACE::http::headers::k2::kMethod) { - ctor.SetMethod(HttpMethodFromString(hvalue)); - stream.CheckUrlComplete(); + const auto method = HttpMethodFromString(hvalue); + if (method == HttpMethod::kConnect) { + // The effective method depends on `:protocol`, which may arrive later in the + // header block. Leaving the method unset defers the url parsing until then. + stream.SetConnect(); + } else { + ctor.SetMethod(method); + stream.CheckUrlComplete(); + } + } else if (hname == USERVER_NAMESPACE::http::headers::k2::kProtocol) { + stream.SetUpgradeProtocol(hvalue); } else if (hname == USERVER_NAMESPACE::http::headers::k2::kPath) { try { ctor.AppendUrl(hvalue); @@ -195,7 +231,14 @@ int Http2Session::OnHeader( int Http2Session::OnStreamClose(nghttp2_session*, int32_t id, uint32_t error_code, void* user_data) { auto& parser = GetParser(user_data); - parser.RemoveStream(parser.GetStreamChecked(Stream::Id{id})); + // The stream is already gone if we rejected it ourselves and reset it right away. + if (auto* stream = parser.FindStream(Stream::Id{id})) { + if (const auto& pipe = stream->GetReadPipe()) { + // The pipe outlives the stream, so the tunnelled protocol still unwinds cleanly. + pipe->Close(); + } + parser.RemoveStream(*stream); + } IncStat(parser.stats_.http2_stats.streams_close); LOG_LIMITED_TRACE("The stream {} was closed with code {}", id, error_code); @@ -226,8 +269,13 @@ int Http2Session::OnDataChunkRecv( ) { auto& parser = GetParser(user_data); auto& stream = parser.GetStreamChecked(Stream::Id{id}); + const auto chunk = ToStringView(data, len); + if (const auto& pipe = stream.GetReadPipe()) { + pipe->Push(chunk); + return 0; + } try { - stream.RequestConstructor().AppendBody(std::string_view{reinterpret_cast(data), len}); + stream.RequestConstructor().AppendBody(chunk); } catch (const std::exception& e) { LOG_LIMITED_WARNING() << "can't append body: " << e; } @@ -311,9 +359,12 @@ void Http2Session::RemoveStream(Stream& stream) { stats_.parsing_request_count.Subtract(1); } +Stream* Http2Session::FindStream(Stream::Id id) { + return static_cast(nghttp2_session_get_stream_user_data(session_.get(), static_cast(id))); +} + Stream& Http2Session::GetStreamChecked(Stream::Id id) { - auto* stream = static_cast< - Stream*>(nghttp2_session_get_stream_user_data(session_.get(), static_cast(id))); + auto* stream = FindStream(id); if (stream == nullptr) { throw std::runtime_error{fmt::format("The stream {} does not exist", id)}; } @@ -358,6 +409,62 @@ void Http2Session::UpgradeToHttp2(std::string_view client_magic) { RegisterStream(kStreamIdAfterUpgradeResponse); } +std::unique_ptr Http2Session::UpgradeStream(Stream::Id id) { + const auto& pipe = GetStreamChecked(id).GetReadPipe(); + UINVARIANT(pipe, "Only a stream that was accepted for an upgrade can be upgraded"); + return std::make_unique( + static_cast(id), pipe, impl::Http2StreamEventProducer{*streaming_queue_, streaming_event_} + ); +} + +void Http2Session::CloseUpgradedStream(Stream::Id id) { + auto* stream = FindStream(id); + if (stream == nullptr) { + // The peer has already closed the stream. + return; + } + stream->SetEnd(true); + if (stream->IsDeferred()) { + const auto res = nghttp2_session_resume_data(session_.get(), static_cast(id)); + ThrowIfErr(res, "Error while resume_data"); + stream->SetDeferred(false); + } + WriteWhileWant(); +} + +void Http2Session::FinalizeCompleteRequest(Stream& stream) { + try { + stream.RequestConstructor().AppendHeaderField(std::string_view{}); + } catch (const std::exception& e) { + IncStat(stats_.http2_stats.streams_parse_error); + LOG_LIMITED_WARNING() << "can't append header field: " << e; + } + FinalizeRequest(stream); +} + +void Http2Session::FinalizeConnectRequest(Stream& stream) { + const auto protocol = stream.GetUpgradeProtocol(); + if (!config_.enable_connect_protocol || protocol != kWebsocketProtocol) { + // Plain CONNECT tunnels of RFC 9110 are not supported, and RFC 8441 is + // implemented for websockets only. + LOG_LIMITED_WARNING() << fmt::format( + "Rejecting the CONNECT stream {}: unsupported ':protocol' value '{}'", stream.GetId(), protocol + ); + IncStat(stats_.http2_stats.streams_parse_error); + SubmitRstStream(stream.GetId(), NGHTTP2_CONNECT_ERROR); + RemoveStream(stream); + return; + } + // Route as a GET so that path-registered websocket handlers match, the same way + // reverse proxies do when converting an HTTP/1.1 upgrade into an extended CONNECT. + stream.RequestConstructor().SetMethod(HttpMethod::kGet); + stream.RequestConstructor().SetWebsocketExtendedConnect(true); + // Buffer the incoming bytes from now on: a client may start sending them before the + // handler has had a chance to accept the stream. + stream.SetReadPipe(std::make_shared()); + FinalizeCompleteRequest(stream); +} + void Http2Session::FinalizeRequest(Stream& stream) { if (!stream.CheckUrlComplete()) { IncStat(stats_.http2_stats.streams_parse_error); @@ -391,17 +498,27 @@ void Http2Session::WriteWhileWant() { engine::SingleConsumerEvent& Http2Session::GetStreamingEvent() { return streaming_event_; } void Http2Session::HandleStreamingEvents() { + // Reset before draining, so that an event pushed while we drain still wakes us up + // again instead of being silently swallowed. + streaming_event_.Reset(); + impl::Http2StreamEvent event; while (streaming_consumer_.PopNoblock(event)) { UASSERT(event.stream_id != -1); - auto& stream = GetStreamChecked(Stream::Id{event.stream_id}); - if (stream.IsDeferred()) { - const auto res = nghttp2_session_resume_data(session_.get(), static_cast(stream.GetId())); + auto* stream = FindStream(Stream::Id{event.stream_id}); + if (stream == nullptr) { + // The peer closed the stream while its producer was still writing. + LOG_LIMITED_DEBUG() << fmt::format("Dropping a streamed chunk of the closed stream {}", event.stream_id); + event = {}; + continue; + } + if (stream->IsDeferred()) { + const auto res = nghttp2_session_resume_data(session_.get(), static_cast(stream->GetId())); ThrowIfErr(res, "Error while resume_data"); - stream.SetDeferred(false); + stream->SetDeferred(false); } - stream.PushChunk(std::move(event.body_part)); - stream.SetEnd(event.is_end); + stream->PushChunk(std::move(event.body_part)); + stream->SetEnd(event.is_end); event = {}; } WriteWhileWant(); diff --git a/core/src/server/http/http2_session.hpp b/core/src/server/http/http2_session.hpp index 91673860d320..02554bb8618f 100644 --- a/core/src/server/http/http2_session.hpp +++ b/core/src/server/http/http2_session.hpp @@ -59,6 +59,13 @@ class Http2Session final : public request::RequestParser { void UpgradeToHttp2(std::string_view client_magic); + /// @brief Hands an already answered stream over to a tunnelled protocol. + /// @returns the stream presented as a stream-like object for that protocol. + std::unique_ptr UpgradeStream(Stream::Id id); + + /// @brief Half-closes an upgraded stream once its tunnelled protocol is over. + void CloseUpgradedStream(Stream::Id id); + engine::SingleConsumerEvent& GetStreamingEvent(); void WriteWhileWant(); @@ -112,11 +119,14 @@ class Http2Session final : public request::RequestParser { void RegisterStream(Stream::Id id); void RemoveStream(Stream& stream); + Stream* FindStream(Stream::Id id); Stream& GetStreamChecked(Stream::Id id); void SubmitRstStream(Stream::Id stream_id, std::uint32_t error_code = NGHTTP2_INTERNAL_ERROR); void FinalizeRequest(Stream& stream); + void FinalizeCompleteRequest(Stream& stream); + void FinalizeConnectRequest(Stream& stream); const net::Http2SessionConfig& config_; @@ -133,7 +143,7 @@ class Http2Session final : public request::RequestParser { engine::io::RwBase* socket_; std::shared_ptr streaming_queue_{nullptr}; - engine::SingleConsumerEvent streaming_event_; + engine::SingleConsumerEvent streaming_event_{engine::SingleConsumerEvent::NoAutoReset{}}; impl::Http2StreamEventQueue::Consumer streaming_consumer_; std::int32_t max_client_stream_id_{0}; bool peer_goaway_received_{false}; diff --git a/core/src/server/http/http2_session_test.cpp b/core/src/server/http/http2_session_test.cpp index c4a40812cae3..8cfd94ad6d61 100644 --- a/core/src/server/http/http2_session_test.cpp +++ b/core/src/server/http/http2_session_test.cpp @@ -1,6 +1,7 @@ #include #include #include +#include #include #include @@ -11,6 +12,7 @@ #include #include #include +#include #include #include @@ -358,6 +360,218 @@ UTEST_F(Http2SessionTest, HeavyHeader) { EXPECT_EQ(request->GetMethod(), HttpMethod::kGet); } +namespace { + +// No HTTP client we can link against speaks the extended CONNECT of RFC 8441, so the +// requests are produced by a real client-side nghttp2 session wired straight to the +// parser under test. +class Http2TestClient final { +public: + Http2TestClient() { + nghttp2_session_callbacks* callbacks{nullptr}; + UINVARIANT(nghttp2_session_callbacks_new(&callbacks) == 0, "Failed to init client callbacks"); + const utils::FastScopeGuard delete_guard{[&callbacks]() noexcept { nghttp2_session_callbacks_del(callbacks); }}; + + nghttp2_session* session{nullptr}; + UINVARIANT(nghttp2_session_client_new(&session, callbacks, this) == 0, "Failed to init client session"); + session_ = SessionPtr{session, nghttp2_session_del}; + + const int rv = nghttp2_submit_settings(session_.get(), NGHTTP2_FLAG_NONE, nullptr, 0); + UINVARIANT(rv == 0, "Failed to submit client settings"); + } + + std::int32_t SubmitRequest(const std::vector& headers, bool end_stream) { + const auto flags = end_stream ? NGHTTP2_FLAG_END_STREAM : NGHTTP2_FLAG_NONE; + const std::int32_t stream_id = + nghttp2_submit_headers(session_.get(), flags, -1, nullptr, headers.data(), headers.size(), nullptr); + UINVARIANT(stream_id > 0, "Failed to submit client headers"); + return stream_id; + } + + void SubmitData(std::int32_t stream_id, std::string_view data, bool end_stream) { + data_to_send_ = data; + nghttp2_data_provider provider{}; + provider.source.ptr = this; + provider.read_callback = ReadData; + const auto flags = end_stream ? NGHTTP2_FLAG_END_STREAM : NGHTTP2_FLAG_NONE; + const int rv = nghttp2_submit_data(session_.get(), flags, stream_id, &provider); + UINVARIANT(rv == 0, "Failed to submit client data"); + } + + /// @returns the bytes the client wants to send, to be fed into the parser. + std::string ExtractOutput() { + std::string output; + while (nghttp2_session_want_write(session_.get())) { + const std::uint8_t* data{nullptr}; + const auto len = nghttp2_session_mem_send(session_.get(), &data); + if (len <= 0) { + break; + } + output.append(reinterpret_cast(data), len); + } + return output; + } + + void Feed(std::string_view data) { + const auto readlen = + nghttp2_session_mem_recv(session_.get(), reinterpret_cast(data.data()), data.size()); + UINVARIANT(readlen >= 0, "Failed to parse the server output"); + } + + std::uint32_t GetRemoteSetting(nghttp2_settings_id id) const { + return nghttp2_session_get_remote_settings(session_.get(), id); + } + +private: + using SessionPtr = std::unique_ptr; + + static ssize_t + ReadData(nghttp2_session*, std::int32_t, std::uint8_t* buf, std::size_t length, std::uint32_t* flags, nghttp2_data_source* source, void*) { + auto& client = *static_cast(source->ptr); + const auto size = std::min(length, client.data_to_send_.size()); + std::memcpy(buf, client.data_to_send_.data(), size); + client.data_to_send_.erase(0, size); + if (client.data_to_send_.empty()) { + *flags |= NGHTTP2_DATA_FLAG_EOF; + } + return static_cast(size); + } + + SessionPtr session_{nullptr, nghttp2_session_del}; + std::string data_to_send_; +}; + +nghttp2_nv MakeHeader(std::string_view name, std::string_view value) { + return { + // NOLINTNEXTLINE(cppcoreguidelines-pro-type-const-cast) + reinterpret_cast(const_cast(name.data())), + // NOLINTNEXTLINE(cppcoreguidelines-pro-type-const-cast) + reinterpret_cast(const_cast(value.data())), + name.size(), + value.size(), + NGHTTP2_NV_FLAG_NONE}; +} + +std::vector MakeExtendedConnectHeaders(std::string_view path) { + return { + MakeHeader(":method", "CONNECT"), + MakeHeader(":protocol", "websocket"), + MakeHeader(":scheme", "https"), + MakeHeader(":path", path), + MakeHeader(":authority", "localhost"), + MakeHeader("sec-websocket-version", "13"), + }; +} + +} // namespace + +// Drives Http2Session directly: RFC 8441 needs no sockets to be exercised, and the +// `enable_connect_protocol` option has to be flipped per test. +class Http2ExtendedConnectTest : public ::testing::Test { +public: + void SetUp() override { MakeSession(/*enable_connect_protocol=*/true); } + + void MakeSession(bool enable_connect_protocol) { + config_.enable_connect_protocol = enable_connect_protocol; + // SETTINGS are deltas, so a client that already learned the setting from an + // earlier session would keep it. Both sides start over together. + client_ = std::make_unique(); + session_ = std::make_unique( + index_, + request_config_, + config_, + [this](std::shared_ptr&& request) { requests_.push_back(std::move(request)); }, + stats_, + accounter_, + engine::io::Sockaddr{} + ); + // The client needs the server SETTINGS before it may use `:protocol` at all. + client_->Feed(PullServerOutput()); + } + + std::string PullServerOutput() { + auto* raw_session = session_->GetNghttp2SessionPtr(); + std::string output; + while (nghttp2_session_want_write(raw_session)) { + const std::uint8_t* data{nullptr}; + const auto len = nghttp2_session_mem_send(raw_session, &data); + if (len <= 0) { + break; + } + output.append(reinterpret_cast(data), len); + } + return output; + } + + void PumpClientToServer() { EXPECT_TRUE(session_->Parse(client_->ExtractOutput())); } + +protected: + HandlerInfoIndex index_; + request::HttpRequestConfig request_config_; + request::ResponseDataAccounter accounter_; + net::ParserStats stats_; + net::Http2SessionConfig config_; + + std::unique_ptr client_; + std::unique_ptr session_; + std::vector> requests_; +}; + +UTEST_F(Http2ExtendedConnectTest, SettingIsAdvertised) { + EXPECT_EQ(1, client_->GetRemoteSetting(NGHTTP2_SETTINGS_ENABLE_CONNECT_PROTOCOL)); +} + +UTEST_F(Http2ExtendedConnectTest, SettingIsNotAdvertisedByDefault) { + MakeSession(/*enable_connect_protocol=*/false); + EXPECT_EQ(0, client_->GetRemoteSetting(NGHTTP2_SETTINGS_ENABLE_CONNECT_PROTOCOL)); +} + +UTEST_F(Http2ExtendedConnectTest, RoutedAsGetWithoutEndStream) { + client_->SubmitRequest(MakeExtendedConnectHeaders("/chat"), /*end_stream=*/false); + PumpClientToServer(); + + ASSERT_EQ(1, requests_.size()); + const auto& request = *requests_.front(); + EXPECT_EQ(HttpMethod::kGet, request.GetMethod()); + EXPECT_EQ("/chat", request.GetRequestPath()); + EXPECT_TRUE(request.IsWebsocketExtendedConnect()); + EXPECT_EQ("13", request.GetHeader("sec-websocket-version")); +} + +UTEST_F(Http2ExtendedConnectTest, DataIsNotARequestBody) { + const auto stream_id = client_->SubmitRequest(MakeExtendedConnectHeaders("/chat"), /*end_stream=*/false); + PumpClientToServer(); + ASSERT_EQ(1, requests_.size()); + + client_->SubmitData(stream_id, "websocket frame bytes", /*end_stream=*/false); + PumpClientToServer(); + + // The bytes belong to the tunnelled protocol, not to the request. + EXPECT_EQ("", requests_.front()->RequestBody()); + EXPECT_EQ(1, requests_.size()); +} + +UTEST_F(Http2ExtendedConnectTest, RejectedWhenDisabled) { + MakeSession(/*enable_connect_protocol=*/false); + + // The client refuses to use `:protocol` unadvertised, so the pseudo-header has to be + // smuggled in as a plain CONNECT to reach the parser at all. + client_->SubmitRequest( + {MakeHeader(":method", "CONNECT"), MakeHeader(":authority", "localhost")}, + /*end_stream=*/false + ); + PumpClientToServer(); + + EXPECT_TRUE(requests_.empty()); +} + +UTEST_F(Http2ExtendedConnectTest, PlainConnectIsRejected) { + client_->SubmitRequest({MakeHeader(":method", "CONNECT"), MakeHeader(":authority", "localhost")}, false); + PumpClientToServer(); + + EXPECT_TRUE(requests_.empty()); +} + UTEST_F(Http2SessionTest, ForCurl) { auto& client = GetClient(); const auto url = GetServer().GetBaseUrl() + "/hello"; diff --git a/core/src/server/http/http2_stream.cpp b/core/src/server/http/http2_stream.cpp index 43e860760544..6332e18b3735 100644 --- a/core/src/server/http/http2_stream.cpp +++ b/core/src/server/http/http2_stream.cpp @@ -1,5 +1,7 @@ #include +#include + #include #include // std::accumulate @@ -55,6 +57,21 @@ bool Stream::IsStreaming() const { return is_streaming_; } void Stream::SetStreaming(bool streaming) { is_streaming_ = streaming; } +void Stream::SetConnect() { is_connect_ = true; } + +bool Stream::IsConnect() const { return is_connect_; } + +void Stream::SetUpgradeProtocol(std::string_view protocol) { upgrade_protocol_ = protocol; } + +std::string_view Stream::GetUpgradeProtocol() const { return upgrade_protocol_; } + +void Stream::SetReadPipe(std::shared_ptr pipe) { + UASSERT(!read_pipe_); + read_pipe_ = std::move(pipe); +} + +const std::shared_ptr& Stream::GetReadPipe() const { return read_pipe_; } + bool Stream::CheckUrlComplete() { if (url_complete_) { return true; diff --git a/core/src/server/http/http2_stream.hpp b/core/src/server/http/http2_stream.hpp index 59eeea06698a..50e55da60405 100644 --- a/core/src/server/http/http2_stream.hpp +++ b/core/src/server/http/http2_stream.hpp @@ -1,5 +1,7 @@ #pragma once +#include + #include #include @@ -15,6 +17,8 @@ class Socket; namespace server::http { +class Http2StreamReadPipe; + class Stream final { public: using Id = utils::StrongTypedef; @@ -40,6 +44,22 @@ class Stream final { bool IsStreaming() const; void SetStreaming(bool streaming); + /// @name RFC 8441 extended CONNECT + /// The `:method` and the `:protocol` pseudo-headers may arrive in any order, so the + /// effective method of a CONNECT stream is only known once the header block is complete. + /// @{ + void SetConnect(); + bool IsConnect() const; + void SetUpgradeProtocol(std::string_view protocol); + std::string_view GetUpgradeProtocol() const; + + /// @brief Hands the stream over to a tunnelled protocol: from now on DATA frames + /// are its incoming bytes rather than a request body. + void SetReadPipe(std::shared_ptr pipe); + /// @returns nullptr unless the stream was handed over to a tunnelled protocol. + const std::shared_ptr& GetReadPipe() const; + /// @} + bool CheckUrlComplete(); void PushChunk(std::string&& chunk); ssize_t GetMaxSize(std::size_t max_len, std::uint32_t* flags); @@ -50,6 +70,10 @@ class Stream final { bool url_complete_{false}; HttpRequestConstructor constructor_; const Id id_; + // Extended CONNECT + bool is_connect_{false}; + std::string upgrade_protocol_{}; + std::shared_ptr read_pipe_{}; // Body sending nghttp2_data_provider nghttp2_provider_{}; boost::container::small_vector chunks_{}; diff --git a/core/src/server/http/http2_stream_rw.cpp b/core/src/server/http/http2_stream_rw.cpp new file mode 100644 index 000000000000..e733ff820e3d --- /dev/null +++ b/core/src/server/http/http2_stream_rw.cpp @@ -0,0 +1,207 @@ +#include + +#include +#include + +#include +#include +#include +#include + +USERVER_NAMESPACE_BEGIN + +namespace server::http { + +namespace { + +[[noreturn]] void ThrowOnWaitFailure(engine::FutureStatus status, std::size_t bytes_transferred) { + UASSERT(status != engine::FutureStatus::kReady); + if (status == engine::FutureStatus::kCancelled) { + throw engine::io::IoCancelled{bytes_transferred}; + } + throw engine::io::IoTimeout{bytes_transferred}; +} + +} // namespace + +Http2StreamReadPipe::Http2StreamReadPipe() + : queue_(Queue::Create()), + producer_(queue_->GetProducer()), + consumer_(queue_->GetConsumer()) +{ + // The peer is already bounded by the flow control window of the stream, and the + // producer runs in the connection task, where it must never block. + queue_->SetSoftMaxSize(Queue::kUnbounded); +} + +void Http2StreamReadPipe::Push(std::string_view data) { + if (data.empty()) { + return; + } + if (!producer_.PushNoblock(std::string{data})) { + // Only happens once the consumer is gone, i.e. the tunnelled protocol is over. + LOG_LIMITED_DEBUG() << "Dropping the incoming bytes of an abandoned upgraded stream"; + return; + } + event_.Send(); +} + +void Http2StreamReadPipe::Close() { + is_closed_.store(true, std::memory_order_release); + event_.Send(); +} + +bool Http2StreamReadPipe::FetchChunk() { + if (pos_in_chunk_ < chunk_.size()) { + return true; + } + chunk_.clear(); + pos_in_chunk_ = 0; + // Push() never enqueues an empty chunk, so a successful pop always means data. + return consumer_.PopNoblock(chunk_); +} + +bool Http2StreamReadPipe::IsExhausted() { + // Drain first: the last chunks may have been pushed just before the close. + return !FetchChunk() && is_closed_.load(std::memory_order_acquire); +} + +std::size_t Http2StreamReadPipe::ReadBuffered(void* buf, std::size_t len) { + auto* out = static_cast(buf); + std::size_t copied = 0; + while (copied < len && FetchChunk()) { + const auto part = std::min(len - copied, chunk_.size() - pos_in_chunk_); + std::memcpy(out + copied, chunk_.data() + pos_in_chunk_, part); + copied += part; + pos_in_chunk_ += part; + } + return copied; +} + +engine::FutureStatus Http2StreamReadPipe::WaitForData(engine::Deadline deadline) { + if (FetchChunk() || is_closed_.load(std::memory_order_acquire)) { + return engine::FutureStatus::kReady; + } + // Reset first and re-check afterwards: a producer that fills the pipe after the + // reset signals again, so no wakeup can be lost. + event_.Reset(); + if (FetchChunk() || is_closed_.load(std::memory_order_acquire)) { + return engine::FutureStatus::kReady; + } + return event_.WaitUntil(deadline); +} + +engine::AwaitableToken Http2StreamReadPipe::GetAwaitableToken() { return event_.GetAwaitableToken(); } + +Http2StreamRw::Http2StreamRw( + std::int32_t stream_id, + std::shared_ptr pipe, + impl::Http2StreamEventProducer producer +) + : stream_id_(stream_id), + pipe_(std::move(pipe)), + producer_(std::move(producer)) +{ + UASSERT(pipe_); + SetReadableAwaitableToken(pipe_->GetAwaitableToken()); + // Writes only enqueue an event for the connection task, so they never block. + SetWritableAwaitableToken(engine::MakeReadyAwaitableToken()); +} + +Http2StreamRw::~Http2StreamRw() = default; + +bool Http2StreamRw::IsValid() const { return !pipe_->IsExhausted(); } + +std::optional Http2StreamRw::ReadNoblock(void* buf, std::size_t len) { + if (len == 0) { + return std::size_t{0}; + } + if (const auto read = pipe_->ReadBuffered(buf, len); read != 0) { + return read; + } + if (pipe_->IsExhausted()) { + return std::size_t{0}; + } + // Unlike a socket, the pipe is filled by another task of this very service, and a + // caller polling it in a busy loop (websocket::WebSocketConnection::TryRecv) would + // otherwise never let the connection task run. + engine::Yield(); + return std::nullopt; +} + +std::size_t Http2StreamRw::ReadSome(void* buf, std::size_t len, engine::Deadline deadline) { + if (len == 0) { + return 0; + } + while (true) { + if (const auto read = pipe_->ReadBuffered(buf, len); read != 0) { + return read; + } + if (pipe_->IsExhausted()) { + return 0; + } + if (const auto status = pipe_->WaitForData(deadline); status != engine::FutureStatus::kReady) { + ThrowOnWaitFailure(status, 0); + } + } +} + +std::size_t Http2StreamRw::ReadAll(void* buf, std::size_t len, engine::Deadline deadline) { + auto* out = static_cast(buf); + std::size_t read = 0; + while (read < len) { + std::size_t part = 0; + try { + part = ReadSome(out + read, len - read, deadline); + } catch (const engine::io::IoCancelled&) { + throw engine::io::IoCancelled{read}; + } catch (const engine::io::IoTimeout&) { + throw engine::io::IoTimeout{read}; + } + if (part == 0) { + break; // the peer half-closed the stream + } + read += part; + } + return read; +} + +bool Http2StreamRw::WaitReadable(engine::Deadline deadline) { + return pipe_->WaitForData(deadline) == engine::FutureStatus::kReady; +} + +std::size_t Http2StreamRw::WriteAll(const void* buf, std::size_t len, engine::Deadline deadline) { + if (len == 0) { + return 0; + } + producer_.PushEvent( + {.stream_id = stream_id_, .body_part = std::string{static_cast(buf), len}, .is_end = false}, + deadline + ); + return len; +} + +std::size_t Http2StreamRw::WriteAll(std::span list, engine::Deadline deadline) { + std::size_t total = 0; + for (const auto& io_data : list) { + total += io_data.len; + } + if (total == 0) { + return 0; + } + + // A single event keeps the parts of one websocket frame in one DATA frame. + std::string body; + body.reserve(total); + for (const auto& io_data : list) { + body.append(static_cast(io_data.data), io_data.len); + } + producer_.PushEvent({.stream_id = stream_id_, .body_part = std::move(body), .is_end = false}, deadline); + return total; +} + +bool Http2StreamRw::WaitWriteable(engine::Deadline) { return true; } + +} // namespace server::http + +USERVER_NAMESPACE_END diff --git a/core/src/server/http/http2_stream_rw.hpp b/core/src/server/http/http2_stream_rw.hpp new file mode 100644 index 000000000000..d874ecdbb633 --- /dev/null +++ b/core/src/server/http/http2_stream_rw.hpp @@ -0,0 +1,115 @@ +#pragma once + +#include +#include +#include +#include +#include +#include +#include +#include + +#include +#include +#include +#include +#include +#include +#include + +USERVER_NAMESPACE_BEGIN + +namespace server::http { + +/// @brief The incoming bytes of an upgraded HTTP/2.0 stream. +/// +/// Filled by the connection task from the DATA frames of the stream and drained by the +/// task that runs the tunnelled protocol. Shared between the two because the peer may +/// close the stream (and with it the server::http::Stream) while the tunnelled protocol +/// is still unwinding. +/// +/// Deliberately lock-free: a tunnelled protocol is free to poll its input in a busy loop +/// (see @ref websocket::WebSocketConnection::TryRecv), and it must not be able to starve +/// the connection task out of filling the pipe. +class Http2StreamReadPipe final { +public: + Http2StreamReadPipe(); + + /// @name Producer side, called from the connection task only. + /// @{ + void Push(std::string_view data); + + /// @brief Makes the consumer observe an end of stream. Idempotent. + void Close(); + /// @} + + /// @name Consumer side, called from the task of the tunnelled protocol only. + /// @{ + /// @returns whether Close() was called and everything buffered has been read. + bool IsExhausted(); + + /// @returns the number of bytes copied, `0` if nothing is buffered. + std::size_t ReadBuffered(void* buf, std::size_t len); + + /// @brief Waits until there is something to read or the stream ends. + [[nodiscard]] engine::FutureStatus WaitForData(engine::Deadline deadline); + + engine::AwaitableToken GetAwaitableToken(); + /// @} + +private: + using Queue = concurrent::SpscQueue; + + /// @returns whether `chunk_` holds unread bytes, taking the next chunk if needed. + bool FetchChunk(); + + std::shared_ptr queue_; + Queue::Producer producer_; + Queue::Consumer consumer_; + // Consumer-only state. + std::string chunk_; + std::size_t pos_in_chunk_{0}; + + std::atomic is_closed_{false}; + // Not auto-resetting: the flag mirrors "readable", and only the consumer clears it, + // which is also what makes it usable as an awaitable. + engine::SingleConsumerEvent event_{engine::SingleConsumerEvent::NoAutoReset{}}; +}; + +/// @brief Presents a single upgraded HTTP/2.0 stream as a stream-like object, so that +/// protocols tunnelled over it (websockets of RFC 8441) run unmodified. +/// +/// Reads come from Http2StreamReadPipe. Writes are pushed as streaming events and are +/// turned into DATA frames by the connection task, because `nghttp2_session` may only +/// ever be touched there. +class Http2StreamRw final : public engine::io::RwBase { +public: + Http2StreamRw( + std::int32_t stream_id, + std::shared_ptr pipe, + impl::Http2StreamEventProducer producer + ); + + ~Http2StreamRw() override; + + bool IsValid() const override; + + std::optional ReadNoblock(void* buf, std::size_t len) override; + std::size_t ReadSome(void* buf, std::size_t len, engine::Deadline deadline) override; + std::size_t ReadAll(void* buf, std::size_t len, engine::Deadline deadline) override; + [[nodiscard]] bool WaitReadable(engine::Deadline deadline) override; + + using engine::io::RwBase::WriteAll; + std::size_t WriteAll(const void* buf, std::size_t len, engine::Deadline deadline) override; + std::size_t WriteAll(std::span list, engine::Deadline deadline) override; + [[nodiscard]] bool WaitWriteable(engine::Deadline deadline) override; + +private: + const std::int32_t stream_id_; + const std::shared_ptr pipe_; + impl::Http2StreamEventProducer producer_; +}; + +} // namespace server::http + +USERVER_NAMESPACE_END diff --git a/core/src/server/http/http2_writer.cpp b/core/src/server/http/http2_writer.cpp index 5bf9eedc6e32..696b991c4306 100644 --- a/core/src/server/http/http2_writer.cpp +++ b/core/src/server/http/http2_writer.cpp @@ -105,18 +105,22 @@ class Http2ResponseWriter final { void WriteHttpResponse() { auto data = response_.ExtractData(); - auto headers = GetHeaders(); - const bool is_body_forbidden = IsBodyForbiddenForStatus(response_.status_); + const auto status = GetStatus(); + auto headers = GetHeaders(status); + const bool is_body_forbidden = IsBodyForbiddenForStatus(status); if (is_body_forbidden && !data.empty()) { LOG_LIMITED_WARNING() - << "Non-empty body provided for response with HTTP2 code " << static_cast(response_.status_) + << "Non-empty body provided for response with HTTP2 code " << static_cast(status) << " which does not allow one, it will be dropped"; } const auto stream_id = response_.GetStreamId().value(); auto& stream = http2_session_.GetStreamChecked(Stream::Id{stream_id}); - stream.SetStreaming(response_.IsBodyStreamed() && data.empty()); + // An upgraded stream is answered with headers only and then stays open for the + // bytes of the tunnelled protocol, exactly like a streamed body. + const bool keeps_stream_open = response_.IsBodyStreamed() || response_.request_.IsUpgradeWebsocket(); + stream.SetStreaming(keeps_stream_open && data.empty()); std::size_t bytes = headers.GetSize(); nghttp2_data_provider* provider{nullptr}; @@ -142,7 +146,22 @@ class Http2ResponseWriter final { } private: - Http2HeaderWriter GetHeaders() const { + HttpStatus GetStatus() const { + // RFC 8441: a 2xx answer to an extended CONNECT tells the client that the tunnel + // is established, so a handler that did not upgrade the stream must not send one. + // The handler cannot fix this up itself: by the time we know whether the upgrade + // happened, its response headers are already frozen. + const auto status = static_cast(response_.status_); + if (response_.request_.IsWebsocketExtendedConnect() && !response_.request_.IsUpgradeWebsocket() && + status >= 200 && status < 300) + { + LOG_LIMITED_WARNING() << "The handler of an extended CONNECT did not upgrade the stream"; + return HttpStatus::kBadGateway; + } + return response_.status_; + } + + Http2HeaderWriter GetHeaders(HttpStatus status) const { // Preallocate space for all headers Http2HeaderWriter header_writer{ response_.system_headers_.size() + response_.user_headers_.size() + response_.cookies_.size() + 3 @@ -150,7 +169,7 @@ class Http2ResponseWriter final { header_writer.AddKeyValue( USERVER_NAMESPACE::http::headers::k2::kStatus, - fmt::to_string(static_cast(response_.status_)) + fmt::to_string(static_cast(status)) ); if (!response_.HasHeader(USERVER_NAMESPACE::http::headers::kDate)) { diff --git a/core/src/server/http/http_request.cpp b/core/src/server/http/http_request.cpp index 33f52033bad9..46ffed23b34a 100644 --- a/core/src/server/http/http_request.cpp +++ b/core/src/server/http/http_request.cpp @@ -291,6 +291,8 @@ HttpResponse& HttpRequest::GetHttpResponse() const noexcept { return pimpl_->res std::chrono::steady_clock::time_point HttpRequest::GetStartTime() const { return pimpl_->start_time; } +bool HttpRequest::IsWebsocketExtendedConnect() const { return pimpl_->is_websocket_extended_connect; } + bool HttpRequest::IsUpgradeWebsocket() const { return static_cast(pimpl_->upgrade_websocket_cb); } void HttpRequest::SetUpgradeWebsocket(UpgradeCallback cb) const { pimpl_->upgrade_websocket_cb = std::move(cb); } diff --git a/core/src/server/http/http_request_builder.cpp b/core/src/server/http/http_request_builder.cpp index a399bfa17640..f742809d8a14 100644 --- a/core/src/server/http/http_request_builder.cpp +++ b/core/src/server/http/http_request_builder.cpp @@ -86,6 +86,11 @@ HttpRequestBuilder& HttpRequestBuilder::SetIsFinal(bool is_final) { return *this; } +HttpRequestBuilder& HttpRequestBuilder::SetWebsocketExtendedConnect(bool is_websocket_extended_connect) { + request_->pimpl_->is_websocket_extended_connect = is_websocket_extended_connect; + return *this; +} + HttpRequestBuilder& HttpRequestBuilder::SetFormDataArgs( utils::impl::TransparentMap, utils::StrCaseHash>&& form_data_args ) { diff --git a/core/src/server/http/http_request_constructor.cpp b/core/src/server/http/http_request_constructor.cpp index 8905a015f6f4..e50d940db29a 100644 --- a/core/src/server/http/http_request_constructor.cpp +++ b/core/src/server/http/http_request_constructor.cpp @@ -183,6 +183,10 @@ void HttpRequestConstructor::SetStreamProducer(impl::Http2StreamEventProducer&& builder_.SetStreamProducer(std::move(producer)); } +void HttpRequestConstructor::SetWebsocketExtendedConnect(bool is_websocket_extended_connect) { + builder_.SetWebsocketExtendedConnect(is_websocket_extended_connect); +} + std::shared_ptr HttpRequestConstructor::Finalize() { FinalizeImpl(); diff --git a/core/src/server/http/http_request_constructor.hpp b/core/src/server/http/http_request_constructor.hpp index abf0981584e4..7b268d49eab4 100644 --- a/core/src/server/http/http_request_constructor.hpp +++ b/core/src/server/http/http_request_constructor.hpp @@ -61,6 +61,7 @@ class HttpRequestConstructor final { // HTTP/2.0 only: void SetStreamProducer(impl::Http2StreamEventProducer&& producer); void SetResponseStreamId(std::int32_t stream_id); + void SetWebsocketExtendedConnect(bool is_websocket_extended_connect); std::shared_ptr Finalize(); diff --git a/core/src/server/http/http_request_impl.hpp b/core/src/server/http/http_request_impl.hpp index a6f48cff5357..a061ad469e4d 100644 --- a/core/src/server/http/http_request_impl.hpp +++ b/core/src/server/http/http_request_impl.hpp @@ -47,6 +47,9 @@ struct HttpRequest::Impl { HeadersMap headers; CookiesMap cookies; bool is_final{false}; + // The request arrived as an RFC 8441 extended CONNECT with `:protocol: websocket`, + // and is routed as a GET so that ordinary websocket handlers match it. + bool is_websocket_extended_connect{false}; #ifndef NDEBUG mutable bool args_referenced{false}; #endif diff --git a/core/src/server/net/connection_config.cpp b/core/src/server/net/connection_config.cpp index 741c6143d6f7..1db388b5c6aa 100644 --- a/core/src/server/net/connection_config.cpp +++ b/core/src/server/net/connection_config.cpp @@ -11,6 +11,7 @@ Http2SessionConfig Parse(const yaml_config::YamlConfig& value, formats::parse::T conf.max_concurrent_streams = value["max_concurrent_streams"].As(conf.max_concurrent_streams); conf.max_frame_size = value["max_frame_size"].As(conf.max_frame_size); conf.initial_window_size = value["initial_window_size"].As(conf.initial_window_size); + conf.enable_connect_protocol = value["enable_connect_protocol"].As(conf.enable_connect_protocol); return conf; } diff --git a/core/src/server/net/connection_config.hpp b/core/src/server/net/connection_config.hpp index 3d0b856b020f..de775f182485 100644 --- a/core/src/server/net/connection_config.hpp +++ b/core/src/server/net/connection_config.hpp @@ -22,6 +22,9 @@ struct Http2SessionConfig final { std::uint32_t max_concurrent_streams = 100; std::uint32_t max_frame_size = 1 << 14; std::uint32_t initial_window_size = 1 << 16; + // Advertises SETTINGS_ENABLE_CONNECT_PROTOCOL, which allows clients to bootstrap + // websockets over HTTP/2.0 with the extended CONNECT method of RFC 8441. + bool enable_connect_protocol = false; }; struct ConnectionConfig { diff --git a/core/src/server/net/http2_connection.cpp b/core/src/server/net/http2_connection.cpp index 12bdcb2ecc44..7187fd11cbee 100644 --- a/core/src/server/net/http2_connection.cpp +++ b/core/src/server/net/http2_connection.cpp @@ -27,13 +27,17 @@ constexpr std::string_view kHttp2Preface = "PRI * HTTP/2.0\r\n\r\nSM\r\n\r\n"; constexpr std::string_view kPrefaceBegin = kHttp2Preface.substr(0, kMinLenPrefaceToDetect); constexpr std::uint64_t kSocketId = std::numeric_limits::max(); +constexpr std::uint64_t kStreamingId = kSocketId - 1; -enum class WakeupKind { kSocketReadable, kTaskComputedResponse }; +enum class WakeupKind { kSocketReadable, kStreamingReady, kTaskComputedResponse }; WakeupKind GetWakeupKind(std::uint64_t id) { if (id == kSocketId) { return WakeupKind::kSocketReadable; } + if (id == kStreamingId) { + return WakeupKind::kStreamingReady; + } return WakeupKind::kTaskComputedResponse; } @@ -92,6 +96,7 @@ void Http2Connection::ListenForRequests() { engine::WaitAnyContext wait_any{}; wait_any.Append(kSocketId, GetSocket().GetReadableBase()); + wait_any.Append(kStreamingId, parser_->GetStreamingEvent()); while (!engine::current_task::ShouldCancel()) { StartAllRequestTasks(wait_any); @@ -117,12 +122,19 @@ void Http2Connection::ListenForRequests() { } wait_any.Append(kSocketId, GetSocket().GetReadableBase()); break; + case WakeupKind::kStreamingReady: + // Chunks produced by handler tasks (a streamed body, or the bytes of a + // protocol tunnelled over an upgraded stream) may only reach nghttp2 here. + parser_->HandleStreamingEvents(); + wait_any.Append(kStreamingId, parser_->GetStreamingEvent()); + break; case WakeupKind::kTaskComputedResponse: - OnRequestTaskFinished(*ready_id); + OnRequestTaskFinished(*ready_id, wait_any); break; } - UASSERT(wait_any.GetSize() <= config_.http2_session_config.max_concurrent_streams + 1); + // One slot per stream, plus the socket and the streaming event. + UASSERT(wait_any.GetSize() <= config_.http2_session_config.max_concurrent_streams + 2); } } @@ -145,9 +157,51 @@ Http2Connection::RequestTaskContext Http2Connection::StartRequestTask(std::share return {.task = ConnectionBase::StartRequestTask(request_ptr), .request = std::move(request_ptr)}; } -void Http2Connection::OnRequestTaskFinished(std::uint64_t event_id) noexcept { - SendResponse(*handler_tasks_[event_id].request); +void Http2Connection::OnRequestTaskFinished(std::uint64_t event_id, engine::WaitAnyContext& wait_any) noexcept { + auto& task_context = handler_tasks_[event_id]; + if (task_context.is_upgraded) { + FinishUpgradedStream(*task_context.request); + handler_tasks_.erase(event_id); + return; + } + + const bool is_upgrade = task_context.request->IsUpgradeWebsocket(); + SendResponse(*task_context.request); + auto request = std::move(task_context.request); handler_tasks_.erase(event_id); + if (is_upgrade) { + StartUpgradedTask(std::move(request), wait_any); + } +} + +void Http2Connection::StartUpgradedTask(HttpRequestPtr&& request_ptr, engine::WaitAnyContext& wait_any) noexcept { + UASSERT(parser_); + try { + const auto stream_id = http::Stream::Id{request_ptr->GetHttpResponse().GetStreamId().value()}; + auto stream_rw = parser_->UpgradeStream(stream_id); + // The tunnelled protocol runs in its own task, so that the connection keeps + // multiplexing the other streams for as long as it lives. + auto task = engine::CriticalAsyncNoTracing( + [request = request_ptr, socket = std::move(stream_rw), peer_name = remote_address_]() mutable { + request->DoUpgrade(std::move(socket), std::move(peer_name)); + } + ); + const auto& [task_context, slot_id] = handler_tasks_.emplace( + RequestTaskContext{.task = std::move(task), .request = std::move(request_ptr), .is_upgraded = true} + ); + wait_any.Append(slot_id, task_context.task); + } catch (const std::exception& ex) { + LOG_ERROR() << "Failed to upgrade a stream on fd " << GetFd() << ": " << ex; + } +} + +void Http2Connection::FinishUpgradedStream(const http::HttpRequest& request) noexcept { + UASSERT(parser_); + try { + parser_->CloseUpgradedStream(http::Stream::Id{request.GetHttpResponse().GetStreamId().value()}); + } catch (const std::exception& ex) { + LOG_WARNING() << "Failed to close an upgraded stream on fd " << GetFd() << ": " << ex; + } } void Http2Connection::SendResponse(http::HttpRequest& request) noexcept { diff --git a/core/src/server/net/http2_connection.hpp b/core/src/server/net/http2_connection.hpp index e6ddb251f905..1dd99902f4b6 100644 --- a/core/src/server/net/http2_connection.hpp +++ b/core/src/server/net/http2_connection.hpp @@ -57,12 +57,17 @@ class Http2Connection final : public ConnectionBase { struct RequestTaskContext final { RequestTask task; HttpRequestPtr request; + // Set for the task that runs a protocol tunnelled over an already answered + // stream; such a task must not produce a response of its own. + bool is_upgraded{false}; }; void ListenForRequests(); RequestTaskContext StartRequestTask(std::shared_ptr&& request_ptr) noexcept; void StartAllRequestTasks(engine::WaitAnyContext& wait_any); - void OnRequestTaskFinished(std::uint64_t event_id) noexcept; + void OnRequestTaskFinished(std::uint64_t event_id, engine::WaitAnyContext& wait_any) noexcept; + void StartUpgradedTask(HttpRequestPtr&& request_ptr, engine::WaitAnyContext& wait_any) noexcept; + void FinishUpgradedStream(const http::HttpRequest& request) noexcept; void SendResponse(http::HttpRequest& request) noexcept; std::unique_ptr MakeParser(); diff --git a/scripts/docs/en/userver/http_server.md b/scripts/docs/en/userver/http_server.md index 5a4c3fceeb9a..84b260d8cbe0 100644 --- a/scripts/docs/en/userver/http_server.md +++ b/scripts/docs/en/userver/http_server.md @@ -15,7 +15,7 @@ * HTTP 1.1/1.0 support; * HTTPS; -* @ref scripts/docs/en/userver/tutorial/websocket_service.md "WebSocket"; +* @ref scripts/docs/en/userver/tutorial/websocket_service.md "WebSocket", over HTTP/1.1 and over HTTP/2.0 (RFC 8441); * Body decompression with "Content-Encoding: gzip"; * HTTP pipelining; * Custom authorization @ref scripts/docs/en/userver/tutorial/auth_postgres.md ; @@ -79,9 +79,32 @@ components_manager: max_concurrent_streams: 100 max_frame_size: 16384 initial_window_size: 65536 + enable_connect_protocol: false ``` You can set some options specific to `HTTP/2.0` in the `http2-session` section. See docs for these options in components::Server +### WebSockets over HTTP/2.0 + +With `enable_connect_protocol: true` the server advertises +`SETTINGS_ENABLE_CONNECT_PROTOCOL` and accepts websockets bootstrapped with the +extended `CONNECT` method of +[RFC 8441](https://datatracker.ietf.org/doc/html/rfc8441), as Chrome, Firefox, +.NET `ClientWebSocket`, HAProxy and Envoy do. + +Such a request carries `:method: CONNECT` and `:protocol: websocket` instead of +the `Upgrade`/`Connection` headers of HTTP/1.1, and is answered with a plain +`200` rather than `101`. It is routed as a `GET`, so +@ref server::handlers::WebsocketHandlerBase handlers need no changes and no +extra registration: the same handler serves both transports. Use +@ref server::http::HttpRequest::IsWebsocketExtendedConnect() if a +@ref server::handlers::WebsocketHandlerBase::HandleHandshake() implementation +has to tell them apart. + +Unlike the HTTP/1.1 upgrade, which takes the whole connection over, the +websocket lives inside a single HTTP/2.0 stream, so ordinary requests keep being +served on the same connection. Note that each live websocket occupies a stream +slot of `max_concurrent_streams` for its whole lifetime. + ## Components diff --git a/scripts/docs/en/userver/tutorial/websocket_service.md b/scripts/docs/en/userver/tutorial/websocket_service.md index 2718237279ad..09e56c1874ba 100644 --- a/scripts/docs/en/userver/tutorial/websocket_service.md +++ b/scripts/docs/en/userver/tutorial/websocket_service.md @@ -33,6 +33,11 @@ Note that all the @ref userver_components "components" and @ref userver_http_handlers "handlers" have their static options additionally described in docs. +The handler above also serves clients that bootstrap the websocket over +HTTP/2.0 with the extended `CONNECT` method of RFC 8441, provided the listener +enables it — see +@ref scripts/docs/en/userver/http_server.md "the HTTP server docs". + ### int main() diff --git a/testsuite/requirements-internal-tests.txt b/testsuite/requirements-internal-tests.txt index 1417a31d199a..2f7998a4aa66 100644 --- a/testsuite/requirements-internal-tests.txt +++ b/testsuite/requirements-internal-tests.txt @@ -1,2 +1,5 @@ httpx >= 0.27.0 h2 >= 4.1.0 +# RFC 6455 framing without a handshake, for the RFC 8441 websocket tests: the +# frames travel inside the DATA frames of an HTTP/2.0 extended CONNECT stream. +wsproto >= 1.2.0 diff --git a/universal/include/userver/http/common_headers.hpp b/universal/include/userver/http/common_headers.hpp index 2c485d77e467..f971a918f42d 100644 --- a/universal/include/userver/http/common_headers.hpp +++ b/universal/include/userver/http/common_headers.hpp @@ -197,6 +197,8 @@ inline constexpr PredefinedHeader kScheme{":scheme"}; inline constexpr PredefinedHeader kAuthority{":authority"}; inline constexpr PredefinedHeader kPath{":path"}; inline constexpr PredefinedHeader kStatus{":status"}; +/// @brief The extended CONNECT pseudo-header of RFC 8441. +inline constexpr PredefinedHeader kProtocol{":protocol"}; } // namespace k2 } // namespace http::headers