new multipart parser

litestar/_kwargs/extractors.py

@@ -340,8 +340,8 @@ async def _extract_multipart(
     connection.scope["_form"] = form_values = (  # type: ignore[typeddict-unknown-key]
         connection.scope["_form"]  # type: ignore[typeddict-item]
         if "_form" in connection.scope
-        else parse_multipart_form(
-            body=await connection.body(),
+        else await parse_multipart_form(
+            stream=connection.stream(),
             boundary=connection.content_type[-1].get("boundary", "").encode(),
             multipart_form_part_limit=multipart_form_part_limit,
             type_decoders=connection.route_handler.resolve_type_decoders(),

litestar/_multipart.py

@@ -1,41 +1,22 @@
-"""The contents of this file were adapted from sanic.
-
-MIT License
-
-Copyright (c) 2016-present Sanic Community
-
-Permission is hereby granted, free of charge, to any person obtaining a copy
-of this software and associated documentation files (the "Software"), to deal
-in the Software without restriction, including without limitation the rights
-to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
-copies of the Software, and to permit persons to whom the Software is
-furnished to do so, subject to the following conditions:
-
-The above copyright notice and this permission notice shall be included in all
-copies or substantial portions of the Software.
-
-THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
-IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
-FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
-AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
-LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
-OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
-SOFTWARE.
-"""
-
 from __future__ import annotations
 
 import re
 from collections import defaultdict
-from email.utils import decode_rfc2231
-from typing import TYPE_CHECKING, Any
-from urllib.parse import unquote
+from typing import TYPE_CHECKING, Any, AsyncGenerator
+
+from multipart import (  # type: ignore[import-untyped]
+    MultipartSegment,
+    ParserError,
+    ParserLimitReached,
+    PushMultipartParser,
+)
 
 from litestar.datastructures.upload_file import UploadFile
-from litestar.exceptions import ValidationException
+from litestar.exceptions import ClientException
 
-__all__ = ("parse_body", "parse_content_header", "parse_multipart_form")
+__all__ = ("parse_content_header", "parse_multipart_form")
 
+from litestar.utils.compat import async_next
 
 if TYPE_CHECKING:
     from litestar.types import TypeDecodersSequence
@@ -67,42 +48,16 @@ def parse_content_header(value: str) -> tuple[str, dict[str, str]]:
     return value.strip().lower(), options
 
 
-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.
-
-    Args:
-        body: The form body.
-        boundary: The boundary used to separate form components.
-        multipart_form_part_limit: The limit of allowed form components
-
-    Returns:
-        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,
+async def parse_multipart_form(  # noqa: C901
+    stream: AsyncGenerator[bytes, None],
     boundary: bytes,
     multipart_form_part_limit: int = 1000,
     type_decoders: TypeDecodersSequence | None = None,
 ) -> dict[str, Any]:
     """Parse multipart form data.
 
     Args:
-        body: Body of the request.
+        stream: Body of the request.
         boundary: Boundary of the multipart message.
         multipart_form_part_limit: Limit of the number of parts allowed.
         type_decoders: A sequence of type decoders to use.
@@ -113,51 +68,55 @@ def parse_multipart_form(
 
     fields: defaultdict[str, list[Any]] = defaultdict(list)
 
-    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[: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)
-            elif post_data:
-                fields[field_name].append(post_data.decode(content_charset))
-            else:
-                fields[field_name].append(None)
+    chunk = await async_next(stream, b"")
+    if not chunk:
+        return fields
+
+    try:
+        with PushMultipartParser(boundary, max_segment_count=multipart_form_part_limit) as parser:
+            segment: MultipartSegment | None = None
+            data: UploadFile | bytes = b""
+            while not parser.closed:
+                for form_part in parser.parse(chunk):
+                    if isinstance(form_part, MultipartSegment):
+                        segment = form_part
+                        if segment.filename:
+                            data = UploadFile(
+                                content_type=segment.content_type or "text/plain",
+                                filename=segment.filename,
+                                headers=dict(segment.headerlist),
+                            )
+                    elif form_part:
+                        if isinstance(data, UploadFile):
+                            await data.write(form_part)
+                        else:
+                            data += form_part
+                    else:
+                        # end of part
+                        if segment is None:
+                            # we have reached the end of a segment before we have
+                            # received a complete header segment
+                            raise ClientException("Unexpected eof in multipart/form-data")
+                        if segment.name:
+                            if isinstance(data, UploadFile):
+                                await data.seek(0)
+                                fields[segment.name].append(data)
+                            elif data:
+                                fields[segment.name].append(data.decode(segment.charset or "utf-8"))
+                            else:
+                                fields[segment.name].append(None)
+
+                        # reset for next part
+                        data = b""
+                        segment = None
+
+                chunk = await async_next(stream, b"")
+
+    except ParserError as exc:
+        raise ClientException("Invalid multipart/form-data") from exc
+    except ParserLimitReached:
+        # FIXME (3.0): This should raise a '413 - Request Entity Too Large', but for
+        # backwards compatibility, we keep it as a 400 for now
+        raise ClientException("Request Entity Too Large") from None
 
     return {k: v if len(v) > 1 else v[0] for k, v in fields.items()}

litestar/connection/request.py

@@ -77,7 +77,7 @@ def __init__(self, scope: Scope, receive: Receive = empty_receive, send: Send =
         """
         super().__init__(scope, receive, send)
         self.is_connected: bool = True
-        self._body: bytes | EmptyType = Empty
+        self._body: bytes | EmptyType = self._connection_state.body
         self._form: FormMultiDict | EmptyType = Empty
         self._json: Any = Empty
         self._msgpack: Any = Empty
@@ -264,8 +264,8 @@ async def form(self) -> FormMultiDict:
             if (form_data := self._connection_state.form) is Empty:
                 content_type, options = self.content_type
                 if content_type == RequestEncodingType.MULTI_PART:
-                    form_data = parse_multipart_form(
-                        body=await self.body(),
+                    form_data = await parse_multipart_form(
+                        stream=self.stream(),
                         boundary=options.get("boundary", "").encode(),
                         multipart_form_part_limit=self.app.multipart_form_part_limit,
                     )