From fd07a7a7d7cdff58ca2a98c6c7a26149562e13a7 Mon Sep 17 00:00:00 2001 From: Sam Bull Date: Tue, 7 Jul 2026 18:24:06 +0100 Subject: [PATCH 1/5] Fix OWS in Content-Disposition (#13069) --- CHANGES/13002.bugfix.rst | 1 + aiohttp/multipart.py | 5 +++-- tests/test_multipart_helpers.py | 10 ++++++++++ 3 files changed, 14 insertions(+), 2 deletions(-) create mode 100644 CHANGES/13002.bugfix.rst diff --git a/CHANGES/13002.bugfix.rst b/CHANGES/13002.bugfix.rst new file mode 100644 index 00000000000..1ac16a04fa7 --- /dev/null +++ b/CHANGES/13002.bugfix.rst @@ -0,0 +1 @@ +Fixed parsing optional whitespace in Content-Disposition -- by :user:`Dreamsorcerer`. diff --git a/aiohttp/multipart.py b/aiohttp/multipart.py index 190f2e2b4cc..3fc2de98d35 100644 --- a/aiohttp/multipart.py +++ b/aiohttp/multipart.py @@ -156,9 +156,10 @@ def unescape(text: str, *, chars: str = "".join(map(re.escape, CHAR))) -> str: else: failed = True - if is_quoted(value): + rstripped = value.rstrip() + if is_quoted(rstripped): failed = False - value = unescape(value[1:-1].lstrip("\\/")) + value = unescape(rstripped[1:-1].lstrip("\\/")) elif is_token(value): failed = False elif parts: diff --git a/tests/test_multipart_helpers.py b/tests/test_multipart_helpers.py index 67bbc86c223..87c33778559 100644 --- a/tests/test_multipart_helpers.py +++ b/tests/test_multipart_helpers.py @@ -30,6 +30,16 @@ def test_semicolon(self) -> None: assert disptype == "form-data" assert params == {"name": "data", "filename": "file ; name.mp4"} + def test_ows_before_separator(self) -> None: + # https://github.com/aio-libs/aiohttp/issues/13002 + # Optional whitespace before the ";" separator (RFC 9110 §5.6.6) must + # not make the quoted-value repair heuristic swallow the next param. + disptype, params = parse_content_disposition( + 'attachment; filename="test.txt" ; name="field"' + ) + assert disptype == "attachment" + assert params == {"filename": "test.txt", "name": "field"} + def test_inlwithasciifilename(self) -> None: disptype, params = parse_content_disposition('inline; filename="foo.html"') assert "inline" == disptype From 4002772f6d1f0700efbac85e507fede89bd76972 Mon Sep 17 00:00:00 2001 From: Sam Bull Date: Tue, 7 Jul 2026 19:07:46 +0100 Subject: [PATCH 2/5] Add chunk tracing to .content (#12889) --- CHANGES/12889.bugfix.rst | 5 + aiohttp/client_reqrep.py | 27 +++-- aiohttp/streams.py | 58 ++++++++- docs/tracing_reference.rst | 9 ++ tests/test_client_response.py | 44 ------- tests/test_client_session.py | 106 +++++++++++++++- tests/test_streams.py | 222 +++++++++++++++++++++++++++++++++- 7 files changed, 410 insertions(+), 61 deletions(-) create mode 100644 CHANGES/12889.bugfix.rst diff --git a/CHANGES/12889.bugfix.rst b/CHANGES/12889.bugfix.rst new file mode 100644 index 00000000000..96f4592888d --- /dev/null +++ b/CHANGES/12889.bugfix.rst @@ -0,0 +1,5 @@ +``TraceConfig.on_response_chunk_received`` now fires for every chunk returned from +``ClientResponse.content`` read methods (``read``, ``readany``, ``readchunk``, +``readuntil``, ``read_nowait``, ``iter_chunked``, ``iter_any``, ``iter_chunks``). +Previously the signal only fired once from ``ClientResponse.read()`` with the entire +body, and direct ``.content`` access bypassed tracing entirely -- by :user:`Dreamsorcerer`. diff --git a/aiohttp/client_reqrep.py b/aiohttp/client_reqrep.py index 5714fb3327f..51da8711c58 100644 --- a/aiohttp/client_reqrep.py +++ b/aiohttp/client_reqrep.py @@ -58,7 +58,7 @@ HttpVersion11, StreamWriter, ) -from .streams import StreamReader +from .streams import EMPTY_PAYLOAD, StreamReader from .typedefs import DEFAULT_JSON_DECODER, JSONDecoder, RawHeaders try: @@ -556,6 +556,9 @@ async def start(self, connection: "Connection") -> "ClientResponse": # payload self.content = payload + if self._traces and payload is not EMPTY_PAYLOAD: + payload._on_chunk_received = self._on_chunk_response_received + # cookies if cookie_hdrs := self.headers._md.getall(hdrs.SET_COOKIE, ()): # Store raw cookie headers for CookieJar @@ -660,8 +663,14 @@ def _cleanup_writer(self) -> None: def _notify_content(self) -> None: content = self.content # content can be None here, but the types are cheated elsewhere. - if content and content.exception() is None: # type: ignore[truthy-bool] - set_exception(content, _CONNECTION_CLOSED_EXCEPTION) + if content: # type: ignore[truthy-bool] + if content.exception() is None: + set_exception(content, _CONNECTION_CLOSED_EXCEPTION) + # The bound method installed in start() captures self, creating a + # response→payload→method→self cycle. Clear it eagerly so the + # response is reclaimable without waiting for cycle GC. + if content._on_chunk_received is not None: + content._on_chunk_received = None self._released = True async def wait_for_close(self) -> None: @@ -677,15 +686,19 @@ async def wait_for_close(self) -> None: raise self.release() + async def _on_chunk_response_received(self, chunk: bytes) -> None: + try: + for trace in self._traces: + await trace.send_response_chunk_received(self.method, self.url, chunk) + except BaseException: + self.close() + raise + async def read(self) -> bytes: """Read response payload.""" if self._body is None: try: self._body = await self.content.read() - for trace in self._traces: - await trace.send_response_chunk_received( - self.method, self.url, self._body - ) except BaseException: self.close() raise diff --git a/aiohttp/streams.py b/aiohttp/streams.py index 72d26e607d7..cdd365cada3 100644 --- a/aiohttp/streams.py +++ b/aiohttp/streams.py @@ -2,7 +2,7 @@ import collections import sys import warnings -from collections.abc import Awaitable, Callable +from collections.abc import Awaitable, Callable, Coroutine from typing import Final, Generic, TypeVar from .base_protocol import BaseProtocol @@ -101,6 +101,7 @@ class StreamReader: "_timer", "_eof_callbacks", "_eof_counter", + "_on_chunk_received", "total_bytes", "total_compressed_bytes", ) @@ -134,6 +135,9 @@ def __init__( self._timer = TimerNoop() if timer is None else timer self._eof_callbacks: list[Callable[[], None]] = [] self._eof_counter = 0 + self._on_chunk_received: ( + Callable[[bytes], Coroutine[None, None, None]] | None + ) = None self.total_bytes = 0 self.total_compressed_bytes: int | None = None @@ -363,6 +367,14 @@ async def _wait(self, func_name: str) -> None: finally: self._waiter = None + async def _fire_chunk_received(self, chunk: bytes) -> None: + cb = self._on_chunk_received + assert cb is not None + # Run under the same per-stream timer that _wait() uses, so a hung + # trace handler is bounded by sock_read just like a hung socket read would be. + with self._timer: + await cb(chunk) + async def readline(self, *, max_line_length: int | None = None) -> bytes: return await self.readuntil(max_size=max_line_length) @@ -403,6 +415,8 @@ async def readuntil( if not_enough: await self._wait("readuntil") + if chunk and self._on_chunk_received is not None: + await self._fire_chunk_received(chunk) return chunk async def read(self, n: int = -1) -> bytes: @@ -414,6 +428,7 @@ async def read(self, n: int = -1) -> bytes: if n < 0: # Reading everything — remove decompression chunk limit. + # readany() fires the chunk hook for each block. self.set_read_chunk_size(sys.maxsize) blocks = [] while True: @@ -430,7 +445,10 @@ async def read(self, n: int = -1) -> bytes: while not self._buffer and not self._eof: await self._wait("read") - return self._read_nowait(n) + chunk = self._read_nowait(n) + if chunk and self._on_chunk_received is not None: + await self._fire_chunk_received(chunk) + return chunk async def readany(self) -> bytes: if self._exception is not None: @@ -442,7 +460,10 @@ async def readany(self) -> bytes: while not self._buffer and not self._eof: await self._wait("readany") - return self._read_nowait(-1) + chunk = self._read_nowait(-1) + if chunk and self._on_chunk_received is not None: + await self._fire_chunk_received(chunk) + return chunk async def readchunk(self) -> tuple[bytes, bool]: """Returns a tuple of (data, end_of_http_chunk). @@ -461,14 +482,20 @@ async def readchunk(self) -> tuple[bytes, bool]: if pos == self._cursor: return (b"", True) if pos > self._cursor: - return (self._read_nowait(pos - self._cursor), True) + chunk = self._read_nowait(pos - self._cursor) + if chunk and self._on_chunk_received is not None: + await self._fire_chunk_received(chunk) + return (chunk, True) internal_logger.warning( "Skipping HTTP chunk end due to data " "consumption beyond chunk boundary" ) if self._buffer: - return (self._read_nowait_chunk(-1), False) + chunk = self._read_nowait_chunk(-1) + if chunk and self._on_chunk_received is not None: + await self._fire_chunk_received(chunk) + return (chunk, False) # return (self._read_nowait(-1), False) if self._eof: @@ -506,7 +533,13 @@ def read_nowait(self, n: int = -1) -> bytes: "Called while some coroutine is waiting for incoming data." ) - return self._read_nowait(n) + chunk = self._read_nowait(n) + if chunk and (cb := self._on_chunk_received) is not None: + # read_nowait is sync but the hook is async; schedule it so the + # observability event still fires. + # TODO: Save and await this task. + asyncio.create_task(cb(chunk)) # type: ignore[unused-awaitable] + return chunk def _read_nowait_chunk(self, n: int) -> bytes: first_buffer = self._buffer[0] @@ -570,6 +603,19 @@ def __init__(self) -> None: self._read_eof_chunk = False self.total_bytes = 0 + # Shadow the inherited slot with a property so the EMPTY_PAYLOAD singleton + # can't be polluted with a per-response hook that would leak across + # requests. EmptyStreamReader never delivers a chunk anyway. + @property + def _on_chunk_received(self) -> None: + return None + + @_on_chunk_received.setter + def _on_chunk_received( + self, value: Callable[[bytes], Coroutine[None, None, None]] | None + ) -> None: + raise AttributeError("EmptyStreamReader._on_chunk_received is read-only") + def __repr__(self) -> str: return "<%s>" % self.__class__.__name__ diff --git a/docs/tracing_reference.rst b/docs/tracing_reference.rst index 21a28042702..528dc62bae7 100644 --- a/docs/tracing_reference.rst +++ b/docs/tracing_reference.rst @@ -237,6 +237,15 @@ Classes .. versionadded:: 3.1 + .. versionchanged:: 4.0 + + The signal now fires for every chunk returned from + ``ClientResponse.content``. Previously it only fired once from + :meth:`ClientResponse.read` with the full body, and direct + ``.content`` access did not fire it at all. For the synchronous + :meth:`StreamReader.read_nowait` the hook runs as a scheduled task, + not inline. + .. attribute:: on_request_redirect Property that gives access to the signals that will be executed when a diff --git a/tests/test_client_response.py b/tests/test_client_response.py index 6288cc23a37..6f3fe961db3 100644 --- a/tests/test_client_response.py +++ b/tests/test_client_response.py @@ -19,7 +19,6 @@ from aiohttp.connector import Connection from aiohttp.helpers import HeadersDictProxy, TimerNoop from aiohttp.multipart import BadContentDispositionHeader -from aiohttp.tracing import Trace class WriterMock(mock.AsyncMock): @@ -1385,49 +1384,6 @@ def test_redirect_history_in_exception() -> None: assert (hist_response,) == cm.value.history -async def test_response_read_triggers_callback(session: ClientSession) -> None: - loop = asyncio.get_running_loop() - trace = mock.create_autospec(Trace, instance=True, spec_set=True) - response_method = "get" - response_url = URL("http://def-cl-resp.org") - response_body = b"This is response" - - response = ClientResponse( - response_method, - response_url, - writer=WriterMock(), - continue100=None, - timer=TimerNoop(), - loop=loop, - session=session, - traces=[trace], - request_headers=CIMultiDict[str](), - original_url=response_url, - stream_writer=mock.create_autospec( - AbstractStreamWriter, spec_set=True, instance=True - ), - ) - - def side_effect(*args: object, **kwargs: object) -> "asyncio.Future[bytes]": - fut = loop.create_future() - fut.set_result(response_body) - return fut - - h = {"Content-Type": "application/json;charset=cp1251"} - response._headers = HeadersDictProxy(CIMultiDict(h)) - content = response.content = mock.Mock() - content.read.side_effect = side_effect - - res = await response.read() - assert res == response_body - assert response._connection is None - - assert trace.send_response_chunk_received.called - assert trace.send_response_chunk_received.call_args == mock.call( - response_method, response_url, response_body - ) - - def test_response_cookies( event_loop: asyncio.AbstractEventLoop, session: ClientSession ) -> None: diff --git a/tests/test_client_session.py b/tests/test_client_session.py index bee0a49263c..10f06d83bb2 100644 --- a/tests/test_client_session.py +++ b/tests/test_client_session.py @@ -1050,6 +1050,107 @@ async def on_request_headers_sent( assert gathered_req_headers["Custom-Header"] == "Custom value" +async def test_response_chunk_received_via_content( + aiohttp_client: AiohttpClient, +) -> None: + body = b"x" * 4096 + + async def handler(request: web.Request) -> web.StreamResponse: + resp = web.StreamResponse() + resp.content_length = len(body) + await resp.prepare(request) + await resp.write(body) + return resp + + chunks: list[bytes] = [] + + async def on_response_chunk_received( + session: object, + context: object, + params: tracing.TraceResponseChunkReceivedParams, + ) -> None: + chunks.append(params.chunk) + + trace_config = aiohttp.TraceConfig() + trace_config.on_response_chunk_received.append(on_response_chunk_received) + + app = web.Application() + app.router.add_get("/", handler) + client = await aiohttp_client(app, trace_configs=[trace_config]) + + async with client.get("/") as resp: + async for _ in resp.content.iter_chunked(512): + pass + # Hook must fire per chunk delivered to the caller; together they + # must concatenate to the full body. + assert chunks + assert b"".join(chunks) == body + + +async def test_response_chunk_received_hook_cleared_on_release( + aiohttp_client: AiohttpClient, +) -> None: + async def handler(request: web.Request) -> web.Response: + return web.Response(body=b"x" * 256) + + async def on_response_chunk_received( + session: object, + context: object, + params: tracing.TraceResponseChunkReceivedParams, + ) -> None: + assert False + + trace_config = aiohttp.TraceConfig() + trace_config.on_response_chunk_received.append(on_response_chunk_received) + + app = web.Application() + app.router.add_get("/", handler) + client = await aiohttp_client(app, trace_configs=[trace_config]) + + resp = await client.get("/") + assert resp.content._on_chunk_received is not None + resp.release() + assert resp.content._on_chunk_received is None + + resp = await client.get("/") # type: ignore[unreachable] + assert resp.content._on_chunk_received is not None + resp.close() + assert resp.content._on_chunk_received is None + + +async def test_response_chunk_received_trace_failure_closes_response( + aiohttp_client: AiohttpClient, +) -> None: + """A raising trace handler must tear down the response from .content too.""" + body = b"x" * 4096 + + async def handler(request: web.Request) -> web.StreamResponse: + resp = web.StreamResponse() + resp.content_length = len(body) + await resp.prepare(request) + await resp.write(body) + return resp + + async def on_response_chunk_received( + session: object, + context: object, + params: tracing.TraceResponseChunkReceivedParams, + ) -> None: + raise RuntimeError("boom") + + trace_config = aiohttp.TraceConfig() + trace_config.on_response_chunk_received.append(on_response_chunk_received) + + app = web.Application() + app.router.add_get("/", handler) + client = await aiohttp_client(app, trace_configs=[trace_config]) + + async with client.get("/") as resp: + with pytest.raises(RuntimeError, match="boom"): + await resp.content.readany() + assert resp.closed + + async def test_request_tracing_url_params(aiohttp_client: AiohttpClient) -> None: async def root_handler(request: web.Request) -> web.Response: return web.Response() @@ -1175,7 +1276,8 @@ def to_url(path: str) -> URL: assert to_trace_urls(on_request_end) == [to_url("/?x=0")] assert to_trace_urls(on_request_exception) == [] assert to_trace_urls(on_request_chunk_sent) == [] - assert to_trace_urls(on_response_chunk_received) == [to_url("/?x=0")] + # Empty response body: no chunks are delivered to the caller. + assert to_trace_urls(on_response_chunk_received) == [] assert to_trace_urls(on_request_headers_sent) == [to_url("/?x=0")] # Redirect @@ -1191,7 +1293,7 @@ def to_url(path: str) -> URL: assert to_trace_urls(on_request_end) == [to_url("/")] assert to_trace_urls(on_request_exception) == [] assert to_trace_urls(on_request_chunk_sent) == [] - assert to_trace_urls(on_response_chunk_received) == [to_url("/")] + assert to_trace_urls(on_response_chunk_received) == [] assert to_trace_urls(on_request_headers_sent) == [ to_url("/redirect?x=0"), to_url("/"), diff --git a/tests/test_streams.py b/tests/test_streams.py index fce10a4ec14..eef6dfe242c 100644 --- a/tests/test_streams.py +++ b/tests/test_streams.py @@ -5,7 +5,7 @@ import gc import types from collections import defaultdict -from collections.abc import Iterator, Sequence +from collections.abc import Callable, Coroutine, Iterator, Sequence from itertools import groupby from typing import TypeVar from unittest import mock @@ -14,7 +14,7 @@ from aiohttp import streams from aiohttp.base_protocol import BaseProtocol -from aiohttp.helpers import DEFAULT_CHUNK_SIZE +from aiohttp.helpers import DEFAULT_CHUNK_SIZE, TimerContext from aiohttp.http_exceptions import LineTooLong DATA: bytes = b"line1\nline2\nline3\n" @@ -1104,6 +1104,210 @@ async def test_unread_empty(self) -> None: assert stream.at_eof() +class TestStreamReaderChunkHook: + """Cover the _on_chunk_received hook used to fan out chunks to client tracing.""" + + def _make_one( + self, cb: Callable[[bytes], Coroutine[None, None, None]] | None = None + ) -> tuple[streams.StreamReader, list[bytes]]: + protocol = mock.create_autospec( + BaseProtocol, spec_set=True, instance=True, _reading_paused=False + ) + stream = streams.StreamReader( + protocol, DEFAULT_CHUNK_SIZE, loop=asyncio.get_running_loop() + ) + seen: list[bytes] = [] + if cb is None: + + async def cb(chunk: bytes) -> None: + seen.append(chunk) + + stream._on_chunk_received = cb + return stream, seen + + async def test_readany_fires(self) -> None: + stream, seen = self._make_one() + + async def feed_in_two_steps() -> None: + stream.feed_data(b"abc") + await asyncio.sleep(0) + stream.feed_data(b"def") + + # Feed blocks one at a time, with readany() between, so each block + # is returned separately rather than drained together. + feeder = asyncio.create_task(feed_in_two_steps()) + assert await stream.readany() == b"abc" + assert await stream.readany() == b"def" + await feeder + stream.feed_eof() + assert await stream.readany() == b"" + assert seen == [b"abc", b"def"] + + async def test_read_n_fires(self) -> None: + stream, seen = self._make_one() + stream.feed_data(b"abcdef") + stream.feed_eof() + + assert await stream.read(3) == b"abc" + assert await stream.read(3) == b"def" + assert seen == [b"abc", b"def"] + + async def test_read_all_fires_via_readany(self) -> None: + # read(-1) loops through readany(); the hook fires per readany() call, + # not per fed block (pre-fed buffer is drained in one shot). + stream, seen = self._make_one() + stream.feed_data(b"abc") + stream.feed_data(b"def") + stream.feed_eof() + + assert await stream.read() == b"abcdef" + assert seen == [b"abcdef"] + + async def test_readchunk_data_fires(self) -> None: + stream, seen = self._make_one() + stream.feed_data(b"abc") + stream.feed_eof() + + data, end = await stream.readchunk() + assert (data, end) == (b"abc", False) + assert seen == [b"abc"] + + async def test_readchunk_http_chunk_eof_does_not_fire(self) -> None: + # After the final http chunk, readchunk returns (b"", False) at EOF; + # that empty terminator must not fire the hook. + stream, seen = self._make_one() + stream.begin_http_chunk_receiving() + stream.feed_data(b"abc") + stream.end_http_chunk_receiving() + stream.feed_eof() + + assert await stream.readchunk() == (b"abc", True) + assert await stream.readchunk() == (b"", False) + assert seen == [b"abc"] + + async def test_readuntil_fires(self) -> None: + stream, seen = self._make_one() + stream.feed_data(b"line1\nline2\n") + stream.feed_eof() + + assert await stream.readuntil() == b"line1\n" + assert await stream.readuntil() == b"line2\n" + assert seen == [b"line1\n", b"line2\n"] + + async def test_readline_fires_once(self) -> None: + # readline chains through readuntil — must not double-fire. + stream, seen = self._make_one() + stream.feed_data(b"line1\n") + stream.feed_eof() + + assert await stream.readline() == b"line1\n" + assert seen == [b"line1\n"] + + async def test_readexactly_fires_via_read(self) -> None: + # readexactly chains through read(n) — each underlying read fires once. + stream, seen = self._make_one() + stream.feed_data(b"abcdef") + stream.feed_eof() + + assert await stream.readexactly(6) == b"abcdef" + assert seen == [b"abcdef"] + + async def test_empty_buffer_eof_does_not_fire(self) -> None: + stream, seen = self._make_one() + stream.feed_eof() + + assert await stream.read() == b"" + assert await stream.readany() == b"" + assert await stream.readchunk() == (b"", False) + assert seen == [] + + async def test_iter_chunked_fires(self) -> None: + stream, seen = self._make_one() + stream.feed_data(b"abcdef") + stream.feed_eof() + + collected = [chunk async for chunk in stream.iter_chunked(3)] + assert b"".join(collected) == b"abcdef" + assert b"".join(seen) == b"abcdef" + + async def test_iter_any_fires(self) -> None: + # _read_nowait(-1) drains the buffer in a single call, so two + # pre-fed blocks come back as one chunk. + stream, seen = self._make_one() + stream.feed_data(b"abc") + stream.feed_data(b"def") + stream.feed_eof() + + collected = [chunk async for chunk in stream.iter_any()] + assert collected == [b"abcdef"] + assert seen == [b"abcdef"] + + async def test_iter_chunks_fires(self) -> None: + stream, seen = self._make_one() + stream.begin_http_chunk_receiving() + stream.feed_data(b"abc") + stream.end_http_chunk_receiving() + stream.begin_http_chunk_receiving() + stream.feed_data(b"def") + stream.end_http_chunk_receiving() + stream.feed_eof() + + collected = [chunk async for chunk, _ in stream.iter_chunks()] + assert b"".join(collected) == b"abcdef" + assert seen == [b"abc", b"def"] + + async def test_hook_runs_under_stream_timer(self) -> None: + # The per-stream timer (driven by sock_read) must bound the trace + # call, not just the wait for incoming bytes. + loop = asyncio.get_running_loop() + timer = TimerContext(loop) + protocol = mock.create_autospec( + BaseProtocol, spec_set=True, instance=True, _reading_paused=False + ) + stream = streams.StreamReader( + protocol, DEFAULT_CHUNK_SIZE, timer=timer, loop=loop + ) + + async def cb(chunk: bytes) -> None: + await asyncio.sleep(60) # simulate a hung exporter + + stream._on_chunk_received = cb + stream.feed_data(b"abc") + stream.feed_eof() + + async def fire_timeout() -> None: + await asyncio.sleep(0) + timer.timeout() + + timeout_task = asyncio.create_task(fire_timeout()) + with pytest.raises(asyncio.TimeoutError): + await stream.readany() + await timeout_task + + async def test_read_nowait_fires_via_task(self) -> None: + # read_nowait is sync; the hook is async. The fire is scheduled, + # so it lands on the next event loop tick — not before read_nowait + # has returned. + stream, seen = self._make_one() + stream.feed_data(b"abc") + + assert stream.read_nowait() == b"abc" + assert seen == [] + await asyncio.sleep(0) + assert seen == [b"abc"] + + async def test_hook_exception_propagates(self) -> None: + async def cb(chunk: bytes) -> None: + raise RuntimeError("boom") + + stream, _ = self._make_one(cb) + stream.feed_data(b"abc") + stream.feed_eof() + + with pytest.raises(RuntimeError, match="boom"): + await stream.readany() + + async def test_empty_stream_reader() -> None: s = streams.EmptyStreamReader() assert str(s) is not None @@ -1134,6 +1338,20 @@ async def test_empty_stream_reader_iter_chunks() -> None: await iter_chunks.__anext__() +async def test_empty_stream_reader_on_chunk_received_is_read_only() -> None: + # EMPTY_PAYLOAD is a module-level singleton; allowing per-response hooks + # to be installed on it would leak across requests. + assert streams.EMPTY_PAYLOAD._on_chunk_received is None + + async def cb(chunk: bytes) -> None: + assert False + + with pytest.raises(AttributeError, match="read-only"): + streams.EMPTY_PAYLOAD._on_chunk_received = cb + # Singleton state untouched after the failed assignment. + assert streams.EMPTY_PAYLOAD._on_chunk_received is None + + @pytest.fixture async def buffer() -> streams.DataQueue[bytes]: return streams.DataQueue(asyncio.get_running_loop()) From 20cde095cf3e8f7fab2926a1bbbbe8e6ad189381 Mon Sep 17 00:00:00 2001 From: Sam Bull Date: Tue, 7 Jul 2026 21:37:19 +0100 Subject: [PATCH 3/5] Fix empty domain issue in DigestAuthMiddleware (#12983) --- CHANGES/12983.bugfix.rst | 1 + aiohttp/client_middleware_digest_auth.py | 8 ++- tests/test_client_middleware_digest_auth.py | 68 +++++++++++++++++++++ 3 files changed, 74 insertions(+), 3 deletions(-) create mode 100644 CHANGES/12983.bugfix.rst diff --git a/CHANGES/12983.bugfix.rst b/CHANGES/12983.bugfix.rst new file mode 100644 index 00000000000..03f4083d83a --- /dev/null +++ b/CHANGES/12983.bugfix.rst @@ -0,0 +1 @@ +Fixed ``DigestAuthMiddleware`` raising an ``IndexError`` on empty domain -- by :user:`Dreamsorcerer`. diff --git a/aiohttp/client_middleware_digest_auth.py b/aiohttp/client_middleware_digest_auth.py index 6c3e37f7c00..5da92fde807 100644 --- a/aiohttp/client_middleware_digest_auth.py +++ b/aiohttp/client_middleware_digest_auth.py @@ -434,21 +434,23 @@ def _authenticate(self, response: ClientResponse) -> bool: # Update protection space based on domain parameter or default to origin origin = response.url.origin() + self._protection_space = [] if domain := self._challenge.get("domain"): # Parse space-separated list of URIs - self._protection_space = [] for uri in domain.split(): # Remove quotes if present uri = uri.strip('"') + if not uri: + continue if uri.startswith("/"): # Path-absolute, relative to origin self._protection_space.append(str(origin.join(URL(uri)))) else: # Absolute URI self._protection_space.append(str(URL(uri))) - else: - # No domain specified, protection space is entire origin + + if not self._protection_space: self._protection_space = [str(origin)] # Return True only if we found at least one challenge parameter diff --git a/tests/test_client_middleware_digest_auth.py b/tests/test_client_middleware_digest_auth.py index ca1a9267b67..c5894f04504 100644 --- a/tests/test_client_middleware_digest_auth.py +++ b/tests/test_client_middleware_digest_auth.py @@ -1497,6 +1497,74 @@ def test_in_protection_space_multiple_spaces( assert digest_auth_mw._in_protection_space(URL("http://example.com/other")) is False +@pytest.mark.parametrize( + "domain_value", + (r'"\""', '"'), + ids=("quoted_escaped_quote", "bare_quote"), +) +def test_authenticate_domain_only_quote_does_not_poison_protection_space( + digest_auth_mw: DigestAuthMiddleware, + domain_value: str, +) -> None: + response = mock.create_autospec(ClientResponse, spec_set=True, instance=True) + response.status = 401 + response.url = URL("http://example.com/resource") + response.headers = { + "www-authenticate": f'Digest realm="test", nonce="abc", domain={domain_value}' + } + + assert digest_auth_mw._authenticate(response) is True + assert digest_auth_mw._challenge["domain"] == '"' + assert digest_auth_mw._protection_space == ["http://example.com"] + assert "" not in digest_auth_mw._protection_space + # Must not raise IndexError and must still scope to the anchor origin. + assert digest_auth_mw._in_protection_space(URL("http://example.com/other")) is True + assert digest_auth_mw._in_protection_space(URL("http://other.com/x")) is False + + +async def test_double_quote_domain_does_not_break_future_requests( + aiohttp_server: AiohttpServer, +) -> None: + """End-to-end regression for a ``domain`` directive of just a double quote. + + The first request triggers the challenge and authenticates on retry. Before + the fix, the bogus ``domain`` left an empty string in the protection space, + so the next request's preemptive-auth check raised ``IndexError``. + """ + digest_auth_mw = DigestAuthMiddleware("user", "pass", preemptive=True) + auth_headers: list[str | None] = [] + + async def handler(request: Request) -> Response: + auth_headers.append(request.headers.get(hdrs.AUTHORIZATION)) + if request.headers.get(hdrs.AUTHORIZATION) is None: + challenge = ( + 'Digest realm="test", nonce="abc123", qop="auth", ' + 'algorithm=MD5, domain="\\""' + ) + return Response( + status=401, + headers={"WWW-Authenticate": challenge}, + text="Unauthorized", + ) + return Response(text="OK") + + app = Application() + app.router.add_get("/path1", handler) + app.router.add_get("/path2", handler) + server = await aiohttp_server(app) + + async with ClientSession(middlewares=(digest_auth_mw,)) as session: + async with session.get(server.make_url("/path1")) as resp: + assert resp.status == 200 + # Previously raised IndexError inside the preemptive-auth check. + async with session.get(server.make_url("/path2")) as resp: + assert resp.status == 200 + + assert auth_headers[0] is None # First request: no auth, gets challenge + assert auth_headers[1] is not None # Retry carries the digest response + assert auth_headers[2] is not None # Second request: preemptive auth + + async def test_case_sensitive_algorithm_server( aiohttp_server: AiohttpServer, ) -> None: From 7d7892044c6fe663a566e78c58e5fd25ac6d57b1 Mon Sep 17 00:00:00 2001 From: Sam Bull Date: Tue, 7 Jul 2026 21:38:46 +0100 Subject: [PATCH 4/5] Fix Digest challenge with multiple schemes (#12984) --- CHANGES/12984.bugfix.rst | 3 + aiohttp/client_middleware_digest_auth.py | 63 ++++++++++++--------- tests/test_client_middleware_digest_auth.py | 31 +++++++++- 3 files changed, 69 insertions(+), 28 deletions(-) create mode 100644 CHANGES/12984.bugfix.rst diff --git a/CHANGES/12984.bugfix.rst b/CHANGES/12984.bugfix.rst new file mode 100644 index 00000000000..a156216397b --- /dev/null +++ b/CHANGES/12984.bugfix.rst @@ -0,0 +1,3 @@ +Fixed :class:`~aiohttp.DigestAuthMiddleware` corrupting the ``Digest`` +challenge when a ``WWW-Authenticate`` response offered more than one +authentication scheme -- by :user:`Dreamsorcerer`. diff --git a/aiohttp/client_middleware_digest_auth.py b/aiohttp/client_middleware_digest_auth.py index 5da92fde807..f4904e363d8 100644 --- a/aiohttp/client_middleware_digest_auth.py +++ b/aiohttp/client_middleware_digest_auth.py @@ -52,27 +52,23 @@ class DigestAuthChallenge(TypedDict, total=False): # Compile the regex pattern once at module level for performance _HEADER_PAIRS_PATTERN = re.compile( - r'(?:^|\s|,\s*)(\w+)\s*=\s*(?:"((?:[^"\\]|\\.)*)"|([^\s,]+))' + r'(?:^|\s|,\s*)(\w+)(?:\s*=\s*(?:"((?:[^"\\]|\\.)*)"|([^\s,]+)))?' if sys.version_info < (3, 11) - else r'(?:^|\s|,\s*)((?>\w+))\s*=\s*(?:"((?:[^"\\]|\\.)*)"|([^\s,]+))' - # +------------|--------|--|-|-|--|----|------|----|--||-----|-> Match valid start/sep - # +--------|--|-|-|--|----|------|----|--||-----|-> alphanumeric key (atomic - # | | | | | | | | || | group reduces backtracking) - # +--|-|-|--|----|------|----|--||-----|-> maybe whitespace - # | | | | | | | || | - # +-|-|--|----|------|----|--||-----|-> = (delimiter) - # +-|--|----|------|----|--||-----|-> maybe whitespace - # | | | | | || | - # +--|----|------|----|--||-----|-> group quoted or unquoted - # | | | | || | - # +----|------|----|--||-----|-> if quoted... - # +------|----|--||-----|-> anything but " or \ - # +----|--||-----|-> escaped characters allowed - # +--||-----|-> or can be empty string - # || | - # +|-----|-> if unquoted... - # +-----|-> anything but , or - # +-> at least one char req'd + else r'(?:^|\s|,\s*)((?>\w+))(?:\s*=\s*(?:"((?:[^"\\]|\\.)*)"|([^\s,]+)))?' + # +------------|--------|--|--||--|--|----|------|---|---||-----|-> Match valid start/sep + # +--------|--|--||--|--|----|------|---|---||-----|-> alphanumeric key (atomic group reduces backtracking) + # +--|--||--|--|----|------|---|---||-----|-> optional value; absent => bare auth-scheme token + # +--||--|--|----|------|---|---||-----|-> maybe whitespace + # +|--|--|----|------|---|---||-----|-> = (delimiter) + # +--|--|----|------|---|---||-----|-> maybe whitespace + # +--|----|------|---|---||-----|-> group quoted or unquoted + # +----|------|---|---||-----|-> if quoted... + # +------|---|---||-----|-> anything but " or \ + # +---|---||-----|-> escaped characters allowed + # +---||-----|-> or can be empty string + # +|-----|-> if unquoted... + # +-----|-> anything but , or + # +-> at least one char req'd ) @@ -117,12 +113,17 @@ def unescape_quotes(value: str) -> str: def parse_header_pairs(header: str) -> dict[str, str]: """ - Parse key-value pairs from WWW-Authenticate or similar HTTP headers. + Parse key-value pairs from the first challenge of a WWW-Authenticate header. This function handles the complex format of WWW-Authenticate header values, supporting both quoted and unquoted values, proper handling of commas in quoted values, and whitespace variations per RFC 7616. + A single header may carry several challenges + (https://www.rfc-editor.org/rfc/rfc7235#section-4.1). Parsing + stops at the next auth-scheme token so a later challenge's parameters cannot + overwrite the first challenge's values; a leading scheme token is skipped. + Examples of supported formats: - key1="value1", key2=value2 - key1 = "value1" , key2="value, with, commas" @@ -135,11 +136,21 @@ def parse_header_pairs(header: str) -> dict[str, str]: Returns: Dictionary mapping parameter names to their values """ - return { - stripped_key: unescape_quotes(quoted_val) if quoted_val else unquoted_val - for key, quoted_val, unquoted_val in _HEADER_PAIRS_PATTERN.findall(header) - if (stripped_key := key.strip()) - } + pairs: dict[str, str] = {} + for match in _HEADER_PAIRS_PATTERN.finditer(header): + key = match.group(1) + quoted_val, unquoted_val = match.group(2), match.group(3) + if quoted_val is None and unquoted_val is None: + # Bare token with no "=value": an auth-scheme name, not a parameter. + # Skip a leading scheme; once parameters exist, a new scheme marks + # the start of the next challenge, so stop here. + if pairs: + break + continue + pairs[key] = ( + unescape_quotes(quoted_val) if quoted_val is not None else unquoted_val + ) + return pairs class DigestAuthMiddleware: diff --git a/tests/test_client_middleware_digest_auth.py b/tests/test_client_middleware_digest_auth.py index c5894f04504..fb0cd0ec876 100644 --- a/tests/test_client_middleware_digest_auth.py +++ b/tests/test_client_middleware_digest_auth.py @@ -121,6 +121,18 @@ def mock_md5_digest() -> Generator[mock.MagicMock, None, None]: True, {"realm": "", "nonce": "abc", "qop": "auth"}, ), + # Multi-scheme header: a second challenge (Basic) in the same + # WWW-Authenticate value must not overwrite the Digest realm/nonce. + # https://www.rfc-editor.org/rfc/rfc7235#section-4.1 + ( + 401, + { + "www-authenticate": 'Digest realm="protected", nonce="n1", ' + 'qop="auth", Basic realm="other"' + }, + True, + {"realm": "protected", "nonce": "n1", "qop": "auth"}, + ), # Non-401 status (200, {}, False, {}), # No challenge should be set ], @@ -469,6 +481,18 @@ async def test_digest_response_exact_match( ), # Empty header ("", {}), + # Multi-scheme header: parsing stops at the next auth-scheme so a later + # challenge cannot overwrite the first's values. + # https://www.rfc-editor.org/rfc/rfc7235#section-4.1 + ( + 'realm="protected", nonce="n1", qop="auth", Basic realm="other"', + {"realm": "protected", "nonce": "n1", "qop": "auth"}, + ), + # Multi-scheme header including the leading scheme token + ( + 'Digest realm="protected", nonce="n1", Basic realm="other"', + {"realm": "protected", "nonce": "n1"}, + ), ], ids=[ "fully_quoted_header", @@ -480,6 +504,8 @@ async def test_digest_response_exact_match( "escaped_quotes", "single_quotes_as_regular_chars", "empty_header", + "multi_scheme_second_challenge_ignored", + "multi_scheme_with_leading_scheme", ], ) def test_parse_header_pairs(header: str, expected_result: dict[str, str]) -> None: @@ -1631,5 +1657,6 @@ def test_regex_performance() -> None: f"Regex took {elapsed * 1000:.1f}ms, " f"expected <{REGEX_TIME_THRESHOLD_SECONDS * 1000:.0f}ms - potential ReDoS issue" ) - # This example shouldn't produce a match either. - assert not matches + # The lone run of word characters matches as a bare auth-scheme token + # with empty value groups, so it never becomes a key=value pair. + assert all(not quoted and not unquoted for _, quoted, unquoted in matches) From b9419c2f1ebc0678a8311018af107e446616cb9b Mon Sep 17 00:00:00 2001 From: Sam Bull Date: Tue, 7 Jul 2026 22:43:49 +0100 Subject: [PATCH 5/5] Ensure all requests go through middleware (#12925) --- CHANGES/3287.breaking.rst | 3 + CHANGES/3287.feature.rst | 1 + aiohttp/web_app.py | 8 +- aiohttp/web_middlewares.py | 12 ++- aiohttp/web_protocol.py | 63 +++++++------- aiohttp/web_request.py | 31 ++++++- aiohttp/web_runner.py | 3 + aiohttp/web_server.py | 15 +++- docs/web_advanced.rst | 42 +++++++++ docs/web_reference.rst | 16 ++++ tests/test_web_app.py | 163 ++++++++++++++++++++++++++++++++++- tests/test_web_middleware.py | 52 ++++++++++- tests/test_web_request.py | 7 ++ tests/test_web_server.py | 124 ++++++-------------------- 14 files changed, 394 insertions(+), 146 deletions(-) create mode 100644 CHANGES/3287.breaking.rst create mode 100644 CHANGES/3287.feature.rst diff --git a/CHANGES/3287.breaking.rst b/CHANGES/3287.breaking.rst new file mode 100644 index 00000000000..ddfcf49d4a9 --- /dev/null +++ b/CHANGES/3287.breaking.rst @@ -0,0 +1,3 @@ +The ``request_factory`` (usually provided by aiohttp) callable must now accept ``pre_handler_error``. + +Users of the low-level ``aiohttp.Server`` (without an ``Application``) now receive malformed requests via their handler with ``BaseRequest.pre_handler_error`` set to an ``HTTPBadRequest`` and must emit the error response themselves; ``Application`` users are unaffected -- by :user:`Dreamsorcerer`. diff --git a/CHANGES/3287.feature.rst b/CHANGES/3287.feature.rst new file mode 100644 index 00000000000..d7c4e98a19f --- /dev/null +++ b/CHANGES/3287.feature.rst @@ -0,0 +1 @@ +Bad request responses caused by parser errors now pass through the application's middleware chain, allowing middleware (and ``on_response_prepare`` handlers) to modify them -- by :user:`Dreamsorcerer`. diff --git a/aiohttp/web_app.py b/aiohttp/web_app.py index 1ff6e48bac2..5a10bc25d96 100644 --- a/aiohttp/web_app.py +++ b/aiohttp/web_app.py @@ -33,9 +33,11 @@ Domain, MaskDomain, MatchedSubAppResource, + MatchInfoError, PrefixedSubAppResource, SystemRoute, UrlDispatcher, + UrlMappingMatchInfo, ) __all__ = ("Application", "CleanupError") @@ -364,7 +366,11 @@ def _prepare_middleware(self) -> Iterator[Middleware]: yield _fix_request_current_app(self) async def _handle(self, request: Request) -> StreamResponse: - match_info = await self._router.resolve(request) + match_info: UrlMappingMatchInfo + if (err := request._pre_handler_error) is not None: + match_info = MatchInfoError(err) + else: + match_info = await self._router.resolve(request) match_info.add_app(self) match_info.freeze() diff --git a/aiohttp/web_middlewares.py b/aiohttp/web_middlewares.py index 5e26fe354cb..cf6f482250f 100644 --- a/aiohttp/web_middlewares.py +++ b/aiohttp/web_middlewares.py @@ -3,10 +3,14 @@ from typing import TYPE_CHECKING, TypeVar from .typedefs import Handler, Middleware -from .web_exceptions import HTTPMove, HTTPPermanentRedirect +from .web_exceptions import ( + HTTPMethodNotAllowed, + HTTPMove, + HTTPNotFound, + HTTPPermanentRedirect, +) from .web_request import Request from .web_response import StreamResponse -from .web_urldispatcher import SystemRoute __all__ = ( "middleware", @@ -81,7 +85,9 @@ def normalize_path_middleware( assert correct_configuration, "Cannot both remove and append slash" async def impl(request: Request, handler: Handler) -> StreamResponse: - if isinstance(request.match_info.route, SystemRoute): + if isinstance( + request.match_info.http_exception, (HTTPNotFound, HTTPMethodNotAllowed) + ): paths_to_check = [] if "?" in request.raw_path: path, query = request.raw_path.split("?", 1) diff --git a/aiohttp/web_protocol.py b/aiohttp/web_protocol.py index 3a5be452090..21bc93811db 100644 --- a/aiohttp/web_protocol.py +++ b/aiohttp/web_protocol.py @@ -11,11 +11,17 @@ from typing import TYPE_CHECKING, Any, Generic, Optional, TypeVar, Union, cast import yarl +from multidict import CIMultiDict from propcache import under_cached_property from .abc import AbstractAccessLogger, AbstractAsyncAccessLogger, AbstractStreamWriter from .base_protocol import PAUSE_RESUME_READING_ERRORS, BaseProtocol -from .helpers import DEFAULT_CHUNK_SIZE, ceil_timeout, frozen_dataclass_decorator +from .helpers import ( + DEFAULT_CHUNK_SIZE, + HeadersDictProxy, + ceil_timeout, + frozen_dataclass_decorator, +) from .http import ( HttpProcessingError, HttpRequestParser, @@ -24,11 +30,10 @@ StreamWriter, WebSocketReader, ) -from .http_exceptions import BadHttpMethod from .log import access_logger, server_logger from .streams import EMPTY_PAYLOAD, StreamReader from .tcp_helpers import tcp_keepalive -from .web_exceptions import HTTPException, HTTPInternalServerError +from .web_exceptions import HTTPBadRequest, HTTPException, HTTPInternalServerError from .web_log import AccessLogger from .web_request import BaseRequest from .web_response import Response, StreamResponse @@ -54,6 +59,7 @@ "RequestHandler[_Request]", AbstractStreamWriter, "asyncio.Task[None]", + HTTPBadRequest | None, ], _Request, ] @@ -68,8 +74,8 @@ "UNKNOWN", "/", HttpVersion10, - {}, # type: ignore[arg-type] - {}, # type: ignore[arg-type] + HeadersDictProxy(CIMultiDict()), + (), True, None, False, @@ -234,7 +240,6 @@ def __init__( ) super().__init__(loop, parser) - # _request_count is the number of requests processed with the same connection. self._request_count = 0 self._keepalive = False self._current_request: _Request | None = None @@ -609,6 +614,13 @@ async def _handle_request( finally: self._current_request = None except HTTPException as exc: + # Uncaught parser error + if request._pre_handler_error is exc: + self.logger.warning( + "Error handling request from %s", + request.remote, + exc_info=exc.__cause__, + ) resp = Response( status=exc.status, reason=exc.reason, text=exc.text, headers=exc.headers ) @@ -678,11 +690,12 @@ async def start(self) -> None: manager.requests_count += 1 writer = StreamWriter(self, loop) - if not isinstance(message, _ErrInfo): - request_handler = self._request_handler - else: - # make request_factory work - request_handler = self._make_error_handler(message) + pre_handler_error: HTTPBadRequest | None = None + if isinstance(message, _ErrInfo): + pre_handler_error = HTTPBadRequest( + text=message.message, content_type="text/plain" + ) + pre_handler_error.__cause__ = message.exc message = ERROR # Important don't hold a reference to the current task @@ -694,10 +707,11 @@ async def start(self) -> None: self, writer, self._task_handler or asyncio.current_task(loop), # type: ignore[arg-type] + pre_handler_error, ) try: # a new task is used for copy context vars (#3406) - coro = self._handle_request(request, start, request_handler) + coro = self._handle_request(request, start, self._request_handler) if sys.version_info >= (3, 12): task = asyncio.Task(coro, loop=loop, eager_start=True) else: @@ -841,18 +855,9 @@ def handle_error( Returns HTTP response with specific status code. Logs additional information. It always closes current connection. """ - if self._request_count == 1 and isinstance(exc, BadHttpMethod): - # BadHttpMethod is common when a client sends non-HTTP - # or encrypted traffic to an HTTP port. This is expected - # to happen when connected to the public internet so we log - # it at the debug level as to not fill logs with noise. - self.logger.debug( - "Error handling request from %s", request.remote, exc_info=exc - ) - else: - self.log_exception( - "Error handling request from %s", request.remote, exc_info=exc - ) + self.log_exception( + "Error handling request from %s", request.remote, exc_info=exc + ) # some data already got sent, connection is broken if request.writer.output_size > 0: @@ -890,13 +895,3 @@ def handle_error( resp.force_close() return resp - - def _make_error_handler( - self, err_info: _ErrInfo - ) -> Callable[[BaseRequest], Awaitable[StreamResponse]]: - async def handler(request: BaseRequest) -> StreamResponse: - return self.handle_error( - request, err_info.status, err_info.exc, err_info.message - ) - - return handler diff --git a/aiohttp/web_request.py b/aiohttp/web_request.py index 745681e699d..642f4e47387 100644 --- a/aiohttp/web_request.py +++ b/aiohttp/web_request.py @@ -9,7 +9,16 @@ from collections.abc import Iterator, Mapping, MutableMapping from re import Pattern from types import MappingProxyType -from typing import TYPE_CHECKING, Any, Final, Optional, TypeVar, cast, overload +from typing import ( + TYPE_CHECKING, + Any, + Final, + Optional, + TypedDict, + TypeVar, + cast, + overload, +) from urllib.parse import parse_qsl from multidict import CIMultiDict, MultiDict, MultiDictProxy @@ -70,6 +79,12 @@ _T = TypeVar("_T") +class _CloneKwargs(TypedDict, total=False): + scheme: str + host: str + remote: str + + @frozen_dataclass_decorator class FileField: name: str @@ -112,6 +127,7 @@ class BaseRequest(MutableMapping[str | RequestKey[Any], Any], HeadersMixin): _post: MultiDictProxy[str | bytes | FileField] | None = None _read_bytes: bytes | None = None + _pre_handler_error: HTTPBadRequest | None = None def __init__( self, @@ -127,10 +143,13 @@ def __init__( scheme: str | None = None, host: str | None = None, remote: str | None = None, + pre_handler_error: HTTPBadRequest | None = None, ) -> None: self._message = message self._protocol = protocol self._payload_writer = payload_writer + if pre_handler_error is not None: + self._pre_handler_error = pre_handler_error self._payload = payload self._headers: HeadersDictProxy = message.headers @@ -207,7 +226,7 @@ def clone( message = self._message._replace(**dct) - kwargs: dict[str, str] = {} + kwargs: _CloneKwargs = {} if scheme is not sentinel: kwargs["scheme"] = scheme if host is not sentinel: @@ -226,6 +245,7 @@ def clone( self._loop, client_max_size=client_max_size, state=self._state.copy(), + pre_handler_error=self._pre_handler_error, **kwargs, ) @@ -249,6 +269,10 @@ def writer(self) -> AbstractStreamWriter: def client_max_size(self) -> int: return self._client_max_size + @property + def pre_handler_error(self) -> HTTPBadRequest | None: + return self._pre_handler_error + @reify def rel_url(self) -> URL: return self._rel_url @@ -868,8 +892,7 @@ def config_dict(self) -> ChainMapProxy: async def _prepare_hook(self, response: StreamResponse) -> None: match_info = self._match_info - if match_info is None: - return + assert match_info is not None for app in match_info._apps: if on_response_prepare := app.on_response_prepare: await on_response_prepare.send(self, response) diff --git a/aiohttp/web_runner.py b/aiohttp/web_runner.py index 82c3bd277f8..d0262793763 100644 --- a/aiohttp/web_runner.py +++ b/aiohttp/web_runner.py @@ -11,6 +11,7 @@ from .streams import StreamReader from .typedefs import PathLike from .web_app import Application +from .web_exceptions import HTTPBadRequest from .web_log import AccessLogger from .web_protocol import RequestHandler from .web_request import BaseRequest, Request @@ -436,6 +437,7 @@ def _make_request( protocol: RequestHandler[Request], writer: AbstractStreamWriter, task: "asyncio.Task[None]", + pre_handler_error: HTTPBadRequest | None, _cls: type[Request] = Request, ) -> Request: loop = asyncio.get_running_loop() @@ -447,6 +449,7 @@ def _make_request( task, loop, client_max_size=self.app._client_max_size, + pre_handler_error=pre_handler_error, ) async def _cleanup_server(self) -> None: diff --git a/aiohttp/web_server.py b/aiohttp/web_server.py index e4decdf0f6f..66a65e0ceb8 100644 --- a/aiohttp/web_server.py +++ b/aiohttp/web_server.py @@ -8,6 +8,7 @@ from .abc import AbstractStreamWriter from .http_parser import RawRequestMessage from .streams import StreamReader +from .web_exceptions import HTTPBadRequest from .web_protocol import RequestHandler from .web_request import BaseRequest from .web_response import StreamResponse @@ -22,6 +23,7 @@ "RequestHandler[_Request]", AbstractStreamWriter, "asyncio.Task[None]", + HTTPBadRequest | None, ], _Request, ] @@ -67,8 +69,6 @@ def __init__( self._loop = asyncio.get_running_loop() self._connections: dict[RequestHandler[_Request], asyncio.Transport] = {} self._kwargs = kwargs - # requests_count is the number of requests being processed by the server - # for the lifetime of the server. self.requests_count = 0 self.request_handler = handler self.request_factory = request_factory or self._make_request # type: ignore[assignment] @@ -101,8 +101,17 @@ def _make_request( protocol: RequestHandler[BaseRequest], writer: AbstractStreamWriter, task: "asyncio.Task[None]", + pre_handler_error: HTTPBadRequest | None, ) -> BaseRequest: - return BaseRequest(message, payload, protocol, writer, task, self._loop) + return BaseRequest( + message, + payload, + protocol, + writer, + task, + self._loop, + pre_handler_error=pre_handler_error, + ) def pre_shutdown(self) -> None: for conn in self._connections: diff --git a/docs/web_advanced.rst b/docs/web_advanced.rst index 44ceed5f1c8..0d7e82b3215 100644 --- a/docs/web_advanced.rst +++ b/docs/web_advanced.rst @@ -665,6 +665,48 @@ Produced output:: Middleware 2 finished Middleware 1 finished +Middlewares also run for responses synthesized from malformed HTTP +requests (HTTP 400). For such a request +:attr:`~aiohttp.web.BaseRequest.pre_handler_error` is set to the +:exc:`HTTPBadRequest` that the handler raises, and its ``__cause__`` +holds the underlying parser exception. As long as an exception doesn't +propagate out of a middleware, the middleware can modify -- or replace +-- every response the server sends. + +.. versionchanged:: 4.0 + + Malformed requests are now dispatched through the middleware chain. + +A middleware can use this to change how malformed requests are +logged. By default aiohttp logs every parser error at the ``warning`` +level; the middleware below changes it to a ``debug`` level:: + + async def log_debug_parser_errors_middleware( + request: web.Request, + handler: Callable[[web.Request], Awaitable[web.StreamResponse]], + ) -> web.StreamResponse: + try: + return await handler(request) + except web.HTTPException as exc: + if request.pre_handler_error is None: + raise # A normal HTTP error from the app, not a parser error. + + request.protocol.logger.debug( + "Error handling request from %s", + # Or for reverse proxy with trusted header: + # request.headers.get("X-Forwarded-For") + request.remote, + exc_info=exc.__cause__, + ) + # Returning a response (rather than re-raising) suppresses + # aiohttp's own error logging with the call above. + return web.Response( + status=exc.status, + reason=exc.reason, + text=exc.text, + headers=exc.headers, + ) + Request Body Stream Consumption ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ diff --git a/docs/web_reference.rst b/docs/web_reference.rst index d4a4f891557..6b6d6aa06c3 100644 --- a/docs/web_reference.rst +++ b/docs/web_reference.rst @@ -167,6 +167,22 @@ and :ref:`aiohttp-web-signals` handlers. Read-only :class:`int` property. + .. attribute:: pre_handler_error + + An :exc:`HTTPBadRequest` set by the protocol when the parser + could not parse the incoming bytes, otherwise ``None``. The + original parser exception is available as ``__cause__``. + + Users of the low-level :class:`aiohttp.web.Server` (without an + :class:`Application`) must inspect this attribute and emit + the error response themselves. :class:`Application` users + receive the error via the middleware chain and do not need + to read this directly. + + Read-only :exc:`HTTPBadRequest` or ``None`` property. + + .. versionadded:: 4.0 + .. attribute:: path_qs The URL including PATH_INFO and the query string. e.g., diff --git a/tests/test_web_app.py b/tests/test_web_app.py index de0e0cb2aac..3012c6b0157 100644 --- a/tests/test_web_app.py +++ b/tests/test_web_app.py @@ -1,4 +1,5 @@ import asyncio +import logging import sys from collections.abc import AsyncIterator, Callable, Iterator from contextlib import asynccontextmanager @@ -6,9 +7,10 @@ from unittest import mock import pytest -from pytest_aiohttp import AiohttpClient +from pytest_aiohttp import AiohttpClient, AiohttpServer -from aiohttp import log, web +from aiohttp import hdrs, log, web +from aiohttp.http_exceptions import HttpProcessingError from aiohttp.typedefs import Handler @@ -649,3 +651,160 @@ def test_forbid_changing_frozen_app() -> None: def test_app_boolean() -> None: app = web.Application() assert app + + +async def _read_response( + reader: asyncio.StreamReader, +) -> tuple[bytes, dict[str, str], bytes]: + head = b"" + while b"\r\n\r\n" not in head: + chunk = await reader.read(4096) + assert chunk, "connection closed before headers complete" + head += chunk + header_block, _, body = head.partition(b"\r\n\r\n") + status_line, *header_lines = header_block.split(b"\r\n") + headers = {} + for line in header_lines: + name, _, value = line.partition(b":") + headers[name.decode().strip().lower()] = value.decode().strip() + return status_line, headers, body + + +@asynccontextmanager +async def _raw_request( + host: str, port: int, payload: bytes +) -> AsyncIterator[asyncio.StreamReader]: + reader, writer = await asyncio.open_connection(host, port) + try: + writer.write(payload) + await writer.drain() + yield reader + finally: + writer.close() + await writer.wait_closed() + + +async def test_parser_error_passes_through_middleware( + aiohttp_server: AiohttpServer, +) -> None: + async def server_header( + request: web.Request, handler: Handler + ) -> web.StreamResponse: + try: + return await handler(request) + except web.HTTPException as exc: + exc.headers[hdrs.SERVER] = "custom" + raise + + app = web.Application(middlewares=[server_header]) + server = await aiohttp_server(app) + + async with _raw_request(server.host, server.port, b"garbage\r\n\r\n") as reader: + status_line, headers, _ = await _read_response(reader) + assert await reader.read() == b"" + + assert status_line.startswith(b"HTTP/1.0 400 ") + assert headers["server"] == "custom" + + +async def test_parser_error_middleware_can_catch( + aiohttp_server: AiohttpServer, +) -> None: + captured: list[BaseException | None] = [] + + async def catcher(request: web.Request, handler: Handler) -> web.StreamResponse: + try: + return await handler(request) + except web.HTTPBadRequest as exc: + captured.append(exc.__cause__) + return web.Response(status=418, text="teapot") + + app = web.Application(middlewares=[catcher]) + server = await aiohttp_server(app) + + async with _raw_request(server.host, server.port, b"garbage\r\n\r\n") as reader: + status_line, _, body = await _read_response(reader) + body += await reader.read() + + assert status_line.startswith(b"HTTP/1.0 418 ") + assert body.endswith(b"teapot") + assert len(captured) == 1 + assert isinstance(captured[0], HttpProcessingError) + + +async def test_parser_error_match_info_exposes_http_exception( + aiohttp_server: AiohttpServer, +) -> None: + seen: list[web.HTTPException | None] = [] + + async def inspector(request: web.Request, handler: Handler) -> web.StreamResponse: + seen.append(request.match_info.http_exception) + return await handler(request) + + app = web.Application(middlewares=[inspector]) + server = await aiohttp_server(app) + + async with _raw_request(server.host, server.port, b"garbage\r\n\r\n") as reader: + await _read_response(reader) + + assert len(seen) == 1 + assert isinstance(seen[0], web.HTTPBadRequest) + + +async def test_parser_error_middleware_suppresses_log( + aiohttp_server: AiohttpServer, +) -> None: + async def swallow(request: web.Request, handler: Handler) -> web.StreamResponse: + try: + return await handler(request) + except web.HTTPBadRequest: + return web.Response(status=204) + + logger = mock.create_autospec(logging.Logger, spec_set=True, instance=True) + app = web.Application(middlewares=[swallow]) + server = await aiohttp_server(app, logger=logger) + + async with _raw_request(server.host, server.port, b"garbage\r\n\r\n") as reader: + status_line, _, _ = await _read_response(reader) + + assert status_line.startswith(b"HTTP/1.0 204 ") + logger.exception.assert_not_called() + logger.debug.assert_not_called() + logger.warning.assert_not_called() + + +async def test_parser_error_propagated_is_logged( + aiohttp_server: AiohttpServer, +) -> None: + logger = mock.create_autospec(logging.Logger, spec_set=True, instance=True) + app = web.Application() + server = await aiohttp_server(app, logger=logger) + + async with _raw_request(server.host, server.port, b"garbage\r\n\r\n") as reader: + await _read_response(reader) + + # garbage triggers BadHttpMethod; logged at warning level by default. + assert logger.warning.called + assert "Error handling request" in logger.warning.call_args.args[0] + + +async def test_parser_error_on_response_prepare_fires( + aiohttp_server: AiohttpServer, +) -> None: + fired = False + + async def hook(request: web.Request, response: web.StreamResponse) -> None: + nonlocal fired + fired = True + response.headers[hdrs.SERVER] = "from-signal" + + app = web.Application() + app.on_response_prepare.append(hook) + server = await aiohttp_server(app) + + async with _raw_request(server.host, server.port, b"garbage\r\n\r\n") as reader: + status_line, headers, _ = await _read_response(reader) + + assert status_line.startswith(b"HTTP/1.0 400 ") + assert fired + assert headers["server"] == "from-signal" diff --git a/tests/test_web_middleware.py b/tests/test_web_middleware.py index e37817b9b87..6e881d2488a 100644 --- a/tests/test_web_middleware.py +++ b/tests/test_web_middleware.py @@ -1,8 +1,9 @@ +import asyncio from collections.abc import Awaitable, Callable, Iterable from typing import NoReturn import pytest -from pytest_aiohttp import AiohttpClient +from pytest_aiohttp import AiohttpClient, AiohttpServer from yarl import URL from aiohttp import web, web_app @@ -415,6 +416,55 @@ async def handle(request: web.Request) -> web.StreamResponse: assert resp.url.query == URL("//google.com").query +async def test_normalize_path_skips_parser_error( + aiohttp_server: AiohttpServer, +) -> None: + # Regression: normalize_path_middleware used to redirect SystemRoute + # matches indiscriminately. After parser errors started flowing through + # the middleware chain (issue #3287) the dummy "/" rel_url on a + # malformed request would match a real "/" route and trigger 301. + async def root(request: web.Request) -> NoReturn: + pytest.fail("root handler should not run for a parser-errored request") + + app = web.Application(middlewares=[web.normalize_path_middleware()]) + app.router.add_get("/", root) + server = await aiohttp_server(app) + + reader, writer = await asyncio.open_connection(server.host, server.port) + try: + writer.write(b"garbage\r\n\r\n") + await writer.drain() + head = b"" + while b"\r\n\r\n" not in head: + chunk = await reader.read(4096) + assert chunk, "connection closed before headers complete" + head += chunk + finally: + writer.close() + await writer.wait_closed() + + assert head.startswith(b"HTTP/1.0 400 ") + + +async def test_normalize_path_redirects_method_not_allowed( + aiohttp_client: AiohttpClient, +) -> None: + # A 405 is a routing failure (the path matched a resource, but not for this + # method), so normalize_path_middleware should still try alternate paths: + # POST /foo has no POST route, but POST /foo/ does. + async def handler(request: web.Request) -> web.Response: + assert False + + app = web.Application(middlewares=[web.normalize_path_middleware()]) + app.router.add_get("/foo", handler) + app.router.add_post("/foo/", handler) + client = await aiohttp_client(app) + + resp = await client.post("/foo", allow_redirects=False) + assert resp.status == 308 + assert resp.headers["Location"] == "/foo/" + + async def test_bug_3669(aiohttp_client: AiohttpClient) -> None: async def paymethod(request: web.Request) -> NoReturn: assert False diff --git a/tests/test_web_request.py b/tests/test_web_request.py index 3ad652916d9..9dd919caf09 100644 --- a/tests/test_web_request.py +++ b/tests/test_web_request.py @@ -851,6 +851,13 @@ def test_clone_override_client_max_size() -> None: assert req2.client_max_size == 2048 +def test_clone_preserves_pre_handler_error() -> None: + req = make_mocked_request("GET", "/path") + err = web.HTTPBadRequest(text="bad") + req._pre_handler_error = err + assert req.clone().pre_handler_error is err + + def test_clone_method() -> None: req = make_mocked_request("GET", "/path") req2 = req.clone(method="POST") diff --git a/tests/test_web_server.py b/tests/test_web_server.py index 52ed10bf00a..25c9b14f9bb 100644 --- a/tests/test_web_server.py +++ b/tests/test_web_server.py @@ -9,7 +9,7 @@ from pytest_aiohttp import AiohttpClient, AiohttpRawServer, AiohttpServer from aiohttp import client, web -from aiohttp.http_exceptions import BadHttpMethod, BadStatusLine +from aiohttp.http_exceptions import BadStatusLine async def test_simple_server( @@ -71,55 +71,10 @@ async def handler(request: web.BaseRequest) -> NoReturn: ) -async def test_raw_server_logs_invalid_method_with_loop_debug( - aiohttp_raw_server: AiohttpRawServer, aiohttp_client: AiohttpClient -) -> None: - exc = BadHttpMethod(b"\x16\x03\x03\x01F\x01".decode(), "error") - - async def handler(request: web.BaseRequest) -> NoReturn: - raise exc - - loop = asyncio.get_running_loop() - loop.set_debug(True) - logger = mock.Mock() - server = await aiohttp_raw_server(handler, logger=logger) - cli = await aiohttp_client(server) # type: ignore[var-annotated] - resp = await cli.get("/path/to") - assert resp.status == 500 - assert resp.headers["Content-Type"].startswith("text/plain") - - txt = await resp.text() - assert "Traceback (most recent call last):\n" in txt - - # BadHttpMethod should be logged as debug - # on the first request since the client may - # be probing for TLS/SSL support which is - # expected to fail - logger.debug.assert_called_with( - "Error handling request from %s", cli.host, exc_info=exc - ) - logger.debug.reset_mock() - - # Now make another connection to the server - # to make sure that the exception is logged - # at debug on a second fresh connection - cli2 = await aiohttp_client(server) # type: ignore[var-annotated] - resp = await cli2.get("/path/to") - assert resp.status == 500 - assert resp.headers["Content-Type"].startswith("text/plain") - # BadHttpMethod should be logged as debug - # on the first request since the client may - # be probing for TLS/SSL support which is - # expected to fail - logger.debug.assert_called_with( - "Error handling request from %s", cli.host, exc_info=exc - ) - - -async def test_raw_server_logs_invalid_method_without_loop_debug( +async def test_raw_server_logs_bad_status_line_as_exception( aiohttp_raw_server: AiohttpRawServer, aiohttp_client: AiohttpClient ) -> None: - exc = BadHttpMethod(b"\x16\x03\x03\x01F\x01".decode(), "error") + exc = BadStatusLine(b"\x16\x03\x03\x01F\x01".decode(), "error") async def handler(request: web.BaseRequest) -> NoReturn: raise exc @@ -136,69 +91,42 @@ async def handler(request: web.BaseRequest) -> NoReturn: txt = await resp.text() assert "Traceback (most recent call last):\n" not in txt - # BadHttpMethod should be logged as debug - # on the first request since the client may - # be probing for TLS/SSL support which is - # expected to fail - logger.debug.assert_called_with( - "Error handling request from %s", cli.host, exc_info=exc - ) - - -async def test_raw_server_logs_invalid_method_second_request( - aiohttp_raw_server: AiohttpRawServer, aiohttp_client: AiohttpClient -) -> None: - exc = BadHttpMethod(b"\x16\x03\x03\x01F\x01".decode(), "error") - request_count = 0 - - async def handler(request: web.BaseRequest) -> web.Response: - nonlocal request_count - request_count += 1 - if request_count == 2: - raise exc - return web.Response() - - loop = asyncio.get_running_loop() - loop.set_debug(False) - logger = mock.Mock() - server = await aiohttp_raw_server(handler, logger=logger) - cli = await aiohttp_client(server) # type: ignore[var-annotated] - resp = await cli.get("/path/to") - assert resp.status == 200 - resp = await cli.get("/path/to") - assert resp.status == 500 - assert resp.headers["Content-Type"].startswith("text/plain") - # BadHttpMethod should be logged as an exception - # if its not the first request since we know - # that the client already was speaking HTTP logger.exception.assert_called_with( "Error handling request from %s", cli.host, exc_info=exc ) -async def test_raw_server_logs_bad_status_line_as_exception( +async def test_raw_server_parser_error_visible_to_handler( aiohttp_raw_server: AiohttpRawServer, aiohttp_client: AiohttpClient ) -> None: - exc = BadStatusLine(b"\x16\x03\x03\x01F\x01".decode(), "error") + async def handler(request: web.BaseRequest) -> web.Response: + if (err := request.pre_handler_error) is not None: + return web.Response(status=err.status, text=f"raw saw: {err.__cause__!r}") + return web.Response(text="ok") - async def handler(request: web.BaseRequest) -> NoReturn: - raise exc + server = await aiohttp_raw_server(handler) - loop = asyncio.get_running_loop() - loop.set_debug(False) - logger = mock.Mock() - server = await aiohttp_raw_server(handler, logger=logger) cli = await aiohttp_client(server) # type: ignore[var-annotated] resp = await cli.get("/path/to") - assert resp.status == 500 - assert resp.headers["Content-Type"].startswith("text/plain") + assert resp.status == 200 + assert await resp.text() == "ok" - txt = await resp.text() - assert "Traceback (most recent call last):\n" not in txt + reader, writer = await asyncio.open_connection(server.host, server.port) + try: + writer.write(b"garbage\r\n\r\n") + await writer.drain() + head = b"" + while b"\r\n\r\n" not in head: + chunk = await reader.read(4096) + assert chunk, "connection closed before headers complete" + head += chunk + body = head.partition(b"\r\n\r\n")[2] + await reader.read() + finally: + writer.close() + await writer.wait_closed() - logger.exception.assert_called_with( - "Error handling request from %s", cli.host, exc_info=exc - ) + assert head.startswith(b"HTTP/1.0 400 ") + assert b"BadHttpMethod" in body async def test_raw_server_handler_timeout(