[pull] master from aio-libs:master

CHANGES/13655.bugfix.rst

@@ -0,0 +1,8 @@
+Fixed unbounded memory growth on a client WebSocket connection. A frame
+protocol error detaches the reader but leaves the connection upgraded, so a
+peer could stream unlimited data into an internal buffer when the application
+never called :meth:`~aiohttp.ClientWebSocketResponse.receive`; that data is
+now discarded, since nothing can parse it. Data arriving before the reader is
+installed is bounded by ``read_bufsize``, which now applies to this buffer as
+well as to :attr:`~aiohttp.ClientResponse.content`
+-- by :user:`bdraco`.

CHANGES/13743.bugfix.rst

@@ -0,0 +1 @@
+13655.bugfix.rst

THREAT_MODEL.md

@@ -538,6 +538,7 @@ client-side, the writer adds masks to outgoing frames.
 | 3.13 | Writer-side: large outbound message as single frame | D | Writer does not auto-fragment; a single `send_str(big_blob)` becomes one frame. Memory pressure on the local side and on intermediaries. | Low |
 | 3.14 | Mask-on-send keys (Cython vs Python parity) | T | Divergence between `mask.pyx` and `helpers.py` `websocket_mask` would silently break receivers (one peer XORs with a different key than the other expects). | Low |
 | 3.15 | Reader Cython vs pure-Python parity | T | Divergence between the two reader backends could let one silently accept a frame the other rejects, weakening protocol enforcement asymmetrically. | Low |
+| 3.16 | Post-error buffering on an upgraded connection | D | `WebSocketReader.feed_data` reports EOF only for a protocol error, and the connection stays upgraded with the reader detached. Were every later byte still buffered with nothing left to drain it, a peer that kept streaming after a deliberate frame error would exhaust memory on a client that never reads. | Medium |
 
 **Mitigations.**
 
@@ -556,6 +557,7 @@ client-side, the writer adds masks to outgoing frames.
 | 3.13 | Writer single-frame size | None — caller-controlled. | **User**: chunk very large outbound payloads (beyond a few MiB) via fragmented messages; a single `send_*` becomes one frame and can pressure intermediaries. |
 | 3.14 | Cython vs pure-Python mask parity | Both implement XOR on the same key cycling; behaviour identical. | Add a parameterised test that runs the mask helper against both backends side-by-side (see [§6.1](#61-highest-leverage-recommendations) #3). |
 | 3.15 | Reader backend parity | `tests/test_websocket_parser.py` imports the single `WebSocketReader` symbol (whichever backend won the import), so each CI run only exercises one. | Parameterise like `tests/test_http_parser.py` does — explicitly import `WebSocketReaderPython` and `WebSocketReaderCython` (when available) and fixture-parametrise over both (see [§6.1](#61-highest-leverage-recommendations) #3). |
+| 3.16 | Post-error buffering bound | After a reader EOF, which only ever means a protocol error, `client_proto.py:data_received` discards what follows; the server closes instead (`web_protocol.py:data_received`). Bytes buffered *before* `set_parser()` are real frames the reader is entitled to, so they are bounded by pausing the transport at `read_bufsize` (default 256 KiB), resumed where the buffer is drained. An upgraded response that never installs a reader keeps that pause, and a paused transport is not told the peer hung up, so it is reclaimed when the response is closed or collected rather than when the peer goes away. | Discarded rather than paused or closed on; see the recap below for what each alternative cost. **User**: set `heartbeat` on long-lived sessions; a peer can keep the client reading and dropping bytes, and `heartbeat` is what reaps such a connection. |
 
 **Past advisories / hardening (recap).**
 
@@ -597,6 +599,18 @@ client-side, the writer adds masks to outgoing frames.
   them once when the frame completes; if a frame arrives in more than
   `max(1024, max_msg_size // 256)` reads, the pending reads are folded into
   a single `bytearray` and cleared.
+- **Issue #13655** — a protocol error detaches the reader but leaves the
+  connection upgraded, so a peer could stream unbounded data into
+  `ResponseHandler._tail`: 32 MiB pushed at a client that never called
+  `receive()` produced 32 MiB of `_tail`. Fixed by discarding what arrives
+  after the reader's EOF (threat 3.16). Two alternatives were measured and
+  rejected: pausing hides the peer's FIN, so sockets accumulate instead of
+  memory; closing makes the transport unwritable, so a message queued before
+  the error cannot be answered, which broke the Autobahn client runner. The
+  pre-parser buffer is bounded by backpressure instead of discarded, since
+  those are frames the reader will want; it looked self-limiting, but the
+  window opens inside `resp.start()` and a `TraceConfig.on_request_end` doing
+  I/O held it open for 32 MiB.
 
 ---
 

aiohttp/base_protocol.py

@@ -24,6 +24,7 @@ class BaseProtocol(asyncio.Protocol):
         "_drain_waiter",
         "_connection_lost",
         "_reading_paused",
+        "_buffer_paused",
         "_upgraded",
         "transport",
     )
@@ -35,6 +36,7 @@ def __init__(
         self._paused = False
         self._drain_waiter: asyncio.Future[None] | None = None
         self._reading_paused = False
+        self._buffer_paused = False
         self._parser = parser
         self._upgraded = False
 
@@ -69,6 +71,25 @@ def pause_reading(self) -> None:
         if not self._upgraded:
             assert self._parser is not None
             self._parser.pause_reading()
+        self._pause_transport_reading()
+
+    def _pause_reading_for_buffer(self) -> None:
+        """Hold the transport for a buffer this protocol cannot drain yet.
+
+        One flag covers every buffer a protocol owns; the server already
+        multiplexes its message queue and its tail through it. Re-check the
+        conditions on resume rather than adding another flag.
+        """
+        self._buffer_paused = True
+        self._pause_transport_reading()
+
+    def _resume_reading_for_buffer(self) -> None:
+        """Release that hold, unless flow control is still holding it too."""
+        self._buffer_paused = False
+        if not self._reading_paused:
+            self._resume_transport_reading()
+
+    def _pause_transport_reading(self) -> None:
         if self.transport is not None:
             try:
                 self.transport.pause_reading()
@@ -77,9 +98,14 @@ def pause_reading(self) -> None:
                 # ignored (see PAUSE_RESUME_READING_ERRORS; do not use suppress).
                 pass
 
-    def _reading_paused_for_msg_queue(self) -> bool:
-        """Keep the transport paused for protocol-specific reasons (overridden)."""
-        return False
+    def _resume_transport_reading(self) -> None:
+        if self.transport is not None:
+            try:
+                self.transport.resume_reading()
+            except PAUSE_RESUME_READING_ERRORS:
+                # Transport lacks flow control; nothing to resume. Intentionally
+                # ignored (see PAUSE_RESUME_READING_ERRORS; do not use suppress).
+                pass
 
     def resume_reading(self, resume_parser: bool = True) -> None:
         self._reading_paused = False
@@ -90,18 +116,8 @@ def resume_reading(self, resume_parser: bool = True) -> None:
 
         # Reading may have been paused again in the above call if there was a lot of
         # compressed data still pending.
-        if (
-            not self._reading_paused
-            and not self._reading_paused_for_msg_queue()
-            and self.transport is not None
-        ):
-            try:
-                self.transport.resume_reading()
-            except PAUSE_RESUME_READING_ERRORS:
-                # Transport lacks flow control; nothing to resume. Intentionally
-                # ignored (see PAUSE_RESUME_READING_ERRORS; do not use suppress).
-                pass
-            self._reading_paused = False
+        if not self._reading_paused and not self._buffer_paused:
+            self._resume_transport_reading()
 
     def connection_made(self, transport: asyncio.BaseTransport) -> None:
         tr = cast(asyncio.Transport, transport)

aiohttp/client_proto.py

@@ -45,6 +45,8 @@ def __init__(self, loop: asyncio.AbstractEventLoop) -> None:
 
         self._timer = None
         self._tail = b""
+        self._payload_parser_failed = False
+        self._read_bufsize = DEFAULT_CHUNK_SIZE
 
         self._read_timeout: float | None = None
         self._read_timeout_handle: asyncio.TimerHandle | None = None
@@ -183,7 +185,9 @@ def connection_lost(self, exc: BaseException | None) -> None:
         self._parser = None
         self._payload = None
         self._payload_parser = None
+        self._payload_parser_failed = False
         self._reading_paused = False
+        self._buffer_paused = False
 
         super().connection_lost(reraised_exc)
 
@@ -198,9 +202,24 @@ def pause_reading(self) -> None:
     def resume_reading(self, resume_parser: bool = True) -> None:
         was_paused = self._reading_paused
         super().resume_reading(resume_parser)
-        if was_paused:
+        # sock_read measures the peer, so it stays stopped while the tail
+        # bound is still holding the transport.
+        if was_paused and not self._buffer_paused:
             self._reschedule_timeout()
 
+    def _drain_tail(self) -> None:
+        if self._tail:
+            data, self._tail = self._tail, b""
+            self.data_received(data)
+        # Tested empty-first so a read_bufsize of 0 cannot wedge the connection.
+        still_bound = bool(self._tail) and len(self._tail) >= self._read_bufsize
+        if self._buffer_paused and not still_bound:
+            self._resume_reading_for_buffer()
+        if self._reading_paused or self._buffer_paused:
+            # The drain restarted sock_read through data_received(); anything
+            # still holding the transport means it stays stopped.
+            self._drop_timeout()
+
     def set_exception(
         self,
         exc: type[BaseException] | BaseException,
@@ -222,9 +241,7 @@ def set_parser(
 
         self._drop_timeout()
 
-        if self._tail:
-            data, self._tail = self._tail, b""
-            self.data_received(data)
+        self._drain_tail()
 
     def set_response_params(
         self,
@@ -242,6 +259,7 @@ def set_response_params(
     ) -> None:
         self._skip_payload = skip_payload
 
+        self._read_bufsize = read_bufsize
         self._read_timeout = read_timeout
 
         self._timeout_ceil_threshold = timeout_ceil_threshold
@@ -260,9 +278,7 @@ def set_response_params(
             max_headers=max_headers,
         )
 
-        if self._tail:
-            data, self._tail = self._tail, b""
-            self.data_received(data)
+        self._drain_tail()
 
     def _drop_timeout(self) -> None:
         if self._read_timeout_handle is not None:
@@ -299,6 +315,12 @@ def _on_read_timeout(self) -> None:
             set_exception(self._payload, exc)
 
     def data_received(self, data: bytes) -> None:
+        if self._payload_parser_failed:
+            # Dropped, not closed on, so queued messages stay answerable and
+            # the peer's FIN still arrives. Discarded bytes are not progress,
+            # so they must not hold off sock_read either.
+            return
+
         # If no data, then we are resuming decompression. We haven't received
         # data from the socket, so we can avoid the reschedule overhead.
         if data:
@@ -308,18 +330,29 @@ def data_received(self, data: bytes) -> None:
         if self._payload_parser is not None:
             if self._data_received_cb is not None:
                 self._data_received_cb()
-            eof, tail = self._payload_parser.feed_data(data)
-            if eof:
+            # WebSocketReader signals EOF only for a protocol error, never
+            # for a clean close, and has already put the error on the queue.
+            protocol_error, _ = self._payload_parser.feed_data(data)
+            if protocol_error:
                 self._payload = None
                 self._payload_parser = None
-
-                if tail:
-                    self.data_received(tail)
+                self._payload_parser_failed = True
             return
 
         if self._upgraded or self._parser is None:
             # i.e. websocket connection, websocket parser is not set yet
             self._tail += data
+            # Nothing drains this until set_parser() runs, and an await
+            # after the 101 can hold that off; stop reading instead.
+            # Tested non-empty so a read_bufsize of 0 cannot pause on nothing.
+            if (
+                not self._buffer_paused
+                and self._tail
+                and len(self._tail) >= self._read_bufsize
+            ):
+                self._pause_reading_for_buffer()
+                # sock_read measures the peer; this pause is ours.
+                self._drop_timeout()
             return
 
         # parse http messages

aiohttp/web_protocol.py

@@ -16,7 +16,7 @@
 from propcache import under_cached_property
 
 from .abc import AbstractAccessLogger, AbstractAsyncAccessLogger, AbstractStreamWriter
-from .base_protocol import PAUSE_RESUME_READING_ERRORS, BaseProtocol
+from .base_protocol import BaseProtocol
 from .helpers import (
     DEFAULT_CHUNK_SIZE,
     HeadersDictProxy,
@@ -182,7 +182,6 @@ class RequestHandler(BaseProtocol, Generic[_Request]):
         "_messages",
         "_max_msg_queue_size",
         "_msg_queue_resume_size",
-        "_msg_queue_paused",
         "_message_tail",
         "_read_bufsize",
         "_handler_waiter",
@@ -227,9 +226,6 @@ def __init__(
         # so we refill in batches instead of churning pause/resume per request.
         self._msg_queue_resume_size = MAX_MSG_QUEUE_SIZE // 2
         self._read_bufsize = read_bufsize
-        # Set before super().__init__ so _reading_paused_for_msg_queue() is safe
-        # if BaseProtocol ever triggers a resume during init.
-        self._msg_queue_paused = False
         parser = HttpRequestParser(
             self,
             loop,
@@ -468,8 +464,8 @@ def set_parser(
             self._payload_parser.feed_data(self._message_tail)
             self._message_tail = b""
 
-        if self._msg_queue_paused:
-            self._resume_msg_queue_reading()
+        if self._buffer_paused:
+            self._resume_reading_if_drained()
 
     def eof_received(self) -> None:
         pass
@@ -502,10 +498,10 @@ def data_received(self, data: bytes) -> None:
             # Queue full: pause the transport (the parser already stopped
             # emitting). start() resumes as it drains the queue.
             if (
-                not self._msg_queue_paused
+                not self._buffer_paused
                 and len(self._messages) >= self._max_msg_queue_size
             ):
-                self._pause_msg_queue_reading()
+                self._pause_reading_for_buffer()
 
             self._upgraded = upgraded
             if upgraded and tail:
@@ -515,10 +511,10 @@ def data_received(self, data: bytes) -> None:
         elif self._payload_parser is None and self._upgraded and data:
             self._message_tail += data
             if (
-                not self._msg_queue_paused
+                not self._buffer_paused
                 and len(self._message_tail) >= self._read_bufsize
             ):
-                self._pause_msg_queue_reading()
+                self._pause_reading_for_buffer()
 
         # feed payload
         elif data:
@@ -528,20 +524,7 @@ def data_received(self, data: bytes) -> None:
             if eof:
                 self.close()
 
-    def _reading_paused_for_msg_queue(self) -> bool:
-        return self._msg_queue_paused
-
-    def _pause_msg_queue_reading(self) -> None:
-        self._msg_queue_paused = True
-        if self.transport is not None:
-            try:
-                self.transport.pause_reading()
-            except PAUSE_RESUME_READING_ERRORS:
-                # Transport lacks flow control; nothing to pause. Intentionally
-                # ignored (see PAUSE_RESUME_READING_ERRORS; do not use suppress).
-                pass
-
-    def _resume_msg_queue_reading(self) -> None:
+    def _resume_reading_if_drained(self) -> None:
         # Tested empty-first so a read_bufsize of 0 cannot wedge the connection.
         if self._message_tail and len(self._message_tail) >= self._read_bufsize:
             return
@@ -553,14 +536,7 @@ def _resume_msg_queue_reading(self) -> None:
             self.data_received(b"")
             if len(self._messages) >= self._max_msg_queue_size:
                 return
-        self._msg_queue_paused = False
-        if not self._reading_paused and self.transport is not None:
-            try:
-                self.transport.resume_reading()
-            except PAUSE_RESUME_READING_ERRORS:
-                # Transport lacks flow control; nothing to resume. Intentionally
-                # ignored (see PAUSE_RESUME_READING_ERRORS; do not use suppress).
-                pass
+        self._resume_reading_for_buffer()
 
     def _replay_message_tail(self) -> None:
         """Re-feed the bytes buffered behind a rejected upgrade.
@@ -614,10 +590,10 @@ def _replay_message_tail(self) -> None:
 
         if len(self._messages) >= self._max_msg_queue_size:
             # Pause the transport, like in data_received().
-            self._pause_msg_queue_reading()
-        elif self._msg_queue_paused:
+            self._pause_reading_for_buffer()
+        elif self._buffer_paused:
             # Resume reading now the tail has been parsed.
-            self._resume_msg_queue_reading()
+            self._resume_reading_if_drained()
 
         # 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.
@@ -766,10 +742,10 @@ async def start(self) -> None:
             if self._parser is not None:
                 self._parser.message_consumed()
             if (
-                self._msg_queue_paused
+                self._buffer_paused
                 and len(self._messages) <= self._msg_queue_resume_size
             ):
-                self._resume_msg_queue_reading()
+                self._resume_reading_if_drained()
 
             # time is only fetched if logging is enabled as otherwise
             # its thrown away and never used.

docs/client_reference.rst

@@ -55,7 +55,7 @@ The client session supports the context manager protocol for self closing.
                          requote_redirect_url=True, \
                          trace_configs=None, \
                          middlewares=(), \
-                         read_bufsize=2**16, \
+                         read_bufsize=2**18, \
                          max_line_size=8190, \
                          max_field_size=8190, \
                          max_headers=128, \
@@ -230,7 +230,9 @@ The client session supports the context manager protocol for self closing.
       .. versionadded:: 3.12
 
    :param int read_bufsize: Size of the read buffer (:attr:`ClientResponse.content`).
-                            64 KiB by default.
+                            256 KiB by default. On a WebSocket connection it
+                            also bounds what is buffered between the handshake
+                            and the reader being installed.
 
       .. versionadded:: 3.7