[pull] master from aio-libs:master

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

CHANGES/12983.bugfix.rst

@@ -0,0 +1 @@
+Fixed ``DigestAuthMiddleware`` raising an ``IndexError`` on empty domain -- by :user:`Dreamsorcerer`.

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

CHANGES/13002.bugfix.rst

@@ -0,0 +1 @@
+Fixed parsing optional whitespace in Content-Disposition -- by :user:`Dreamsorcerer`.

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

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

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 <space>
-    #                                                            +-> 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 <space>
+    #                                                               +-> 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:
@@ -434,21 +445,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

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

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:

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__
 

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()