From ff7ee7b005779a0b922922435ddc7f6a74ced85b Mon Sep 17 00:00:00 2001 From: SSE4 Date: Mon, 20 Jul 2026 11:32:58 +0700 Subject: [PATCH 1/4] feat core: restore response body streaming for HTTP/2.0 Body streaming over h2 was disabled (SetStreamBody() hit a UINVARIANT whose throw escaped into std::terminate) after the connection refactor dropped the streaming-event wiring from the connection loop, and because the old serial loop head-of-line-blocked concurrent streams. Restore it on top of the concurrent WaitAnyContext loop: * Http2Connection additionally waits on the session's streaming SingleConsumerEvent (NoAutoReset, reset only at the wakeup point where its awaitable has been dropped out of the context) and drains the event queue into the per-stream chunk buffers, resuming deferred nghttp2 data providers. * The response of a streaming handler is submitted upon the first streaming event of its stream (SetHeadersEnd() always precedes the first PushBodyChunk()), so headers go out early while the handler keeps producing; task completion only finalizes stats/access logs. * Events for streams already reset by the client are dropped instead of tearing down the connection; submit errors fail only that request. * An h2c-upgrade first request with a streaming handler (no stream id at SetStreamBody() time, so it used the HTTP/1.1 queue) degrades to a buffered send instead of hanging on a forever-deferred provider. Head-of-line blocking is gone by construction: handler tasks already run concurrently and NGHTTP2_ERR_DEFERRED starves only the chunk-less stream, which nghttp2 skips while serving the other multiplexed streams. Un-skip the three TAXICOMMON-10258 functional tests and add a head-of-line-blocking guard test (slow streamed response must not delay fast requests on the same connection). --- core/functional_tests/http2server/service.cpp | 4 +- .../http2server/tests/test_http2_streaming.py | 46 +++++++++-- core/src/server/http/http2_session.cpp | 33 ++++---- core/src/server/http/http2_session.hpp | 12 ++- core/src/server/http/http2_writer.cpp | 16 +++- core/src/server/http/http_response.cpp | 1 - core/src/server/net/http2_connection.cpp | 81 ++++++++++++++++++- core/src/server/net/http2_connection.hpp | 14 ++++ 8 files changed, 178 insertions(+), 29 deletions(-) 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..e6bfd9fe55c4 100644 --- a/core/functional_tests/http2server/tests/test_http2_streaming.py +++ b/core/functional_tests/http2server/tests/test_http2_streaming.py @@ -1,11 +1,8 @@ import asyncio -import pytest - DEFAULT_PATH = '/http2server-stream' -@pytest.mark.skip(reason='TAXICOMMON-10258') async def test_body_stream(http2_client, service_client, dynamic_config): part = 'part' count = 100 @@ -25,7 +22,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 +30,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 +39,44 @@ 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_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 diff --git a/core/src/server/http/http2_session.cpp b/core/src/server/http/http2_session.cpp index 1ed167b28280..cde008b8c5d7 100644 --- a/core/src/server/http/http2_session.cpp +++ b/core/src/server/http/http2_session.cpp @@ -58,7 +58,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"); @@ -390,21 +389,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 = static_cast(nghttp2_session_get_stream_user_data(session_.get(), 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; + } + if (stream->IsDeferred()) { + const auto res = nghttp2_session_resume_data(session_.get(), event.stream_id); + ThrowIfErr(res, "Error while resume_data"); + stream->SetDeferred(false); } - WriteWhileWant(); + 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 91673860d320..83a5dd03da1a 100644 --- a/core/src/server/http/http2_session.hpp +++ b/core/src/server/http/http2_session.hpp @@ -62,7 +62,13 @@ class Http2Session final : public request::RequestParser { 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; @@ -133,7 +139,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_writer.cpp b/core/src/server/http/http2_writer.cpp index 5bf9eedc6e32..3320bd0f75d1 100644 --- a/core/src/server/http/http2_writer.cpp +++ b/core/src/server/http/http2_writer.cpp @@ -105,6 +105,20 @@ class Http2ResponseWriter final { void WriteHttpResponse() { auto data = response_.ExtractData(); + 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; + } + auto headers = GetHeaders(); const bool is_body_forbidden = IsBodyForbiddenForStatus(response_.status_); @@ -116,7 +130,7 @@ class Http2ResponseWriter final { const auto stream_id = response_.GetStreamId().value(); auto& stream = http2_session_.GetStreamChecked(Stream::Id{stream_id}); - stream.SetStreaming(response_.IsBodyStreamed() && data.empty()); + stream.SetStreaming(response_.IsBodyStreamed() && !buffered_h1_stream && data.empty()); std::size_t bytes = headers.GetSize(); nghttp2_data_provider* provider{nullptr}; diff --git a/core/src/server/http/http_response.cpp b/core/src/server/http/http_response.cpp index 817366707767..18498f007db7 100644 --- a/core/src/server/http/http_response.cpp +++ b/core/src/server/http/http_response.cpp @@ -462,7 +462,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/net/http2_connection.cpp b/core/src/server/net/http2_connection.cpp index 12bdcb2ecc44..f6ec0f8b2513 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); 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,69 @@ 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); + auto& request = *handler_tasks_[event_id].request; + 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); + } handler_tasks_.erase(event_id); } +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()); request.SetStartSendResponseTime(); @@ -177,6 +247,9 @@ void Http2Connection::SendResponse(http::HttpRequest& request) noexcept { } else { response.SetSendFailed(std::chrono::steady_clock::now()); } +} + +void Http2Connection::FinalizeResponse(http::HttpRequest& request) noexcept { request.SetFinishSendResponseTime(); stats_.active_request_count.Subtract(1); ++stats_.requests_processed_count; diff --git a/core/src/server/net/http2_connection.hpp b/core/src/server/net/http2_connection.hpp index e6ddb251f905..99a0eccf0107 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 @@ -59,11 +60,23 @@ class Http2Connection final : public ConnectionBase { HttpRequestPtr request; }; + // 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 HandleStreamingEvents(); + void SubmitStreamedResponseIfPending(std::int32_t stream_id) 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 +96,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 From c6cfbc8f08c5670ec69b61f026746cc69c382828 Mon Sep 17 00:00:00 2001 From: SSE4 Date: Mon, 20 Jul 2026 13:41:20 +0700 Subject: [PATCH 2/4] test core: harden HTTP/2.0 body streaming coverage; fix debug assert race New functional tests for the restored h2 streaming: * client reset mid-stream: the handler keeps producing after RST_STREAM; its events must be dropped and the connection must stay usable; * h2c upgrade whose first response is streamed: must degrade to a buffered send, not hang; * flow-control backpressure: a 100 KiB streamed body behind a 1 KiB stream window (and past the connection window) must arrive intact, WINDOW_UPDATE-gated. New unit tests (there were none for the streaming machinery): event queue FIFO/signal contract, WaitAny compatibility of the streaming event, dropped events for unknown streams, and SetStreamBody() picking the h2 producer (previously a process-fatal UINVARIANT). Under TSan the old PushBodyChunk() assert read the response data while the connection coroutine's ExtractData() moved it (the response is now submitted concurrently with the producing handler). Check the data only before the first chunk is announced - after that SetBody() is already impossible and asserts in SetBody() itself. --- .../http2server/tests/test_http2_streaming.py | 132 ++++++++++++++++++ core/src/server/http/http2_session_test.cpp | 78 +++++++++++ .../server/http/http_response_body_stream.cpp | 9 +- 3 files changed, 218 insertions(+), 1 deletion(-) diff --git a/core/functional_tests/http2server/tests/test_http2_streaming.py b/core/functional_tests/http2server/tests/test_http2_streaming.py index e6bfd9fe55c4..73bba45f967a 100644 --- a/core/functional_tests/http2server/tests/test_http2_streaming.py +++ b/core/functional_tests/http2server/tests/test_http2_streaming.py @@ -1,8 +1,23 @@ import asyncio +import h2.connection +import h2.events +import h2.settings + +import utils + DEFAULT_PATH = '/http2server-stream' +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 @@ -80,3 +95,120 @@ async def test_body_stream_no_head_of_line_blocking( 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/src/server/http/http2_session_test.cpp b/core/src/server/http/http2_session_test.cpp index c4a40812cae3..353c1cb882d7 100644 --- a/core/src/server/http/http2_session_test.cpp +++ b/core/src/server/http/http2_session_test.cpp @@ -11,6 +11,7 @@ #include #include #include +#include #include #include @@ -382,6 +383,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/http_response_body_stream.cpp b/core/src/server/http/http_response_body_stream.cpp index de2f5c3d6cb4..e7ded0820ade 100644 --- a/core/src/server/http/http_response_body_stream.cpp +++ b/core/src/server/http/http_response_body_stream.cpp @@ -24,7 +24,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(); From 999a084d49a500a49444a8fb93aaa399071e9e5c Mon Sep 17 00:00:00 2001 From: SSE4 Date: Mon, 20 Jul 2026 14:10:50 +0700 Subject: [PATCH 3/4] test core: verify streamed body integrity across multiplexed streams The concurrent streaming tests used identical payloads on every stream, so bytes leaking between concurrently multiplexed streams (or chunks reordered within one stream) could not be detected. Echo a payload whose every block encodes the stream number and block position, 20 streams multiplexed on one connection, and require byte-exact responses. --- .../http2server/tests/test_http2_streaming.py | 18 ++++++++++++++++++ 1 file changed, 18 insertions(+) diff --git a/core/functional_tests/http2server/tests/test_http2_streaming.py b/core/functional_tests/http2server/tests/test_http2_streaming.py index 73bba45f967a..885bb53ab5b0 100644 --- a/core/functional_tests/http2server/tests/test_http2_streaming.py +++ b/core/functional_tests/http2server/tests/test_http2_streaming.py @@ -56,6 +56,24 @@ async def test_body_stream_concurrent( 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, From f92f542cbd07db4c4473a8ff37007a0d89daac7f Mon Sep 17 00:00:00 2001 From: SSE4 Date: Tue, 18 Aug 2026 22:39:08 +0700 Subject: [PATCH 4/4] feat core: support websockets over HTTP/2.0 (RFC 8441) Accept websockets bootstrapped with the extended CONNECT method of RFC 8441, as Chrome, Firefox, .NET ClientWebSocket, HAProxy and Envoy send them. Enabled per listener with the new `http2-session.enable_connect_protocol` option (default off), which makes the server advertise SETTINGS_ENABLE_CONNECT_PROTOCOL. Such a request carries `:method: CONNECT` and `:protocol: websocket` instead of the Upgrade/Connection headers, and is answered with a plain 200 rather than a 101. It is routed as a GET so that path-registered handlers match, the way reverse proxies do when converting an HTTP/1.1 upgrade, so server::handlers::WebsocketHandlerBase subclasses need no changes at all: the same handler now serves both transports. HttpRequest::IsWebsocketExtendedConnect() tells them apart where it matters. Unlike the HTTP/1.1 upgrade, which hands the socket over and stops doing HTTP, the websocket lives inside a single stream, so the connection keeps multiplexing ordinary requests around it. The new Http2StreamRw presents that stream as an engine::io::RwBase: incoming DATA frames land in a lock-free pipe that outlives the stream (the peer may reset it while the handler unwinds), and writes are pushed as streaming events, since nghttp2_session may only be touched on the connection task. This is why the change sits on top of the body-streaming restore: it reuses the same event queue and the same connection-loop drain. Notable details: * `:method` and `:protocol` may arrive in any order, so a CONNECT stream leaves its method unset until its header block ends, which defers url parsing and handler matching to the point where the effective method is known. * The read pipe is created when the request is finalized rather than on upgrade: a client may send websocket bytes before the handler accepted the stream. * Http2StreamRw::ReadNoblock yields. Over HTTP/1.1 the busy loop of WebSocketConnection::TryRecv is self-sufficient because the handler owns the socket; over HTTP/2.0 the pipe is filled by the connection task, which a non-yielding spinner starves. * A handler that does not upgrade must not answer 2xx to an extended CONNECT, as that claims the tunnel is established; Http2ResponseWriter rewrites it. * OnStreamClose now tolerates an already-removed stream. Rejecting a CONNECT (reset + remove) otherwise threw out of an nghttp2 callback and killed the whole connection -- the same latent bug the FinalizeRequest error path had. Tested by core/functional_tests/websocket_http2 (the same service as the HTTP/1.1 websocket test, driven by a hand-rolled h2 + wsproto client, as no mainstream Python websocket client speaks RFC 8441), including multiplexing an ordinary request onto a connection with a live websocket, and by unit tests for the settings, the routing and the negative cases. Green under asan/ubsan and TSAN. --- core/functional_tests/CMakeLists.txt | 3 + core/functional_tests/websocket/service.cpp | 14 ++ .../websocket/static_config.yaml | 4 + .../websocket_http2/CMakeLists.txt | 8 + .../websocket_http2/static_config.yaml | 72 ++++++ .../websocket_http2/tests/conftest.py | 196 ++++++++++++++++ .../tests/test_websocket_http2.py | 122 ++++++++++ .../userver/server/http/http_request.hpp | 7 + .../server/http/http_request_builder.hpp | 2 + core/src/server/component.yaml | 5 + .../src/server/handlers/websocket_handler.cpp | 57 +++-- core/src/server/http/http2_session.cpp | 136 +++++++++-- core/src/server/http/http2_session.hpp | 10 + core/src/server/http/http2_session_test.cpp | 214 ++++++++++++++++++ core/src/server/http/http2_stream.cpp | 17 ++ core/src/server/http/http2_stream.hpp | 24 ++ core/src/server/http/http2_stream_rw.cpp | 207 +++++++++++++++++ core/src/server/http/http2_stream_rw.hpp | 115 ++++++++++ core/src/server/http/http2_writer.cpp | 32 ++- core/src/server/http/http_request.cpp | 2 + core/src/server/http/http_request_builder.cpp | 5 + .../server/http/http_request_constructor.cpp | 4 + .../server/http/http_request_constructor.hpp | 1 + core/src/server/http/http_request_impl.hpp | 3 + core/src/server/net/connection_config.cpp | 1 + core/src/server/net/connection_config.hpp | 3 + core/src/server/net/http2_connection.cpp | 49 +++- core/src/server/net/http2_connection.hpp | 7 +- scripts/docs/en/userver/http_server.md | 25 +- .../en/userver/tutorial/websocket_service.md | 5 + testsuite/requirements-internal-tests.txt | 3 + .../include/userver/http/common_headers.hpp | 2 + 32 files changed, 1314 insertions(+), 41 deletions(-) create mode 100644 core/functional_tests/websocket_http2/CMakeLists.txt create mode 100644 core/functional_tests/websocket_http2/static_config.yaml create mode 100644 core/functional_tests/websocket_http2/tests/conftest.py create mode 100644 core/functional_tests/websocket_http2/tests/test_websocket_http2.py create mode 100644 core/src/server/http/http2_stream_rw.cpp create mode 100644 core/src/server/http/http2_stream_rw.hpp diff --git a/core/functional_tests/CMakeLists.txt b/core/functional_tests/CMakeLists.txt index 85d2d9353eb7..42b39e3fb2d3 100644 --- a/core/functional_tests/CMakeLists.txt +++ b/core/functional_tests/CMakeLists.txt @@ -56,6 +56,9 @@ add_dependencies(${PROJECT_NAME} ${PROJECT_NAME}-cache-update) add_subdirectory(websocket) add_dependencies(${PROJECT_NAME} ${PROJECT_NAME}-websocket) +add_subdirectory(websocket_http2) +add_dependencies(${PROJECT_NAME} ${PROJECT_NAME}-websocket-http2) + # WebSocket client requires curl >= 7.86 with WebSocket support if(CURL_VERSION_STRING VERSION_GREATER_EQUAL "7.86") add_subdirectory(websocket_client) diff --git a/core/functional_tests/websocket/service.cpp b/core/functional_tests/websocket/service.cpp index 86a66daf434e..c2025fb21863 100644 --- a/core/functional_tests/websocket/service.cpp +++ b/core/functional_tests/websocket/service.cpp @@ -146,10 +146,24 @@ class WebsocketsPingPongHandler final : public server::handlers::WebsocketHandle } }; +/// Only configured by the HTTP/2.0 variant of this service, to check what an extended +/// CONNECT of RFC 8441 aimed at an ordinary handler answers. +class PlainHandler final : public server::handlers::HttpHandlerBase { +public: + static constexpr std::string_view kName = "plain-handler"; + + using HttpHandlerBase::HttpHandlerBase; + + std::string HandleRequestThrow(const server::http::HttpRequest&, server::request::RequestContext&) const override { + return "Not a websocket handler"; + } +}; + int main(int argc, char* argv[]) { const auto component_list = components::MinimalServerComponentList() .Append() + .Append() .Append() .Append() .Append() diff --git a/core/functional_tests/websocket/static_config.yaml b/core/functional_tests/websocket/static_config.yaml index 8730de66ec27..e848c2b65845 100644 --- a/core/functional_tests/websocket/static_config.yaml +++ b/core/functional_tests/websocket/static_config.yaml @@ -45,6 +45,10 @@ components_manager: task_processor: main-task-processor max-remote-payload: 100000 fragment-size: 10 + plain-handler: # Unused here; exercised by the HTTP/2.0 variant of this service. + path: /plain + method: GET + task_processor: main-task-processor testsuite-support: diff --git a/core/functional_tests/websocket_http2/CMakeLists.txt b/core/functional_tests/websocket_http2/CMakeLists.txt new file mode 100644 index 000000000000..138884adc364 --- /dev/null +++ b/core/functional_tests/websocket_http2/CMakeLists.txt @@ -0,0 +1,8 @@ +project(userver-core-tests-websocket-http2 CXX) + +# The very same handlers as the HTTP/1.1 websocket test: a websocket handler is not +# supposed to notice which transport bootstrapped it. +add_executable(${PROJECT_NAME} "${CMAKE_CURRENT_SOURCE_DIR}/../websocket/service.cpp") +target_link_libraries(${PROJECT_NAME} userver::core) + +userver_chaos_testsuite_add() diff --git a/core/functional_tests/websocket_http2/static_config.yaml b/core/functional_tests/websocket_http2/static_config.yaml new file mode 100644 index 000000000000..a27a001eb51f --- /dev/null +++ b/core/functional_tests/websocket_http2/static_config.yaml @@ -0,0 +1,72 @@ +components_manager: + + task_processors: # Task processor is an executor for coroutine tasks + main-task-processor: # Make a task processor for CPU-bound coroutine tasks. + worker_threads: 4 # Process tasks in 4 threads. + fs-task-processor: # Make a separate task processor for filesystem bound tasks. + worker_threads: 4 + + default_task_processor: main-task-processor + + components: # Configuring components that were registered via component_list + server: + listener: # configuring the main listening socket... + connection: + http-version: 2 + http2-session: + enable_connect_protocol: true + port: 8080 # ...to listen on this port and... + task_processor: main-task-processor # ...process incoming requests on this task processor. + logging: + fs-task-processor: fs-task-processor + loggers: + default: + file_path: '@stderr' + level: debug + overflow_behavior: discard # Drop logs if the system is too busy to write them down. + + websocket-handler: # Finally! Websocket handler. + path: /chat # Registering handlers '/*' find files. + method: GET # Handle only GET requests. + task_processor: main-task-processor # Run it on CPU bound task processor + max-remote-payload: 100000 + fragment-size: 10 + websocket-handler-alt: # Finally! Websocket handler. + path: /handler-alt # Registering handlers '/*' find files. + method: GET # Handle only GET requests. + task_processor: main-task-processor # Run it on CPU bound task processor + max-remote-payload: 100000 + fragment-size: 10 + websocket-duplex-handler: # Finally! Websocket handler. + path: /duplex # Registering handlers '/*' find files. + method: GET # Handle only GET requests. + task_processor: main-task-processor # Run it on CPU bound task processor + max-remote-payload: 100000 + fragment-size: 10 + websocket-ping-pong-handler: + path: /ping-pong + method: GET + task_processor: main-task-processor + max-remote-payload: 100000 + fragment-size: 10 + plain-handler: # An ordinary handler, to check that it never accepts a tunnel. + path: /plain + method: GET + task_processor: main-task-processor + + testsuite-support: + + http-client: + http-client-core: + fs-task-processor: main-task-processor + dns-client: + fs-task-processor: fs-task-processor + + tests-control: + method: POST + path: /tests/{action} + skip-unregistered-testpoints: true + task_processor: main-task-processor + testpoint-timeout: 10s + testpoint-url: $mockserver/testpoint + throttling_enabled: false diff --git a/core/functional_tests/websocket_http2/tests/conftest.py b/core/functional_tests/websocket_http2/tests/conftest.py new file mode 100644 index 000000000000..1daa1ba83329 --- /dev/null +++ b/core/functional_tests/websocket_http2/tests/conftest.py @@ -0,0 +1,196 @@ +import contextlib +import socket + +import h2.config +import h2.connection +import h2.events +import pytest +import wsproto.connection +import wsproto.events + +pytest_plugins = ['pytest_userver.plugins.core'] + + +class Rfc8441Error(Exception): + pass + + +class Rfc8441Client: + """A websocket bootstrapped over HTTP/2.0 with the extended CONNECT of RFC 8441. + + Hand-rolled because no mainstream Python websocket client speaks RFC 8441: the + handshake is done with `h2`, and `wsproto` provides plain RFC 6455 framing for + the bytes that then flow inside the DATA frames of the stream. + """ + + # Generous: a sanitized service on a loaded machine is slow, and a real hang still + # fails the test, only later. + def __init__(self, host: str, port: int, timeout: float = 60.0): + self._authority = f'{host}:{port}' + self._sock = socket.create_connection((host, port), timeout=timeout) + self._conn = h2.connection.H2Connection( + config=h2.config.H2Configuration(client_side=True), + ) + self._conn.initiate_connection() + self._flush() + self._ws = wsproto.connection.Connection( + wsproto.connection.ConnectionType.CLIENT, + ) + self._stream_id = None + self._pending = [] + self._got_settings = False + + # --- HTTP/2.0 plumbing --- + + def _flush(self): + self._sock.sendall(self._conn.data_to_send()) + + def _pump(self): + """Reads one batch of server bytes and returns the resulting h2 events.""" + data = self._sock.recv(65535) + if not data: + raise Rfc8441Error('the server closed the connection') + events = self._conn.receive_data(data) + self._flush() + return events + + @property + def enable_connect_protocol(self) -> int: + # The local default is 0, so wait for the server SETTINGS to actually arrive + # instead of reporting "not advertised" before it had a chance to. + while not self._got_settings: + for event in self._pump(): + if isinstance(event, h2.events.RemoteSettingsChanged): + self._got_settings = True + self._remember(event) + return self._conn.remote_settings.enable_connect_protocol + + def request(self, path: str, method: str = 'GET') -> int: + """Starts an ordinary request on its own stream and returns its id.""" + stream_id = self._conn.get_next_available_stream_id() + self._conn.send_headers( + stream_id, + [ + (':method', method), + (':scheme', 'http'), + (':path', path), + (':authority', self._authority), + ], + end_stream=True, + ) + self._flush() + return stream_id + + def status_of(self, stream_id: int) -> str: + status = None + while status is None: + # The whole batch has to be consumed even once the status is known: the + # server may well have put the response headers and the first bytes of the + # tunnelled protocol into one TCP segment. + for event in self._pump(): + if isinstance(event, h2.events.ResponseReceived) and event.stream_id == stream_id: + status = dict(event.headers)[b':status'].decode() + else: + self._remember(event) + return status + + # --- RFC 8441 --- + + def connect(self, path: str, extra_headers=()) -> str: + """Sends the extended CONNECT and returns the response `:status`.""" + assert self._stream_id is None, 'the client drives a single websocket' + self._stream_id = self._conn.get_next_available_stream_id() + self._conn.send_headers( + self._stream_id, + [ + (':method', 'CONNECT'), + (':protocol', 'websocket'), + (':scheme', 'http'), + (':path', path), + (':authority', self._authority), + ('sec-websocket-version', '13'), + *extra_headers, + ], + # The stream stays open for the lifetime of the websocket. + end_stream=False, + ) + self._flush() + + return self.status_of(self._stream_id) + + def _remember(self, event): + if isinstance(event, h2.events.DataReceived) and event.stream_id == self._stream_id: + if event.flow_controlled_length: + self._conn.acknowledge_received_data( + event.flow_controlled_length, + event.stream_id, + ) + self._flush() + if event.data: + self._ws.receive_data(event.data) + self._pending.extend(self._ws.events()) + elif isinstance(event, h2.events.StreamEnded) and event.stream_id == self._stream_id: + # Half-closing the stream is how the transport under the websocket goes away. + self._ws.receive_data(None) + self._pending.extend(self._ws.events()) + elif isinstance(event, h2.events.StreamReset) and event.stream_id == self._stream_id: + raise Rfc8441Error(f'the websocket stream was reset: {event.error_code}') + elif isinstance(event, h2.events.ConnectionTerminated): + raise Rfc8441Error(f'the connection was terminated: {event.error_code}') + + def send(self, event): + self._conn.send_data(self._stream_id, self._ws.send(event)) + self._flush() + + def send_text(self, payload: str): + self.send(wsproto.events.TextMessage(data=payload)) + + def send_bytes(self, payload: bytes): + self.send(wsproto.events.BytesMessage(data=payload)) + + def recv(self): + """Returns one websocket event, reassembling fragmented messages.""" + parts = None + while True: + while self._pending: + event = self._pending.pop(0) + if not isinstance(event, wsproto.events.Message): + return event + parts = event.data if parts is None else parts + event.data + if event.message_finished: + return type(event)(data=parts) + for event in self._pump(): + self._remember(event) + + def recv_message(self): + event = self.recv() + assert isinstance(event, wsproto.events.Message), event + return event.data + + def close(self, code: int = 1000): + self.send(wsproto.events.CloseConnection(code=code)) + return self.recv() + + def disconnect(self): + self._sock.close() + + +@pytest.fixture(name='rfc8441_client') +async def _rfc8441_client(service_client, service_port): + # `service_client` is required so that the daemon is up before we connect: the + # client speaks to the listener directly, bypassing the testsuite plumbing. + clients = [] + + @contextlib.contextmanager + def make_client(): + client = Rfc8441Client('localhost', service_port) + clients.append(client) + try: + yield client + finally: + client.disconnect() + + yield make_client + + for client in clients: + client.disconnect() diff --git a/core/functional_tests/websocket_http2/tests/test_websocket_http2.py b/core/functional_tests/websocket_http2/tests/test_websocket_http2.py new file mode 100644 index 000000000000..bdeb38717feb --- /dev/null +++ b/core/functional_tests/websocket_http2/tests/test_websocket_http2.py @@ -0,0 +1,122 @@ +"""Websockets bootstrapped over HTTP/2.0 with the extended CONNECT of RFC 8441. + +The service is the very same one the HTTP/1.1 websocket test drives; only the +listener speaks HTTP/2.0 with `enable_connect_protocol` on. +""" + +import pytest +import wsproto.events + +from conftest import Rfc8441Error + + +async def test_setting_is_advertised(rfc8441_client): + with rfc8441_client() as client: + assert client.enable_connect_protocol == 1 + + +async def test_echo(rfc8441_client): + with rfc8441_client() as client: + assert client.connect('/chat') == '200' + client.send_text('hello') + assert client.recv_message() == 'hello' + client.send_text('second message') + assert client.recv_message() == 'second message' + + +async def test_echo_bin(rfc8441_client): + with rfc8441_client() as client: + assert client.connect('/chat') == '200' + client.send_bytes(b'\x00\x01\x02\xff') + assert client.recv_message() == b'\x00\x01\x02\xff' + + +async def test_handshake_hook_sees_the_request(rfc8441_client): + with rfc8441_client() as client: + # The handler echoes the Origin back as its first message, which proves + # HandleHandshake() ran and saw the real headers. + assert client.connect('/chat', extra_headers=[('origin', 'localhost')]) == '200' + assert client.recv_message() == 'localhost' + + +async def test_close_handshake(rfc8441_client): + with rfc8441_client() as client: + assert client.connect('/chat') == '200' + client.send_text('hello') + assert client.recv_message() == 'hello' + assert isinstance(client.close(), wsproto.events.CloseConnection) + + +async def test_server_initiated_close(rfc8441_client): + with rfc8441_client() as client: + assert client.connect('/chat') == '200' + client.send_text('close') + closed = client.recv() + assert isinstance(closed, wsproto.events.CloseConnection) + assert closed.code == 1001 + + +async def test_multiplexing_with_a_plain_request(rfc8441_client, service_client): + """The property the HTTP/1.1 transport can never have.""" + with rfc8441_client() as client: + assert client.connect('/chat') == '200' + client.send_text('before') + assert client.recv_message() == 'before' + + # Another stream of the *same* connection, while the websocket is open. + assert client.status_of(client.request('/ping')) == '404' + + client.send_text('after') + assert client.recv_message() == 'after' + + +async def test_two_websockets_on_one_connection(rfc8441_client): + with rfc8441_client() as first, rfc8441_client() as second: + assert first.connect('/chat') == '200' + assert second.connect('/handler-alt') == '200' + + first.send_text('to first') + second.send_text('to second') + assert first.recv_message() == 'to first' + assert second.recv_message() == 'to second' + + +async def test_unknown_path_is_not_upgraded(rfc8441_client): + with rfc8441_client() as client: + assert client.connect('/no-such-handler') == '404' + + +async def test_non_websocket_handler_does_not_answer_2xx(rfc8441_client): + """A 2xx to an extended CONNECT would tell the client the tunnel is up.""" + with rfc8441_client() as client: + assert client.connect('/plain') == '502' + + +async def test_non_websocket_protocol_is_rejected(rfc8441_client): + with rfc8441_client() as client: + client._stream_id = client._conn.get_next_available_stream_id() # noqa: SLF001 + client._conn.send_headers( # noqa: SLF001 + client._stream_id, # noqa: SLF001 + [ + (':method', 'CONNECT'), + (':protocol', 'mqtt'), + (':scheme', 'http'), + (':path', '/chat'), + (':authority', client._authority), # noqa: SLF001 + ], + end_stream=False, + ) + client._flush() # noqa: SLF001 + with pytest.raises(Rfc8441Error): + client.recv() + + +async def test_service_survives_an_abandoned_websocket(rfc8441_client, service_client): + with rfc8441_client() as client: + assert client.connect('/chat') == '200' + client.send_text('hello') + assert client.recv_message() == 'hello' + client.disconnect() + + response = await service_client.get('/ping') + assert response.status in (404, 200) diff --git a/core/include/userver/server/http/http_request.hpp b/core/include/userver/server/http/http_request.hpp index 0343c6fad8d1..ee6aba2a6af2 100644 --- a/core/include/userver/server/http/http_request.hpp +++ b/core/include/userver/server/http/http_request.hpp @@ -258,6 +258,13 @@ class HttpRequest final { /// Get approximate time point of request handling start std::chrono::steady_clock::time_point GetStartTime() const; + /// @brief Whether the request is a websocket bootstrapped over HTTP/2.0 with the + /// extended CONNECT method of RFC 8441. + /// + /// Such a request is routed as a `GET` so that ordinary websocket handlers match + /// it, and it carries neither `Upgrade`/`Connection` nor `Sec-WebSocket-Key`. + bool IsWebsocketExtendedConnect() const; + /// @cond void MarkAsInternalServerError() const; diff --git a/core/include/userver/server/http/http_request_builder.hpp b/core/include/userver/server/http/http_request_builder.hpp index 201df9c8866b..b619a4e89705 100644 --- a/core/include/userver/server/http/http_request_builder.hpp +++ b/core/include/userver/server/http/http_request_builder.hpp @@ -47,6 +47,8 @@ class HttpRequestBuilder final { /// @cond HttpRequestBuilder& SetIsFinal(bool is_final); + HttpRequestBuilder& SetWebsocketExtendedConnect(bool is_websocket_extended_connect); + HttpRequestBuilder& SetFormDataArgs( utils::impl::TransparentMap, utils::StrCaseHash>&& form_data_args ); diff --git a/core/src/server/component.yaml b/core/src/server/component.yaml index dc85f4307afd..75c0f8a33394 100644 --- a/core/src/server/component.yaml +++ b/core/src/server/component.yaml @@ -198,6 +198,11 @@ properties: type: integer description: the initial window size of the server default: 65536 + enable_connect_protocol: + type: boolean + description: set to true to accept websockets bootstrapped over HTTP/2.0 with the extended + CONNECT method of RFC 8441 + default: false shards: type: integer description: how many concurrent tasks harvest data from a single socket; do not set if not sure what it is doing diff --git a/core/src/server/handlers/websocket_handler.cpp b/core/src/server/handlers/websocket_handler.cpp index fd5cb6d2fe08..a42ffe6770ac 100644 --- a/core/src/server/handlers/websocket_handler.cpp +++ b/core/src/server/handlers/websocket_handler.cpp @@ -21,6 +21,25 @@ USERVER_NAMESPACE_BEGIN namespace server::handlers { +namespace { + +/// @brief Checks the `Sec-WebSocket-Key` of the RFC 6455 handshake over HTTP/1.1. +/// The extended CONNECT of RFC 8441 has no counterpart: opening the stream is itself +/// the proof of intent. +const std::string& GetCheckedWebsocketKey(const server::http::HttpRequest& request) { + const std::string& sec_websocket_key = request.GetHeader(USERVER_NAMESPACE::http::headers::kWebsocketKey); + + // We are fine if `secWebsocketKey` is not properly base64-ecoded + static constexpr std::size_t kLengthOfBase64Encoded16Bytes = 24; + if (kLengthOfBase64Encoded16Bytes != sec_websocket_key.size()) { + LOG_WARNING() << "Empty or invalid Websocket Key"; + throw server::handlers::ClientError(); + } + return sec_websocket_key; +} + +} // namespace + WebsocketHandlerBase::WebsocketHandlerBase( const components::ComponentConfig& config, const components::ComponentContext& context @@ -34,6 +53,12 @@ WebsocketHandlerBase::WebsocketHandlerBase( } bool WebsocketHandlerBase::IsWebsocketRequest(const server::http::HttpRequest& request) const { + if (request.IsWebsocketExtendedConnect()) { + // RFC 8441 carries no Upgrade/Connection headers, and the extended CONNECT form + // has already been validated while parsing the HTTP/2.0 stream. + return true; + } + constexpr auto kIcaseEq = utils::StrIcaseEqual(); return request.GetMethod() == server::http::HttpMethod::kGet && @@ -45,14 +70,10 @@ void WebsocketHandlerBase::HandleWebsocketRequest( server::http::HttpRequest& request, server::request::RequestContext& context ) const { - const std::string& sec_websocket_key = request.GetHeader(USERVER_NAMESPACE::http::headers::kWebsocketKey); - - // We are fine if `secWebsocketKey` is not properly base64-ecoded - static constexpr std::size_t kLengthOfBase64Encoded16Bytes = 24; - if (kLengthOfBase64Encoded16Bytes != sec_websocket_key.size()) { - LOG_WARNING() << "Empty or invalid Websocket Key"; - throw server::handlers::ClientError(); - } + const bool is_extended_connect = request.IsWebsocketExtendedConnect(); + // Checked before anything else, as an invalid key is a bad request no matter what + // websocket version was asked for. + const std::string sec_websocket_key = is_extended_connect ? std::string{} : GetCheckedWebsocketKey(request); auto& response = request.GetHttpResponse(); @@ -68,13 +89,19 @@ void WebsocketHandlerBase::HandleWebsocketRequest( return; } - response.SetStatus(server::http::HttpStatus::kSwitchingProtocols); - response.SetHeader(USERVER_NAMESPACE::http::headers::kConnection, "Upgrade"); - response.SetHeader(USERVER_NAMESPACE::http::headers::kUpgrade, "websocket"); - response.SetHeader( - USERVER_NAMESPACE::http::headers::kWebsocketAccept, - websocket::impl::WebsocketSecAnswer(sec_websocket_key) - ); + if (is_extended_connect) { + // RFC 8441: the stream is accepted with a plain 200 and stays open. There is no + // protocol switch to announce, as the connection keeps speaking HTTP/2.0. + response.SetStatus(server::http::HttpStatus::kOk); + } else { + response.SetStatus(server::http::HttpStatus::kSwitchingProtocols); + response.SetHeader(USERVER_NAMESPACE::http::headers::kConnection, "Upgrade"); + response.SetHeader(USERVER_NAMESPACE::http::headers::kUpgrade, "websocket"); + response.SetHeader( + USERVER_NAMESPACE::http::headers::kWebsocketAccept, + websocket::impl::WebsocketSecAnswer(sec_websocket_key) + ); + } request.SetUpgradeWebsocket([context = std::make_shared(std::move(context)), this](std::unique_ptr socket, engine::io::Sockaddr&& peer_name) { diff --git a/core/src/server/http/http2_session.cpp b/core/src/server/http/http2_session.cpp index cde008b8c5d7..6d2cbb8059a2 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))}; @@ -78,11 +84,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"); @@ -96,17 +106,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: { @@ -170,8 +196,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); @@ -194,7 +229,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); @@ -225,8 +267,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; } @@ -310,9 +357,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)}; } @@ -357,6 +407,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); @@ -395,7 +501,7 @@ bool Http2Session::PopStreamingEventNoblock(impl::Http2StreamEvent& event) { void Http2Session::ApplyStreamingEvent(impl::Http2StreamEvent&& event) { UASSERT(event.stream_id != -1); - auto* stream = static_cast(nghttp2_session_get_stream_user_data(session_.get(), event.stream_id)); + 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. diff --git a/core/src/server/http/http2_session.hpp b/core/src/server/http/http2_session.hpp index 83a5dd03da1a..0b2721d6eeb9 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(); @@ -118,11 +125,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_; diff --git a/core/src/server/http/http2_session_test.cpp b/core/src/server/http/http2_session_test.cpp index 353c1cb882d7..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 @@ -12,6 +13,7 @@ #include #include #include +#include #include #include @@ -359,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"; 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 3320bd0f75d1..fee8e98ad69f 100644 --- a/core/src/server/http/http2_writer.cpp +++ b/core/src/server/http/http2_writer.cpp @@ -119,18 +119,23 @@ class Http2ResponseWriter final { buffered_h1_stream = true; } - 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() && !buffered_h1_stream && 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}; @@ -156,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 @@ -164,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 33f52033bad9..46ffed23b34a 100644 --- a/core/src/server/http/http_request.cpp +++ b/core/src/server/http/http_request.cpp @@ -291,6 +291,8 @@ HttpResponse& HttpRequest::GetHttpResponse() const noexcept { return pimpl_->res std::chrono::steady_clock::time_point HttpRequest::GetStartTime() const { return pimpl_->start_time; } +bool HttpRequest::IsWebsocketExtendedConnect() const { return pimpl_->is_websocket_extended_connect; } + bool HttpRequest::IsUpgradeWebsocket() const { return static_cast(pimpl_->upgrade_websocket_cb); } void HttpRequest::SetUpgradeWebsocket(UpgradeCallback cb) const { pimpl_->upgrade_websocket_cb = std::move(cb); } diff --git a/core/src/server/http/http_request_builder.cpp b/core/src/server/http/http_request_builder.cpp index a399bfa17640..f742809d8a14 100644 --- a/core/src/server/http/http_request_builder.cpp +++ b/core/src/server/http/http_request_builder.cpp @@ -86,6 +86,11 @@ HttpRequestBuilder& HttpRequestBuilder::SetIsFinal(bool is_final) { return *this; } +HttpRequestBuilder& HttpRequestBuilder::SetWebsocketExtendedConnect(bool is_websocket_extended_connect) { + request_->pimpl_->is_websocket_extended_connect = is_websocket_extended_connect; + return *this; +} + HttpRequestBuilder& HttpRequestBuilder::SetFormDataArgs( utils::impl::TransparentMap, utils::StrCaseHash>&& form_data_args ) { diff --git a/core/src/server/http/http_request_constructor.cpp b/core/src/server/http/http_request_constructor.cpp index 8905a015f6f4..e50d940db29a 100644 --- a/core/src/server/http/http_request_constructor.cpp +++ b/core/src/server/http/http_request_constructor.cpp @@ -183,6 +183,10 @@ void HttpRequestConstructor::SetStreamProducer(impl::Http2StreamEventProducer&& builder_.SetStreamProducer(std::move(producer)); } +void HttpRequestConstructor::SetWebsocketExtendedConnect(bool is_websocket_extended_connect) { + builder_.SetWebsocketExtendedConnect(is_websocket_extended_connect); +} + std::shared_ptr HttpRequestConstructor::Finalize() { FinalizeImpl(); diff --git a/core/src/server/http/http_request_constructor.hpp b/core/src/server/http/http_request_constructor.hpp index abf0981584e4..7b268d49eab4 100644 --- a/core/src/server/http/http_request_constructor.hpp +++ b/core/src/server/http/http_request_constructor.hpp @@ -61,6 +61,7 @@ class HttpRequestConstructor final { // HTTP/2.0 only: void SetStreamProducer(impl::Http2StreamEventProducer&& producer); void SetResponseStreamId(std::int32_t stream_id); + void SetWebsocketExtendedConnect(bool is_websocket_extended_connect); std::shared_ptr Finalize(); diff --git a/core/src/server/http/http_request_impl.hpp b/core/src/server/http/http_request_impl.hpp index a6f48cff5357..a061ad469e4d 100644 --- a/core/src/server/http/http_request_impl.hpp +++ b/core/src/server/http/http_request_impl.hpp @@ -47,6 +47,9 @@ struct HttpRequest::Impl { HeadersMap headers; CookiesMap cookies; bool is_final{false}; + // The request arrived as an RFC 8441 extended CONNECT with `:protocol: websocket`, + // and is routed as a GET so that ordinary websocket handlers match it. + bool is_websocket_extended_connect{false}; #ifndef NDEBUG mutable bool args_referenced{false}; #endif diff --git a/core/src/server/net/connection_config.cpp b/core/src/server/net/connection_config.cpp index 741c6143d6f7..1db388b5c6aa 100644 --- a/core/src/server/net/connection_config.cpp +++ b/core/src/server/net/connection_config.cpp @@ -11,6 +11,7 @@ Http2SessionConfig Parse(const yaml_config::YamlConfig& value, formats::parse::T conf.max_concurrent_streams = value["max_concurrent_streams"].As(conf.max_concurrent_streams); conf.max_frame_size = value["max_frame_size"].As(conf.max_frame_size); conf.initial_window_size = value["initial_window_size"].As(conf.initial_window_size); + conf.enable_connect_protocol = value["enable_connect_protocol"].As(conf.enable_connect_protocol); return conf; } diff --git a/core/src/server/net/connection_config.hpp b/core/src/server/net/connection_config.hpp index 3d0b856b020f..de775f182485 100644 --- a/core/src/server/net/connection_config.hpp +++ b/core/src/server/net/connection_config.hpp @@ -22,6 +22,9 @@ struct Http2SessionConfig final { std::uint32_t max_concurrent_streams = 100; std::uint32_t max_frame_size = 1 << 14; std::uint32_t initial_window_size = 1 << 16; + // Advertises SETTINGS_ENABLE_CONNECT_PROTOCOL, which allows clients to bootstrap + // websockets over HTTP/2.0 with the extended CONNECT method of RFC 8441. + bool enable_connect_protocol = false; }; struct ConnectionConfig { diff --git a/core/src/server/net/http2_connection.cpp b/core/src/server/net/http2_connection.cpp index f6ec0f8b2513..f3f6fde85138 100644 --- a/core/src/server/net/http2_connection.cpp +++ b/core/src/server/net/http2_connection.cpp @@ -134,7 +134,7 @@ void Http2Connection::ListenForRequests() { wait_any.Append(kStreamingId, parser_->GetStreamingEvent()); break; case WakeupKind::kTaskComputedResponse: - OnRequestTaskFinished(*ready_id); + OnRequestTaskFinished(*ready_id, wait_any); break; } @@ -171,8 +171,16 @@ Http2Connection::RequestTaskContext Http2Connection::StartRequestTask(std::share return {.task = std::move(task), .request = std::move(request_ptr)}; } -void Http2Connection::OnRequestTaskFinished(std::uint64_t event_id) noexcept { - auto& request = *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 @@ -190,7 +198,42 @@ void Http2Connection::OnRequestTaskFinished(std::uint64_t event_id) noexcept { } 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() { diff --git a/core/src/server/net/http2_connection.hpp b/core/src/server/net/http2_connection.hpp index 99a0eccf0107..8215e09b29af 100644 --- a/core/src/server/net/http2_connection.hpp +++ b/core/src/server/net/http2_connection.hpp @@ -58,6 +58,9 @@ 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()`). @@ -71,9 +74,11 @@ class Http2Connection final : public ConnectionBase { 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; 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