From 0a0f845786316ed4dd9fdd785f5f694dee74bba5 Mon Sep 17 00:00:00 2001 From: "J. Nick Koston" Date: Sun, 7 Jun 2026 00:39:29 -0500 Subject: [PATCH] [PR #12828/13b635d7 backport][3.14] Bounded unread compressed drain (#12845) CVE: CVE-2026-54278 Upstream-Status: Backport [https://github.com/aio-libs/aiohttp/commit/4f7480e474cccc6a8cc2c92ad3f17a31dedf8232] (cherry picked from commit 4f7480e474cccc6a8cc2c92ad3f17a31dedf8232) Signed-off-by: Darsh Kelaiya --- CHANGES/12828.bugfix.rst | 1 + aiohttp/streams.py | 17 +++++++--- tests/test_streams.py | 28 ++++++++++++++++ tests/test_web_functional.py | 63 ++++++++++++++++++++++++++++++++++++ 4 files changed, 104 insertions(+), 5 deletions(-) create mode 100644 CHANGES/12828.bugfix.rst diff --git a/CHANGES/12828.bugfix.rst b/CHANGES/12828.bugfix.rst new file mode 100644 index 000000000..9893577a5 --- /dev/null +++ b/CHANGES/12828.bugfix.rst @@ -0,0 +1 @@ +Fixed :meth:`~aiohttp.StreamReader.readany` and :meth:`~aiohttp.StreamReader.read_nowait` joining data fed back into the buffer during the call (when draining below the low water mark resumes reading) into a single unbounded :class:`bytes`; a call now returns only the chunks that were buffered when it started, keeping the drain of an unread auto-decompressed request body bounded by the read buffer -- by :user:`bdraco`. diff --git a/aiohttp/streams.py b/aiohttp/streams.py index 921827eb3..b52c20742 100644 --- a/aiohttp/streams.py +++ b/aiohttp/streams.py @@ -572,14 +572,21 @@ class StreamReader(AsyncStreamReaderMixin): """Read not more than n bytes, or whole buffer if n == -1""" self._timer.assert_timeout() - chunks = [] + if n == -1: + # Drain only chunks present now; _read_nowait_chunk() can + # re-entrantly resume_reading() and refill the buffer. + count = len(self._buffer) + if count == 1: + return self._read_nowait_chunk(-1) + return b"".join([self._read_nowait_chunk(-1) for _ in range(count)]) + + chunks: list[bytes] = [] while self._buffer: chunk = self._read_nowait_chunk(n) chunks.append(chunk) - if n != -1: - n -= len(chunk) - if n == 0: - break + n -= len(chunk) + if n == 0: + break return b"".join(chunks) if chunks else b"" diff --git a/tests/test_streams.py b/tests/test_streams.py index 93686746e..8b4fcf322 100644 --- a/tests/test_streams.py +++ b/tests/test_streams.py @@ -1723,3 +1723,31 @@ async def test_stream_reader_small_limit_resumes_reading( protocol.resume_reading.assert_called() assert protocol._reading_paused is False + + +async def test_readany_does_not_drain_reentrant_refill( + protocol: mock.Mock, +) -> None: + """A single readany() must not reassemble data fed re-entrantly. + + Draining below the low water mark resumes reading, which can synchronously + refill the buffer (e.g. decompressing another chunk). Joining that refill in + one call would reassemble an unbounded body. + """ + loop = asyncio.get_running_loop() + stream = streams.StreamReader(protocol, limit=4, loop=loop) + + refills = [b"second", b"third"] + + def resume_reading() -> None: + if refills: + stream.feed_data(refills.pop(0)) + + protocol.resume_reading.side_effect = resume_reading + + stream.feed_data(b"first") + + # Popping "first" refills "second", but this readany() returns only "first". + assert await stream.readany() == b"first" + assert await stream.readany() == b"second" + assert await stream.readany() == b"third" diff --git a/tests/test_web_functional.py b/tests/test_web_functional.py index fe9cce27b..e2fd40432 100644 --- a/tests/test_web_functional.py +++ b/tests/test_web_functional.py @@ -5,6 +5,8 @@ import pathlib import socket import sys from typing import Any, Dict, Generator, NoReturn, Optional, Tuple +import zlib +from contextlib import suppress from unittest import mock import pytest @@ -23,7 +25,9 @@ from aiohttp import ( ) from aiohttp.compression_utils import ZLibBackend, ZLibCompressObjProtocol from aiohttp.hdrs import CONTENT_LENGTH, CONTENT_TYPE, TRANSFER_ENCODING +from aiohttp.helpers import DEFAULT_CHUNK_SIZE from aiohttp.pytest_plugin import AiohttpClient, AiohttpServer +from aiohttp.streams import StreamReader from aiohttp.typedefs import Handler from aiohttp.web_protocol import RequestHandler @@ -1683,6 +1687,65 @@ async def test_response_prepared_with_clone(aiohttp_client) -> None: await resp.release() +@pytest.mark.parametrize("decompressed_size", [4 * 1024 * 1024, 32 * 1024 * 1024]) +async def test_unread_compressed_body_drain_is_bounded( + aiohttp_server: AiohttpServer, + monkeypatch: pytest.MonkeyPatch, + decompressed_size: int, +) -> None: + """Draining an unread compressed body stays bounded by the read buffer. + + A handler that rejects before reading still drains the payload during + lingering close; a small compressed body must not force a large transient + allocation (a deflate-bomb style DoS). + """ + drain_reads: list[int] = [] + drained = asyncio.Event() + readany = StreamReader.readany + + async def record_readany(self: StreamReader) -> bytes: + data = await readany(self) + assert data + drain_reads.append(len(data)) + drained.set() + return data + + monkeypatch.setattr(StreamReader, "readany", record_readany) + + async def handler(request: web.Request) -> web.Response: + return web.Response(status=401) + + app = web.Application(client_max_size=1024) + app.router.add_post("/", handler) + server = await aiohttp_server(app) + + body = zlib.compress(b"a" * decompressed_size) + assert len(body) < decompressed_size + head = ( + b"POST / HTTP/1.1\r\n" + b"Host: localhost\r\n" + b"Content-Encoding: deflate\r\n" + b"Content-Length: %d\r\n" + b"Connection: keep-alive\r\n\r\n" + ) % len(body) + + reader, writer = await asyncio.open_connection(server.host, server.port) + try: + writer.write(head + body) + await writer.drain() + status_line = await asyncio.wait_for(reader.readline(), 5) + assert status_line.startswith(b"HTTP/1.1 401 ") + await asyncio.wait_for(drained.wait(), 5) + finally: + writer.close() + with suppress(ConnectionResetError, BrokenPipeError): + await writer.wait_closed() + + # Bounded by the buffer, not the decompressed size. + assert max(drain_reads) <= 3 * DEFAULT_CHUNK_SIZE + assert max(drain_reads) < decompressed_size + + async def test_app_max_client_size(aiohttp_client) -> None: async def handler(request): await request.post()