CVE-2023-25578 (Python, published 2023-02-15T15:15Z)
Description: value: Starlite is an Asynchronous Server Gateway Interface (ASGI) framework. Prior to version 1.5.2, the request body parsing in `starlite` allows a potentially unauthenticated attacker to consume a large amount of CPU time and RAM. The multipart body parser processes an unlimited number of file parts and an unlimited number of field parts. This is a remote, potentially unauthenticated Denial of Service vulnerability. This vulnerability affects applications with a request handler that accepts a `Body(media_type=RequestEncodingType.MULTI_PART)`. The large amount of CPU time required for processing requests can block all available worker processes and significantly delay or slow down the processing of legitimate user requests. The large amount of RAM accumulated while processing requests can lead to Out-Of-Memory kills. Complete DoS is achievable by sending many concurrent multipart requests in a loop. Version 1.51.2 contains a patch for this issue.

Severity: nan (CVSS v3 base score: 7.5)
Weakness: CWE-770 \u2014 Allocation of Resources Without Limits or Throttling
The product allocates a reusable resource or group of resources on behalf of an actor without imposing any restrictions on the size or number of resources that can be allocated, in violation of the intended security policy for that actor.
Repository: https://github.com/starlite-api/starlite
Files changed:
starlite/app.py
starlite/config/app.py
starlite/connection/request.py
starlite/kwargs/extractors.py
starlite/multipart.py
starlite/params.py
starlite/testing/create_test_client.py
tests/app/test_app_config.py
tests/kwargs/test_multipart_data.py
Commit message: Merge pull request from GHSA-p24m-863f-fm6q

* implement limitation to form multipart parts

* Adds multipart_form_part_limit config to Body function.

A limit defined through the Body function takes precedence over the app limit.

* Replace "DDoS" with "DoS" in docstrings.

* Provide `maxsplit` arg to `body.split()`.

`multipart_form_part_limit + 3` parts are required to determine if body
exceeds limit.

We discard the first item of the result via [1:] (+1) and we discard the
last item of the result [:-1] (+1) and need to observe whether there are
N+1 items (+1), which gets us to 3 extra items after multipart_form_part_limit. (thanks @das7pad).

* Fix type errors.

---------

Co-authored-by: Na'aman Hirschfeld <nhirschfeld@gmail.com>

Vulnerable code:
                    body=await self.body(), boundary=options.get("boundary", "").encode()
                body=await connection.body(), boundary=connection.content_type[-1].get("boundary", "").encode()
from starlite.exceptions import SerializationException
_token, _quoted = r"([\w!#$%&'*+\-.^_`|~]+)", r'"([^"]*)"'
def parse_multipart_form(body: bytes, boundary: bytes) -> Dict[str, Any]:
    if body and boundary:
        form_parts = body.split(boundary)
        for form_part in form_parts[1:-1]:
            file_name = None
            content_type = "text/plain"
            content_charset = "utf-8"
            field_name = None
            line_index = 2
            line_end_index = 0
            headers: List[Tuple[str, str]] = []

            while line_end_index != -1:
                line_end_index = form_part.find(b"\r\n", line_index)
                form_line = form_part[line_index:line_end_index].decode("utf-8")

                if not form_line:
                    break

                line_index = line_end_index + 2
                colon_index = form_line.index(":")
                current_idx = colon_index + 2
                form_header_field = form_line[0:colon_index].lower()
                form_header_value, form_parameters = parse_content_header(form_line[current_idx:])

                if form_header_field == "content-disposition":
                    field_name = form_parameters.get("name")
                    file_name = form_parameters.get("filename")

                    if file_name is None and (filename_with_asterisk := form_parameters.get("filename*")):
                        encoding, _, value = decode_rfc2231(filename_with_asterisk)
                        file_name = unquote(value, encoding=encoding or content_charset)

                elif form_header_field == "content-type":
                    content_type = form_header_value
                    content_charset = form_parameters.get("charset", "utf-8")
                headers.append((form_header_field, form_header_value))

            if field_name:
                post_data = form_part[line_index:-4].lstrip(b"\r\n")
                if file_name:
                    form_file = UploadFile(
                        content_type=content_type, filename=file_name, file_data=post_data, headers=dict(headers)
                    )
                    fields[field_name].append(form_file)
                else:
                    try:
                        fields[field_name].append(decode_json(post_data))
                    except SerializationException:
                        fields[field_name].append(post_data.decode(content_charset))
    regex: Optional[str] = None
from starlite.status_codes import HTTP_201_CREATED

Fixed code:
        "multipart_form_part_limit",
        multipart_form_part_limit: int = 1000,
            multipart_form_part_limit: The maximal number of allowed parts in a multipart/formdata request.
                This limit is intended to protect from DoS attacks.
            multipart_form_part_limit=multipart_form_part_limit,
        self.multipart_form_part_limit = config.multipart_form_part_limit
    multipart_form_part_limit: int
    """The maximal number of allowed parts in a multipart/formdata request. This limit is intended to protect from DoS attacks."""
                    body=await self.body(),
                    boundary=options.get("boundary", "").encode(),
                    multipart_form_part_limit=self.app.multipart_form_part_limit,
from starlite.params import BodyKwarg
    body_kwarg_multipart_form_part_limit: Optional[int] = None
    if signature_field.kwarg_model and isinstance(signature_field.kwarg_model, BodyKwarg):
        body_kwarg_multipart_form_part_limit = signature_field.kwarg_model.multipart_form_part_limit
        multipart_form_part_limit = (
            body_kwarg_multipart_form_part_limit
            if body_kwarg_multipart_form_part_limit is not None
            else connection.app.multipart_form_part_limit
        )
                body=await connection.body(),
                boundary=connection.content_type[-1].get("boundary", "").encode(),
                multipart_form_part_limit=multipart_form_part_limit,
from starlite.exceptions import SerializationException, ValidationException
_token = r"([\w!#$%&'*+\-.^_`|~]+)"
_quoted = r'"([^"]*)"'
def parse_body(body: bytes, boundary: bytes, multipart_form_part_limit: int) -> List[bytes]:
    """Split the body using the boundary
        and validate the number of form parts is within the allowed limit.

    :param body: The form body.
    :param boundary: The boundary used to separate form components.
    :param multipart_form_part_limit: The limit of allowed form components
    :return:
        A list of form components.
    """
    if not (body and boundary):
        return []

    form_parts = body.split(boundary, multipart_form_part_limit + 3)[1:-1]

    if len(form_parts) > multipart_form_part_limit:
        raise ValidationException(
            f"number of multipart components exceeds the allowed limit of {multipart_form_part_limit}, "
            f"this potentially indicates a DoS attack"
        )

    return form_parts


def parse_multipart_form(body: bytes, boundary: bytes, multipart_form_part_limit: int = 1000) -> Dict[str, Any]:
        multipart_form_part_limit: Limit of the number of parts allowed.
    for form_part in parse_body(body=body, boundary=boundary, multipart_form_part_limit=multipart_form_part_limit):
        file_name = None
        content_type = "text/plain"
        content_charset = "utf-8"
        field_name = None
        line_index = 2
        line_end_index = 0
        headers: List[Tuple[str, str]] = []

        while line_end_index != -1:
            line_end_index = form_part.find(b"\r\n", line_index)
            form_line = form_part[line_index:line_end_index].decode("utf-8")

            if not form_line:
                break

            line_index = line_end_index + 2
            colon_index = form_line.index(":")
            current_idx = colon_index + 2
            form_header_field = form_line[0:colon_index].lower()
            form_header_value, form_parameters = parse_content_header(form_line[current_idx:])

            if form_header_field == "content-disposition":
                field_name = form_parameters.get("name")
                file_name = form_parameters.get("filename")

                if file_name is None and (filename_with_asterisk := form_parameters.get("filename*")):
                    encoding, _, value = decode_rfc2231(filename_with_asterisk)
                    file_name = unquote(value, encoding=encoding or content_charset)

            elif form_header_field == "content-type":
                content_type = form_header_value
                content_charset = form_parameters.get("charset", "utf-8")
            headers.append((form_header_field, form_header_value))

        if field_name:
            post_data = form_part[line_index:-4].lstrip(b"\r\n")
            if file_name:
                form_file = UploadFile(
                    content_type=content_type, filename=file_name, file_data=post_data, headers=dict(headers)
                )
                fields[field_name].append(form_file)
            else:
                try:
                    fields[field_name].append(decode_json(post_data))
                except SerializationException:
                    fields[field_name].append(post_data.decode(content_charset))
    multipart_form_part_limit: Optional[int] = field(default=None)
    """The maximal number of allowed parts in a multipart/formdata request. This limit is intended to protect from DoS attacks."""
    regex: Optional[str] = None,
    multipart_form_part_limit: Optional[int] = None
        multipart_form_part_limit: The maximal number of allowed parts in a multipart/formdata request.
            This limit is intended to protect from DoS attacks.
        multipart_form_part_limit=multipart_form_part_limit,
    multipart_form_part_limit: int = 1000,
        multipart_form_part_limit: The maximal number of allowed parts in a multipart/formdata request.
            This limit is intended to protect from DoS attacks.
            multipart_form_part_limit=multipart_form_part_limit,
        multipart_form_part_limit=1000,
from starlite.status_codes import HTTP_201_CREATED, HTTP_400_BAD_REQUEST


@pytest.mark.parametrize("limit", (1000, 100, 10))
def test_multipart_form_part_limit(limit: int) -> None:
    @post("/")
    async def hello_world(data: List[UploadFile] = Body(media_type=RequestEncodingType.MULTI_PART)) -> None:
        assert len(data) == limit

    with create_test_client(route_handlers=[hello_world], multipart_form_part_limit=limit) as client:
        data = {str(i): "a" for i in range(limit)}
        response = client.post("/", files=data)
        assert response.status_code == HTTP_201_CREATED

        data = {str(i): "a" for i in range(limit)}
        data[str(limit + 1)] = "b"
        response = client.post("/", files=data)
        assert response.status_code == HTTP_400_BAD_REQUEST


def test_multipart_form_part_limit_body_param_precedence() -> None:
    app_limit = 100
    route_limit = 10

    @post("/")
    async def hello_world(
        data: List[UploadFile] = Body(media_type=RequestEncodingType.MULTI_PART, multipart_form_part_limit=route_limit)
    ) -> None:
        assert len(data) == route_limit

    with create_test_client(route_handlers=[hello_world], multipart_form_part_limit=app_limit) as client:
        data = {str(i): "a" for i in range(route_limit)}
        response = client.post("/", files=data)
        assert response.status_code == HTTP_201_CREATED

        data = {str(i): "a" for i in range(route_limit + 1)}
        response = client.post("/", files=data)
        assert response.status_code == HTTP_400_BAD_REQUEST