CVE-2023-30798 (Python, published 2023-04-21T16:15Z)
Description: value: There MultipartParser usage in Encode's Starlette python framework before versions 0.25.0 allows an unauthenticated and remote attacker to specify any number of form fields or files which can cause excessive memory usage resulting in denial of service of the HTTP service.
Severity: nan (CVSS v3 base score: 7.5)
Weakness: CWE-400 \u2014 Uncontrolled Resource Consumption
The product does not properly control the allocation and maintenance of a limited resource, thereby enabling an actor to influence the amount of resources consumed, eventually leading to the exhaustion of available resources.
Repository: https://github.com/encode/starlette
Files changed:
docs/requests.md
starlette/formparsers.py
starlette/requests.py
tests/test_formparsers.py
Commit message: Merge pull request from GHSA-74m5-2c7w-9w3x

* ♻️ Refactor multipart parser logic to support limiting max fields and files

* ✨ Add support for new request.form() parameters max_files and max_fields

* ✅ Add tests for limiting max fields and files in form data

* 📝 Add docs about request.form() with new parameters max_files and max_fields

* 📝 Update `docs/requests.md`

Co-authored-by: Marcelo Trylesinski <marcelotryle@gmail.com>

* 📝 Tweak docs for request.form()

* ✏ Fix typo in `starlette/formparsers.py`

Co-authored-by: Adrian Garcia Badaracco <1755071+adriangb@users.noreply.github.com>

---------

Co-authored-by: Marcelo Trylesinski <marcelotryle@gmail.com>
Co-authored-by: Adrian Garcia Badaracco <1755071+adriangb@users.noreply.github.com>

Vulnerable code:
class MultiPartMessage(Enum):
    PART_BEGIN = 1
    PART_DATA = 2
    PART_END = 3
    HEADER_FIELD = 4
    HEADER_VALUE = 5
    HEADER_END = 6
    HEADERS_FINISHED = 7
    END = 8
        self, headers: Headers, stream: typing.AsyncGenerator[bytes, None]
        self.messages: typing.List[typing.Tuple[MultiPartMessage, bytes]] = []
        message = (MultiPartMessage.PART_BEGIN, b"")
        self.messages.append(message)
        message = (MultiPartMessage.PART_DATA, data[start:end])
        self.messages.append(message)
        message = (MultiPartMessage.PART_END, b"")
        self.messages.append(message)
        message = (MultiPartMessage.HEADER_FIELD, data[start:end])
        self.messages.append(message)
        message = (MultiPartMessage.HEADER_VALUE, data[start:end])
        self.messages.append(message)
        message = (MultiPartMessage.HEADER_END, b"")
        self.messages.append(message)
        message = (MultiPartMessage.HEADERS_FINISHED, b"")
        self.messages.append(message)
        message = (MultiPartMessage.END, b"")
        self.messages.append(message)
        header_field = b""
        header_value = b""
        content_disposition = None
        field_name = ""
        data = b""
        file: typing.Optional[UploadFile] = None

        items: typing.List[typing.Tuple[str, typing.Union[str, UploadFile]]] = []
        item_headers: typing.List[typing.Tuple[bytes, bytes]] = []

            messages = list(self.messages)
            self.messages.clear()
            for message_type, message_bytes in messages:
                if message_type == MultiPartMessage.PART_BEGIN:
                    content_disposition = None
                    data = b""
                    item_headers = []
                elif message_type == MultiPartMessage.HEADER_FIELD:
                    header_field += message_bytes
                elif message_type == MultiPartMessage.HEADER_VALUE:
                    header_value += message_bytes
                elif message_type == MultiPartMessage.HEADER_END:
                    field = header_field.lower()
                    if field == b"content-disposition":
                        content_disposition = header_value
                    item_headers.append((field, header_value))
                    header_field = b""
                    header_value = b""
                elif message_type == MultiPartMessage.HEADERS_FINISHED:
                    disposition, options = parse_options_header(content_disposition)
                    try:
                        field_name = _user_safe_decode(options[b"name"], charset)
                    except KeyError:
                        raise MultiPartException(
                            'The Content-Disposition header field "name" must be '
                            "provided."
                        )
                    if b"filename" in options:
                        filename = _user_safe_decode(options[b"filename"], charset)
                        tempfile = SpooledTemporaryFile(max_size=self.max_file_size)
                        file = UploadFile(
                            file=tempfile,  # type: ignore[arg-type]
                            size=0,
                            filename=filename,
                            headers=Headers(raw=item_headers),
                        )
                    else:
                        file = None
                elif message_type == MultiPartMessage.PART_DATA:
                    if file is None:
                        data += message_bytes
                    else:
                        await file.write(message_bytes)
                elif message_type == MultiPartMessage.PART_END:
                    if file is None:
                        items.append((field_name, _user_safe_decode(data, charset)))
                    else:
                        await file.seek(0)
                        items.append((field_name, file))
        return FormData(items)
    async def _get_form(self) -> FormData:
                    multipart_parser = MultiPartParser(self.headers, self.stream())
    def form(self) -> AwaitableOrContextManager[FormData]:
        return AwaitableOrContextManagerWrapper(self._get_form())

Fixed code:
Signature: `request.form(max_files=1000, max_fields=1000)`

You can configure the number of maximum fields or files with the parameters `max_files` and `max_fields`:

```python
async with request.form(max_files=1000, max_fields=1000):
    ...
```

!!! info
    These limits are for security reasons, allowing an unlimited number of fields or files could lead to a denial of service attack by consuming a lot of CPU and memory parsing too many empty fields.

from dataclasses import dataclass, field
@dataclass
class MultipartPart:
    content_disposition: typing.Optional[bytes] = None
    field_name: str = ""
    data: bytes = b""
    file: typing.Optional[UploadFile] = None
    item_headers: typing.List[typing.Tuple[bytes, bytes]] = field(default_factory=list)
        self,
        headers: Headers,
        stream: typing.AsyncGenerator[bytes, None],
        *,
        max_files: typing.Union[int, float] = 1000,
        max_fields: typing.Union[int, float] = 1000,
        self.max_files = max_files
        self.max_fields = max_fields
        self.items: typing.List[typing.Tuple[str, typing.Union[str, UploadFile]]] = []
        self._current_files = 0
        self._current_fields = 0
        self._current_partial_header_name: bytes = b""
        self._current_partial_header_value: bytes = b""
        self._current_part = MultipartPart()
        self._charset = ""
        self._file_parts_to_write: typing.List[typing.Tuple[MultipartPart, bytes]] = []
        self._file_parts_to_finish: typing.List[MultipartPart] = []
        self._current_part = MultipartPart()
        message_bytes = data[start:end]
        if self._current_part.file is None:
            self._current_part.data += message_bytes
        else:
            self._file_parts_to_write.append((self._current_part, message_bytes))
        if self._current_part.file is None:
            self.items.append(
                (
                    self._current_part.field_name,
                    _user_safe_decode(self._current_part.data, self._charset),
                )
            )
        else:
            self._file_parts_to_finish.append(self._current_part)
            # The file can be added to the items right now even though it's not
            # finished yet, because it will be finished in the `parse()` method, before
            # self.items is used in the return value.
            self.items.append((self._current_part.field_name, self._current_part.file))
        self._current_partial_header_name += data[start:end]
        self._current_partial_header_value += data[start:end]
        field = self._current_partial_header_name.lower()
        if field == b"content-disposition":
            self._current_part.content_disposition = self._current_partial_header_value
        self._current_part.item_headers.append(
            (field, self._current_partial_header_value)
        )
        self._current_partial_header_name = b""
        self._current_partial_header_value = b""
        disposition, options = parse_options_header(
            self._current_part.content_disposition
        )
        try:
            self._current_part.field_name = _user_safe_decode(
                options[b"name"], self._charset
            )
        except KeyError:
            raise MultiPartException(
                'The Content-Disposition header field "name" must be ' "provided."
            )
        if b"filename" in options:
            self._current_files += 1
            if self._current_files > self.max_files:
                raise MultiPartException(
                    f"Too many files. Maximum number of files is {self.max_files}."
                )
            filename = _user_safe_decode(options[b"filename"], self._charset)
            tempfile = SpooledTemporaryFile(max_size=self.max_file_size)
            self._current_part.file = UploadFile(
                file=tempfile,  # type: ignore[arg-type]
                size=0,
                filename=filename,
                headers=Headers(raw=self._current_part.item_headers),
            )
        else:
            self._current_fields += 1
            if self._current_fields > self.max_fields:
                raise MultiPartException(
                    f"Too many fields. Maximum number of fields is {self.max_fields}."
                )
            self._current_part.file = None
        pass
        self._charset = charset
            # Write file data, it needs to use await with the UploadFile methods that
            # call the corresponding file methods *in a threadpool*, otherwise, if
            # they were called directly in the callback methods above (regular,
            # non-async functions), that would block the event loop in the main thread.
            for part, data in self._file_parts_to_write:
                assert part.file  # for type checkers
                await part.file.write(data)
            for part in self._file_parts_to_finish:
                assert part.file  # for type checkers
                await part.file.seek(0)
            self._file_parts_to_write.clear()
            self._file_parts_to_finish.clear()
        return FormData(self.items)
    async def _get_form(
        self,
        *,
        max_files: typing.Union[int, float] = 1000,
        max_fields: typing.Union[int, float] = 1000,
    ) -> FormData:
                    multipart_parser = MultiPartParser(
                        self.headers,
                        self.stream(),
                        max_files=max_files,
                        max_fields=max_fields,
                    )
    def form(
        self,
        *,
        max_files: typing.Union[int, float] = 1000,
        max_fields: typing.Union[int, float] = 1000,
    ) -> AwaitableOrContextManager[FormData]:
        return AwaitableOrContextManagerWrapper(
            self._get_form(max_files=max_files, max_fields=max_fields)
        )
def make_app_max_parts(max_files: int = 1000, max_fields: int = 1000):
    async def app(scope, receive, send):
        request = Request(scope, receive)
        data = await request.form(max_files=max_files, max_fields=max_fields)
        output: typing.Dict[str, typing.Any] = {}
        for key, value in data.items():
            if isinstance(value, UploadFile):
                content = await value.read()
                output[key] = {
                    "filename": value.filename,
                    "size": value.size,
                    "content": content.decode(),
                    "content_type": value.content_type,
                }
            else:
                output[key] = value
        await request.close()
        response = JSONResponse(output)
        await response(scope, receive, send)

    return app




@pytest.mark.parametrize(
    "app,expectation",
    [
        (app, pytest.raises(MultiPartException)),
        (Starlette(routes=[Mount("/", app=app)]), does_not_raise()),
    ],
)
def test_too_many_fields_raise(app, expectation, test_client_factory):
    client = test_client_factory(app)
    fields = []
    for i in range(1001):
        fields.append(
            "--B\r\n" f'Content-Disposition: form-data; name="N{i}";\r\n\r\n' "\r\n"
        )
    data = "".join(fields).encode("utf-8")
    with expectation:
        res = client.post(
            "/",
            data=data,
            headers={"Content-Type": ("multipart/form-data; boundary=B")},
        )
        assert res.status_code == 400
        assert res.text == "Too many fields. Maximum number of fields is 1000."


@pytest.mark.parametrize(
    "app,expectation",
    [
        (app, pytest.raises(MultiPartException)),
        (Starlette(routes=[Mount("/", app=app)]), does_not_raise()),
    ],
)
def test_too_many_files_raise(app, expectation, test_client_factory):
    client = test_client_factory(app)
    fields = []
    for i in range(1001):
        fields.append(
            "--B\r\n"
            f'Content-Disposition: form-data; name="N{i}"; filename="F{i}";\r\n\r\n'
            "\r\n"
        )
    data = "".join(fields).encode("utf-8")
    with expectation:
        res = client.post(
            "/",
            data=data,
            headers={"Content-Type": ("multipart/form-data; boundary=B")},
        )
        assert res.status_code == 400
        assert res.text == "Too many files. Maximum number of files is 1000."


@pytest.mark.parametrize(
    "app,expectation",
    [
        (app, pytest.raises(MultiPartException)),
        (Starlette(routes=[Mount("/", app=app)]), does_not_raise()),
    ],
)
def test_too_many_files_single_field_raise(app, expectation, test_client_factory):
    client = test_client_factory(app)
    fields = []
    for i in range(1001):
        # This uses the same field name "N" for all files, equivalent to a
        # multifile upload form field
        fields.append(
            "--B\r\n"
            f'Content-Disposition: form-data; name="N"; filename="F{i}";\r\n\r\n'
            "\r\n"
        )
    data = "".join(fields).encode("utf-8")
    with expectation:
        res = client.post(
            "/",
            data=data,
            headers={"Content-Type": ("multipart/form-data; boundary=B")},
        )
        assert res.status_code == 400
        assert res.text == "Too many files. Maximum number of files is 1000."


@pytest.mark.parametrize(
    "app,expectation",
    [
        (app, pytest.raises(MultiPartException)),
        (Starlette(routes=[Mount("/", app=app)]), does_not_raise()),
    ],
)
def test_too_many_files_and_fields_raise(app, expectation, test_client_factory):
    client = test_client_factory(app)
    fields = []
    for i in range(1001):
        fields.append(
            "--B\r\n"
            f'Content-Disposition: form-data; name="F{i}"; filename="F{i}";\r\n\r\n'
            "\r\n"
        )
        fields.append(
            "--B\r\n" f'Content-Disposition: form-data; name="N{i}";\r\n\r\n' "\r\n"
        )
    data = "".join(fields).encode("utf-8")
    with expectation:
        res = client.post(
            "/",
            data=data,
            headers={"Content-Type": ("multipart/form-data; boundary=B")},
        )
        assert res.status_code == 400
        assert res.text == "Too many files. Maximum number of files is 1000."


@pytest.mark.parametrize(
    "app,expectation",
    [
        (make_app_max_parts(max_fields=1), pytest.raises(MultiPartException)),
        (
            Starlette(routes=[Mount("/", app=make_app_max_parts(max_fields=1))]),
            does_not_raise(),
        ),
    ],
)
def test_max_fields_is_customizable_low_raises(app, expectation, test_client_factory):
    client = test_client_factory(app)
    fields = []
    for i in range(2):
        fields.append(
            "--B\r\n" f'Content-Disposition: form-data; name="N{i}";\r\n\r\n' "\r\n"
        )
    data = "".join(fields).encode("utf-8")
    with expectation:
        res = client.post(
            "/",
            data=data,
            headers={"Content-Type": ("multipart/form-data; boundary=B")},
        )
        assert res.status_code == 400
        assert res.text == "Too many fields. Maximum number of fields is 1."


@pytest.mark.parametrize(
    "app,expectation",
    [
        (make_app_max_parts(max_files=1), pytest.raises(MultiPartException)),
        (
            Starlette(routes=[Mount("/", app=make_app_max_parts(max_files=1))]),
            does_not_raise(),
        ),
    ],
)
def test_max_files_is_customizable_low_raises(app, expectation, test_client_factory):
    client = test_client_factory(app)
    fields = []
    for i in range(2):
        fields.append(
            "--B\r\n"
            f'Content-Disposition: form-data; name="F{i}"; filename="F{i}";\r\n\r\n'
            "\r\n"
        )
    data = "".join(fields).encode("utf-8")
    with expectation:
        res = client.post(
            "/",
            data=data,
            headers={"Content-Type": ("multipart/form-data; boundary=B")},
        )
        assert res.status_code == 400
        assert res.text == "Too many files. Maximum number of files is 1."


def test_max_fields_is_customizable_high(test_client_factory):
    client = test_client_factory(make_app_max_parts(max_fields=2000, max_files=2000))
    fields = []
    for i in range(2000):
        fields.append(
            "--B\r\n" f'Content-Disposition: form-data; name="N{i}";\r\n\r\n' "\r\n"
        )
        fields.append(
            "--B\r\n"
            f'Content-Disposition: form-data; name="F{i}"; filename="F{i}";\r\n\r\n'
            "\r\n"
        )
    data = "".join(fields).encode("utf-8")
    data += b"--B--\r\n"
    res = client.post(
        "/",
        data=data,
        headers={"Content-Type": ("multipart/form-data; boundary=B")},
    )
    assert res.status_code == 200
    res_data = res.json()
    assert res_data["N1999"] == ""
    assert res_data["F1999"] == {
        "filename": "F1999",
        "size": 0,
        "content": "",
        "content_type": None,
    }