aiohttp: Unread Compressed Request Bodies Bypass client_max_size During Cleanup

@@ -560,14 +560,21 @@ def _read_nowait(self, n: int) -> bytes:
         """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""
 

@@ -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`.

@@ -1723,3 +1723,31 @@ def resume_reading() -> None:
 
     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"

@@ -4,7 +4,9 @@
 import pathlib
 import socket
 import sys
+import zlib
 from collections.abc import Generator
+from contextlib import suppress
 from typing import Any, NoReturn
 from unittest import mock
 
@@ -24,7 +26,9 @@
 )
 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
 
@@ -1711,6 +1715,65 @@ async def handler(request):
     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()