From 7c9340cd2be5c8dd6d829a62220bb9129e7a9d8a Mon Sep 17 00:00:00 2001 From: Sam Bull Date: Sat, 3 Jan 2026 15:56:02 +0000 Subject: [PATCH] Use decompressor max_length parameter (#11898) (#11918) --------- CVE: CVE-2025-69223 Upstream-Status: Backport [https://github.com/aio-libs/aiohttp/commit/2b920c39002cee0ec5b402581779bbaaf7c9138a] Backport Changes: - Adapted the decompression base class to aiohttp 3.9.5's direct `zlib` implementation because the newer upstream buffer and backend wrappers are absent. - Kept `BodyPartReader.decode()` and `_decode_content()` synchronous. Applied the output limit through `ZLibDecompressor.decompress_sync()` to preserve the aiohttp 3.9.5 API while bounding gzip and deflate output. - Omitted the upstream `aiohttp/web_request.py` change from `field.decode(chunk)` to `await field.decode(chunk)`. That change is required only for upstream's asynchronous `BodyPartReader.decode()` conversion. This backport keeps synchronous decoding, so the existing call remains valid. - Did not carry the Brotli 1.2 dependency updates from `pyproject.toml` and `requirements/runtime-deps.in`. Scarthgap supplies python3-brotli 1.1.0, which lacks the bounded-output API required by the upstream fix. - Disabled Brotli decoding by forcing `HAS_BROTLI` to `False`. As a result, `Content-Encoding: br` is rejected before a Brotli decoder is created. This closes the decompression-bomb path at the cost of Brotli support. - Omitted upstream Zstandard implementation, dependency, and test changes because aiohttp 3.9.5 does not support Zstandard content decoding. - Adapted the client, parser, and multipart tests to the aiohttp 3.9.5 fixtures and synchronous decoding API. (cherry picked from commit 92477c5a74c43dfe0474bd24f8de11875daa2298) Co-authored-by: J. Nick Koston (cherry picked from commit 2b920c39002cee0ec5b402581779bbaaf7c9138a) Signed-off-by: Darsh Kelaiya --- CHANGES/11898.breaking.rst | 3 ++ aiohttp/compression_utils.py | 92 +++++++++++++++++++++------------ aiohttp/http_exceptions.py | 4 ++ aiohttp/http_parser.py | 29 +++++++++-- aiohttp/multipart.py | 19 +++++-- docs/spelling_wordlist.txt | 1 + tests/test_client_functional.py | 80 +++++++++++++++++++++++++++- tests/test_http_parser.py | 44 +++++++++++++++- tests/test_multipart.py | 92 +++++++++++++++++++++++++-------- 9 files changed, 299 insertions(+), 65 deletions(-) create mode 100644 CHANGES/11898.breaking.rst diff --git a/CHANGES/11898.breaking.rst b/CHANGES/11898.breaking.rst new file mode 100644 index 000000000..228b69baa --- /dev/null +++ b/CHANGES/11898.breaking.rst @@ -0,0 +1,3 @@ +``Brotli`` decoding is disabled in the Scarthgap backport because its +``python3-brotli`` recipe provides version 1.1.0, not the required 1.2. +Decompression now has a default maximum output size of 32MiB per decompress call -- by :user:`Dreamsorcerer`. diff --git a/aiohttp/compression_utils.py b/aiohttp/compression_utils.py index 9631d377e..fe762c755 100644 --- a/aiohttp/compression_utils.py +++ b/aiohttp/compression_utils.py @@ -1,5 +1,6 @@ import asyncio import zlib +from abc import ABC, abstractmethod from concurrent.futures import Executor from typing import Optional, cast @@ -13,7 +14,17 @@ try: except ImportError: # pragma: no cover HAS_BROTLI = False -MAX_SYNC_CHUNK_SIZE = 1024 +# Scarthgap provides python3-brotli 1.1.0, whose Decompressor API does not +# support the max_length argument required by the bounded decoder below. +# Do not advertise or instantiate Brotli decoding until the recipe provides +# Brotli 1.2 or newer. +HAS_BROTLI = False + +MAX_SYNC_CHUNK_SIZE = 4096 +DEFAULT_MAX_DECOMPRESS_SIZE = 2**25 # 32MiB + +# Unlimited decompression constant +ZLIB_MAX_LENGTH_UNLIMITED = 0 # zlib uses 0 to mean unlimited def encoding_to_mode( @@ -26,19 +37,37 @@ def encoding_to_mode( return -zlib.MAX_WBITS if suppress_deflate_header else zlib.MAX_WBITS -class ZlibBaseHandler: +class DecompressionBaseHandler(ABC): def __init__( self, - mode: int, executor: Optional[Executor] = None, max_sync_chunk_size: Optional[int] = MAX_SYNC_CHUNK_SIZE, ): - self._mode = mode + """Base class for decompression handlers.""" self._executor = executor self._max_sync_chunk_size = max_sync_chunk_size + @abstractmethod + def decompress_sync( + self, data: bytes, max_length: int = ZLIB_MAX_LENGTH_UNLIMITED + ) -> bytes: + """Decompress the given data.""" + + async def decompress( + self, data: bytes, max_length: int = ZLIB_MAX_LENGTH_UNLIMITED + ) -> bytes: + """Decompress the given data.""" + if ( + self._max_sync_chunk_size is not None + and len(data) > self._max_sync_chunk_size + ): + return await asyncio.get_event_loop().run_in_executor( + self._executor, self.decompress_sync, data, max_length + ) + return self.decompress_sync(data, max_length) + -class ZLibCompressor(ZlibBaseHandler): +class ZLibCompressor: def __init__( self, encoding: Optional[str] = None, @@ -49,12 +78,12 @@ class ZLibCompressor(ZlibBaseHandler): executor: Optional[Executor] = None, max_sync_chunk_size: Optional[int] = MAX_SYNC_CHUNK_SIZE, ): - super().__init__( - mode=encoding_to_mode(encoding, suppress_deflate_header) + self._executor = executor + self._max_sync_chunk_size = max_sync_chunk_size + self._mode = ( + encoding_to_mode(encoding, suppress_deflate_header) if wbits is None - else wbits, - executor=executor, - max_sync_chunk_size=max_sync_chunk_size, + else wbits ) if level is None: self._compressor = zlib.compressobj(wbits=self._mode, strategy=strategy) @@ -86,7 +115,7 @@ class ZLibCompressor(ZlibBaseHandler): return self._compressor.flush(mode) -class ZLibDecompressor(ZlibBaseHandler): +class ZLibDecompressor(DecompressionBaseHandler): def __init__( self, encoding: Optional[str] = None, @@ -94,26 +123,15 @@ class ZLibDecompressor(ZlibBaseHandler): executor: Optional[Executor] = None, max_sync_chunk_size: Optional[int] = MAX_SYNC_CHUNK_SIZE, ): - super().__init__( - mode=encoding_to_mode(encoding, suppress_deflate_header), - executor=executor, - max_sync_chunk_size=max_sync_chunk_size, - ) + super().__init__(executor=executor, max_sync_chunk_size=max_sync_chunk_size) + self._mode = encoding_to_mode(encoding, suppress_deflate_header) self._decompressor = zlib.decompressobj(wbits=self._mode) - def decompress_sync(self, data: bytes, max_length: int = 0) -> bytes: + def decompress_sync( + self, data: bytes, max_length: int = ZLIB_MAX_LENGTH_UNLIMITED + ) -> bytes: return self._decompressor.decompress(data, max_length) - async def decompress(self, data: bytes, max_length: int = 0) -> bytes: - if ( - self._max_sync_chunk_size is not None - and len(data) > self._max_sync_chunk_size - ): - return await asyncio.get_event_loop().run_in_executor( - self._executor, self.decompress_sync, data, max_length - ) - return self.decompress_sync(data, max_length) - def flush(self, length: int = 0) -> bytes: return ( self._decompressor.flush(length) @@ -134,24 +152,34 @@ class ZLibDecompressor(ZlibBaseHandler): return self._decompressor.unused_data -class BrotliDecompressor: +class BrotliDecompressor(DecompressionBaseHandler): # Supports both 'brotlipy' and 'Brotli' packages # since they share an import name. The top branches # are for 'brotlipy' and bottom branches for 'Brotli' - def __init__(self) -> None: + def __init__( + self, + executor: Optional[Executor] = None, + max_sync_chunk_size: Optional[int] = MAX_SYNC_CHUNK_SIZE, + ) -> None: + """Decompress data using the Brotli library.""" if not HAS_BROTLI: raise RuntimeError( "The brotli decompression is not available. " "Please install `Brotli` module" ) self._obj = brotli.Decompressor() + super().__init__(executor=executor, max_sync_chunk_size=max_sync_chunk_size) - def decompress_sync(self, data: bytes) -> bytes: + def decompress_sync( + self, data: bytes, max_length: int = ZLIB_MAX_LENGTH_UNLIMITED + ) -> bytes: + """Decompress the given data.""" if hasattr(self._obj, "decompress"): - return cast(bytes, self._obj.decompress(data)) - return cast(bytes, self._obj.process(data)) + return cast(bytes, self._obj.decompress(data, max_length)) + return cast(bytes, self._obj.process(data, max_length)) def flush(self) -> bytes: + """Flush the decompressor.""" if hasattr(self._obj, "flush"): return cast(bytes, self._obj.flush()) return b"" diff --git a/aiohttp/http_exceptions.py b/aiohttp/http_exceptions.py index 72eac3a3c..877b07d4c 100644 --- a/aiohttp/http_exceptions.py +++ b/aiohttp/http_exceptions.py @@ -75,6 +75,10 @@ class ContentLengthError(PayloadEncodingError): """Not enough data for satisfy content length header.""" +class DecompressSizeError(PayloadEncodingError): + """Decompressed size exceeds the configured limit.""" + + class LineTooLong(BadHttpMessage): def __init__( self, line: str, limit: str = "Unknown", actual_size: str = "Unknown" diff --git a/aiohttp/http_parser.py b/aiohttp/http_parser.py index 5768bd623..cdf3fc89a 100644 --- a/aiohttp/http_parser.py +++ b/aiohttp/http_parser.py @@ -26,7 +26,12 @@ from yarl import URL from . import hdrs from .base_protocol import BaseProtocol -from .compression_utils import HAS_BROTLI, BrotliDecompressor, ZLibDecompressor +from .compression_utils import ( + DEFAULT_MAX_DECOMPRESS_SIZE, + HAS_BROTLI, + BrotliDecompressor, + ZLibDecompressor, +) from .helpers import ( _EXC_SENTINEL, DEBUG, @@ -41,6 +46,7 @@ from .http_exceptions import ( BadStatusLine, ContentEncodingError, ContentLengthError, + DecompressSizeError, InvalidHeader, InvalidURLError, LineTooLong, @@ -959,7 +965,12 @@ class DeflateBuffer: decompressor: Any - def __init__(self, out: StreamReader, encoding: Optional[str]) -> None: + def __init__( + self, + out: StreamReader, + encoding: Optional[str], + max_decompress_size: int = DEFAULT_MAX_DECOMPRESS_SIZE, + ) -> None: self.out = out self.size = 0 self.encoding = encoding @@ -976,6 +987,8 @@ class DeflateBuffer: else: self.decompressor = ZLibDecompressor(encoding=encoding) + self._max_decompress_size = max_decompress_size + def set_exception( self, exc: BaseException, @@ -1004,7 +1017,10 @@ class DeflateBuffer: ) try: - chunk = self.decompressor.decompress_sync(chunk) + # Decompress with limit + 1 so we can detect if output exceeds limit + chunk = self.decompressor.decompress_sync( + chunk, max_length=self._max_decompress_size + 1 + ) except Exception: raise ContentEncodingError( "Can not decode content-encoding: %s" % self.encoding @@ -1012,6 +1028,13 @@ class DeflateBuffer: self._started_decoding = True + # Check if decompression limit was exceeded + if len(chunk) > self._max_decompress_size: + raise DecompressSizeError( + "Decompressed data exceeds the configured limit of %d bytes" + % self._max_decompress_size + ) + if chunk: self.out.feed_data(chunk, len(chunk)) diff --git a/aiohttp/multipart.py b/aiohttp/multipart.py index 9e5ff9b41..baf07fd16 100644 --- a/aiohttp/multipart.py +++ b/aiohttp/multipart.py @@ -27,7 +27,12 @@ from urllib.parse import parse_qsl, unquote, urlencode from multidict import CIMultiDict, CIMultiDictProxy -from .compression_utils import ZLibCompressor, ZLibDecompressor +from .abc import AbstractStreamWriter +from .compression_utils import ( + DEFAULT_MAX_DECOMPRESS_SIZE, + ZLibCompressor, + ZLibDecompressor, +) from .hdrs import ( CONTENT_DISPOSITION, CONTENT_ENCODING, @@ -263,6 +268,7 @@ class BodyPartReader: *, subtype: str = "mixed", default_charset: Optional[str] = None, + max_decompress_size: int = DEFAULT_MAX_DECOMPRESS_SIZE, ) -> None: self.headers = headers self._boundary = boundary @@ -278,6 +284,7 @@ class BodyPartReader: self._prev_chunk: Optional[bytes] = None self._content_eof = 0 self._cache: Dict[str, Any] = {} + self._max_decompress_size = max_decompress_size def __aiter__(self) -> AsyncIterator["BodyPartReader"]: return self # type: ignore[return-value] @@ -471,7 +478,7 @@ class BodyPartReader: return ZLibDecompressor( encoding=encoding, suppress_deflate_header=True, - ).decompress_sync(data) + ).decompress_sync(data, max_length=self._max_decompress_size) raise RuntimeError(f"unknown content encoding: {encoding}") @@ -528,7 +535,7 @@ class BodyPartReaderPayload(Payload): if params: self.set_content_disposition("attachment", True, **params) - async def write(self, writer: Any) -> None: + async def write(self, writer: AbstractStreamWriter) -> None: field = self._value chunk = await field.read_chunk(size=2**16) while chunk: @@ -927,7 +934,9 @@ class MultipartWriter(Payload): total += 2 + len(self._boundary) + 4 # b'--'+self._boundary+b'--\r\n' return total - async def write(self, writer: Any, close_boundary: bool = True) -> None: + async def write( + self, writer: AbstractStreamWriter, close_boundary: bool = True + ) -> None: """Write body.""" for part, encoding, te_encoding in self._parts: if self._is_form_data: @@ -956,7 +965,7 @@ class MultipartWriter(Payload): class MultipartPayloadWriter: - def __init__(self, writer: Any) -> None: + def __init__(self, writer: AbstractStreamWriter) -> None: self._writer = writer self._encoding: Optional[str] = None self._compress: Optional[ZLibCompressor] = None diff --git a/docs/spelling_wordlist.txt b/docs/spelling_wordlist.txt index 514477e8f..34399e6ba 100644 --- a/docs/spelling_wordlist.txt +++ b/docs/spelling_wordlist.txt @@ -182,6 +182,7 @@ lowercased Mako manylinux metadata +MiB microservice middleware middlewares diff --git a/tests/test_client_functional.py b/tests/test_client_functional.py index dbb2dff5a..7d126d185 100644 --- a/tests/test_client_functional.py +++ b/tests/test_client_functional.py @@ -8,9 +8,18 @@ import json import pathlib import socket import ssl +import zlib from typing import Any, AsyncIterator from unittest import mock +try: + try: + import brotlicffi as brotli + except ImportError: + import brotli +except ImportError: + brotli = None # pragma: no cover + import pytest from multidict import MultiDict from yarl import URL @@ -19,6 +28,8 @@ import aiohttp from aiohttp import Fingerprint, ServerFingerprintMismatch, hdrs, web from aiohttp.abc import AbstractResolver from aiohttp.client_exceptions import TooManyRedirects +from aiohttp.compression_utils import DEFAULT_MAX_DECOMPRESS_SIZE, HAS_BROTLI +from aiohttp.http_exceptions import DecompressSizeError from aiohttp.pytest_plugin import AiohttpClient, TestClient from aiohttp.test_utils import unused_port @@ -1903,8 +1914,73 @@ async def test_bad_payload_compression(aiohttp_client) -> None: resp.close() -async def test_bad_payload_chunked_encoding(aiohttp_client) -> None: - async def handler(request): +async def test_payload_decompress_size_limit(aiohttp_client: AiohttpClient) -> None: + """Test that decompression size limit triggers DecompressSizeError. + + When a compressed payload expands beyond the configured limit, + we raise DecompressSizeError. + """ + # Create a highly compressible payload that exceeds the decompression limit. + # 64MiB of repeated bytes compresses to ~32KB but expands beyond the + # 32MiB per-call limit. + original = b"A" * (64 * 2**20) + compressed = zlib.compress(original) + assert len(original) > DEFAULT_MAX_DECOMPRESS_SIZE + + async def handler(request: web.Request) -> web.Response: + # Send compressed data with Content-Encoding header + resp = web.Response(body=compressed) + resp.headers["Content-Encoding"] = "deflate" + return resp + + app = web.Application() + app.router.add_get("/", handler) + client = await aiohttp_client(app) + + async with client.get("/") as resp: + assert resp.status == 200 + + with pytest.raises(aiohttp.ClientPayloadError) as exc_info: + await resp.read() + + assert isinstance(exc_info.value.__cause__, DecompressSizeError) + assert "Decompressed data exceeds" in str(exc_info.value.__cause__) + + +@pytest.mark.skipif( + brotli is None or not HAS_BROTLI, reason="brotli decoding is unavailable" +) +async def test_payload_decompress_size_limit_brotli( + aiohttp_client: AiohttpClient, +) -> None: + """Test that brotli decompression size limit triggers DecompressSizeError.""" + assert brotli is not None + # Create a highly compressible payload that exceeds the decompression limit. + original = b"A" * (64 * 2**20) + compressed = brotli.compress(original) + assert len(original) > DEFAULT_MAX_DECOMPRESS_SIZE + + async def handler(request: web.Request) -> web.Response: + resp = web.Response(body=compressed) + resp.headers["Content-Encoding"] = "br" + return resp + + app = web.Application() + app.router.add_get("/", handler) + client = await aiohttp_client(app) + + async with client.get("/") as resp: + assert resp.status == 200 + + with pytest.raises(aiohttp.ClientPayloadError) as exc_info: + await resp.read() + + assert isinstance(exc_info.value.__cause__, DecompressSizeError) + assert "Decompressed data exceeds" in str(exc_info.value.__cause__) + + +async def test_bad_payload_chunked_encoding(aiohttp_client: AiohttpClient) -> None: + async def handler(request: web.Request) -> web.StreamResponse: resp = web.StreamResponse() resp.force_close() resp._length_check = False diff --git a/tests/test_http_parser.py b/tests/test_http_parser.py index 0fcefdefd..9449c4061 100644 --- a/tests/test_http_parser.py +++ b/tests/test_http_parser.py @@ -2,6 +2,7 @@ import asyncio import re +import zlib from contextlib import nullcontext from typing import Any, Dict, List from unittest import mock @@ -14,6 +15,7 @@ from yarl import URL import aiohttp from aiohttp import http_exceptions, streams from aiohttp.base_protocol import BaseProtocol +from aiohttp.compression_utils import HAS_BROTLI from aiohttp.http_parser import ( NO_EXTENSIONS, DeflateBuffer, @@ -561,7 +563,9 @@ def test_compression_gzip(parser) -> None: assert msg.compression == "gzip" -@pytest.mark.skipif(brotli is None, reason="brotli is not installed") +@pytest.mark.skipif( + brotli is None or not HAS_BROTLI, reason="brotli decoding is unavailable" +) def test_compression_brotli(parser) -> None: text = b"GET /test HTTP/1.1\r\n" b"content-encoding: br\r\n\r\n" messages, upgrade, tail = parser.feed_data(text) @@ -1736,7 +1740,9 @@ class TestParsePayload: assert p.done assert out.is_eof() - @pytest.mark.skipif(brotli is None, reason="brotli is not installed") + @pytest.mark.skipif( + brotli is None or not HAS_BROTLI, reason="brotli decoding is unavailable" + ) async def test_http_payload_brotli(self, stream) -> None: compressed = brotli.compress(b"brotli data") out = aiohttp.FlowControlDataQueue( @@ -1816,6 +1822,7 @@ class TestDeflateBuffer: dbuf.feed_eof() assert [b"line"] == list(d for d, _ in buf._buffer) + @pytest.mark.skipif(not HAS_BROTLI, reason="brotli decoding is unavailable") async def test_feed_eof_no_err_brotli(self, stream) -> None: buf = aiohttp.FlowControlDataQueue( stream, 2**16, loop=asyncio.get_event_loop() @@ -1837,3 +1844,36 @@ class TestDeflateBuffer: dbuf.feed_eof() assert buf.at_eof() + + @pytest.mark.parametrize( + "chunk_size", + [1024, 2**14, 2**16], # 1KB, 16KB, 64KB + ids=["1KB", "16KB", "64KB"], + ) + async def test_streaming_decompress_large_payload( + self, protocol: BaseProtocol, chunk_size: int + ) -> None: + """Test that large payloads decompress correctly when streamed in chunks. + + This simulates real HTTP streaming where compressed data arrives in + small network chunks. Each chunk's decompressed output should be within + the max_decompress_size limit, allowing full recovery of the original data. + """ + # Create a large payload (3MiB) that compresses well + original = b"A" * (3 * 2**20) + compressed = zlib.compress(original) + + buf = aiohttp.StreamReader(protocol, 2**16, loop=asyncio.get_running_loop()) + dbuf = DeflateBuffer(buf, "deflate") + + # Feed compressed data in chunks (simulating network streaming) + for i in range(0, len(compressed), chunk_size): + chunk = compressed[i : i + chunk_size] + dbuf.feed_data(chunk, len(chunk)) + + dbuf.feed_eof() + + # Read all decompressed data + result = b"".join(buf._buffer) + assert len(result) == len(original) + assert result == original diff --git a/tests/test_multipart.py b/tests/test_multipart.py index e4a2be1f3..553085ca5 100644 --- a/tests/test_multipart.py +++ b/tests/test_multipart.py @@ -9,6 +9,7 @@ import pytest import aiohttp from aiohttp import payload +from aiohttp.abc import AbstractStreamWriter from aiohttp.hdrs import ( CONTENT_DISPOSITION, CONTENT_ENCODING, @@ -32,14 +33,14 @@ def buf(): @pytest.fixture -def stream(buf): - writer = mock.Mock() +def stream(buf: bytearray) -> AbstractStreamWriter: + writer = mock.create_autospec(AbstractStreamWriter, instance=True, spec_set=True) async def write(chunk): buf.extend(chunk) writer.write.side_effect = write - return writer + return writer # type: ignore[no-any-return] @pytest.fixture @@ -336,6 +337,17 @@ class TestPartReader: result = await obj.read(decode=True) assert b"Time to Relax!" == result + def test_decode_remains_synchronous(self) -> None: + data = b"\x0b\xc9\xccMU(\xc9W\x08J\xcdI\xacP\x04\x00" + with Stream(b"") as stream: + obj = aiohttp.BodyPartReader( + BOUNDARY, + {CONTENT_ENCODING: "deflate"}, + stream, + ) + result = obj.decode(data) + assert b"Time to Relax!" == result + async def test_read_with_content_encoding_identity(self) -> None: thing = ( b"\x1f\x8b\x08\x00\x00\x00\x00\x00\x00\x03\x0b\xc9\xccMU" @@ -1012,7 +1024,9 @@ async def test_writer(writer) -> None: assert writer.boundary == ":" -async def test_writer_serialize_io_chunk(buf, stream, writer) -> None: +async def test_writer_serialize_io_chunk( + buf: bytearray, stream: AbstractStreamWriter, writer: aiohttp.MultipartWriter +) -> None: with io.BytesIO(b"foobarbaz") as file_handle: writer.append(file_handle) await writer.write(stream) @@ -1022,7 +1036,9 @@ async def test_writer_serialize_io_chunk(buf, stream, writer) -> None: ) -async def test_writer_serialize_json(buf, stream, writer) -> None: +async def test_writer_serialize_json( + buf: bytearray, stream: AbstractStreamWriter, writer: aiohttp.MultipartWriter +) -> None: writer.append_json({"привет": "мир"}) await writer.write(stream) assert ( @@ -1031,7 +1047,9 @@ async def test_writer_serialize_json(buf, stream, writer) -> None: ) -async def test_writer_serialize_form(buf, stream, writer) -> None: +async def test_writer_serialize_form( + buf: bytearray, stream: AbstractStreamWriter, writer: aiohttp.MultipartWriter +) -> None: data = [("foo", "bar"), ("foo", "baz"), ("boo", "zoo")] writer.append_form(data) await writer.write(stream) @@ -1039,7 +1057,9 @@ async def test_writer_serialize_form(buf, stream, writer) -> None: assert b"foo=bar&foo=baz&boo=zoo" in buf -async def test_writer_serialize_form_dict(buf, stream, writer) -> None: +async def test_writer_serialize_form_dict( + buf: bytearray, stream: AbstractStreamWriter, writer: aiohttp.MultipartWriter +) -> None: data = {"hello": "мир"} writer.append_form(data) await writer.write(stream) @@ -1047,7 +1067,9 @@ async def test_writer_serialize_form_dict(buf, stream, writer) -> None: assert b"hello=%D0%BC%D0%B8%D1%80" in buf -async def test_writer_write(buf, stream, writer) -> None: +async def test_writer_write( + buf: bytearray, stream: AbstractStreamWriter, writer: aiohttp.MultipartWriter +) -> None: writer.append("foo-bar-baz") writer.append_json({"test": "passed"}) writer.append_form({"test": "passed"}) @@ -1093,7 +1115,9 @@ async def test_writer_write(buf, stream, writer) -> None: ) == bytes(buf) -async def test_writer_write_no_close_boundary(buf, stream) -> None: +async def test_writer_write_no_close_boundary( + buf: bytearray, stream: AbstractStreamWriter +) -> None: writer = aiohttp.MultipartWriter(boundary=":") writer.append("foo-bar-baz") writer.append_json({"test": "passed"}) @@ -1125,12 +1149,18 @@ async def test_writer_write_no_close_boundary(buf, stream) -> None: ) == bytes(buf) -async def test_writer_write_no_parts(buf, stream, writer) -> None: +async def test_writer_write_no_parts( + buf: bytearray, stream: AbstractStreamWriter, writer: aiohttp.MultipartWriter +) -> None: await writer.write(stream) assert b"--:--\r\n" == bytes(buf) -async def test_writer_serialize_with_content_encoding_gzip(buf, stream, writer): +async def test_writer_serialize_with_content_encoding_gzip( + buf: bytearray, + stream: AbstractStreamWriter, + writer: aiohttp.MultipartWriter, +) -> None: writer.append("Time to Relax!", {CONTENT_ENCODING: "gzip"}) await writer.write(stream) headers, message = bytes(buf).split(b"\r\n\r\n", 1) @@ -1146,7 +1176,9 @@ async def test_writer_serialize_with_content_encoding_gzip(buf, stream, writer): assert b"Time to Relax!" == data -async def test_writer_serialize_with_content_encoding_deflate(buf, stream, writer): +async def test_writer_serialize_with_content_encoding_deflate( + buf: bytearray, stream: AbstractStreamWriter, writer: aiohttp.MultipartWriter +) -> None: writer.append("Time to Relax!", {CONTENT_ENCODING: "deflate"}) await writer.write(stream) headers, message = bytes(buf).split(b"\r\n\r\n", 1) @@ -1160,7 +1192,9 @@ async def test_writer_serialize_with_content_encoding_deflate(buf, stream, write assert thing == message -async def test_writer_serialize_with_content_encoding_identity(buf, stream, writer): +async def test_writer_serialize_with_content_encoding_identity( + buf: bytearray, stream: AbstractStreamWriter, writer: aiohttp.MultipartWriter +) -> None: thing = b"\x0b\xc9\xccMU(\xc9W\x08J\xcdI\xacP\x04\x00" writer.append(thing, {CONTENT_ENCODING: "identity"}) await writer.write(stream) @@ -1175,12 +1209,16 @@ async def test_writer_serialize_with_content_encoding_identity(buf, stream, writ assert thing == message.split(b"\r\n")[0] -def test_writer_serialize_with_content_encoding_unknown(buf, stream, writer): +def test_writer_serialize_with_content_encoding_unknown( + buf: bytearray, stream: AbstractStreamWriter, writer: aiohttp.MultipartWriter +) -> None: with pytest.raises(RuntimeError): writer.append("Time to Relax!", {CONTENT_ENCODING: "snappy"}) -async def test_writer_with_content_transfer_encoding_base64(buf, stream, writer): +async def test_writer_with_content_transfer_encoding_base64( + buf: bytearray, stream: AbstractStreamWriter, writer: aiohttp.MultipartWriter +) -> None: writer.append("Time to Relax!", {CONTENT_TRANSFER_ENCODING: "base64"}) await writer.write(stream) headers, message = bytes(buf).split(b"\r\n\r\n", 1) @@ -1193,7 +1231,9 @@ async def test_writer_with_content_transfer_encoding_base64(buf, stream, writer) assert b"VGltZSB0byBSZWxheCE=" == message.split(b"\r\n")[0] -async def test_writer_content_transfer_encoding_quote_printable(buf, stream, writer): +async def test_writer_content_transfer_encoding_quote_printable( + buf: bytearray, stream: AbstractStreamWriter, writer: aiohttp.MultipartWriter +) -> None: writer.append("Привет, мир!", {CONTENT_TRANSFER_ENCODING: "quoted-printable"}) await writer.write(stream) headers, message = bytes(buf).split(b"\r\n\r\n", 1) @@ -1209,7 +1249,9 @@ async def test_writer_content_transfer_encoding_quote_printable(buf, stream, wri ) -def test_writer_content_transfer_encoding_unknown(buf, stream, writer) -> None: +def test_writer_content_transfer_encoding_unknown( + buf: bytearray, stream: AbstractStreamWriter, writer: aiohttp.MultipartWriter +) -> None: with pytest.raises(RuntimeError): writer.append("Time to Relax!", {CONTENT_TRANSFER_ENCODING: "unknown"}) @@ -1333,7 +1375,9 @@ class TestMultipartWriter: with aiohttp.MultipartWriter(boundary=":") as writer: writer.append(None) - async def test_write_preserves_content_disposition(self, buf, stream) -> None: + async def test_write_preserves_content_disposition( + self, buf: bytearray, stream: AbstractStreamWriter + ) -> None: with aiohttp.MultipartWriter(boundary=":") as writer: part = writer.append(b"foo", headers={CONTENT_TYPE: "test/passed"}) part.set_content_disposition("form-data", filename="bug") @@ -1350,7 +1394,9 @@ class TestMultipartWriter: ) assert message == b"foo\r\n--:--\r\n" - async def test_preserve_content_disposition_header(self, buf, stream): + async def test_preserve_content_disposition_header( + self, buf: bytearray, stream: AbstractStreamWriter + ) -> None: # https://github.com/aio-libs/aiohttp/pull/3475#issuecomment-451072381 with pathlib.Path(__file__).open("rb") as fobj: with aiohttp.MultipartWriter("form-data", boundary=":") as writer: @@ -1374,7 +1420,9 @@ class TestMultipartWriter: b'Content-Disposition: attachments; filename="bug.py"' ) - async def test_set_content_disposition_override(self, buf, stream): + async def test_set_content_disposition_override( + self, buf: bytearray, stream: AbstractStreamWriter + ) -> None: # https://github.com/aio-libs/aiohttp/pull/3475#issuecomment-451072381 with pathlib.Path(__file__).open("rb") as fobj: with aiohttp.MultipartWriter("form-data", boundary=":") as writer: @@ -1398,7 +1446,9 @@ class TestMultipartWriter: b'Content-Disposition: attachments; filename="bug.py"' ) - async def test_reset_content_disposition_header(self, buf, stream): + async def test_reset_content_disposition_header( + self, buf: bytearray, stream: AbstractStreamWriter + ) -> None: # https://github.com/aio-libs/aiohttp/pull/3475#issuecomment-451072381 with pathlib.Path(__file__).open("rb") as fobj: with aiohttp.MultipartWriter("form-data", boundary=":") as writer: -- 2.35.6