Fix parser when paused at end of content-length (aio-libs#13349)

Author: Dreamsorcerer

CHANGES/13348.bugfix.rst

@@ -0,0 +1 @@
+Fixed the HTTP parser raising :exc:`~aiohttp.ClientPayloadError` when a fully received ``Content-Length`` body was pending completion -- by :user:`Dreamsorcerer`.

aiohttp/_http_parser.pyx

@@ -554,8 +554,11 @@ cdef class HttpParser:
         self._messages.append((msg, payload))
 
     cdef _on_message_complete(self):
-        self._payload.feed_eof()
-        self._payload = None
+        # The payload is None when feed_eof() already completed a fully
+        # received content-length body.
+        if self._payload is not None:
+            self._payload.feed_eof()
+            self._payload = None
 
     cdef _on_chunk_header(self):
         self._payload.begin_http_chunk_receiving()
@@ -595,7 +598,8 @@ cdef class HttpParser:
             if self._cparser.flags & cparser.F_CHUNKED:
                 raise TransferEncodingError(
                     "Not enough data to satisfy transfer length header.")
-            elif self._cparser.flags & cparser.F_CONTENT_LENGTH:
+            elif (self._cparser.flags & cparser.F_CONTENT_LENGTH
+                  and self._cparser.content_length):
                 received = self._content_length_expected - self._cparser.content_length
                 raise ContentLengthError(
                     f"Not enough data to satisfy content length header "
@@ -604,6 +608,8 @@ cdef class HttpParser:
                 desc = cparser.llhttp_get_error_reason(self._cparser)
                 raise PayloadEncodingError(desc.decode('latin-1'))
             else:
+                # Reading until EOF, or a content-length body that was fully
+                # received but the parser paused.
                 self._eof_pending = True
                 while self._more_data_available:
                     if self._paused:

aiohttp/http_parser.py

@@ -914,11 +914,20 @@ def feed_eof(self) -> None:
             self.done = True
             self._eof_pending = False
         elif self._type == ParseState.PARSE_LENGTH:
-            received = self._length_expected - self._length
-            raise ContentLengthError(
-                f"Not enough data to satisfy content length header "
-                f"(received {received} of {self._length_expected} bytes)."
-            )
+            if self._length:
+                received = self._length_expected - self._length
+                raise ContentLengthError(
+                    f"Not enough data to satisfy content length header "
+                    f"(received {received} of {self._length_expected} bytes)."
+                )
+            # Body has already been received, but parser paused.
+            while self._more_data_available:
+                if self._paused:
+                    self._paused = False
+                    return  # Will resume via feed_data(b"") later
+                self._more_data_available = self.payload.feed_data(b"")
+            self.payload.feed_eof()
+            self.done = True
         elif self._type == ParseState.PARSE_CHUNKED:
             raise TransferEncodingError(
                 "Not enough data to satisfy transfer length header."

tests/test_http_parser.py

@@ -1564,6 +1564,79 @@ async def test_compressed_until_eof_with_pending(response: HttpResponseParser) -
     assert result == original
 
 
+async def test_content_length_eof_while_paused(response: HttpResponseParser) -> None:
+    """EOF right after a fully received content-length body must complete it.
+
+    Regression test for #13348:
+    feeding the final body bytes pauses the parser for flow control before
+    the message can complete; a server closing the connection in that state
+    raised ContentLengthError despite received == expected.
+    """
+    # Must be large enough to exceed the high water mark so the parser
+    # pauses with the message not yet complete.
+    body = b"x" * (1024 * 1024)
+    headers = b"HTTP/1.1 200 OK\r\nContent-Length: %d\r\n\r\n" % len(body)
+
+    msgs, upgrade, tail = response.feed_data(headers + body)
+    payload = msgs[0][-1]
+    # The server has sent everything and closed the connection.
+    response.feed_eof()
+
+    result = await payload.read()
+    assert result == body
+    assert payload.is_eof()
+    assert payload.exception() is None
+
+
+async def test_compressed_content_length_eof_while_paused(
+    response: HttpResponseParser,
+) -> None:
+    """EOF with pending decompressed data on a complete content-length body.
+
+    Like test_content_length_eof_while_paused, but the decompressor still
+    holds pending data at EOF, so completion is deferred until the reader
+    drains it.
+    """
+    # Must be large enough to exceed high water mark.
+    original = b"B" * 5 * 1024 * 1024
+    compressed = zlib.compress(original)
+    headers = (
+        b"HTTP/1.1 200 OK\r\n"
+        b"Content-Length: " + str(len(compressed)).encode() + b"\r\n"
+        b"Content-Encoding: deflate\r\n"
+        b"\r\n"
+    )
+
+    msgs, upgrade, tail = response.feed_data(headers + compressed)
+    payload = msgs[0][-1]
+    response.feed_eof()
+
+    # Check that .feed_eof() hasn't decompressed entire payload into memory.
+    assert sum(len(b) for b in payload._buffer) <= (2 * 1024 * 1024)
+
+    result = await payload.read()
+    assert len(result) == len(original)
+    assert result == original
+    assert payload.is_eof()
+    assert payload.exception() is None
+
+
+async def test_content_length_eof_while_paused_incomplete(
+    response: HttpResponseParser,
+) -> None:
+    """EOF on a paused parser with a genuinely incomplete body still raises."""
+    body = b"x" * (1024 * 1024)
+    headers = b"HTTP/1.1 200 OK\r\nContent-Length: %d\r\n\r\n" % (len(body) + 1)
+
+    response.feed_data(headers + body)
+
+    with pytest.raises(
+        http_exceptions.ContentLengthError,
+        match=r"received 1048576 of 1048577 bytes",
+    ):
+        response.feed_eof()
+
+
 async def test_compressed_until_eof_high_water(
     response_cls: type[HttpResponseParser],
 ) -> None:
@@ -2809,6 +2882,41 @@ async def test_parse_length_payload_partial_data(
         ):
             p.feed_eof()
 
+    async def test_parse_length_payload_eof_completes_after_pause(
+        self, protocol: BaseProtocol
+    ) -> None:
+        """feed_eof() completes a fully received length payload despite a pause.
+
+        Regression test for #13348:
+        The parser paused for flow control with pending decompressed data
+        when EOF arrived; the fully received body must complete instead of
+        raising ContentLengthError.
+        """
+        out = aiohttp.StreamReader(protocol, 2**16, loop=asyncio.get_running_loop())
+        original = b"x" * (1024 * 1024)
+        compressed = zlib.compress(original)
+
+        p = HttpPayloadParser(
+            out,
+            length=len(compressed),
+            compression="deflate",
+            headers_parser=HeadersParser(),
+        )
+        p.pause_reading()  # flow control kicked in before the final bytes
+        state, tail = p.feed_data(compressed)
+        assert state is PayloadState.PAYLOAD_HAS_PENDING_INPUT
+        assert not p.done
+
+        # All bytes were received, so EOF drains the pending data and
+        # completes the payload.
+        p.feed_eof()
+
+        assert p.done
+        assert out.is_eof()  # type: ignore[unreachable]
+        assert out.exception() is None
+        result = await out.read()
+        assert result == original
+
     async def test_parse_chunked_payload_size_error(
         self, protocol: BaseProtocol
     ) -> None: