diff --git a/CHANGES/13352.bugfix.rst b/CHANGES/13352.bugfix.rst index a10a0d7767d..dd0d94e3b0c 100644 --- a/CHANGES/13352.bugfix.rst +++ b/CHANGES/13352.bugfix.rst @@ -1 +1 @@ -Capped the number of per-read payload fragments the WebSocket reader retains in memory -- by :user:`Dreamsorcerer`. +Bounded the per-read object overhead the WebSocket reader retains while reassembling a frame delivered across many small reads; the reads are joined once when the frame completes, and folded into a single buffer if they exceed a fragment cap, so a frame dribbled in tiny reads cannot pin unbounded per-read overhead -- by :user:`Dreamsorcerer` and :user:`bdraco`. diff --git a/CHANGES/13488.bugfix.rst b/CHANGES/13488.bugfix.rst new file mode 120000 index 00000000000..955ee549b59 --- /dev/null +++ b/CHANGES/13488.bugfix.rst @@ -0,0 +1 @@ +13352.bugfix.rst \ No newline at end of file diff --git a/CHANGES/13609.bugfix.rst b/CHANGES/13609.bugfix.rst new file mode 100644 index 00000000000..b956bba913a --- /dev/null +++ b/CHANGES/13609.bugfix.rst @@ -0,0 +1 @@ +Capped reflected payload in parser errors to ~100 bytes -- by :user:`Dreamsorcerer`. diff --git a/THREAT_MODEL.md b/THREAT_MODEL.md index 41d0ef51c7b..11529668f82 100644 --- a/THREAT_MODEL.md +++ b/THREAT_MODEL.md @@ -267,7 +267,7 @@ into `StreamReader`) is then handed to `web_protocol.RequestHandler` and | 1.8 | `Transfer-Encoding` lenience | `_is_chunked_te` requires `chunked` to be the last value; duplicate `chunked` rejected (`#10611`). Request parser strict. | None. | | 1.9 | Chunk-size DoS | The parser doesn't cap chunk size, but **server-side body length is bounded by `client_max_size` (default `1 MiB`)** in `web_request.py:BaseRequest.read`. Client-side responses are bounded by user-supplied `max_body_size` / streaming reads. | None. If a cap is ever needed at the parser level, plumb it through `HttpPayloadParser`. | | 1.10 | Chunk-extension DoS | Chunk-extension content is bounded by the same wire-level size constraints (it shares the chunk-size line with `max_line_size`). | **Add an explicit test that chunk-extension flooding cannot blow past `max_line_size`.** | -| 1.11 | Parser error reflection | `http_parser.py` truncates to `[:100]` only for `LineTooLong`; `BadStatusLine` / `InvalidHeader` / `TransferEncodingError` carry the offending line up to `max_line_size` / `max_field_size`. | **Audit any aiohttp path where `BadHttpMessage` content is reflected to the client unsanitised.** **User**: Review custom `web_log` configurations and any middleware that reflects parser exception messages back to the peer. | +| 1.11 | Parser error reflection | `http_parser.py` truncates to `[:100]` only for `LineTooLong`; `BadStatusLine` / `InvalidHeader` / `TransferEncodingError` carry the offending line up to `max_line_size` / `max_field_size`. `_http_parser.pyx` bounds its snippet to 50 bytes either side of the error position, so input with no CRLF to delimit the offending line is not quoted in full (`test_c_parser_error_message_bounded_for_crlf_free_input`). | **Audit any aiohttp path where `BadHttpMessage` content is reflected to the client unsanitised.** **User**: Review custom `web_log` configurations and any middleware that reflects parser exception messages back to the peer. | | 1.12 | Cython ⇄ pure-Python divergence | `tests/test_http_parser.py` parameterises tests over `REQUEST_PARSERS` / `RESPONSE_PARSERS` (pure-Python always; Cython when the extension imports). The high-leverage attack vectors are already covered under both backends: CL+TE (`test_content_length_transfer_encoding`), CL×N (`test_duplicate_singleton_header_rejected`), obs-fold (`test_reject_obsolete_line_folding`, `test_http_response_parser_obs_line_folding*`), CR/LF/NUL (`test_bad_headers`, `test_http_response_parser_null_byte_in_header_value`, `test_http_response_parser_bad_crlf`), version regex (`test_http_request_parser_bad_version*`, `test_http_response_parser_bad_version*`), bare-LF line endings (`test_reject_bare_lf_no_cross_request_leak`), control characters in the request target (`test_http_request_parser_ctl_in_request_target`). | None. When new attack vectors emerge, add them to the parameterised tests. | | 1.13 | llhttp version drift | Manual upgrade via `make generate-llhttp`; vendor pinned in `vendor/llhttp/package.json`. | Track upstream releases (e.g. via Dependabot rule for `vendor/llhttp/package.json`), bump on every llhttp release, regenerate in CI. | | 1.14 | npm-side compromise of `llhttp` | The vendored output is checked into git, so a compromise during a future regen would be detectable in PR review. See [§5.19](#519-build--release-supply-chain). | **Make the llhttp build reproducible: pin Node.js version, commit the npm lockfile, and on every bump verify the regenerated C against upstream's release tarballs before committing.** | @@ -548,7 +548,7 @@ client-side, the writer adds masks to outgoing frames. | 3.3 | RSV bits | `reader_py.py:WebSocketReader._feed_data` gates RSV1 on the PMCE-negotiated `_compress` flag; RSV2/3 always rejected. | None. | | 3.4 | Unknown opcode | Rejected. | None. | | 3.5–3.7 | Control-frame and fragmentation rules | All enforced at reader. | None. | -| 3.8 | Fragment memory bound | (a) Declared byte size — `max_msg_size` enforced pre-FIN and at assembly (default 4 MiB). (b) `WebSocketReader.__init__` caps the total fragment count at `max(1024, max_msg_size // 256)` and pauses reading for backpressure once exceeded. | **User**: set a smaller `max_msg_size` for protocols where messages are bounded (e.g. chat); the 4 MiB default suits arbitrary payloads. | +| 3.8 | Fragment memory bound | (a) Declared byte size — `max_msg_size` enforced pre-FIN and at assembly (default 4 MiB). (b) The reader collects a frame's reads in a list and joins 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`. | **User**: set a smaller `max_msg_size` for protocols where messages are bounded (e.g. chat); the 4 MiB default suits arbitrary payloads. | | 3.9 | PMCE decompression bomb | `WebSocketReader._handle_frame` decompresses with a `max_length` of `max_msg_size + 1` and checks the result; on overflow, raises `MESSAGE_TOO_BIG` (1009). This `max_length` post-decompress check was introduced by PR #11898 (v3.13.3). | **Documented known limitation.** Some backends (notably `isal_zlib`) do not strictly honour `max_length` in `decompress()` and may overshoot by up to one zlib block before the post-decompress size check fires. The post-check still catches it before the bytes reach the application, but a transient over-allocation is possible. Document and monitor. | | 3.10 | PMCE context retention | Default extensions request context takeover (per RFC 7692 default); user can negotiate `server_no_context_takeover` / `client_no_context_takeover` via handshake. | Documented design decision: keep the RFC 7692 default (context takeover). **Document the memory tradeoff in user-facing WebSocket docs.** **User**: configure no-context-takeover on long-lived sessions running on memory-constrained hosts. | | 3.11 | UTF-8 validation | Strict `bytes.decode("utf-8")` post-assembly. | None. | @@ -575,13 +575,6 @@ client-side, the writer adds masks to outgoing frames. RFC 6455 §5.2 requires failing such frames). Fixed by passing `compress=bool(compress)` in `client.py:_ws_connect` and removing the `compress` / `decode_text` defaults on `WebSocketReader.__init__`. -- **PR #13350** (follow-up to CVE-2026-54274) — the per-frame - `max_msg_size` byte cap still let a size-legal frame dribbled in tiny - transport reads pin ~28x its on-wire size in per-read `bytes` objects - (`_payload_fragments`). `WebSocketReader` now caps the retained - fragment count (`max(1024, max_msg_size // 256)`) and pauses reading for - backpressure once exceeded, mirroring the HTTP chunk-splits limit in - `StreamReader` (PR #11894). - **PR #13393** — queue accounting only counted payload bytes, so a flood of empty/tiny frames could pin unbounded per-message object overhead in `WebSocketDataQueue` before the `_limit` high-water mark fired. Each @@ -597,6 +590,13 @@ client-side, the writer adds masks to outgoing frames. Callers constructing a `WebSocketReader` for `set_parser()` must hold a strong reference to it (both in-tree response classes do, via `_parser`). +- **PR #13488** — the per-frame + `max_msg_size` byte cap still let a size-legal frame dribbled in tiny + transport reads pin ~28x its on-wire size in per-read `bytes` objects + (`_payload_fragments`). The reader now collects reads in a list and joins + 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. --- diff --git a/aiohttp/_http_parser.pyx b/aiohttp/_http_parser.pyx index cab0054e4c5..c78d9a30be4 100644 --- a/aiohttp/_http_parser.pyx +++ b/aiohttp/_http_parser.pyx @@ -692,10 +692,13 @@ cdef class HttpParser: else: error_pos = cparser.llhttp_get_error_pos(self._cparser) error_off = error_pos - base + + # Bounded window either side of the error position of 50 bytes. before = data[:error_off] - after = data[error_off:].split(b"\r\n", 1)[0] - before = before.rsplit(b"\r\n", 1)[-1] + before = before.rsplit(b"\r\n", 1)[-1][-50:] + after = data[error_off:].split(b"\r\n", 1)[0][:50] data = before + after + pointer = " " * (len(repr(before))-1) + "^" ex = parser_error_from_errno(self._cparser, data, pointer) self._payload = None diff --git a/aiohttp/_websocket/reader_c.pxd b/aiohttp/_websocket/reader_c.pxd index 19c7b3150d9..b9c07f71d23 100644 --- a/aiohttp/_websocket/reader_c.pxd +++ b/aiohttp/_websocket/reader_c.pxd @@ -87,6 +87,7 @@ cdef class WebSocketReader: cdef bint _frame_fin cdef int _frame_opcode cdef list _payload_fragments + cdef bytearray _payload_buffer cdef Py_ssize_t _max_fragments cdef Py_ssize_t _frame_payload_len diff --git a/aiohttp/_websocket/reader_py.py b/aiohttp/_websocket/reader_py.py index 569be86f3fe..00588632dc2 100644 --- a/aiohttp/_websocket/reader_py.py +++ b/aiohttp/_websocket/reader_py.py @@ -209,10 +209,12 @@ def __init__( self._opcode: int = OP_CODE_NOT_SET self._frame_fin = False self._frame_opcode: int = OP_CODE_NOT_SET + # Reads of an in-flight frame, joined once when it completes. self._payload_fragments: list[bytes] = [] - # Limit number of fragments, so a large number of tiny fragments - # doesn't exceed reasonable memory usage. + # Fold reads into _payload_buffer past this count to bound the object + # count (bytes are bounded by max_msg_size). self._max_fragments = max(1024, max_msg_size // 256) if max_msg_size else 0 + self._payload_buffer = bytearray() self._frame_payload_len = 0 self._tail: bytes = b"" @@ -572,22 +574,29 @@ def _feed_data(self, data: bytes) -> None: start_pos = f_end_pos if self._payload_bytes_to_read != 0: - # If we don't have a complete frame, we need to save the - # data for the next call to feed_data. - self._payload_fragments.append(data_cstr[f_start_pos:f_end_pos]) + if f_start_pos < f_end_pos: # skip a header-only read + self._payload_fragments.append(data_cstr[f_start_pos:f_end_pos]) if ( self._max_fragments and len(self._payload_fragments) > self._max_fragments - and not self.queue._protocol._reading_paused ): - self.queue._protocol.pause_reading() + # Fold to bound the object count. Not a pause: nothing + # resumes reading until the frame is queued. + self._payload_buffer += b"".join(self._payload_fragments) + self._payload_fragments.clear() break payload: bytes | bytearray if had_fragments: - # We have to join the payload fragments get the payload self._payload_fragments.append(data_cstr[f_start_pos:f_end_pos]) - if self._has_mask: + if self._payload_buffer: # folded prefix + self._payload_buffer += b"".join(self._payload_fragments) + if self._has_mask: + assert self._frame_mask is not None + websocket_mask(self._frame_mask, self._payload_buffer) + payload = self._payload_buffer + self._payload_buffer = bytearray() # detach; payload aliases it + elif self._has_mask: assert self._frame_mask is not None payload_bytearray = bytearray(b"".join(self._payload_fragments)) websocket_mask(self._frame_mask, payload_bytearray) diff --git a/tests/test_http_parser.py b/tests/test_http_parser.py index 701622cd1e4..846f8392759 100644 --- a/tests/test_http_parser.py +++ b/tests/test_http_parser.py @@ -2364,6 +2364,23 @@ def test_c_parser_error_snippet_at_buffer_end_request( parser.feed_data(text) +@pytest.mark.skipif(NO_EXTENSIONS, reason="Python parser lacks error pos output") +def test_c_parser_error_message_bounded_for_crlf_free_input( + event_loop: asyncio.AbstractEventLoop, + server: Server[Request], +) -> None: + """Garbage with no CRLF must not be echoed back whole.""" + protocol = RequestHandler(server, loop=event_loop) + parser = HttpRequestParserC( + protocol, event_loop, 2**16, max_line_size=8190, max_field_size=8190 + ) + protocol._parser = parser + with pytest.raises(http_exceptions.BadHttpMethod) as exc_info: + parser.feed_data(b"A" * (64 * 1024)) + # Two bounded windows, escaped by repr, plus the pointer line beneath them. + assert len(exc_info.value.message) < 2048 + + @pytest.mark.skipif(NO_EXTENSIONS, reason="Only tests C parser.") @pytest.mark.parametrize("split", [1, 2, 3, 5]) @pytest.mark.parametrize(("body", "snippet", "reason"), _BAD_CHUNKED_RESPONSES) diff --git a/tests/test_web_functional.py b/tests/test_web_functional.py index 1519a2d8aa6..f129cbe90df 100644 --- a/tests/test_web_functional.py +++ b/tests/test_web_functional.py @@ -1974,7 +1974,8 @@ async def send_until_paused() -> None: await sender finally: release_handler.set() - writer.close() + # Abort so the server doesn't get stuck waiting for the client to read. + writer.transport.abort() with suppress(ConnectionResetError, BrokenPipeError): await writer.wait_closed() diff --git a/tests/test_websocket_parser.py b/tests/test_websocket_parser.py index a4942bc0460..4bd1568194a 100644 --- a/tests/test_websocket_parser.py +++ b/tests/test_websocket_parser.py @@ -21,7 +21,11 @@ ) from aiohttp._websocket.models import WS_DEFLATE_TRAILING from aiohttp._websocket.reader import WebSocketDataQueue -from aiohttp._websocket.reader_py import MSG_SIZE_OVERHEAD +from aiohttp._websocket.reader_py import ( + MSG_SIZE_OVERHEAD, + WebSocketDataQueue as PyWebSocketDataQueue, + WebSocketReader as PyWebSocketReader, +) from aiohttp.base_protocol import BaseProtocol from aiohttp.compression_utils import ZLibBackend, ZLibBackendWrapper from aiohttp.helpers import DEFAULT_CHUNK_SIZE @@ -986,48 +990,174 @@ def test_flow_control_multi_byte_text( assert protocol._reading_paused is True -async def test_incomplete_frame_pauses_when_fragment_limit_exceeded( +async def test_incomplete_frame_retained_objects_stay_bounded( protocol: BaseProtocol, ) -> None: + """Reads fold into one buffer past the cap, bounding the object count. + + Uses the pure-Python reader so the internals are reachable. + """ max_msg_size = 64 * 1024 loop = asyncio.get_running_loop() - out = WebSocketDataQueue(protocol, 2**16, loop=loop) - parser = WebSocketReader(out, max_msg_size, compress=False, decode_text=False) + out = PyWebSocketDataQueue(protocol, 2**16, loop=loop) + parser = PyWebSocketReader(out, max_msg_size, compress=False, decode_text=False) payload_len = 32 * 1024 parser.feed_data(PACK_LEN2(0x80 | WSMsgType.BINARY, 126, payload_len)) - assert protocol._reading_paused is False - # Feed the payload two bytes per read so the pause is - # driven purely by the fragment count, not the total byte count. - paused_after = None - for i in range(payload_len // 2 - 1): # pragma: no branch + # Two bytes per read runs past the cap; the list folds and stays bounded. + for _ in range(payload_len // 2 - 1): parser.feed_data(b"xx") - if protocol._reading_paused: - paused_after = i + 1 # type: ignore[unreachable] - break + assert len(parser._payload_fragments) <= parser._max_fragments + + parser.feed_data(b"xx") + msg = await out.read() + assert msg.data == b"x" * payload_len + assert len(parser._payload_fragments) == 0 + assert len(parser._payload_buffer) == 0 - assert paused_after is not None - # Paused long before the frame could complete (16384 two-byte reads). - assert paused_after < payload_len // 2 # type: ignore[unreachable] +def _fold_cap(max_msg_size: int) -> int: + """The fragment count the reader folds at (single source of truth).""" + return PyWebSocketReader( + mock.Mock(), max_msg_size, compress=False, decode_text=False + )._max_fragments + + +@pytest.mark.parametrize("mask", [False, True]) +async def test_frame_split_across_many_reads_is_delivered_intact( + protocol: BaseProtocol, mask: bool +) -> None: + """A frame arriving two bytes per read (past the fold cap) is delivered whole.""" + max_msg_size = 64 * 1024 + loop = asyncio.get_running_loop() + out = WebSocketDataQueue(protocol, 2**16, loop=loop) + parser = WebSocketReader(out, max_msg_size, compress=False, decode_text=False) -async def test_incomplete_frame_not_paused_for_normal_reads( + payload = bytes(range(256)) * 32 # 8 KiB of varied, distinctive bytes + frame = build_frame(payload, WSMsgType.BINARY, mask=mask) + + # Guard the reads stay past the fold cap even if the cap constants change. + assert len(frame) // 2 > _fold_cap(max_msg_size) + chunks = [frame[i : i + 2] for i in range(0, len(frame), 2)] + for chunk in chunks[:-1]: + parser.feed_data(chunk) + assert not out._buffer + assert protocol._reading_paused is False + + parser.feed_data(chunks[-1]) + msg = await out.read() + assert msg.data == payload + + +async def test_consecutive_folded_frames_do_not_bleed( protocol: BaseProtocol, ) -> None: + """Two frames folded on one reader stay independent (buffer detached, not cleared).""" max_msg_size = 64 * 1024 loop = asyncio.get_running_loop() out = WebSocketDataQueue(protocol, 2**16, loop=loop) parser = WebSocketReader(out, max_msg_size, compress=False, decode_text=False) - # Normal traffic (a frame delivered in a handful of reasonably sized reads) - # must never trip the fragment-limit backpressure. + first = bytes(range(256)) * 32 # 8 KiB + second = bytes(reversed(range(256))) * 24 # 6 KiB, distinct content + + for payload in (first, second): + frame = build_frame(payload, WSMsgType.BINARY, mask=True) + assert len(frame) // 2 > _fold_cap(max_msg_size) # reads exceed the cap + for i in range(0, len(frame), 2): + parser.feed_data(frame[i : i + 2]) + # .clear() would deliver frame 1 empty; buffer reuse would corrupt frame 2. + msg = await out.read() + assert msg.data == payload + + +async def test_fragmented_message_survives_fold(protocol: BaseProtocol) -> None: + """A multi-frame message whose frames each fold reassembles correctly. + + A non-final folded frame hands a bytearray to ``self._partial += payload``. + """ + max_msg_size = 64 * 1024 + loop = asyncio.get_running_loop() + out = WebSocketDataQueue(protocol, 2**16, loop=loop) + parser = WebSocketReader(out, max_msg_size, compress=False, decode_text=False) + + first = bytes(range(256)) * 32 # 8 KiB + second = bytes(reversed(range(256))) * 24 # 6 KiB + frames = [ + build_frame(first, WSMsgType.BINARY, is_fin=False, mask=True), + build_frame(second, WSMsgType.CONTINUATION, is_fin=True, mask=True), + ] + for frame in frames: + assert len(frame) // 2 > _fold_cap(max_msg_size) + for i in range(0, len(frame), 2): + parser.feed_data(frame[i : i + 2]) + + msg = await out.read() + assert msg.data == first + second + + +async def test_compressed_frame_survives_fold(protocol: BaseProtocol) -> None: + """A PMCE frame folded past the cap still decompresses (folded bytearray path).""" + max_msg_size = 64 * 1024 + loop = asyncio.get_running_loop() + out = WebSocketDataQueue(protocol, 2**16, loop=loop) + parser = WebSocketReader(out, max_msg_size, compress=True, decode_text=False) + + # Incompressible, so the deflated frame stays large enough to fold. + rng = random.Random(0) + payload = bytes(rng.getrandbits(8) for _ in range(8 * 1024)) + frame = build_frame(payload, WSMsgType.BINARY, ZLibBackend=ZLibBackend, mask=True) + assert len(frame) // 2 > _fold_cap(max_msg_size) + + for i in range(0, len(frame), 2): + parser.feed_data(frame[i : i + 2]) + msg = await out.read() + assert msg.data == payload + + +async def test_frame_split_under_cap_uses_list_without_folding( + protocol: BaseProtocol, +) -> None: + """A frame under the cap stays in the list, never folds, never pauses.""" + max_msg_size = 64 * 1024 + loop = asyncio.get_running_loop() + out = PyWebSocketDataQueue(protocol, 2**16, loop=loop) + parser = PyWebSocketReader(out, max_msg_size, compress=False, decode_text=False) + payload_len = 32 * 1024 parser.feed_data(PACK_LEN2(0x80 | WSMsgType.BINARY, 126, payload_len)) - # 32 KiB in 4 KiB reads -> 8 fragments, far below the cap. - for _ in range(payload_len // 4096): + reads = payload_len // 4096 + assert reads < parser._max_fragments # stays under the cap: no fold + for _ in range(reads): parser.feed_data(b"x" * 4096) - assert protocol._reading_paused is False + assert protocol._reading_paused is False + assert len(parser._payload_buffer) == 0 # never folded + + msg = await out.read() + assert msg.data == b"x" * payload_len + + +async def test_max_msg_size_zero_opts_out_of_the_fold( + protocol: BaseProtocol, +) -> None: + """max_msg_size=0 disables every limit, including the fold.""" + loop = asyncio.get_running_loop() + out = PyWebSocketDataQueue(protocol, 2**16, loop=loop) + parser = PyWebSocketReader(out, 0, compress=False, decode_text=False) + assert parser._max_fragments == 0 + + payload_len = 4096 # well past the 1024 floor + parser.feed_data(PACK_LEN2(0x80 | WSMsgType.BINARY, 126, payload_len)) + for _ in range(payload_len - 1): + parser.feed_data(b"x") + # No fold: every read accumulates in the list, the buffer is untouched. + assert len(parser._payload_fragments) == payload_len - 1 + assert len(parser._payload_buffer) == 0 + + parser.feed_data(b"x") + msg = await out.read() + assert msg.data == b"x" * payload_len def _compressed_burst(payload: bytes, count: int) -> bytes: