CVE-2023-25578: Allocation of Resources Without Limits or Throttling (CWE-770)
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.

CWE description: 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.
CVSS v3 base score: 7.5
Published: 2023-02-15T15:15Z
Repository: https://github.com/starlite-api/starlite
Language: Python
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 date: 2023-02-15 11:05:56 +1000
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>
Diff: diff --git a/starlite/app.py b/starlite/app.py
index 45f06e03..121bf495 100644
--- a/starlite/app.py
+++ b/starlite/app.py
@@ -133,10 +133,11 @@ class Starlite(Router):
         "csrf_config",
         "debug",
         "get_logger",
         "logger",
         "logging_config",
+        "multipart_form_part_limit",
         "on_shutdown",
         "on_startup",
         "openapi_config",
         "openapi_schema",
         "plugins",
@@ -173,10 +174,11 @@ class Starlite(Router):
         exception_handlers: Optional["ExceptionHandlersMap"] = None,
         guards: Optional[List["Guard"]] = None,
         initial_state: Optional["InitialStateType"] = None,
         logging_config: Union["BaseLoggingConfig", "EmptyType", None] = Empty,
         middleware: Optional[List["Middleware"]] = None,
+        multipart_form_part_limit: int = 1000,
         on_app_init: Optional[List["OnAppInitHandler"]] = None,
         on_shutdown: Optional[List["LifeSpanHandler"]] = None,
         on_startup: Optional[List["LifeSpanHandler"]] = None,
         openapi_config: Optional[OpenAPIConfig] = DEFAULT_OPENAPI_CONFIG,
         opt: Optional[Dict[str, Any]] = None,
@@ -236,10 +238,12 @@ class Starlite(Router):
             exception_handlers: A dictionary that maps handler functions to status codes and/or exception types.
             guards: A list of :class:`Guard <starlite.types.Guard>` callables.
             initial_state: An object from which to initialize the app state.
             logging_config: A subclass of :class:`BaseLoggingConfig <starlite.config.logging.BaseLoggingConfig>`.
             middleware: A list of :class:`Middleware <starlite.types.Middleware>`.
+            multipart_form_part_limit: The maximal number of allowed parts in a multipart/formdata request.
+                This limit is intended to protect from DoS attacks.
             on_app_init: A sequence of :class:`OnAppInitHandler <starlite.types.OnAppInitHandler>` instances. Handlers receive
                 an instance of :class:`AppConfig <starlite.config.app.AppConfig>` that will have been initially populated with
                 the parameters passed to :class:`Starlite <starlite.app.Starlite>`, and must return an instance of same. If more
                 than one handler is registered they are called in the order they are provided.
             on_shutdown: A list of :class:`LifeSpanHandler <starlite.types.LifeSpanHandler>` called during
@@ -298,10 +302,11 @@ class Starlite(Router):
             exception_handlers=exception_handlers or {},
             guards=guards or [],
             initial_state=initial_state or {},
             logging_config=logging_config if logging_config is not Empty else LoggingConfig() if debug else None,  # type: ignore[arg-type]
             middleware=middleware or [],
+            multipart_form_part_limit=multipart_form_part_limit,
             on_shutdown=on_shutdown or [],
             on_startup=on_startup or [],
             openapi_config=openapi_config,
             opt=opt or {},
             parameters=parameters or {},
@@ -341,10 +346,11 @@ class Starlite(Router):
         self.request_class = config.request_class or Request
         self.state = State(config.initial_state, deep_copy=True)
         self.static_files_config = config.static_files_config
         self.template_engine = config.template_config.engine_instance if config.template_config else None
         self.websocket_class = config.websocket_class or WebSocket
+        self.multipart_form_part_limit = config.multipart_form_part_limit
 
         super().__init__(
             after_request=config.after_request,
             after_response=config.after_response,
             before_request=config.before_request,
diff --git a/starlite/config/app.py b/starlite/config/app.py
index 623b2ffd..dba28b5c 100644
--- a/starlite/config/app.py
+++ b/starlite/config/app.py
@@ -176,10 +176,12 @@ class AppConfig(BaseModel):
     """An instance of :class:`TemplateConfig <starlite.config.TemplateConfig>`."""
     type_encoders: Optional[TypeEncodersMap] = None
     """A mapping of types to callables that transform them into types supported for serialization."""
     websocket_class: Optional[Type[WebSocket]]
     """An optional subclass of :class:`WebSocket <starlite.connection.websocket.WebSocket>` to use for websocket connections."""
+    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."""
 
     @validator("allowed_hosts", always=True)
     def validate_allowed_hosts(  # pylint: disable=no-self-argument
         cls, value: Optional[Union[List[str], AllowedHostsConfig]]
     ) -> Optional[AllowedHostsConfig]:
diff --git a/starlite/connection/request.py b/starlite/connection/request.py
index 8a7f2d76..8c932d6a 100644
--- a/starlite/connection/request.py
+++ b/starlite/connection/request.py
@@ -148,11 +148,13 @@ class Request(Generic[User, Auth], ASGIConnection["HTTPRouteHandler", User, Auth
         """
         if self._form is Empty:
             content_type, options = self.content_type
             if content_type == RequestEncodingType.MULTI_PART:
                 self._form = self.scope["_form"] = form_values = parse_multipart_form(  # type: ignore[typeddict-item]
-                    body=await self.body(), boundary=options.get("boundary", "").encode()
+                    body=await self.body(),
+                    boundary=options.get("boundary", "").encode(),
+                    multipart_form_part_limit=self.app.multipart_form_part_limit,
                 )
                 return FormMultiDict(form_values)
             if content_type == RequestEncodingType.URL_ENCODED:
                 self._form = self.scope["_form"] = form_values = parse_url_encoded_form_data(  # type: ignore[typeddict-item]
                     await self.body(),
diff --git a/starlite/kwargs/extractors.py b/starlite/kwargs/extractors.py
index 7d4964b2..5603b09e 100644
--- a/starlite/kwargs/extractors.py
+++ b/starlite/kwargs/extractors.py
@@ -17,10 +17,11 @@ from typing import (
 
 from starlite.datastructures.upload_file import UploadFile
 from starlite.enums import ParamType, RequestEncodingType
 from starlite.exceptions import ValidationException
 from starlite.multipart import parse_multipart_form
+from starlite.params import BodyKwarg
 from starlite.parsers import (
     parse_headers,
     parse_query_string,
     parse_url_encoded_form_data,
 )
@@ -287,19 +288,29 @@ def create_multipart_extractor(
         is_data_optional: Boolean dictating whether the field is optional.
 
     Returns:
         An extractor function.
     """
+    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
 
     async def extract_multipart(
         connection: "Request[Any, Any]",
     ) -> Any:
+        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
+        )
         connection.scope["_form"] = form_values = (  # type: ignore[typeddict-item]
             connection.scope["_form"]  # type: ignore[typeddict-item]
             if "_form" in connection.scope
             else parse_multipart_form(
-                body=await connection.body(), boundary=connection.content_type[-1].get("boundary", "").encode()
+                body=await connection.body(),
+                boundary=connection.content_type[-1].get("boundary", "").encode(),
+                multipart_form_part_limit=multipart_form_part_limit,
             )
         )
 
         if signature_field.is_non_string_sequence:
             return list(form_values.values())
diff --git a/starlite/multipart.py b/starlite/multipart.py
index 644e4200..4718c0bd 100644
--- a/starlite/multipart.py
+++ b/starlite/multipart.py
@@ -28,14 +28,15 @@ from collections import defaultdict
 from email.utils import decode_rfc2231
 from typing import Any, DefaultDict, Dict, List, Tuple
 from urllib.parse import unquote
 
 from starlite.datastructures.upload_file import UploadFile
-from starlite.exceptions import SerializationException
+from starlite.exceptions import SerializationException, ValidationException
 from starlite.utils.serialization import decode_json
 
-_token, _quoted = r"([\w!#$%&'*+\-.^_`|~]+)", r'"([^"]*)"'
+_token = r"([\w!#$%&'*+\-.^_`|~]+)"
+_quoted = r'"([^"]*)"'
 _param = re.compile(rf";\s*{_token}=(?:{_token}|{_quoted})", re.ASCII)
 _firefox_quote_escape = re.compile(r'\\"(?!; |\s*$)')
 
 
 def parse_content_header(value: str) -> Tuple[str, Dict[str, str]]:
@@ -57,69 +58,92 @@ def parse_content_header(value: str) -> Tuple[str, Dict[str, str]]:
         }
         value = value[:pos]
     return value.strip().lower(), options
 
 
-def parse_multipart_form(body: bytes, boundary: bytes) -> Dict[str, Any]:
+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]:
     """Parse multipart form data.
 
     Args:
         body: Body of the request.
         boundary: Boundary of the multipart message.
+        multipart_form_part_limit: Limit of the number of parts allowed.
 
     Returns:
         A dictionary of parsed results.
     """
 
     fields: DefaultDict[str, List[Any]] = defaultdict(list)
 
-    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))
+    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))
 
     return {k: v if len(v) > 1 else v[0] for k, v in fields.items()}
diff --git a/starlite/params.py b/starlite/params.py
index 3f606d91..ddf5fdf7 100644
--- a/starlite/params.py
+++ b/starlite/params.py
@@ -284,10 +284,12 @@ class BodyKwarg:
     regex: Optional[str] = field(default=None)
     """A string representing a regex against which the given string will be matched.
 
     Equivalent to pattern in the OpenAPI specification.
     """
+    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."""
 
     def __hash__(self) -> int:  # pragma: no cover
         """Hash the dataclass in a safe way.
 
         Returns:
@@ -313,11 +315,12 @@ def Body(
     multiple_of: Optional[float] = None,
     min_items: Optional[int] = None,
     max_items: Optional[int] = None,
     min_length: Optional[int] = None,
     max_length: Optional[int] = None,
-    regex: Optional[str] = None
+    regex: Optional[str] = None,
+    multipart_form_part_limit: Optional[int] = None
 ) -> Any:
     """Create an extended request body kwarg definition.
 
     Args:
         media_type: Defaults to RequestEncodingType.JSON.
@@ -352,10 +355,12 @@ def Body(
             minLength in the OpenAPI specification.
         max_length: Constrict a string or bytes value to have a maximum length. Equivalent to
             maxLength in the OpenAPI specification.
         regex: A string representing a regex against which the given string will be matched.
             Equivalent to pattern in the OpenAPI specification.
+        multipart_form_part_limit: The maximal number of allowed parts in a multipart/formdata request.
+            This limit is intended to protect from DoS attacks.
     """
     return BodyKwarg(
         media_type=media_type,
         examples=examples,
         external_docs=external_docs,
@@ -372,10 +377,11 @@ def Body(
         min_items=min_items,
         max_items=max_items,
         min_length=min_length,
         max_length=max_length,
         regex=regex,
+        multipart_form_part_limit=multipart_form_part_limit,
     )
 
 
 @dataclass(frozen=True)
 class DependencyKwarg:
diff --git a/starlite/testing/create_test_client.py b/starlite/testing/create_test_client.py
index 28f5d3fc..2f4c9b1d 100644
--- a/starlite/testing/create_test_client.py
+++ b/starlite/testing/create_test_client.py
@@ -76,10 +76,11 @@ def create_test_client(
     exception_handlers: Optional["ExceptionHandlersMap"] = None,
     guards: Optional[List["Guard"]] = None,
     initial_state: Optional[Union["ImmutableState", Dict[str, Any], Iterable[Tuple[str, Any]]]] = None,
     logging_config: Optional["BaseLoggingConfig"] = None,
     middleware: Optional[List["Middleware"]] = None,
+    multipart_form_part_limit: int = 1000,
     on_app_init: Optional[List["OnAppInitHandler"]] = None,
     on_shutdown: Optional[List["LifeSpanHandler"]] = None,
     on_startup: Optional[List["LifeSpanHandler"]] = None,
     openapi_config: Optional["OpenAPIConfig"] = None,
     parameters: Optional["ParametersMap"] = None,
@@ -158,10 +159,12 @@ def create_test_client(
         exception_handlers: A dictionary that maps handler functions to status codes and/or exception types.
         guards: A list of :class:`Guard <starlite.types.Guard>` callables.
         initial_state: An object from which to initialize the app state.
         logging_config: A subclass of :class:`BaseLoggingConfig <starlite.config.logging.BaseLoggingConfig>`.
         middleware: A list of :class:`Middleware <starlite.types.Middleware>`.
+        multipart_form_part_limit: The maximal number of allowed parts in a multipart/formdata request.
+            This limit is intended to protect from DoS attacks.
         on_app_init:  A sequence of :class:`OnAppInitHandler <starlite.types.OnAppInitHandler>` instances. Handlers receive
                 an instance of :class:`AppConfig <starlite.config.app.AppConfig>` that will have been initially populated with
                 the parameters passed to :class:`Starlite <starlite.app.Starlite>`, and must return an instance of same. If more
                 than one handler is registered they are called in the order they are provided.
         on_shutdown: A list of :class:`LifeSpanHandler <starlite.types.LifeSpanHandler>` called during
@@ -208,10 +211,11 @@ def create_test_client(
             exception_handlers=exception_handlers,
             guards=guards,
             initial_state=initial_state,
             logging_config=logging_config,
             middleware=middleware,
+            multipart_form_part_limit=multipart_form_part_limit,
             on_app_init=on_app_init,
             on_shutdown=on_shutdown,
             on_startup=on_startup,
             openapi_config=openapi_config,
             parameters=parameters,
diff --git a/tests/app/test_app_config.py b/tests/app/test_app_config.py
index c396c7fa..64e7a2e8 100644
--- a/tests/app/test_app_config.py
+++ b/tests/app/test_app_config.py
@@ -33,10 +33,11 @@ def app_config_object() -> AppConfig:
         exception_handlers={},
         guards=[],
         initial_state={},
         logging_config=None,
         middleware=[],
+        multipart_form_part_limit=1000,
         on_shutdown=[],
         on_startup=[],
         openapi_config=None,
         opt={},
         parameters={},
diff --git a/tests/kwargs/test_multipart_data.py b/tests/kwargs/test_multipart_data.py
index 10f11305..1613af7b 100644
--- a/tests/kwargs/test_multipart_data.py
+++ b/tests/kwargs/test_multipart_data.py
@@ -8,11 +8,11 @@ from typing import Any, DefaultDict, Dict, List, Optional, Type
 import pytest
 from pydantic import BaseConfig, BaseModel
 
 from starlite import Body, Request, RequestEncodingType, post
 from starlite.datastructures import UploadFile
-from starlite.status_codes import HTTP_201_CREATED
+from starlite.status_codes import HTTP_201_CREATED, HTTP_400_BAD_REQUEST
 from starlite.testing import create_test_client
 from tests import Person, PersonFactory
 from tests.kwargs import Form
 
 
@@ -403,5 +403,42 @@ def test_optional_formdata() -> None:
             await data.read()
 
     with create_test_client(route_handlers=[hello_world]) as client:
         response = client.post("/")
         assert response.status_code == HTTP_201_CREATED
+
+
+@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

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