AIOHTTP vulnerable to DoS when bypassing asserts

@@ -361,11 +361,8 @@ async def read_chunk(self, size: int = chunk_size) -> bytes:
         self._read_bytes += len(chunk)
         if self._read_bytes == self._length:
             self._at_eof = True
-        if self._at_eof:
-            clrf = await self._content.readline()
-            assert (
-                b"\r\n" == clrf
-            ), "reader did not read all the data or it is malformed"
+        if self._at_eof and await self._content.readline() != b"\r\n":
+            raise ValueError("Reader did not read all the data or it is malformed")
         return chunk
 
     async def _read_chunk_from_length(self, size: int) -> bytes:
@@ -395,7 +392,8 @@ async def _read_chunk_from_stream(self, size: int) -> bytes:
         while len(chunk) < self._boundary_len:
             chunk += await self._content.read(size)
             self._content_eof += int(self._content.at_eof())
-            assert self._content_eof < 3, "Reading after EOF"
+            if self._content_eof > 2:
+                raise ValueError("Reading after EOF")
             if self._content_eof:
                 break
         if len(chunk) > size:

@@ -722,12 +722,12 @@ async def post(self) -> "MultiDictProxy[Union[str, bytes, FileField]]":
             max_size = self._client_max_size
 
             size = 0
-            field = await multipart.next()
-            while field is not None:
+            while (field := await multipart.next()) is not None:
                 field_ct = field.headers.get(hdrs.CONTENT_TYPE)
 
                 if isinstance(field, BodyPartReader):
-                    assert field.name is not None
+                    if field.name is None:
+                        raise ValueError("Multipart field missing name.")
 
                     # Note that according to RFC 7578, the Content-Type header
                     # is optional, even for files, so we can't assume it's
@@ -779,8 +779,6 @@ async def post(self) -> "MultiDictProxy[Union[str, bytes, FileField]]":
                     raise ValueError(
                         "To decode nested multipart you need to use custom reader",
                     )
-
-                field = await multipart.next()
         else:
             data = await self.read()
             if data:

@@ -221,11 +221,21 @@ async def test_read_incomplete_body_chunked(self) -> None:
         with Stream(data) as stream:
             obj = aiohttp.BodyPartReader(BOUNDARY, {}, stream)
             result = b""
-            with pytest.raises(AssertionError):
+            with pytest.raises(ValueError):
                 for _ in range(4):
                     result += await obj.read_chunk(7)
         assert data == result
 
+    async def test_read_with_content_length_malformed_crlf(self) -> None:
+        # Content-Length is correct but data after content is not \r\n
+        content = b"Hello"
+        h = CIMultiDictProxy(CIMultiDict({"CONTENT-LENGTH": str(len(content))}))
+        # Malformed: "XX" instead of "\r\n" after content
+        with Stream(content + b"XX--:--") as stream:
+            obj = aiohttp.BodyPartReader(BOUNDARY, h, stream)
+            with pytest.raises(ValueError, match="malformed"):
+                await obj.read()
+
     async def test_read_boundary_with_incomplete_chunk(self) -> None:
         with Stream(b"") as stream:
 

@@ -12,6 +12,7 @@
 from yarl import URL
 
 from aiohttp import HttpVersion
+from aiohttp.base_protocol import BaseProtocol
 from aiohttp.http_parser import RawRequestMessage
 from aiohttp.streams import StreamReader
 from aiohttp.test_utils import make_mocked_request
@@ -845,7 +846,28 @@ async def test_multipart_formdata(protocol) -> None:
     assert dict(result) == {"a": "b", "c": "d"}
 
 
-async def test_multipart_formdata_file(protocol) -> None:
+async def test_multipart_formdata_field_missing_name(protocol: BaseProtocol) -> None:
+    # Ensure ValueError is raised when Content-Disposition has no name
+    payload = StreamReader(protocol, 2**16, loop=asyncio.get_event_loop())
+    payload.feed_data(
+        b"-----------------------------326931944431359\r\n"
+        b"Content-Disposition: form-data\r\n"  # Missing name!
+        b"\r\n"
+        b"value\r\n"
+        b"-----------------------------326931944431359--\r\n"
+    )
+    content_type = (
+        "multipart/form-data; boundary=---------------------------326931944431359"
+    )
+    payload.feed_eof()
+    req = make_mocked_request(
+        "POST", "/", headers={"CONTENT-TYPE": content_type}, payload=payload
+    )
+    with pytest.raises(ValueError, match="Multipart field missing name"):
+        await req.post()
+
+
+async def test_multipart_formdata_file(protocol: BaseProtocol) -> None:
     # Make sure file uploads work, even without a content type
     payload = StreamReader(protocol, 2**16, loop=asyncio.get_event_loop())
     payload.feed_data(