[pull] master from aio-libs:master

CHANGES/13426.bugfix.rst

@@ -0,0 +1 @@
+Switched multipart handling to use spooled temporary files to reduce number of file descriptors needed -- by :user:`Dreamsorcerer`.

aiohttp/web_request.py

@@ -4,12 +4,12 @@
 import re
 import string
 import sys
-import tempfile
 import types
 from collections.abc import Iterator, Mapping, MutableMapping
 from re import Pattern
 from types import MappingProxyType
 from typing import (
+    IO,
     TYPE_CHECKING,
     Any,
     Final,
@@ -63,10 +63,32 @@
 from .web_response import StreamResponse
 
 if sys.version_info >= (3, 11):
+    from tempfile import SpooledTemporaryFile
     from typing import Self
 else:
+    import tempfile
+
     Self = Any
 
+    class SpooledTemporaryFile(tempfile.SpooledTemporaryFile[bytes], io.IOBase):
+        """Make the spooled file satisfy the documented `FileField.file` type.
+
+        `tempfile.SpooledTemporaryFile` only became an `io.IOBase` subclass in
+        3.11 (python/cpython#70363), and never grew the three capability
+        predicates before then. Both underlying files (`io.BytesIO` before
+        rollover, a `w+b` temp file after) are readable, writable and seekable.
+        """
+
+        def readable(self) -> bool:
+            return True
+
+        def writable(self) -> bool:
+            return True
+
+        def seekable(self) -> bool:
+            return True
+
+
 __all__ = ("BaseRequest", "FileField", "Request")
 
 
@@ -85,11 +107,18 @@ class _CloneKwargs(TypedDict, total=False):
     remote: str
 
 
+# Multipart file parts are buffered in memory and only spilled to a temporary
+# file once they outgrow this size. A temp file per part would let a body made
+# of many tiny parts exhaust the process's file descriptors; spooling bounds
+# that at ``client_max_size // _FILE_SPOOL_MAX_SIZE`` descriptors per request.
+_FILE_SPOOL_MAX_SIZE: Final[int] = 1024**2
+
+
 @frozen_dataclass_decorator
 class FileField:
     name: str
     filename: str
-    file: io.BufferedReader
+    file: IO[bytes]
     content_type: str
     headers: HeadersDictProxy
 
@@ -740,8 +769,13 @@ async def post(self) -> "MultiDictProxy[str | bytes | FileField]":
             multipart = await self.multipart()
             max_size = self._client_max_size
 
-            size = 0
+            payload = self._payload
             while (field := await multipart.next()) is not None:
+                # This check is needed for empty payloads, which still add
+                # overhead without entering the loop and the check below.
+                if 0 < max_size < payload.total_bytes:
+                    raise HTTPRequestEntityTooLarge(max_size)
+
                 field_ct = field.headers.get(hdrs.CONTENT_TYPE)
 
                 if isinstance(field, BodyPartReader):
@@ -753,28 +787,43 @@ async def post(self) -> "MultiDictProxy[str | bytes | FileField]":
                     # present.
                     # https://tools.ietf.org/html/rfc7578#section-4.4
                     if field.filename:
-                        # store file in temp file
-                        tmp = await self._loop.run_in_executor(
-                            None, tempfile.TemporaryFile
-                        )
+                        tmp = SpooledTemporaryFile(_FILE_SPOOL_MAX_SIZE)
+                        # rolled means the temp file now uses the disk, at
+                        # which point we want to run in the executor.
+                        rolled = False
                         while chunk := await field.read_chunk(size=DEFAULT_CHUNK_SIZE):
+                            # Bounds one part, mid-read.
+                            if 0 < max_size < payload.total_bytes:
+                                if rolled:
+                                    await self._loop.run_in_executor(None, tmp.close)
+                                else:
+                                    tmp.close()
+                                raise HTTPRequestEntityTooLarge(max_size)
                             async for decoded_chunk in field.decode_iter(chunk):
-                                await self._loop.run_in_executor(
-                                    None, tmp.write, decoded_chunk
+                                # Update before writing, so we know if next
+                                # write is going to hit the disk.
+                                rolled = rolled or (
+                                    tmp.tell() + len(decoded_chunk)
+                                    > _FILE_SPOOL_MAX_SIZE
                                 )
-                                size += len(decoded_chunk)
-                                if 0 < max_size < size:
-                                    await self._loop.run_in_executor(None, tmp.close)
-                                    raise HTTPRequestEntityTooLarge(max_size)
-                        await self._loop.run_in_executor(None, tmp.seek, 0)
+                                if rolled:
+                                    await self._loop.run_in_executor(
+                                        None, tmp.write, decoded_chunk
+                                    )
+                                else:
+                                    tmp.write(decoded_chunk)
+                        if rolled:
+                            await self._loop.run_in_executor(None, tmp.seek, 0)
+                        else:
+                            tmp.seek(0)
 
                         if field_ct is None:
                             field_ct = "application/octet-stream"
 
                         ff = FileField(
                             field.name,
                             field.filename,
-                            cast(io.BufferedReader, tmp),
+                            tmp,
                             field_ct,
                             field.headers,
                         )
@@ -783,8 +832,7 @@ async def post(self) -> "MultiDictProxy[str | bytes | FileField]":
                         # deal with ordinary data
                         raw_data = bytearray()
                         while chunk := await field.read_chunk():
-                            size += len(chunk)
-                            if 0 < max_size < size:
+                            if 0 < max_size < payload.total_bytes:
                                 raise HTTPRequestEntityTooLarge(max_size)
                             raw_data.extend(chunk)
 
@@ -848,10 +896,7 @@ def _finish(self) -> None:
         if self._post is None or self.content_type != "multipart/form-data":
             return
 
-        # NOTE: Release file descriptors for the
-        # NOTE: `tempfile.Temporaryfile`-created `_io.BufferedRandom`
-        # NOTE: instances of files sent within multipart request body
-        # NOTE: via HTTP POST request.
+        # Release the temp files created within multipart request body.
         for file_name, file_field_object in self._post.items():
             if isinstance(file_field_object, FileField):
                 file_field_object.file.close()

tests/test_helpers.py

@@ -4,7 +4,6 @@
 import ipaddress
 import itertools
 import sys
-import time
 import weakref
 from collections.abc import Iterator
 from math import ceil, modf
@@ -353,22 +352,20 @@ def test_timeout_handle(event_loop: asyncio.AbstractEventLoop) -> None:
     assert not handle._callbacks
 
 
-@pytest.mark.skipif(
-    time.get_clock_info("monotonic").resolution > 0.001,
-    reason="loop.time() resolution is coarser than the test's 1ms tolerance",
-)
 def test_when_timeout_smaller_second(event_loop: asyncio.AbstractEventLoop) -> None:
     timeout = 0.1
 
     handle = helpers.TimeoutHandle(event_loop, timeout)
-    timer = event_loop.time() + timeout
+    before = event_loop.time()
     start_handle = handle.start()
+    after = event_loop.time()
     assert start_handle is not None
     when = start_handle.when()
     handle.close()
 
+    # Below the ceil threshold the deadline keeps sub-second precision.
     assert isinstance(when, float)
-    assert when - timer == pytest.approx(0, abs=0.001)
+    assert before + timeout <= when <= after + timeout
 
 
 def test_when_timeout_smaller_second_with_low_threshold(
@@ -377,14 +374,15 @@ def test_when_timeout_smaller_second_with_low_threshold(
     timeout = 0.1
 
     handle = helpers.TimeoutHandle(event_loop, timeout, 0.01)
-    timer = event_loop.time() + timeout
+    before = event_loop.time()
     start_handle = handle.start()
+    after = event_loop.time()
     assert start_handle is not None
     when = start_handle.when()
     handle.close()
 
     assert isinstance(when, int)
-    assert when == ceil(timer)
+    assert ceil(before + timeout) <= when <= ceil(after + timeout)
 
 
 def test_timeout_handle_cb_exc(event_loop: asyncio.AbstractEventLoop) -> None:

tests/test_web_functional.py

@@ -1,6 +1,8 @@
 import asyncio
+import base64
 import io
 import json
+import os
 import pathlib
 import socket
 import sys
@@ -33,6 +35,7 @@
 from aiohttp.streams import StreamReader
 from aiohttp.typedefs import Handler, Middleware
 from aiohttp.web_protocol import MAX_MSG_QUEUE_SIZE, RequestHandler
+from aiohttp.web_request import _FILE_SPOOL_MAX_SIZE
 
 try:
     import brotlicffi as brotli
@@ -2144,6 +2147,165 @@ async def handler(request: web.Request) -> NoReturn:
     resp.release()
 
 
+def _multipart_file_parts(count: int, body: bytes = b"") -> bytes:
+    """Raw multipart/form-data body of ``count`` file parts, boundary ``b``."""
+    return (
+        b"".join(
+            b'--b\r\nContent-Disposition: form-data; name="f%d"; '
+            b'filename="x"\r\n\r\n' % index + body + b"\r\n"
+            for index in range(count)
+        )
+        + b"--b--\r\n"
+    )
+
+
+async def test_post_max_client_size_counts_multipart_headers(
+    aiohttp_client: AiohttpClient,
+) -> None:
+    async def handler(request: web.Request) -> NoReturn:
+        await request.post()
+        assert False
+
+    app = web.Application(client_max_size=1)
+    app.router.add_post("/", handler)
+    client = await aiohttp_client(app)
+
+    body = _multipart_file_parts(55)
+    assert len(body) > 1
+    async with client.post(
+        "/", data=body, headers={CONTENT_TYPE: "multipart/form-data; boundary=b"}
+    ) as resp:
+        assert resp.status == 413
+
+
+@pytest.mark.parametrize(
+    "max_size",
+    (_FILE_SPOOL_MAX_SIZE, 2 * _FILE_SPOOL_MAX_SIZE),
+    ids=("spooled", "rolled-over"),
+)
+async def test_post_max_client_size_within_single_part(
+    aiohttp_client: AiohttpClient, max_size: int
+) -> None:
+    async def handler(request: web.Request) -> NoReturn:
+        await request.post()
+        assert False
+
+    app = web.Application(client_max_size=max_size)
+    app.router.add_post("/", handler)
+    client = await aiohttp_client(app)
+
+    # A sole oversized part is never followed by a boundary, so only the
+    # per-chunk check can reject it, and it must do so before buffering. The
+    # larger limit lets the part reach the disk first, so the spool is then
+    # closed off the event loop too.
+    data = FormData()
+    with io.BytesIO(b"x" * (3 * _FILE_SPOOL_MAX_SIZE)) as file_handle:
+        data.add_field("file", file_handle, filename="x.bin")
+        async with client.post("/", data=data) as resp:
+            assert resp.status == 413
+
+
+@pytest.mark.parametrize("filename", (b"", b'; filename="x"'), ids=("field", "file"))
+async def test_post_max_client_size_counts_undecoded_bytes(
+    aiohttp_client: AiohttpClient, filename: bytes
+) -> None:
+    async def handler(request: web.Request) -> NoReturn:
+        await request.post()
+        assert False
+
+    limit = 64 * 1024
+    app = web.Application(client_max_size=limit)
+    app.router.add_post("/", handler)
+    client = await aiohttp_client(app)
+
+    # base64 decodes to exactly the limit but arrives 4/3 larger. The limit is
+    # on the request body, so both kinds of part must reject it alike.
+    encoded = base64.b64encode(b"A" * limit)
+    assert len(encoded) > limit
+    body = (
+        b'--b\r\nContent-Disposition: form-data; name="f"' + filename + b"\r\n"
+        b"Content-Transfer-Encoding: base64\r\n\r\n" + encoded + b"\r\n--b--\r\n"
+    )
+    async with client.post(
+        "/", data=body, headers={CONTENT_TYPE: "multipart/form-data; boundary=b"}
+    ) as resp:
+        assert resp.status == 413
+
+
+@pytest.mark.skipif(not os.path.isdir("/dev/fd"), reason="needs /dev/fd to count fds")
+@pytest.mark.parametrize(
+    ("parts", "part_body", "expected_fds"),
+    (
+        (55, b"payload", 0),
+        (1, b"x" * (_FILE_SPOOL_MAX_SIZE + 1), 1),
+    ),
+    ids=("many-small-parts", "one-oversized-part"),
+)
+async def test_post_file_fields_descriptor_cost(
+    aiohttp_client: AiohttpClient, parts: int, part_body: bytes, expected_fds: int
+) -> None:
+    """Only a part past the spool size may cost a descriptor."""
+
+    async def handler(request: web.Request) -> web.Response:
+        before = set(await asyncio.to_thread(os.listdir, "/dev/fd"))
+        data = await request.post()
+        after = set(await asyncio.to_thread(os.listdir, "/dev/fd"))
+        assert len(data) == parts
+        return web.Response(text=str(len(after - before)))
+
+    app = web.Application(client_max_size=8 * 1024**2)
+    app.router.add_post("/", handler)
+    client = await aiohttp_client(app)
+
+    with io.BytesIO(_multipart_file_parts(parts, part_body)) as body:
+        async with client.post(
+            "/",
+            data=body,
+            headers={CONTENT_TYPE: "multipart/form-data; boundary=b"},
+        ) as resp:
+            assert resp.status == 200
+            assert await resp.text() == str(expected_fds)
+
+
+@pytest.mark.parametrize(
+    "size",
+    (
+        0,
+        1,
+        _FILE_SPOOL_MAX_SIZE - 1,
+        _FILE_SPOOL_MAX_SIZE,
+        _FILE_SPOOL_MAX_SIZE + 1,
+    ),
+)
+async def test_post_file_field_spool_rollover(
+    aiohttp_client: AiohttpClient, size: int
+) -> None:
+    """A part is byte-identical either side of the spool/temp-file boundary."""
+    payload = (b"0123456789abcdef" * (size // 16 + 1))[:size]
+
+    async def handler(request: web.Request) -> web.Response:
+        field = (await request.post())["file"]
+        assert isinstance(field, aiohttp.web_request.FileField)
+        assert isinstance(field.file, io.IOBase)
+        assert field.file.readable()
+        assert field.file.writable()
+        assert field.file.seekable()
+        content = await asyncio.to_thread(field.file.read)
+        assert len(content) == size
+        assert content == payload
+        return web.Response()
+
+    app = web.Application(client_max_size=8 * 1024**2)
+    app.router.add_post("/", handler)
+    client = await aiohttp_client(app)
+
+    data = FormData()
+    with io.BytesIO(payload) as file_handle:
+        data.add_field("file", file_handle, filename="x.bin")
+        async with client.post("/", data=data) as resp:
+            assert resp.status == 200
+
+
 async def test_response_with_bodypart(aiohttp_client: AiohttpClient) -> None:
     async def handler(request: web.Request) -> web.Response:
         reader = await request.multipart()