From d460210c19b97fde184843d17487e336698d59da Mon Sep 17 00:00:00 2001 From: "patchback[bot]" <45432694+patchback[bot]@users.noreply.github.com> Date: Sun, 9 Aug 2026 17:17:34 +0100 Subject: [PATCH] [PR #13356/72eaa429 backport][3.14] Stop handing back pipelined requests the parser buffered (#13360) **This is a backport of PR #13356 as merged into master (72eaa429cf89b1e20212590f3e704444239569fb).** --------- Co-authored-by: Rodrigo Nogueira CVE: CVE-2026-54273 Upstream-Status: Backport [https://github.com/aio-libs/aiohttp/commit/47babd8c23ca79e2bae6cc8dcb5752b61e369fc7] Backport Changes: - Adapted the required 8943d343 replay behavior to aiohttp 3.9.5 `_request_parser` and `_upgrade` fields. Replayed messages are retained and queued, and the returned parser tail is preserved. - Applied the 47babd8c queue-pause check after replay and retained its pure-Python parser tail fix, change note, parser test, and declined-upgrade functional regression test. (cherry picked from commit 47babd8c23ca79e2bae6cc8dcb5752b61e369fc7) Signed-off-by: Darsh Kelaiya --- CHANGES/13356.bugfix.rst | 4 +++ aiohttp/http_parser.py | 2 ++ aiohttp/web_protocol.py | 18 +++++++++++-- tests/test_http_parser.py | 18 +++++++++++++ tests/test_web_functional.py | 51 ++++++++++++++++++++++++++++++++++++ 5 files changed, 91 insertions(+), 2 deletions(-) create mode 100644 CHANGES/13356.bugfix.rst diff --git a/CHANGES/13356.bugfix.rst b/CHANGES/13356.bugfix.rst new file mode 100644 index 0000000..a961723 --- /dev/null +++ b/CHANGES/13356.bugfix.rst @@ -0,0 +1,4 @@ +Fixed requests pipelined behind a request whose upgrade the handler declined +going unanswered once there were more of them than the per-connection queue +holds. With the pure-Python parser the same requests were also served more +than once -- by :user:`rodrigobnogueira`. diff --git a/aiohttp/http_parser.py b/aiohttp/http_parser.py index 6cf49c7..37a0dff 100644 --- a/aiohttp/http_parser.py +++ b/aiohttp/http_parser.py @@ -357,6 +357,8 @@ class HttpParser(abc.ABC, Generic[_MsgT]): # any preceding body is consumed before the next request # line. Resumes via feed_data(b"") when the queue drains. self._tail = data[start_pos:] + # The remainder now lives in self._tail only. Don't return it. + data = EMPTY break pos = data.find(SEP, start_pos) # consume \r\n diff --git a/aiohttp/web_protocol.py b/aiohttp/web_protocol.py index baa0290..39a1b68 100644 --- a/aiohttp/web_protocol.py +++ b/aiohttp/web_protocol.py @@ -687,8 +687,22 @@ class RequestHandler(BaseProtocol): self._request_parser.set_upgraded(False) self._upgrade = False if self._message_tail: - self._request_parser.feed_data(self._message_tail) - self._message_tail = b"" + messages, _upgraded, tail = self._request_parser.feed_data( + self._message_tail + ) + self._message_tail = tail + for msg, payload in messages: + self._request_count += 1 + self._messages.append((msg, payload)) + # Pause the transport, like in data_received(). + if ( + not self._msg_queue_paused + and len(self._messages) >= self._max_msg_queue_size + ): + self._pause_msg_queue_reading() + # This shouldn't be possible. If a future refactor results in this + # failing, then the code may need to be updated to set the waiter. + assert self._waiter is None try: prepare_meth = resp.prepare except AttributeError: diff --git a/tests/test_http_parser.py b/tests/test_http_parser.py index 213761c..b29518a 100644 --- a/tests/test_http_parser.py +++ b/tests/test_http_parser.py @@ -142,6 +142,24 @@ def test_max_msg_queue_size_caps_emitted_messages( assert not upgraded +async def test_max_msg_queue_size_keeps_tail_to_itself( + request_cls: type[HttpRequestParser], + protocol: BaseProtocol, +) -> None: + """The remainder is buffered for the next feed, so it must not be returned. + + Handing it back as well gives the caller a second copy of bytes the parser + is already holding, and both copies get parsed. + """ + loop = asyncio.get_running_loop() + parser = _build_request_parser(request_cls, protocol, loop, 4) + + messages, _upgraded, tail = parser.feed_data(_PIPELINED_GET * 10) + + assert len(messages) == 4 + assert tail == b"" + + def test_max_msg_queue_size_resumes_after_consume( request_cls: type[HttpRequestParser], protocol: BaseProtocol, diff --git a/tests/test_web_functional.py b/tests/test_web_functional.py index 533b132..a0e3a8e 100644 --- a/tests/test_web_functional.py +++ b/tests/test_web_functional.py @@ -1751,6 +1751,57 @@ async def test_http1_pipelined_queue_resumes_after_drain( assert len(handled) == pipelined_requests + 1 +async def test_http1_pipelined_behind_declined_upgrade_served_once( + aiohttp_server: AiohttpServer, +) -> None: + """Requests pipelined behind a declined upgrade are each served once. + + The bytes following an upgrade request are buffered whole, then re-fed once + the handler answers it normally. More of them than the queue holds must + still be served, and none of them twice. + """ + pipelined_requests = MAX_MSG_QUEUE_SIZE + 8 + handled: list[str] = [] + all_handled = asyncio.Event() + + async def handler(request: web.Request) -> web.Response: + handled.append(request.path) + if len(handled) == pipelined_requests + 1: + all_handled.set() + return web.Response() + + app = web.Application() + app.router.add_get("/{tail:.*}", handler) + server = await aiohttp_server(app) + + def raw_get(path: str) -> bytes: + return ( + f"GET {path} HTTP/1.1\r\nHost: localhost\r\n" + "Connection: keep-alive\r\n\r\n" + ).encode("ascii") + + # An upgrade the handler answers normally, then the pipeline, in one write. + upgrade = ( + b"GET /upgrade HTTP/1.1\r\nHost: localhost\r\n" + b"Connection: Upgrade\r\nUpgrade: websocket\r\n\r\n" + ) + + reader, writer = await asyncio.open_connection(server.host, server.port) + try: + writer.write( + upgrade + b"".join(raw_get(f"/r{i}") for i in range(pipelined_requests)) + ) + await writer.drain() + await asyncio.wait_for(all_handled.wait(), 10) + finally: + writer.close() + with suppress(ConnectionResetError, BrokenPipeError): + await writer.wait_closed() + + assert len(handled) == pipelined_requests + 1 + assert len(set(handled)) == len(handled) + + @pytest.mark.parametrize("decompressed_size", [4 * 1024 * 1024, 32 * 1024 * 1024]) async def test_unread_compressed_body_drain_is_bounded( aiohttp_server: AiohttpServer, -- 2.35.6