[pull] master from aio-libs:master

CHANGES/12985.bugfix.rst

@@ -0,0 +1 @@
+Fixed client not closing cleanly after an exception -- by :user:`Dreamsorcerer`.

CHANGES/13001.bugfix.rst

@@ -0,0 +1 @@
+Fixed an :exc:`IndexError` in the pure-Python HTTP parser -- by :user:`Dreamsorcerer`.

aiohttp/client.py

@@ -587,6 +587,7 @@ async def _request(
             await trace.send_request_start(method, url.update_query(params), headers)
 
         timer = tm.timer()
+        req: ClientRequest | None = None
         try:
             with timer:
                 # https://www.rfc-editor.org/rfc/rfc9112.html#name-retrying-requests
@@ -869,6 +870,9 @@ async def _request(
                 handle.cancel()
                 handle = None
 
+            if req is not None and req._body is not None:
+                await req._body.close()
+
             for trace in traces:
                 await trace.send_request_exception(
                     method, url.update_query(params), headers, e

aiohttp/helpers.py

@@ -332,7 +332,7 @@ def parse_mimetype(mimetype: str) -> MimeType:
     parts = mimetype.split(";")
     params: MultiDict[str] = MultiDict()
     for item in parts[1:]:
-        if not item:
+        if not item.strip():
             continue
         key, _, value = item.partition("=")
         params.add(key.lower().strip(), value.strip(' "'))

aiohttp/http_parser.py

@@ -1146,19 +1146,20 @@ def feed_data(self, chunk: bytes) -> bool:
         self.size += len(chunk)
         self.out.total_compressed_bytes = self.size
 
-        # RFC1950
-        # bits 0..3 = CM = 0b1000 = 8 = "deflate"
-        # bits 4..7 = CINFO = 1..7 = windows size.
-        if (
-            not self._started_decoding
-            and self.encoding == "deflate"
-            and chunk[0] & 0xF != 8
-        ):
-            # Change the decoder to decompress incorrectly compressed data
-            # Actually we should issue a warning about non-RFC-compliant data.
-            self.decompressor = ZLibDecompressor(
-                encoding=self.encoding, suppress_deflate_header=True
-            )
+        # Inspect the first real byte once to choose the decompressor. An empty
+        # chunk (e.g. a chunk-size line arriving without body bytes) has no
+        # header to sniff, so skip it and wait for the first data byte.
+        if not self._started_decoding and chunk:
+            # RFC1950
+            # bits 0..3 = CM = 0b1000 = 8 = "deflate"
+            # bits 4..7 = CINFO = 1..7 = windows size.
+            if self.encoding == "deflate" and chunk[0] & 0xF != 8:
+                # Change the decoder to decompress incorrectly compressed data
+                # Actually we should issue a warning about non-RFC-compliant data.
+                self.decompressor = ZLibDecompressor(
+                    encoding=self.encoding, suppress_deflate_header=True
+                )
+            self._started_decoding = True
 
         low_water = self.out._low_water
         max_length = (
@@ -1171,8 +1172,6 @@ def feed_data(self, chunk: bytes) -> bool:
                 "Can not decode content-encoding: %s" % self.encoding
             )
 
-        self._started_decoding = True
-
         if chunk:
             self.out.feed_data(chunk)
         return self.decompressor.data_available

tests/test_client_functional.py

@@ -5430,6 +5430,62 @@ async def redirect_handler(request: web.Request) -> web.Response:
     ), "Payload.close() was not called when InvalidUrlRedirectClientError (invalid origin) was raised"
 
 
+async def test_request_body_closed_on_server_disconnect() -> None:
+    async def drop(reader: asyncio.StreamReader, writer: asyncio.StreamWriter) -> None:
+        # Slam the connection shut without sending a response.
+        writer.close()
+
+    server = await asyncio.start_server(drop, "127.0.0.1", 0)
+    port = server.sockets[0].getsockname()[1]
+    payload = MockedBytesPayload(b"x" * 1024)
+    try:
+        async with aiohttp.ClientSession() as session:
+            with pytest.raises(aiohttp.ClientError):
+                await session.post(f"http://127.0.0.1:{port}/", data=payload)
+    finally:
+        server.close()
+        await server.wait_closed()
+
+    assert (
+        payload.close_called
+    ), "Payload.close() was not called after a mid-upload disconnect"
+
+
+async def test_request_body_closed_on_cancellation() -> None:
+    accepted = asyncio.Event()
+
+    async def stall(reader: asyncio.StreamReader, writer: asyncio.StreamWriter) -> None:
+        accepted.set()
+        try:
+            await reader.read()  # wait for client EOF; never respond
+        finally:
+            writer.close()
+
+    server = await asyncio.start_server(stall, "127.0.0.1", 0)
+    port = server.sockets[0].getsockname()[1]
+    payload = MockedBytesPayload(b"y" * 1024)
+    try:
+        async with aiohttp.ClientSession() as session:
+            task = asyncio.create_task(
+                session.post(f"http://127.0.0.1:{port}/", data=payload)
+            )
+            await accepted.wait()
+            task.cancel()
+            with pytest.raises(asyncio.CancelledError):
+                await task
+    finally:
+        server.close()
+        await server.wait_closed()
+
+    assert payload.close_called, "Payload.close() was not called after cancellation"
+
+
+async def test_request_error_before_body_created_does_not_mask() -> None:
+    async with aiohttp.ClientSession() as session:
+        with pytest.raises(InvalidUrlClientError):
+            await session.get("http:///path")
+
+
 async def test_amazon_like_cookie_scenario(aiohttp_client: AiohttpClient) -> None:
     """Test real-world cookie scenario similar to Amazon."""
 

tests/test_helpers.py

@@ -74,6 +74,27 @@
                 "text", "plain", "", MultiDictProxy(MultiDict({"base64": ""}))
             ),
         ),
+        (
+            "text/html; ",
+            helpers.MimeType("text", "html", "", MultiDictProxy(MultiDict())),
+        ),
+        (
+            "text/html;  ",
+            helpers.MimeType("text", "html", "", MultiDictProxy(MultiDict())),
+        ),
+        (
+            "text/html;\t",
+            helpers.MimeType("text", "html", "", MultiDictProxy(MultiDict())),
+        ),
+        (
+            "text/html; charset=utf-8; ",
+            helpers.MimeType(
+                "text",
+                "html",
+                "",
+                MultiDictProxy(MultiDict({"charset": "utf-8"})),
+            ),
+        ),
     ],
 )
 def test_parse_mimetype(mimetype: str, expected: helpers.MimeType) -> None:

tests/test_http_parser.py

@@ -1402,6 +1402,37 @@ async def test_compressed_chunked_with_pending(response: HttpResponseParser) ->
     assert result == original
 
 
+async def test_compressed_chunked_split_chunk_size_line(
+    response: HttpResponseParser,
+) -> None:
+    """First chunk-size line arrives in a feed that carries no body bytes.
+
+    Regression test for an ``IndexError`` in the pure-Python parser: the
+    chunked parser fed an empty chunk to ``DeflateBuffer`` before any data
+    had been decoded, and the deflate header sniff indexed ``chunk[0]`` on it.
+    """
+    original = b"Hello, world! " * 4
+    compressed = zlib.compress(original)
+    size_line = hex(len(compressed))[2:].encode() + b"\r\n"
+    headers = (
+        b"HTTP/1.1 200 OK\r\n"
+        b"Transfer-Encoding: chunked\r\n"
+        b"Content-Encoding: deflate\r\n"
+        b"\r\n"
+    )
+
+    msgs, upgrade, tail = response.feed_data(headers)
+    payload = msgs[0][-1]
+    # The chunk-size line lands with no body bytes in the same feed, so the
+    # parser transitions into the chunk body with an empty buffer.
+    response.feed_data(size_line)
+    response.feed_data(compressed + b"\r\n0\r\n\r\n")
+
+    result = await payload.read()
+    assert result == original
+    assert payload.exception() is None
+
+
 async def test_compressed_until_eof_with_pending(response: HttpResponseParser) -> None:
     """Test read-until-eof + compressed with pause."""
     # Must be large enough to exceed high water mark.