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/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/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/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/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/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/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/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/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/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_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.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/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; } 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/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; 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