Feat/upload file max size config

docs/requests.md

@@ -122,6 +122,13 @@ async with request.form(max_files=1000, max_fields=1000, max_part_size=1024*1024
     ...
 ```
 
+You can configure maximum [spooled file](https://docs.python.org/3/library/tempfile.html#tempfile.SpooledTemporaryFile) size per file uploaded with the parameter `max_file_size`:
+
+```python
+async with request.form(max_file_spool_size=100*1024*1024):  # 100 MB limit per file
+    ...
+```
+
 !!! 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.
 

starlette/formparsers.py

@@ -135,6 +135,7 @@ def __init__(
         max_files: int | float = 1000,
         max_fields: int | float = 1000,
         max_part_size: int = 1024 * 1024,  # 1MB
+        spool_max_size: int = 1024 * 1024,  # 1MB
     ) -> None:
         assert multipart is not None, "The `python-multipart` library must be installed to use form parsing."
         self.headers = headers
@@ -152,6 +153,7 @@ def __init__(
         self._file_parts_to_finish: list[MultipartPart] = []
         self._files_to_close_on_error: list[SpooledTemporaryFile[bytes]] = []
         self.max_part_size = max_part_size
+        self.spool_max_size = spool_max_size
 
     def on_part_begin(self) -> None:
         self._current_part = MultipartPart()

starlette/requests.py

@@ -255,6 +255,7 @@ async def _get_form(
         max_files: int | float = 1000,
         max_fields: int | float = 1000,
         max_part_size: int = 1024 * 1024,
+        max_file_spool_size: int = 1024 * 1024,
     ) -> FormData:
         if self._form is None:  # pragma: no branch
             assert parse_options_header is not None, (
@@ -271,6 +272,7 @@ async def _get_form(
                         max_files=max_files,
                         max_fields=max_fields,
                         max_part_size=max_part_size,
+                        spool_max_size=max_file_spool_size,
                     )
                     self._form = await multipart_parser.parse()
                 except MultiPartException as exc:
@@ -290,9 +292,15 @@ def form(
         max_files: int | float = 1000,
         max_fields: int | float = 1000,
         max_part_size: int = 1024 * 1024,
+        max_file_spool_size: int = 1024 * 1024,
     ) -> AwaitableOrContextManager[FormData]:
         return AwaitableOrContextManagerWrapper(
-            self._get_form(max_files=max_files, max_fields=max_fields, max_part_size=max_part_size)
+            self._get_form(
+                max_files=max_files,
+                max_fields=max_fields,
+                max_part_size=max_part_size,
+                max_file_spool_size=max_file_spool_size,
+            )
         )
 
     async def close(self) -> None:

tests/test_formparsers.py

@@ -4,6 +4,7 @@
 import typing
 from contextlib import nullcontext as does_not_raise
 from pathlib import Path
+from secrets import token_bytes
 
 import pytest
 
@@ -127,6 +128,14 @@ async def app(scope: Scope, receive: Receive, send: Send) -> None:
     return app
 
 
+def make_app_max_spool_file_size(max_file_size: int) -> ASGIApp:
+    async def app(scope: Scope, receive: Receive, send: Send) -> None:
+        request = Request(scope, receive)
+        await request.form(max_file_spool_size=max_file_size)
+
+    return app
+
+
 def test_multipart_request_data(tmpdir: Path, test_client_factory: TestClientFactory) -> None:
     client = test_client_factory(app)
     response = client.post("/", data={"some": "data"}, files=FORCE_MULTIPART)
@@ -580,6 +589,30 @@ def test_too_many_files_and_fields_raise(
         assert res.text == "Too many files. Maximum number of files is 1000."
 
 
+@pytest.mark.parametrize(
+    "app,expectation",
+    [
+        (make_app_max_spool_file_size(1024), pytest.raises(MultiPartException)),
+        (Starlette(routes=[Mount("/", app=make_app_max_spool_file_size(1024))]), does_not_raise()),
+    ],
+)
+def test_max_part_file_size_raise(
+    tmpdir: Path,
+    app: ASGIApp,
+    expectation: typing.ContextManager[Exception],
+    test_client_factory: TestClientFactory,
+) -> None:
+    path = os.path.join(tmpdir, "test.txt")
+    with open(path, "wb") as file:
+        file.write(token_bytes(1024 + 1))
+
+    client = test_client_factory(app)
+    with open(path, "rb") as f, expectation:
+        response = client.post("/", files={"test": f})
+        assert response.status_code == 400
+        assert response.text == "File exceeds maximum size of 1024 bytes."
+
+
 @pytest.mark.parametrize(
     "app,expectation",
     [