CVE-2024-30251 (Python, published 2024-05-02T14:15Z)
Description: value: aiohttp is an asynchronous HTTP client/server framework for asyncio and Python. In affected versions an attacker can send a specially crafted POST (multipart/form-data) request. When the aiohttp server processes it, the server will enter an infinite loop and be unable to process any further requests. An attacker can stop the application from serving requests after sending a single request. This issue has been addressed in version 3.9.4. Users are advised to upgrade. Users unable to upgrade may manually apply a patch to their systems. Please see the linked GHSA for instructions.
Severity: nan (CVSS v3 base score: )
Weakness: NVD-CWE-noinfo \u2014 Insufficient Information
There is insufficient information about the issue to classify it; details are unkown or unspecified.
Repository: https://github.com/aio-libs/aiohttp
Files changed:
CHANGES/8280.bugfix.rst
CHANGES/8280.deprecation.rst
aiohttp/formdata.py
aiohttp/multipart.py
tests/test_client_functional.py
tests/test_multipart.py
tests/test_web_functional.py
Commit message: Fix handling of multipart/form-data (#8280) (#8302)

https://datatracker.ietf.org/doc/html/rfc7578
(cherry picked from commit 7d0be3fee540a3d4161ac7dc76422f1f5ea60104)

Vulnerable code:
            headers[hdrs.CONTENT_TRANSFER_ENCODING] = content_transfer_encoding
        self, boundary: bytes, headers: "CIMultiDictProxy[str]", content: StreamReader
        length = self.headers.get(CONTENT_LENGTH, None)
        if CONTENT_ENCODING in self.headers:
        return mimetype.parameters.get("charset", default)
        self._last_part = await self.fetch_next_part()
            return self.part_reader_cls(self._boundary, headers, self._content)

    def _get_boundary(self) -> str:
        mimetype = parse_mimetype(self.headers[CONTENT_TYPE])

        assert mimetype.type == "multipart", "multipart/* content type expected"

        if "boundary" not in mimetype.parameters:
            raise ValueError(
                "boundary missed for Content-Type: %s" % self.headers[CONTENT_TYPE]
        boundary = mimetype.parameters["boundary"]
        # compression
        encoding: Optional[str] = payload.headers.get(
            CONTENT_ENCODING,
            "",
        ).lower()
        if encoding and encoding not in ("deflate", "gzip", "identity"):
            raise RuntimeError(f"unknown content encoding: {encoding}")
        if encoding == "identity":
            encoding = None

        # te encoding
        te_encoding: Optional[str] = payload.headers.get(
            CONTENT_TRANSFER_ENCODING,
            "",
        ).lower()
        if te_encoding not in ("", "base64", "quoted-printable", "binary"):
            raise RuntimeError(
                "unknown content transfer encoding: {}" "".format(te_encoding)
        if te_encoding == "binary":
            te_encoding = None

        # size
        size = payload.size
        if size is not None and not (encoding or te_encoding):
            payload.headers[CONTENT_LENGTH] = str(size)
async def test_POST_DATA_with_context_transfer_encoding(aiohttp_client) -> None:
    async def handler(request):
        data = await request.post()
        assert data["name"] == "text"
        return web.Response(text=data["name"])

    app = web.Application()
    app.router.add_post("/", handler)
    client = await aiohttp_client(app)

    form = aiohttp.FormData()
    form.add_field("name", "text", content_transfer_encoding="base64")

    resp = await client.post("/", data=form)
    assert 200 == resp.status
    content = await resp.text()
    assert content == "text"
    resp.close()


async def test_POST_DATA_with_content_type_context_transfer_encoding(aiohttp_client):
    async def handler(request):
        data = await request.post()
        assert data["name"] == "text"
        return web.Response(body=data["name"])

    app = web.Application()
    app.router.add_post("/", handler)
    client = await aiohttp_client(app)

    form = aiohttp.FormData()
    form.add_field(
        "name", "text", content_type="text/plain", content_transfer_encoding="base64"
    )

    resp = await client.post("/", data=form)
    assert 200 == resp.status
    content = await resp.text()
    assert content == "text"
    resp.close()


            "/", data={"some": f, "test": b"data"}, chunked=True
            content_length = part.size
            b'Content-Disposition: attachments; filename="bug.py"\r\n'
            b"Content-Length: %s"
            b"" % (str(content_length).encode(),)
            content_length = part.size
            b'Content-Disposition: attachments; filename="bug.py"\r\n'
            b"Content-Length: %s"
            b"" % (str(content_length).encode(),)
            content_length = part.size

            b' attachments; filename="bug.py"\r\n'
            b"Content-Length: %s"
            b"" % (str(content_length).encode(),)
    form.add_field("name", b"123", content_transfer_encoding="base64")
async def test_POST_DATA_with_content_transfer_encoding(aiohttp_client) -> None:
    async def handler(request):
        data = await request.post()
        assert b"123" == data["name"]
        return web.Response()

    app = web.Application()
    app.router.add_post("/", handler)
    client = await aiohttp_client(app)

    form = FormData()
    form.add_field("name", b"123", content_transfer_encoding="base64")

    resp = await client.post("/", data=form)
    assert 200 == resp.status

    await resp.release()


    form.add_field("name", b"123", content_transfer_encoding="base64")
    resp = await client.post("/", data={"file": data})

Fixed code:
Fixed ``multipart/form-data`` compliance with :rfc:`7578` -- by :user:`Dreamsorcerer`.
Deprecated ``content_transfer_encoding`` parameter in :py:meth:`FormData.add_field()
<aiohttp.FormData.add_field>` -- by :user:`Dreamsorcerer`.
import warnings
            msg = (
                "In v4, passing bytes will no longer create a file field. "
                "Please explicitly use the filename parameter or pass a BytesIO object."
            )
                warnings.warn(msg, DeprecationWarning)
            msg = (
                "content_transfer_encoding is deprecated. "
                "To maintain compatibility with v4 please pass a BytesPayload."
            )
            warnings.warn(msg, DeprecationWarning)
        self,
        boundary: bytes,
        headers: "CIMultiDictProxy[str]",
        content: StreamReader,
        *,
        subtype: str = "mixed",
        default_charset: Optional[str] = None,
        self._default_charset = default_charset
        self._is_form_data = subtype == "form-data"
        # https://datatracker.ietf.org/doc/html/rfc7578#section-4.8
        length = None if self._is_form_data else self.headers.get(CONTENT_LENGTH, None)
        if self._content.at_eof():
            self._at_eof = True
        # https://datatracker.ietf.org/doc/html/rfc7578#section-4.8
        if not self._is_form_data and CONTENT_ENCODING in self.headers:
        return mimetype.parameters.get("charset", self._default_charset or default)
        self._mimetype = parse_mimetype(headers[CONTENT_TYPE])
        assert self._mimetype.type == "multipart", "multipart/* content type expected"
        if "boundary" not in self._mimetype.parameters:
            raise ValueError(
                "boundary missed for Content-Type: %s" % headers[CONTENT_TYPE]
            )

        self._default_charset: Optional[str] = None

        part = await self.fetch_next_part()
        # https://datatracker.ietf.org/doc/html/rfc7578#section-4.6
        if (
            self._last_part is None
            and self._mimetype.subtype == "form-data"
            and isinstance(part, BodyPartReader)
        ):
            _, params = parse_content_disposition(part.headers.get(CONTENT_DISPOSITION))
            if params.get("name") == "_charset_":
                # Longest encoding in https://encoding.spec.whatwg.org/encodings.json
                # is 19 characters, so 32 should be more than enough for any valid encoding.
                charset = await part.read_chunk(32)
                if len(charset) > 31:
                    raise RuntimeError("Invalid default charset")
                self._default_charset = charset.strip().decode()
                part = await self.fetch_next_part()
        self._last_part = part
            return self.part_reader_cls(
                self._boundary,
                headers,
                self._content,
                subtype=self._mimetype.subtype,
                default_charset=self._default_charset,
    def _get_boundary(self) -> str:
        boundary = self._mimetype.parameters["boundary"]
        self._is_form_data = subtype == "form-data"
        encoding: Optional[str] = None
        te_encoding: Optional[str] = None
        if self._is_form_data:
            # https://datatracker.ietf.org/doc/html/rfc7578#section-4.7
            # https://datatracker.ietf.org/doc/html/rfc7578#section-4.8
            assert CONTENT_DISPOSITION in payload.headers
            assert "name=" in payload.headers[CONTENT_DISPOSITION]
            assert (
                not {CONTENT_ENCODING, CONTENT_LENGTH, CONTENT_TRANSFER_ENCODING}
                & payload.headers.keys()
        else:
            # compression
            encoding = payload.headers.get(CONTENT_ENCODING, "").lower()
            if encoding and encoding not in ("deflate", "gzip", "identity"):
                raise RuntimeError(f"unknown content encoding: {encoding}")
            if encoding == "identity":
                encoding = None

            # te encoding
            te_encoding = payload.headers.get(CONTENT_TRANSFER_ENCODING, "").lower()
            if te_encoding not in ("", "base64", "quoted-printable", "binary"):
                raise RuntimeError(f"unknown content transfer encoding: {te_encoding}")
            if te_encoding == "binary":
                te_encoding = None

            # size
            size = payload.size
            if size is not None and not (encoding or te_encoding):
                payload.headers[CONTENT_LENGTH] = str(size)
            "/", data={"some": f, "test": io.BytesIO(b"data")}, chunked=True
    async def test_read_form_default_encoding(self) -> None:
        with Stream(
            b"--:\r\n"
            b'Content-Disposition: form-data; name="_charset_"\r\n\r\n'
            b"ascii"
            b"\r\n"
            b"--:\r\n"
            b'Content-Disposition: form-data; name="field1"\r\n\r\n'
            b"foo"
            b"\r\n"
            b"--:\r\n"
            b"Content-Type: text/plain;charset=UTF-8\r\n"
            b'Content-Disposition: form-data; name="field2"\r\n\r\n'
            b"foo"
            b"\r\n"
            b"--:\r\n"
            b'Content-Disposition: form-data; name="field3"\r\n\r\n'
            b"foo"
            b"\r\n"
        ) as stream:
            reader = aiohttp.MultipartReader(
                {CONTENT_TYPE: 'multipart/form-data;boundary=":"'},
                stream,
            )
            field1 = await reader.next()
            assert field1.name == "field1"
            assert field1.get_charset("default") == "ascii"
            field2 = await reader.next()
            assert field2.name == "field2"
            assert field2.get_charset("default") == "UTF-8"
            field3 = await reader.next()
            assert field3.name == "field3"
            assert field3.get_charset("default") == "ascii"

    async def test_read_form_invalid_default_encoding(self) -> None:
        with Stream(
            b"--:\r\n"
            b'Content-Disposition: form-data; name="_charset_"\r\n\r\n'
            b"this-value-is-too-long-to-be-a-charset"
            b"\r\n"
            b"--:\r\n"
            b'Content-Disposition: form-data; name="field1"\r\n\r\n'
            b"foo"
            b"\r\n"
        ) as stream:
            reader = aiohttp.MultipartReader(
                {CONTENT_TYPE: 'multipart/form-data;boundary=":"'},
                stream,
            )
            with pytest.raises(RuntimeError, match="Invalid default charset"):
                await reader.next()

            b'Content-Disposition: attachments; filename="bug.py"'
            b'Content-Disposition: attachments; filename="bug.py"'
            b' attachments; filename="bug.py"'
    with pytest.warns(DeprecationWarning, match="BytesPayload"):
        form.add_field("name", b"123", content_transfer_encoding="base64")
    with pytest.warns(DeprecationWarning, match="BytesPayload"):
        form.add_field("name", b"123", content_transfer_encoding="base64")
    resp = await client.post("/", data={"file": io.BytesIO(data)})