diff --git a/core/functional_tests/CMakeLists.txt b/core/functional_tests/CMakeLists.txt index 5dcf1ee83a3f..27c4a203eeef 100644 --- a/core/functional_tests/CMakeLists.txt +++ b/core/functional_tests/CMakeLists.txt @@ -61,6 +61,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/service.cpp b/core/functional_tests/http2server/service.cpp index 8fd5dd2da7ec..248d67dfcd44 100644 --- a/core/functional_tests/http2server/service.cpp +++ b/core/functional_tests/http2server/service.cpp @@ -53,13 +53,15 @@ class HandlerHttp2Stream final : public server::handlers::HttpHandlerBase { const auto& count_str = req.GetArg("count"); const std::size_t count = std::stoi(count_str); UASSERT(count != 0); + const auto& delay_str = req.GetArg("delay_ms"); + const std::chrono::milliseconds delay{delay_str.empty() ? 2 : std::stoi(delay_str)}; stream.SetStatusCode(200); stream.SetEndOfHeaders(); for (std::size_t i = 0; i < count - 1; i++) { std::string part{body_part}; stream.PushBodyChunk(std::move(part), {}); // Some pause... - engine::SleepFor(std::chrono::milliseconds{2}); + engine::SleepFor(delay); } std::string part{body_part}; stream.PushBodyChunk(std::move(part), {}); diff --git a/core/functional_tests/http2server/tests/test_http2_streaming.py b/core/functional_tests/http2server/tests/test_http2_streaming.py index 101273dc0404..885bb53ab5b0 100644 --- a/core/functional_tests/http2server/tests/test_http2_streaming.py +++ b/core/functional_tests/http2server/tests/test_http2_streaming.py @@ -1,11 +1,23 @@ import asyncio -import pytest +import h2.connection +import h2.events +import h2.settings + +import utils DEFAULT_PATH = '/http2server-stream' -@pytest.mark.skip(reason='TAXICOMMON-10258') +def _stream_headers(query: str) -> list: + return [ + (':method', 'GET'), + (':path', f'{DEFAULT_PATH}?{query}'), + (':scheme', 'http'), + (':authority', 'localhost'), + ] + + async def test_body_stream(http2_client, service_client, dynamic_config): part = 'part' count = 100 @@ -25,7 +37,6 @@ async def _stream_request(client, req_per_client): assert data == r.text -@pytest.mark.skip(reason='TAXICOMMON-10258') async def test_body_stream_small_pieces( http2_client, service_client, @@ -34,7 +45,6 @@ async def test_body_stream_small_pieces( await _stream_request(http2_client, 1) -@pytest.mark.skip(reason='TAXICOMMON-10258') async def test_body_stream_concurrent( http2_client, service_client, @@ -44,3 +54,179 @@ async def test_body_stream_concurrent( req_per_client = 10 tasks = [_stream_request(http2_client, req_per_client) for _ in range(clients_count)] await asyncio.gather(*tasks) + + +async def test_body_stream_concurrent_unique_bodies( + http2_client, + service_client, + dynamic_config, +): + # Each stream echoes a payload whose every 8-byte block encodes the + # stream number and the block position. Unlike identical payloads, this + # detects bytes leaking between concurrently multiplexed streams as well + # as chunk reordering within one stream. + async def echo_stream(i): + data = ''.join(f'{i:02d}:{j:04d};' for j in range(128)) # 1 KiB + r = await http2_client.get(DEFAULT_PATH, params={'type': 'ne'}, data=data) + assert 200 == r.status_code + assert data == r.text + + await asyncio.gather(*[echo_stream(i) for i in range(20)]) + + +async def test_body_stream_no_head_of_line_blocking( + http2_client, + service_client, + dynamic_config, +): + # A slow streamed response (~5s) on one stream must not delay other + # requests multiplexed on the same connection. If it did, each "fast" + # request below would complete only after the slow stream finishes and + # trip its timeout. + part = 'x' + count = 50 + slow = asyncio.create_task( + http2_client.get( + DEFAULT_PATH, + params={ + 'type': 'eq', + 'body_part': part, + 'count': count, + 'delay_ms': 100, + }, + timeout=30.0, + ), + ) + try: + for _ in range(5): + r = await asyncio.wait_for( + http2_client.get( + '/http2server', + params={'type': 'echo-body'}, + data='ping', + ), + timeout=2.0, + ) + assert 200 == r.status_code + assert 'ping' == r.text + finally: + r = await slow + assert 200 == r.status_code + assert part * count == r.text + + +async def test_reset_mid_stream_keeps_connection_usable( + create_connection, + service_client, +): + async with create_connection() as (sock, conn): + # A slow stream: the handler will keep producing for ~3s after the + # client resets the stream; those events must be dropped, not tear + # down the connection or the process. + stream_id = conn.get_next_available_stream_id() + conn.send_headers( + stream_id, + _stream_headers('type=eq&body_part=part&count=30&delay_ms=100'), + end_stream=True, + ) + await sock.sendall(conn.data_to_send()) + + events = [] + while not any(isinstance(event, h2.events.DataReceived) for event in events): + events += await utils.send_and_receive(sock, conn) + + conn.reset_stream(stream_id, error_code=0x8) # CANCEL + await sock.sendall(conn.data_to_send()) + + # The same connection must still serve requests, concurrently with + # the handler of the reset stream still pushing body parts. + echo_stream_id = conn.get_next_available_stream_id() + conn.send_headers( + echo_stream_id, + [ + (':method', 'GET'), + (':path', '/http2server?type=echo-header'), + (':scheme', 'http'), + (':authority', 'localhost'), + ('echo-header', 'still-alive'), + ], + end_stream=True, + ) + await sock.sendall(conn.data_to_send()) + + events = await utils.receive_until_stream_ended(sock, conn) + assert b'still-alive' == utils.response_data(events) + + +async def test_h2c_upgrade_with_streamed_response(create_socket, service_client): + # The first request of an h2c upgrade is parsed as HTTP/1.1, so the + # streamed response has no HTTP/2 producer; it must degrade to a + # buffered send instead of hanging on a forever-deferred provider. + async with create_socket() as sock: + conn = h2.connection.H2Connection() + settings_header = conn.initiate_upgrade_connection().decode('ascii') + request = ( + f'GET {DEFAULT_PATH}?type=eq&body_part=part&count=10 HTTP/1.1\r\n' + 'Host: localhost\r\n' + 'Connection: Upgrade, HTTP2-Settings\r\n' + 'Upgrade: h2c\r\n' + f'HTTP2-Settings: {settings_header}\r\n' + '\r\n' + ) + await sock.sendall(request.encode('ascii')) + + receive = b'' + while utils.HTTP1_HEADERS_END not in receive: + receive += await sock.recv(utils.RECEIVE_SIZE) + headers, _, http2_data = receive.partition(utils.HTTP1_HEADERS_END) + assert headers.startswith(b'HTTP/1.1 101 Switching Protocols') + + events = conn.receive_data(http2_data) if http2_data else [] + await sock.sendall(conn.data_to_send()) + while not any(isinstance(event, h2.events.StreamEnded) for event in events): + events += await utils.send_and_receive(sock, conn) + + assert b'part' * 10 == utils.response_data(events) + + +async def test_flow_control_backpressure(create_connection, service_client): + # With a tiny stream window the server may only produce as fast as the + # client opens the window with WINDOW_UPDATEs; the deferred provider must + # resume each time instead of stalling or flooding. + window = 1024 + part = 'x' * 1024 + count = 100 # total body is 100 KiB, also exceeds the connection window + + async with create_connection() as (sock, conn): + conn.update_settings( + {h2.settings.SettingCodes.INITIAL_WINDOW_SIZE: window}, + ) + stream_id = conn.get_next_available_stream_id() + conn.send_headers( + stream_id, + _stream_headers(f'type=eq&body_part={part}&count={count}&delay_ms=0'), + end_stream=True, + ) + await sock.sendall(conn.data_to_send()) + + body = b'' + ended = False + while not ended: + receive = await sock.recv(utils.RECEIVE_SIZE) + if not receive: + raise RuntimeError('Socket connection was closed by the other side') + for event in conn.receive_data(receive): + if isinstance(event, h2.events.DataReceived): + body += event.data + conn.acknowledge_received_data( + event.flow_controlled_length, + event.stream_id, + ) + elif isinstance(event, h2.events.StreamEnded): + ended = True + data = conn.data_to_send() + if data: + await sock.sendall(data) + + assert len(body) == len(part) * count + assert part.encode() * count == body 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 9316b30748f9..1dc91497c65b 100644 --- a/core/include/userver/server/http/http_request.hpp +++ b/core/include/userver/server/http/http_request.hpp @@ -256,6 +256,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 b31276b78732..478e3e452f76 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,6 @@ Http2Session::Http2Session( streaming_consumer_(streaming_queue_->GetConsumer()) { UASSERT(streaming_queue_); - UASSERT(streaming_event_.IsAutoReset()); nghttp2_session_callbacks* callbacks{nullptr}; UINVARIANT(nghttp2_session_callbacks_new(&callbacks) == 0, "Failed to init callbacks for HTTP/2.0"); @@ -80,11 +85,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"); @@ -98,17 +107,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: { @@ -191,8 +216,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); @@ -215,7 +249,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); @@ -246,8 +287,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; } @@ -331,9 +377,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)}; } @@ -385,6 +434,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); @@ -420,21 +525,25 @@ void Http2Session::WriteWhileWant() { engine::SingleConsumerEvent& Http2Session::GetStreamingEvent() { return streaming_event_; } -void Http2Session::HandleStreamingEvents() { - 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())); - ThrowIfErr(res, "Error while resume_data"); - stream.SetDeferred(false); - } - stream.PushChunk(std::move(event.body_part)); - stream.SetEnd(event.is_end); - event = {}; +bool Http2Session::PopStreamingEventNoblock(impl::Http2StreamEvent& event) { + return streaming_consumer_.PopNoblock(event); +} + +void Http2Session::ApplyStreamingEvent(impl::Http2StreamEvent&& event) { + UASSERT(event.stream_id != -1); + auto* stream = FindStream(Stream::Id{event.stream_id}); + if (stream == nullptr) { + // The stream is already closed (e.g. reset by the client) while the + // handler was still producing body parts. Drop the event. + return; } - WriteWhileWant(); + if (stream->IsDeferred()) { + const auto res = nghttp2_session_resume_data(session_.get(), event.stream_id); + ThrowIfErr(res, "Error while resume_data"); + stream->SetDeferred(false); + } + stream->PushChunk(std::move(event.body_part)); + stream->SetEnd(event.is_end); } } // namespace server::http diff --git a/core/src/server/http/http2_session.hpp b/core/src/server/http/http2_session.hpp index f35ae2a8d97f..f73bc1f6da98 100644 --- a/core/src/server/http/http2_session.hpp +++ b/core/src/server/http/http2_session.hpp @@ -59,10 +59,23 @@ 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(); - void HandleStreamingEvents(); + + // Returns false if there are no pending streaming events. + [[nodiscard]] bool PopStreamingEventNoblock(impl::Http2StreamEvent& event); + + // Applies a body-streaming event to its stream. Events for streams that + // are already closed (e.g. reset by the client) are dropped. + void ApplyStreamingEvent(impl::Http2StreamEvent&& event); bool ConnectionIsOk() const; @@ -119,11 +132,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); bool MemRecv(std::string_view data); @@ -142,7 +158,9 @@ class Http2Session final : public request::RequestParser { engine::io::RwBase* socket_; std::shared_ptr streaming_queue_{nullptr}; - engine::SingleConsumerEvent streaming_event_; + // No-auto-reset: the event is awaited through WaitAny in the connection + // loop, which resets it manually before draining the queue. + 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..93d98f0c4b82 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,8 @@ #include #include #include +#include +#include #include #include @@ -358,6 +361,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"; @@ -382,6 +597,83 @@ UTEST_F(Http2SessionTest, ForCurl) { EXPECT_EQ(request->GetMethod(), HttpMethod::kPost); } +UTEST(Http2SessionStreaming, EventQueueIsFifoPerStreamAndSignals) { + const auto queue = impl::Http2StreamEventQueue::Create(); + engine::SingleConsumerEvent event{engine::SingleConsumerEvent::NoAutoReset()}; + impl::Http2StreamEventProducer producer{*queue, event}; + auto consumer = queue->GetConsumer(); + + producer.PushEvent({1, "first"}); + producer.PushEvent({1, "second"}); + producer.CloseStream(1); + EXPECT_TRUE(event.IsReady()); + + impl::Http2StreamEvent popped; + ASSERT_TRUE(consumer.PopNoblock(popped)); + EXPECT_EQ(popped.stream_id, 1); + EXPECT_EQ(popped.body_part, "first"); + EXPECT_FALSE(popped.is_end); + ASSERT_TRUE(consumer.PopNoblock(popped)); + EXPECT_EQ(popped.body_part, "second"); + ASSERT_TRUE(consumer.PopNoblock(popped)); + EXPECT_TRUE(popped.is_end); + EXPECT_FALSE(consumer.PopNoblock(popped)); +} + +UTEST(Http2SessionStreaming, StreamingEventIsWaitAnyCompatible) { + auto parser = CreateTestParser([](std::shared_ptr&&) {}, USERVER_NAMESPACE::http::HttpVersion::k2); + auto& session = dynamic_cast(*parser); + + auto& event = session.GetStreamingEvent(); + EXPECT_FALSE(event.IsAutoReset()); + // Auto-reset events UINVARIANT-abort here; the connection loop appends + // this token to its WaitAnyContext. + EXPECT_FALSE(event.GetAwaitableToken().IsEmpty()); +} + +UTEST(Http2SessionStreaming, EventForUnknownStreamIsDropped) { + auto parser = CreateTestParser([](std::shared_ptr&&) {}, USERVER_NAMESPACE::http::HttpVersion::k2); + auto& session = dynamic_cast(*parser); + + impl::Http2StreamEvent event; + EXPECT_FALSE(session.PopStreamingEventNoblock(event)); + + // A handler may still be producing body parts after the client reset the + // stream; such events must be dropped, not tear down the connection. + impl::Http2StreamEvent late{42, "late chunk", true}; + EXPECT_NO_THROW(session.ApplyStreamingEvent(std::move(late))); +} + +UTEST(Http2SessionStreaming, SetStreamBodyPicksHttp2Producer) { + const auto queue = impl::Http2StreamEventQueue::Create(); + engine::SingleConsumerEvent event{engine::SingleConsumerEvent::NoAutoReset()}; + + request::ResponseDataAccounter accounter; + const auto request = HttpRequestBuilder{accounter} + .SetMethod(HttpMethod::kGet) + .SetHttpMajor(2) + .SetHttpMinor(0) + .SetUrl("/") + .SetResponseStreamId(1) + .SetStreamProducer(impl::Http2StreamEventProducer{*queue, event}) + .Build(); + auto& response = request->GetHttpResponse(); + + // Used to UINVARIANT-abort the whole process for HTTP/2 responses. + response.SetStreamBody(); + EXPECT_TRUE(response.IsBodyStreamed()); + + auto producer = response.GetBodyProducer(); + ASSERT_TRUE(std::holds_alternative(producer)); + + auto consumer = queue->GetConsumer(); + std::get(producer).PushEvent({1, "chunk"}); + impl::Http2StreamEvent popped; + ASSERT_TRUE(consumer.PopNoblock(popped)); + EXPECT_EQ(popped.body_part, "chunk"); + EXPECT_TRUE(event.IsReady()); +} + } // namespace server::http USERVER_NAMESPACE_END diff --git a/core/src/server/http/http2_stream.cpp b/core/src/server/http/http2_stream.cpp index 2ec5636d2b3e..0286b8ce6247 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 b5ea4c88c3dc..fd964bb2eb78 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 #include @@ -17,6 +19,8 @@ class Socket; namespace server::http { +class Http2StreamReadPipe; + class Stream final { public: using Id = utils::StrongTypedef; @@ -42,6 +46,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); void PushChunk(request::impl::ChunkStorage&& chunk); @@ -53,6 +73,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 3d9ca11a6b16..fa760cd1f95a 100644 --- a/core/src/server/http/http2_writer.cpp +++ b/core/src/server/http/http2_writer.cpp @@ -105,18 +105,37 @@ class Http2ResponseWriter final { void WriteHttpResponse() { const auto& data = response_.GetData(); - auto headers = GetHeaders(); - const bool is_body_forbidden = IsBodyForbiddenForStatus(response_.status_); + bool buffered_h1_stream = false; + if (response_.IsBodyStreamed() && response_.body_stream_.has_value()) { + // The handler streamed into the HTTP/1.1 queue because the stream + // id was assigned only now, at send time (h2c upgrade). The + // handler has already finished, so buffer the parts and send a + // regular response. + std::string body_part; + while (response_.body_stream_->Pop(body_part)) { + data.append(body_part); + } + response_.body_stream_.reset(); + buffered_h1_stream = true; + } + + 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() && !buffered_h1_stream) || response_.request_.IsUpgradeWebsocket(); + stream.SetStreaming(keeps_stream_open && data.empty()); std::size_t bytes = headers.GetSize(); nghttp2_data_provider* provider{nullptr}; @@ -142,7 +161,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 +184,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 3007cf595cd4..5dc7c0013821 100644 --- a/core/src/server/http/http_request.cpp +++ b/core/src/server/http/http_request.cpp @@ -292,6 +292,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 40da68663436..76d5c542e758 100644 --- a/core/src/server/http/http_request_impl.hpp +++ b/core/src/server/http/http_request_impl.hpp @@ -42,6 +42,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_response.cpp b/core/src/server/http/http_response.cpp index f0e3828d4cc0..31e7a12daaf7 100644 --- a/core/src/server/http/http_response.cpp +++ b/core/src/server/http/http_response.cpp @@ -480,7 +480,6 @@ void SetThrottleReason(http::HttpResponse& http_response, std::string log_reason void HttpResponse::SetStreamBody() { UASSERT(body_stream_producer_.index() == 0); if (GetStreamId().has_value()) { - UINVARIANT(false, "Streaming in HTTP/2.0 is not supported currently."); body_stream_producer_.emplace(GetStreamProducer()); } else { UASSERT(!body_stream_); diff --git a/core/src/server/http/http_response_body_stream.cpp b/core/src/server/http/http_response_body_stream.cpp index 2c5bac24e6f4..e2bc9c6decef 100644 --- a/core/src/server/http/http_response_body_stream.cpp +++ b/core/src/server/http/http_response_body_stream.cpp @@ -30,7 +30,14 @@ ResponseBodyStream::~ResponseBodyStream() { void ResponseBodyStream::PushBodyChunk(std::string&& chunk, engine::Deadline deadline) { UASSERT_MSG(headers_ended_, "SetEndOfHeaders() was not called before PushBodyChunk()"); - UASSERT_MSG(http_response_.GetData().empty(), "PushBodyChunk() was called after SetBody()"); + // Only check before the first chunk is announced: after that the + // connection coroutine may be sending the response concurrently and reads + // of the response data would race with it. SetBody() after the first + // chunk is already asserted in SetBody() itself. + UASSERT_MSG( + headers_end_sent_ || http_response_.GetData().empty(), + "PushBodyChunk() was called after SetBody()" + ); if (headers_ended_ && !headers_end_sent_) { http_response_.SetHeadersEnd(); 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 871069ea405c..b1d7ac160671 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 = std::numeric_limits::max() - 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,23 @@ void Http2Connection::ListenForRequests() { } wait_any.Append(kSocketId, GetSocket().GetReadableBase()); break; + case WakeupKind::kStreamingReady: + // The completed awaitable was dropped out of `wait_any`, so the + // no-auto-reset event has no active awaiter and may be reset + // here. Resetting before the drain keeps a signal arriving + // mid-drain for the next round. `Reset()` is not allowed while + // the event is appended (active awaiter), which is why the + // drain in `OnRequestTaskFinished` leaves the signal alone. + parser_->GetStreamingEvent().Reset(); + 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); + UASSERT(wait_any.GetSize() <= config_.http2_session_config.max_concurrent_streams + 2); } } @@ -142,15 +158,112 @@ Http2Connection::RequestTaskContext Http2Connection::StartRequestTask(std::share stats_.active_request_count.Add(1); - return {.task = ConnectionBase::StartRequestTask(request_ptr), .request = std::move(request_ptr)}; + auto task = ConnectionBase::StartRequestTask(request_ptr); + + // `SetStreamBody()` is called synchronously in `StartRequestTask` before + // the handler task is spawned, so `IsBodyStreamed()` is reliable here. + // Requests without a stream id (h2c upgrade) keep the buffered send path. + const auto& response = request_ptr->GetHttpResponse(); + if (response.IsBodyStreamed() && response.GetStreamId().has_value()) { + streamed_requests_.emplace(*response.GetStreamId(), StreamedRequestContext{request_ptr, false}); + } + + return {.task = std::move(task), .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; + } + + auto& request = *task_context.request; + const bool is_upgrade = request.IsUpgradeWebsocket(); + const auto stream_id = request.GetHttpResponse().GetStreamId(); + if (stream_id.has_value() && streamed_requests_.find(*stream_id) != streamed_requests_.end()) { + // Drain the remaining body parts. `ResponseBodyStream` always pushes a + // final event before the handler task completes, so this also submits + // the response if no streaming event was processed for it yet. + try { + HandleStreamingEvents(); + } catch (const std::exception& ex) { + LOG_ERROR() << "Error while sending streamed body parts: " << ex; + request.GetHttpResponse().SetSendFailed(std::chrono::steady_clock::now()); + } + SubmitStreamedResponseIfPending(*stream_id); + FinalizeResponse(request); + streamed_requests_.erase(*stream_id); + } else { + SendResponse(request); + } + + auto request_ptr = std::move(task_context.request); handler_tasks_.erase(event_id); + if (is_upgrade) { + StartUpgradedTask(std::move(request_ptr), 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::HandleStreamingEvents() { + http::impl::Http2StreamEvent event; + while (parser_->PopStreamingEventNoblock(event)) { + // The first event for a stream means its headers are complete + // (`SetHeadersEnd()` precedes the first `PushBodyChunk()`), so the + // response with its deferred body provider is submitted here. + SubmitStreamedResponseIfPending(event.stream_id); + parser_->ApplyStreamingEvent(std::move(event)); + event = {}; + } + parser_->WriteWhileWant(); +} + +void Http2Connection::SubmitStreamedResponseIfPending(std::int32_t stream_id) noexcept { + const auto it = streamed_requests_.find(stream_id); + if (it == streamed_requests_.end() || it->second.submit_attempted) { + return; + } + it->second.submit_attempted = true; + SubmitResponse(*it->second.request); } void Http2Connection::SendResponse(http::HttpRequest& request) noexcept { + SubmitResponse(request); + FinalizeResponse(request); +} + +void Http2Connection::SubmitResponse(http::HttpRequest& request) noexcept { auto& response = request.GetHttpResponse(); UASSERT(!response.IsSent()); if (IsResponseChainValid()) { diff --git a/core/src/server/net/http2_connection.hpp b/core/src/server/net/http2_connection.hpp index e6ddb251f905..8215e09b29af 100644 --- a/core/src/server/net/http2_connection.hpp +++ b/core/src/server/net/http2_connection.hpp @@ -3,6 +3,7 @@ #include #include #include +#include #include #include @@ -57,13 +58,30 @@ 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}; + }; + + // A request whose handler streams the response body (`IsBodyStreamed()`). + // Its response is submitted upon the first streaming event instead of on + // handler task completion. + struct StreamedRequestContext final { + HttpRequestPtr request; + bool submit_attempted{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 HandleStreamingEvents(); + void SubmitStreamedResponseIfPending(std::int32_t stream_id) noexcept; + void StartUpgradedTask(HttpRequestPtr&& request_ptr, engine::WaitAnyContext& wait_any) noexcept; + void FinishUpgradedStream(const http::HttpRequest& request) noexcept; void SendResponse(http::HttpRequest& request) noexcept; + void SubmitResponse(http::HttpRequest& request) noexcept; + void FinalizeResponse(http::HttpRequest& request) noexcept; std::unique_ptr MakeParser(); void EnsureHttp2(); @@ -83,6 +101,7 @@ class Http2Connection final : public ConnectionBase { engine::io::Sockaddr remote_address_; std::unique_ptr parser_; utils::SlotMap handler_tasks_; + std::unordered_map streamed_requests_; }; } // namespace server::net 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 37a22b5cf972..019175b431af 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