diff --git a/CHANGES/13426.bugfix.rst b/CHANGES/13426.bugfix.rst new file mode 100644 index 00000000000..8b9ee295f2d --- /dev/null +++ b/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`. diff --git a/aiohttp/web_request.py b/aiohttp/web_request.py index bafbb5fdb86..b35f6124c53 100644 --- a/aiohttp/web_request.py +++ b/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,20 +787,35 @@ 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" @@ -774,7 +823,7 @@ async def post(self) -> "MultiDictProxy[str | bytes | FileField]": 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() diff --git a/tests/test_helpers.py b/tests/test_helpers.py index 9e2092fc36b..40b128a78c6 100644 --- a/tests/test_helpers.py +++ b/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: diff --git a/tests/test_web_functional.py b/tests/test_web_functional.py index b8f61e1c2bc..0f5f7d11109 100644 --- a/tests/test_web_functional.py +++ b/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()