Validate Content-Length format in ClientRequest

Author: jmestwa-coder

aiohttp/client_reqrep.py

@@ -78,6 +78,7 @@
 
 _CONNECTION_CLOSED_EXCEPTION = ClientConnectionError("Connection closed")
 _CONTAINS_CONTROL_CHAR_RE = re.compile(r"[^-!#$%&'*+.^_`|~0-9a-zA-Z]")
+_DIGITS_RE = re.compile(r"\d+", re.ASCII)
 
 
 def _gen_default_accept_encoding() -> str:
@@ -753,12 +754,11 @@ def _get_content_length(self) -> int | None:
             return None
 
         content_length_hdr = self.headers[hdrs.CONTENT_LENGTH]
-        try:
-            return int(content_length_hdr)
-        except ValueError:
+        if not _DIGITS_RE.fullmatch(content_length_hdr):
             raise ValueError(
-                f"Invalid Content-Length header: {content_length_hdr}"
-            ) from None
+                f"Invalid Content-Length header: {content_length_hdr!r}"
+            )
+        return int(content_length_hdr)
 
     @property
     def _writer(self) -> asyncio.Task[None] | None:

tests/test_client_request.py

@@ -1911,7 +1911,24 @@ async def test_get_content_length(make_client_request: _RequestMaker) -> None:
 
     # Invalid Content-Length header
     req.headers["Content-Length"] = "invalid"
-    with pytest.raises(ValueError, match="Invalid Content-Length header: invalid"):
+    with pytest.raises(ValueError, match="Invalid Content-Length header"):
+        req._get_content_length()
+
+
+async def test_get_content_length_invalid_formats(
+    make_client_request: _RequestMaker,
+) -> None:
+    req = make_client_request("get", URL("http://python.org/"))
+
+    req.headers["Content-Length"] = "100"
+    assert req._get_content_length() == 100
+
+    req.headers["Content-Length"] = "-100"
+    with pytest.raises(ValueError, match="Invalid Content-Length header"):
+        req._get_content_length()
+
+    req.headers["Content-Length"] = " 100"
+    with pytest.raises(ValueError, match="Invalid Content-Length header"):
         req._get_content_length()
 
 

tests/test_web_functional.py

@@ -196,6 +196,29 @@ async def handler(request: web.Request) -> web.Response:
     resp.release()
 
 
+async def test_content_length_invalid(
+    aiohttp_client: AiohttpClient,
+) -> None:
+    async def handler(request: web.Request) -> web.Response:
+        body = await request.read()
+        return web.Response(body=body)
+
+    app = web.Application()
+    app.router.add_post("/", handler)
+    client = await aiohttp_client(app)
+
+    resp = await client.post("/", data=b"hello world", headers={CONTENT_LENGTH: "11"})
+    assert resp.status == 200
+    assert await resp.read() == b"hello world"
+    resp.release()
+
+    with pytest.raises(ValueError, match="Invalid Content-Length header"):
+        await client.post("/", data=b"hello world", headers={CONTENT_LENGTH: "-100"})
+
+    with pytest.raises(ValueError, match="Invalid Content-Length header"):
+        await client.post("/", data=b"hello world", headers={CONTENT_LENGTH: " 100"})
+
+
 @pytest.mark.skipif(sys.version_info < (3, 11), reason="Needs Task.cancelling()")
 async def test_cancel_shutdown(aiohttp_client: AiohttpClient) -> None:
     async def handler(request: web.Request) -> web.Response: