Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/aiohttp/multipart.py: 19%
Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1import base64
2import binascii
3import builtins
4import json
5import re
6import sys
7import uuid
8import warnings
9from collections import deque
10from collections.abc import AsyncIterator, Iterator, Mapping, Sequence
11from types import TracebackType
12from typing import TYPE_CHECKING, Any, TypeVar, Union, cast
13from urllib.parse import parse_qsl, unquote, urlencode
15from multidict import CIMultiDict, CIMultiDictProxy
17from .abc import AbstractStreamWriter
18from .compression_utils import ZLibCompressor, ZLibDecompressor
19from .hdrs import (
20 CONTENT_DISPOSITION,
21 CONTENT_ENCODING,
22 CONTENT_LENGTH,
23 CONTENT_TRANSFER_ENCODING,
24 CONTENT_TYPE,
25)
26from .helpers import CHAR, DEFAULT_CHUNK_SIZE, TOKEN, parse_mimetype, reify
27from .http import HeadersParser
28from .http_exceptions import BadHttpMessage
29from .log import internal_logger
30from .payload import (
31 JsonPayload,
32 LookupError,
33 Order,
34 Payload,
35 StringPayload,
36 get_payload,
37 payload_type,
38)
39from .streams import StreamReader
41if sys.version_info >= (3, 11):
42 from typing import Self
43else:
44 Self = TypeVar("Self", bound="BodyPartReader")
46if sys.version_info >= (3, 12):
47 from collections.abc import Buffer
48else:
49 Buffer = Union[bytes, bytearray, "memoryview[int]", "memoryview[bytes]"]
51_Buffer = TypeVar("_Buffer", bound=Buffer)
53__all__ = (
54 "MultipartReader",
55 "MultipartWriter",
56 "BodyPartReader",
57 "BadContentDispositionHeader",
58 "BadContentDispositionParam",
59 "parse_content_disposition",
60 "content_disposition_filename",
61)
64# The base64 alphabet plus the padding character.
65_BASE64_CHARS = frozenset(
66 b"ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/="
67)
68_NON_BASE64_BYTES = bytes(b for b in range(256) if b not in _BASE64_CHARS)
71if TYPE_CHECKING:
72 from .client_reqrep import ClientResponse
75class BadContentDispositionHeader(RuntimeWarning):
76 pass
79class BadContentDispositionParam(RuntimeWarning):
80 pass
83def parse_content_disposition(
84 header: str | None,
85) -> tuple[str | None, dict[str, str]]:
86 def is_token(string: str) -> bool:
87 return bool(string) and TOKEN >= set(string)
89 def is_quoted(string: str) -> bool:
90 return len(string) >= 2 and string[0] == string[-1] == '"'
92 def is_rfc5987(string: str) -> bool:
93 return is_token(string) and string.count("'") == 2
95 def is_extended_param(string: str) -> bool:
96 return string.endswith("*")
98 def is_continuous_param(string: str) -> bool:
99 pos = string.find("*") + 1
100 if not pos:
101 return False
102 substring = string[pos:-1] if string.endswith("*") else string[pos:]
103 return substring.isdigit()
105 def unescape(text: str, *, chars: str = "".join(map(re.escape, CHAR))) -> str:
106 return re.sub(f"\\\\([{chars}])", "\\1", text)
108 if not header:
109 return None, {}
111 # https://www.rfc-editor.org/info/rfc9110/#section-5.6.6-2
112 disptype, *parts = header.split(";")
113 disptype = disptype.strip()
114 if not is_token(disptype):
115 warnings.warn(BadContentDispositionHeader(header))
116 return None, {}
118 params: dict[str, str] = {}
119 while parts:
120 item = parts.pop(0)
122 if not item: # To handle trailing semicolons
123 warnings.warn(BadContentDispositionHeader(header))
124 continue
126 if "=" not in item:
127 warnings.warn(BadContentDispositionHeader(header))
128 return None, {}
130 key, value = item.split("=", 1)
131 key = key.lower().strip()
132 value = value.lstrip()
134 if key in params:
135 warnings.warn(BadContentDispositionHeader(header))
136 return None, {}
138 if not is_token(key):
139 warnings.warn(BadContentDispositionParam(item))
140 continue
142 elif is_continuous_param(key):
143 if is_quoted(value):
144 value = unescape(value[1:-1])
145 elif not is_token(value):
146 warnings.warn(BadContentDispositionParam(item))
147 continue
149 elif is_extended_param(key):
150 if is_rfc5987(value):
151 encoding, _, value = value.split("'", 2)
152 encoding = encoding or "utf-8"
153 else:
154 warnings.warn(BadContentDispositionParam(item))
155 continue
157 try:
158 value = unquote(value, encoding, "strict")
159 except (builtins.LookupError, UnicodeDecodeError):
160 # The charset is attacker-controlled here; an unknown name
161 # raises the builtin LookupError (the bare name is shadowed in
162 # this module by payload.LookupError).
163 warnings.warn(BadContentDispositionParam(item))
164 continue
166 else:
167 failed = True
168 rstripped = value.rstrip()
169 if is_quoted(rstripped):
170 failed = False
171 value = unescape(rstripped[1:-1].lstrip("\\/"))
172 elif is_token(value):
173 failed = False
174 elif parts:
175 # maybe just ; in filename, in any case this is just
176 # one case fix, for proper fix we need to redesign parser
177 _value = f"{value};{parts[0]}"
178 if is_quoted(_value):
179 parts.pop(0)
180 value = unescape(_value[1:-1].lstrip("\\/"))
181 failed = False
183 if failed:
184 warnings.warn(BadContentDispositionHeader(header))
185 return None, {}
187 params[key] = value
189 return disptype.lower(), params
192def content_disposition_filename(
193 params: Mapping[str, str], name: str = "filename"
194) -> str | None:
195 name_suf = "%s*" % name
196 if not params:
197 return None
198 elif name_suf in params:
199 return params[name_suf]
200 elif name in params:
201 return params[name]
202 else:
203 parts = []
204 fnparams = sorted(
205 (key, value) for key, value in params.items() if key.startswith(name_suf)
206 )
207 for num, (key, value) in enumerate(fnparams):
208 _, tail = key.split("*", 1)
209 if tail.endswith("*"):
210 tail = tail[:-1]
211 if tail == str(num):
212 parts.append(value)
213 else:
214 break
215 if not parts:
216 return None
217 value = "".join(parts)
218 if "'" in value:
219 encoding, _, value = value.split("'", 2)
220 encoding = encoding or "utf-8"
221 try:
222 return unquote(value, encoding, "strict")
223 except (builtins.LookupError, UnicodeDecodeError):
224 # Both the charset name and the octets are attacker-controlled
225 # here; an unknown encoding raises the builtin LookupError
226 # (shadowed in this module by payload.LookupError) and
227 # undecodable bytes raise UnicodeDecodeError.
228 return None
229 return value
232class MultipartResponseWrapper:
233 """Wrapper around the MultipartReader.
235 It takes care about
236 underlying connection and close it when it needs in.
237 """
239 def __init__(
240 self,
241 resp: "ClientResponse",
242 stream: "MultipartReader",
243 ) -> None:
244 self.resp = resp
245 self.stream = stream
247 def __aiter__(self) -> "MultipartResponseWrapper":
248 return self
250 async def __anext__(
251 self,
252 ) -> Union["MultipartReader", "BodyPartReader"]:
253 part = await self.next()
254 if part is None:
255 raise StopAsyncIteration
256 return part
258 def at_eof(self) -> bool:
259 """Returns True when all response data had been read."""
260 return self.resp.content.at_eof()
262 async def next(
263 self,
264 ) -> Union["MultipartReader", "BodyPartReader"] | None:
265 """Emits next multipart reader object."""
266 item = await self.stream.next()
267 if self.stream.at_eof():
268 await self.release()
269 return item
271 async def release(self) -> None:
272 """Release the connection gracefully.
274 All remaining content is read to the void.
275 """
276 await self.resp.release()
279class BodyPartReader:
280 """Multipart reader for single body part."""
282 chunk_size = 8192
284 def __init__(
285 self,
286 boundary: bytes,
287 headers: "CIMultiDictProxy[str]",
288 content: StreamReader,
289 *,
290 subtype: str = "mixed",
291 default_charset: str | None = None,
292 max_decompress_size: int = DEFAULT_CHUNK_SIZE,
293 client_max_size: int = sys.maxsize,
294 max_size_error_cls: type[Exception] = ValueError,
295 ) -> None:
296 self.headers = headers
297 self._boundary = boundary
298 self._boundary_len = len(boundary) + 2 # Boundary + \r\n
299 self._content = content
300 self._default_charset = default_charset
301 self._at_eof = False
302 self._is_form_data = subtype == "form-data"
303 # https://datatracker.ietf.org/doc/html/rfc7578#section-4.8
304 length = None if self._is_form_data else self.headers.get(CONTENT_LENGTH, None)
305 if length is not None and not (length.isascii() and length.isdigit()):
306 # Reject sign prefixes, underscores, whitespace and non-ASCII
307 # digits that int() would otherwise accept.
308 # https://www.rfc-editor.org/rfc/rfc9110#section-8.6
309 raise ValueError(f"invalid Content-Length: {length!r}")
310 self._length = int(length) if length is not None else None
311 self._read_bytes = 0
312 self._b64_carry = b""
313 self._unread: deque[bytes] = deque()
314 self._prev_chunk: bytes | None = None
315 self._content_eof = 0
316 self._cache: dict[str, Any] = {}
317 self._max_decompress_size = max_decompress_size
318 self._client_max_size = client_max_size
319 self._max_size_error_cls = max_size_error_cls
321 def __aiter__(self: Self) -> Self:
322 return self
324 async def __anext__(self) -> bytes:
325 part = await self.next()
326 if part is None:
327 raise StopAsyncIteration
328 return part
330 async def next(self) -> bytes | None:
331 item = await self.read()
332 if not item:
333 return None
334 return item
336 async def read(self, *, decode: bool = False) -> bytes:
337 """Reads body part data.
339 decode: Decodes data following by encoding
340 method from Content-Encoding header. If it missed
341 data remains untouched
342 """
343 if self._at_eof:
344 return b""
345 data = bytearray()
346 while not self._at_eof:
347 data.extend(await self.read_chunk(self.chunk_size))
348 if len(data) > self._client_max_size:
349 raise self._max_size_error_cls(self._client_max_size)
350 if decode:
351 decoded_data = bytearray()
352 async for d in self.decode_iter(data):
353 decoded_data.extend(d)
354 if len(decoded_data) > self._client_max_size:
355 raise self._max_size_error_cls(self._client_max_size)
356 return decoded_data
357 return data
359 async def read_chunk(self, size: int = chunk_size) -> bytes:
360 """Reads body part content chunk of the specified size.
362 size: chunk size
363 """
364 if self._at_eof:
365 return b""
366 carry = self._b64_carry
367 want = size - len(carry)
368 if carry:
369 self._b64_carry = b""
370 want = max(want, self._boundary_len)
371 if self._length is not None:
372 fresh = await self._read_chunk_from_length(want)
373 else:
374 fresh = await self._read_chunk_from_stream(want)
375 chunk = carry + fresh
376 self._read_bytes += len(fresh)
378 # base64 decodes in quartets and every chunk is decoded on its own, so
379 # a chunk should not end mid-quartet.
380 encoding = self.headers.get(CONTENT_TRANSFER_ENCODING)
381 if encoding and encoding.lower() == "base64":
382 chunk = self._align_base64_chunk(chunk, len(carry) + want)
384 if self._read_bytes == self._length:
385 self._at_eof = True
386 if self._at_eof and await self._content.readline() != b"\r\n":
387 raise ValueError("Reader did not read all the data or it is malformed")
388 return chunk
390 def _align_base64_chunk(self, chunk: bytes, size: int) -> bytes:
391 at_end = self._at_eof or (
392 self._length is not None and self._read_bytes >= self._length
393 )
394 if not at_end and len(chunk) > size:
395 self._b64_carry = chunk[size:]
396 chunk = chunk[:size]
398 remainder = len(chunk.translate(None, _NON_BASE64_BYTES)) % 4
399 if not remainder or at_end:
400 return chunk
402 # Walk back over the trailing partial quartet and carry it into the
403 # next chunk.
404 cut = len(chunk)
405 left = remainder
406 while left:
407 cut -= 1
408 if chunk[cut] in _BASE64_CHARS:
409 left -= 1
410 if not cut:
411 # No whole quartet to hand back, and carrying the lot would make
412 # no progress: the caller asked for this many bytes, and a part
413 # that holds no quartet within them holds none to give.
414 return chunk
416 self._b64_carry = chunk[cut:] + self._b64_carry
417 return chunk[:cut]
419 async def _read_chunk_from_length(self, size: int) -> bytes:
420 # Reads body part content chunk of the specified size.
421 # The body part must has Content-Length header with proper value.
422 assert self._length is not None, "Content-Length required for chunked read"
423 chunk_size = min(size, self._length - self._read_bytes)
424 chunk = await self._content.read(chunk_size)
425 if self._content.at_eof():
426 self._at_eof = True
427 return chunk
429 async def _read_chunk_from_stream(self, size: int) -> bytes:
430 # Reads content chunk of body part with unknown length.
431 # The Content-Length header for body part is not necessary.
432 assert (
433 size >= self._boundary_len
434 ), "Chunk size must be greater or equal than boundary length + 2"
435 first_chunk = self._prev_chunk is None
436 if first_chunk:
437 # We need to re-add the CRLF that got removed from headers parsing.
438 self._prev_chunk = b"\r\n" + await self._content.read(size)
440 chunk = b""
441 # content.read() may return less than size, so we need to loop to ensure
442 # we have enough data to detect the boundary.
443 while len(chunk) < self._boundary_len:
444 chunk += await self._content.read(size)
445 self._content_eof += int(self._content.at_eof())
446 if self._content_eof > 2:
447 raise ValueError("Reading after EOF")
448 if self._content_eof:
449 break
450 if len(chunk) > size:
451 self._content.unread_data(chunk[size:])
452 chunk = chunk[:size]
454 assert self._prev_chunk is not None
455 window = self._prev_chunk + chunk
456 sub = b"\r\n" + self._boundary
457 if first_chunk:
458 idx = window.find(sub)
459 else:
460 idx = window.find(sub, max(0, len(self._prev_chunk) - len(sub)))
461 if idx >= 0:
462 # pushing boundary back to content
463 with warnings.catch_warnings():
464 warnings.filterwarnings("ignore", category=DeprecationWarning)
465 self._content.unread_data(window[idx:])
466 self._prev_chunk = self._prev_chunk[:idx]
467 chunk = window[len(self._prev_chunk) : idx]
468 if not chunk:
469 self._at_eof = True
470 result = self._prev_chunk[2 if first_chunk else 0 :] # Strip initial CRLF
471 self._prev_chunk = chunk
472 return result
474 async def readline(self) -> bytes:
475 """Reads body part by line by line."""
476 if self._at_eof:
477 return b""
479 if self._unread:
480 line = self._unread.popleft()
481 else:
482 line = await self._content.readline()
484 if line.startswith(self._boundary):
485 # the very last boundary may not come with \r\n,
486 # so set single rules for everyone
487 sline = line.rstrip(b"\r\n")
488 boundary = self._boundary
489 last_boundary = self._boundary + b"--"
490 # ensure that we read exactly the boundary, not something alike
491 if sline == boundary or sline == last_boundary:
492 self._at_eof = True
493 self._unread.append(line)
494 return b""
495 else:
496 next_line = await self._content.readline()
497 if next_line.startswith(self._boundary):
498 line = line[:-2] # strip CRLF but only once
499 self._unread.append(next_line)
501 return line
503 async def release(self) -> None:
504 """Like read(), but reads all the data to the void."""
505 if self._at_eof:
506 return
507 while not self._at_eof:
508 await self.read_chunk(self.chunk_size)
510 async def text(self, *, encoding: str | None = None) -> str:
511 """Like read(), but assumes that body part contains text data."""
512 data = await self.read(decode=True)
513 # see https://www.w3.org/TR/html5/forms.html#multipart/form-data-encoding-algorithm
514 # and https://dvcs.w3.org/hg/xhr/raw-file/tip/Overview.html#dom-xmlhttprequest-send
515 encoding = encoding or self.get_charset(default="utf-8")
516 return data.decode(encoding)
518 async def json(self, *, encoding: str | None = None) -> dict[str, Any] | None:
519 """Like read(), but assumes that body parts contains JSON data."""
520 data = await self.read(decode=True)
521 if not data:
522 return None
523 encoding = encoding or self.get_charset(default="utf-8")
524 return cast(dict[str, Any], json.loads(data.decode(encoding)))
526 async def form(self, *, encoding: str | None = None) -> list[tuple[str, str]]:
527 """Like read(), but assumes that body parts contain form urlencoded data."""
528 data = await self.read(decode=True)
529 if not data:
530 return []
531 if encoding is not None:
532 real_encoding = encoding
533 else:
534 real_encoding = self.get_charset(default="utf-8")
535 try:
536 decoded_data = data.rstrip().decode(real_encoding)
537 except UnicodeDecodeError:
538 raise ValueError("data cannot be decoded with %s encoding" % real_encoding)
540 return parse_qsl(
541 decoded_data,
542 keep_blank_values=True,
543 encoding=real_encoding,
544 )
546 def at_eof(self) -> bool:
547 """Returns True if the boundary was reached or False otherwise."""
548 return self._at_eof
550 def _apply_content_transfer_decoding(self, data: _Buffer) -> _Buffer | bytes:
551 """Apply Content-Transfer-Encoding decoding if header is present."""
552 if CONTENT_TRANSFER_ENCODING in self.headers:
553 return self._decode_content_transfer(data)
554 return data
556 def _needs_content_decoding(self) -> bool:
557 """Check if Content-Encoding decoding should be applied."""
558 # https://datatracker.ietf.org/doc/html/rfc7578#section-4.8
559 return not self._is_form_data and CONTENT_ENCODING in self.headers
561 def decode(self, data: _Buffer) -> _Buffer | bytes:
562 """Decodes data synchronously.
564 Decodes data according the specified Content-Encoding
565 or Content-Transfer-Encoding headers value.
567 Note: For large payloads, consider using decode_iter() instead
568 to avoid blocking the event loop during decompression.
569 """
570 decoded = self._apply_content_transfer_decoding(data)
571 if self._needs_content_decoding():
572 return self._decode_content(decoded)
573 return decoded
575 async def decode_iter(self, data: _Buffer) -> AsyncIterator[_Buffer | bytes]:
576 """Async generator that yields decoded data chunks.
578 Decodes data according the specified Content-Encoding
579 or Content-Transfer-Encoding headers value.
581 This method offloads decompression to an executor for large payloads
582 to avoid blocking the event loop.
583 """
584 decoded = self._apply_content_transfer_decoding(data)
585 if self._needs_content_decoding():
586 async for d in self._decode_content_async(decoded):
587 yield d
588 else:
589 yield decoded
591 def _decode_content(self, data: _Buffer) -> _Buffer | bytes:
592 encoding = self.headers.get(CONTENT_ENCODING, "").lower()
593 if encoding == "identity":
594 return data
595 if encoding in {"deflate", "gzip"}:
596 return ZLibDecompressor(
597 encoding=encoding,
598 suppress_deflate_header=True,
599 ).decompress_sync(data, max_length=self._max_decompress_size)
601 raise RuntimeError(f"unknown content encoding: {encoding}")
603 async def _decode_content_async(
604 self, data: _Buffer
605 ) -> AsyncIterator[_Buffer | bytes]:
606 encoding = self.headers.get(CONTENT_ENCODING, "").lower()
607 if encoding == "identity":
608 yield data
609 elif encoding in {"deflate", "gzip"}:
610 d = ZLibDecompressor(
611 encoding=encoding,
612 suppress_deflate_header=True,
613 )
614 yield await d.decompress(data, max_length=self._max_decompress_size)
615 while d.data_available:
616 yield await d.decompress(b"", max_length=self._max_decompress_size)
617 else:
618 raise RuntimeError(f"unknown content encoding: {encoding}")
620 def _decode_content_transfer(self, data: _Buffer) -> _Buffer | bytes:
621 encoding = self.headers.get(CONTENT_TRANSFER_ENCODING, "").lower()
623 if encoding == "base64":
624 return base64.b64decode(data)
625 elif encoding == "quoted-printable":
626 return binascii.a2b_qp(data)
627 elif encoding in ("binary", "8bit", "7bit"):
628 return data
629 else:
630 raise RuntimeError(f"unknown content transfer encoding: {encoding}")
632 def get_charset(self, default: str) -> str:
633 """Returns charset parameter from Content-Type header or default."""
634 ctype = self.headers.get(CONTENT_TYPE, "")
635 mimetype = parse_mimetype(ctype)
636 return mimetype.parameters.get("charset", self._default_charset or default)
638 @reify
639 def name(self) -> str | None:
640 """Returns name specified in Content-Disposition header.
642 If the header is missing or malformed, returns None.
643 """
644 _, params = parse_content_disposition(self.headers.get(CONTENT_DISPOSITION))
645 return content_disposition_filename(params, "name")
647 @reify
648 def filename(self) -> str | None:
649 """Returns filename specified in Content-Disposition header.
651 Returns None if the header is missing or malformed.
652 """
653 _, params = parse_content_disposition(self.headers.get(CONTENT_DISPOSITION))
654 return content_disposition_filename(params, "filename")
657@payload_type(BodyPartReader, order=Order.try_first)
658class BodyPartReaderPayload(Payload):
659 _value: BodyPartReader
660 # _autoclose = False (inherited) - Streaming reader that may have resources
662 def __init__(self, value: BodyPartReader, *args: Any, **kwargs: Any) -> None:
663 super().__init__(value, *args, **kwargs)
665 params: dict[str, str] = {}
666 if value.name is not None:
667 params["name"] = value.name
668 if value.filename is not None:
669 params["filename"] = value.filename
671 if params:
672 self.set_content_disposition("attachment", True, **params)
674 def decode(self, encoding: str = "utf-8", errors: str = "strict") -> str:
675 raise TypeError("Unable to decode.")
677 async def as_bytes(self, encoding: str = "utf-8", errors: str = "strict") -> bytes:
678 """Raises TypeError as body parts should be consumed via write().
680 This is intentional: BodyPartReader payloads are designed for streaming
681 large data (potentially gigabytes) and must be consumed only once via
682 the write() method to avoid memory exhaustion. They cannot be buffered
683 in memory for reuse.
684 """
685 raise TypeError("Unable to read body part as bytes. Use write() to consume.")
687 async def write(self, writer: AbstractStreamWriter) -> None:
688 field = self._value
689 # Reading the part drains the underlying stream irreversibly, so mark the
690 # payload consumed up front: even an interrupted write leaves nothing that
691 # a retry or redirect could replay.
692 self._consumed = True
693 while chunk := await field.read_chunk(size=DEFAULT_CHUNK_SIZE):
694 async for d in field.decode_iter(chunk):
695 await writer.write(d)
698class MultipartReader:
699 """Multipart body reader."""
701 #: Response wrapper, used when multipart readers constructs from response.
702 response_wrapper_cls = MultipartResponseWrapper
703 #: Multipart reader class, used to handle multipart/* body parts.
704 #: None points to type(self)
705 multipart_reader_cls: type["MultipartReader"] | None = None
706 #: Body part reader class for non multipart/* content types.
707 part_reader_cls = BodyPartReader
709 def __init__(
710 self,
711 headers: Mapping[str, str],
712 content: StreamReader,
713 *,
714 client_max_size: int = sys.maxsize,
715 max_field_size: int = 8190,
716 max_headers: int = 128,
717 max_size_error_cls: type[Exception] = ValueError,
718 ) -> None:
719 self._mimetype = parse_mimetype(headers[CONTENT_TYPE])
720 assert self._mimetype.type == "multipart", "multipart/* content type expected"
721 if "boundary" not in self._mimetype.parameters:
722 raise ValueError(
723 "boundary missed for Content-Type: %s" % headers[CONTENT_TYPE]
724 )
726 self.headers = headers
727 self._boundary = ("--" + self._get_boundary()).encode()
728 self._client_max_size = client_max_size
729 self._content = content
730 self._default_charset: str | None = None
731 self._last_part: MultipartReader | BodyPartReader | None = None
732 self._max_field_size = max_field_size
733 self._max_headers = max_headers
734 self._max_size_error_cls = max_size_error_cls
735 self._at_eof = False
736 self._at_bof = True
737 self._unread: list[bytes] = []
739 def __aiter__(self: Self) -> Self:
740 return self
742 async def __anext__(
743 self,
744 ) -> Union["MultipartReader", BodyPartReader] | None:
745 part = await self.next()
746 if part is None:
747 raise StopAsyncIteration
748 return part
750 @classmethod
751 def from_response(
752 cls,
753 response: "ClientResponse",
754 ) -> MultipartResponseWrapper:
755 """Constructs reader instance from HTTP response.
757 :param response: :class:`~aiohttp.client.ClientResponse` instance
758 """
759 obj = cls.response_wrapper_cls(
760 response, cls(response.headers, response.content)
761 )
762 return obj
764 def at_eof(self) -> bool:
765 """Returns True if the final boundary was reached, false otherwise."""
766 return self._at_eof
768 async def next(
769 self,
770 ) -> Union["MultipartReader", BodyPartReader] | None:
771 """Emits the next multipart body part."""
772 # So, if we're at BOF, we need to skip till the boundary.
773 if self._at_eof:
774 return None
775 await self._maybe_release_last_part()
776 if self._at_bof:
777 await self._read_until_first_boundary()
778 self._at_bof = False
779 else:
780 await self._read_boundary()
781 if self._at_eof: # we just read the last boundary, nothing to do there
782 return None
784 part = await self.fetch_next_part()
785 # https://datatracker.ietf.org/doc/html/rfc7578#section-4.6
786 if (
787 self._last_part is None
788 and self._mimetype.subtype == "form-data"
789 and isinstance(part, BodyPartReader)
790 ):
791 _, params = parse_content_disposition(part.headers.get(CONTENT_DISPOSITION))
792 if params.get("name") == "_charset_":
793 # Longest encoding in https://encoding.spec.whatwg.org/encodings.json
794 # is 19 characters, so 32 should be more than enough for any valid encoding.
795 charset = await part.read_chunk(32)
796 if len(charset) > 31:
797 raise RuntimeError("Invalid default charset")
798 self._default_charset = charset.strip().decode()
799 part = await self.fetch_next_part()
800 self._last_part = part
801 return self._last_part
803 async def release(self) -> None:
804 """Reads all the body parts to the void till the final boundary."""
805 while not self._at_eof:
806 item = await self.next()
807 if item is None:
808 break
809 await item.release()
811 async def fetch_next_part(
812 self,
813 ) -> Union["MultipartReader", BodyPartReader]:
814 """Returns the next body part reader."""
815 headers = await self._read_headers()
816 return self._get_part_reader(headers)
818 def _get_part_reader(
819 self,
820 headers: "CIMultiDictProxy[str]",
821 ) -> Union["MultipartReader", BodyPartReader]:
822 """Dispatches the response by the `Content-Type` header.
824 Returns a suitable reader instance.
826 :param dict headers: Response headers
827 """
828 ctype = headers.get(CONTENT_TYPE, "")
829 mimetype = parse_mimetype(ctype)
831 if mimetype.type == "multipart":
832 if self.multipart_reader_cls is None:
833 return type(self)(
834 headers,
835 self._content,
836 client_max_size=self._client_max_size,
837 max_field_size=self._max_field_size,
838 max_headers=self._max_headers,
839 max_size_error_cls=self._max_size_error_cls,
840 )
841 return self.multipart_reader_cls(
842 headers,
843 self._content,
844 client_max_size=self._client_max_size,
845 max_field_size=self._max_field_size,
846 max_headers=self._max_headers,
847 max_size_error_cls=self._max_size_error_cls,
848 )
849 else:
850 return self.part_reader_cls(
851 self._boundary,
852 headers,
853 self._content,
854 subtype=self._mimetype.subtype,
855 default_charset=self._default_charset,
856 client_max_size=self._client_max_size,
857 max_size_error_cls=self._max_size_error_cls,
858 )
860 def _get_boundary(self) -> str:
861 boundary = self._mimetype.parameters["boundary"]
862 if len(boundary) > 70:
863 raise ValueError("boundary %r is too long (70 chars max)" % boundary)
865 return boundary
867 async def _readline(self) -> bytes:
868 if self._unread:
869 return self._unread.pop()
870 return await self._content.readline()
872 async def _read_until_first_boundary(self) -> None:
873 while True:
874 chunk = await self._readline()
875 if chunk == b"":
876 raise ValueError(
877 "Could not find starting boundary %r" % (self._boundary)
878 )
879 chunk = chunk.rstrip()
880 if chunk == self._boundary:
881 return
882 elif chunk == self._boundary + b"--":
883 self._at_eof = True
884 return
886 async def _read_boundary(self) -> None:
887 chunk = (await self._readline()).rstrip()
888 if chunk == self._boundary:
889 pass
890 elif chunk == self._boundary + b"--":
891 self._at_eof = True
892 epilogue = await self._readline()
893 next_line = await self._readline()
895 # the epilogue is expected and then either the end of input or the
896 # parent multipart boundary, if the parent boundary is found then
897 # it should be marked as unread and handed to the parent for
898 # processing
899 if next_line[:2] == b"--":
900 self._unread.append(next_line)
901 # otherwise the request is likely missing an epilogue and both
902 # lines should be passed to the parent for processing
903 # (this handles the old behavior gracefully)
904 else:
905 self._unread.extend([next_line, epilogue])
906 else:
907 raise ValueError(f"Invalid boundary {chunk!r}, expected {self._boundary!r}")
909 async def _read_headers(self) -> "CIMultiDictProxy[str]":
910 lines = []
911 while True:
912 chunk = await self._content.readline(max_line_length=self._max_field_size)
913 chunk = chunk.rstrip(b"\r\n")
914 lines.append(chunk)
915 if not chunk:
916 break
917 if len(lines) > self._max_headers:
918 raise BadHttpMessage("Too many headers received")
919 parser = HeadersParser(max_field_size=self._max_field_size)
920 headers, raw_headers = parser.parse_headers(lines)
921 return headers
923 async def _maybe_release_last_part(self) -> None:
924 """Ensures that the last read body part is read completely."""
925 if self._last_part is not None:
926 if not self._last_part.at_eof():
927 await self._last_part.release()
928 self._unread.extend(self._last_part._unread)
929 self._last_part = None
932_Part = tuple[Payload, str, str]
935class MultipartWriter(Payload):
936 """Multipart body writer."""
938 _value: None
939 # _consumed = False (inherited) - Can be encoded multiple times
940 _autoclose = True # No file handles, just collects parts in memory
942 def __init__(self, subtype: str = "mixed", boundary: str | None = None) -> None:
943 boundary = boundary if boundary is not None else uuid.uuid4().hex
944 # The underlying Payload API demands a str (utf-8), not bytes,
945 # so we need to ensure we don't lose anything during conversion.
946 # As a result, require the boundary to be ASCII only.
947 # In both situations.
949 try:
950 self._boundary = boundary.encode("ascii")
951 except UnicodeEncodeError:
952 raise ValueError("boundary should contain ASCII only chars") from None
953 ctype = f"multipart/{subtype}; boundary={self._boundary_value}"
955 super().__init__(None, content_type=ctype)
957 self._parts: list[_Part] = []
958 self._is_form_data = subtype == "form-data"
960 def __enter__(self) -> "MultipartWriter":
961 return self
963 def __exit__(
964 self,
965 exc_type: type[BaseException] | None,
966 exc_val: BaseException | None,
967 exc_tb: TracebackType | None,
968 ) -> None:
969 pass
971 @property
972 def consumed(self) -> bool:
973 """Whether the writer or any of its parts can no longer be replayed."""
974 return self._consumed or any(part.consumed for part, _, _ in self._parts)
976 def __iter__(self) -> Iterator[_Part]:
977 return iter(self._parts)
979 def __len__(self) -> int:
980 return len(self._parts)
982 def __bool__(self) -> bool:
983 return True
985 _valid_tchar_regex = re.compile(rb"\A[!#$%&'*+\-.^_`|~\w]+\Z")
986 _invalid_qdtext_char_regex = re.compile(rb"[\x00-\x08\x0A-\x1F\x7F]")
988 @property
989 def _boundary_value(self) -> str:
990 """Wrap boundary parameter value in quotes, if necessary.
992 Reads self.boundary and returns a unicode string.
993 """
994 # Refer to RFCs 7231, 7230, 5234.
995 #
996 # parameter = token "=" ( token / quoted-string )
997 # token = 1*tchar
998 # quoted-string = DQUOTE *( qdtext / quoted-pair ) DQUOTE
999 # qdtext = HTAB / SP / %x21 / %x23-5B / %x5D-7E / obs-text
1000 # obs-text = %x80-FF
1001 # quoted-pair = "\" ( HTAB / SP / VCHAR / obs-text )
1002 # tchar = "!" / "#" / "$" / "%" / "&" / "'" / "*"
1003 # / "+" / "-" / "." / "^" / "_" / "`" / "|" / "~"
1004 # / DIGIT / ALPHA
1005 # ; any VCHAR, except delimiters
1006 # VCHAR = %x21-7E
1007 value = self._boundary
1008 if re.match(self._valid_tchar_regex, value):
1009 return value.decode("ascii") # cannot fail
1011 if re.search(self._invalid_qdtext_char_regex, value):
1012 raise ValueError("boundary value contains invalid characters")
1014 # escape %x5C and %x22
1015 quoted_value_content = value.replace(b"\\", b"\\\\")
1016 quoted_value_content = quoted_value_content.replace(b'"', b'\\"')
1018 return '"' + quoted_value_content.decode("ascii") + '"'
1020 @property
1021 def boundary(self) -> str:
1022 return self._boundary.decode("ascii")
1024 def append(self, obj: Any, headers: Mapping[str, str] | None = None) -> Payload:
1025 if headers is None:
1026 headers = CIMultiDict()
1028 if isinstance(obj, Payload):
1029 obj.headers.update(headers)
1030 return self.append_payload(obj)
1031 else:
1032 try:
1033 payload = get_payload(obj, headers=headers)
1034 except LookupError:
1035 raise TypeError("Cannot create payload from %r" % obj)
1036 else:
1037 return self.append_payload(payload)
1039 def append_payload(self, payload: Payload) -> Payload:
1040 """Adds a new body part to multipart writer."""
1041 encoding: str | None = None
1042 te_encoding: str | None = None
1043 if self._is_form_data:
1044 # https://datatracker.ietf.org/doc/html/rfc7578#section-4.7
1045 # https://datatracker.ietf.org/doc/html/rfc7578#section-4.8
1046 assert (
1047 not {CONTENT_ENCODING, CONTENT_LENGTH, CONTENT_TRANSFER_ENCODING}
1048 & payload.headers.keys()
1049 )
1050 # Set default Content-Disposition in case user doesn't create one
1051 if CONTENT_DISPOSITION not in payload.headers:
1052 name = f"section-{len(self._parts)}"
1053 payload.set_content_disposition("form-data", name=name)
1054 else:
1055 # compression
1056 encoding = payload.headers.get(CONTENT_ENCODING, "").lower()
1057 if encoding and encoding not in ("deflate", "gzip", "identity"):
1058 raise RuntimeError(f"unknown content encoding: {encoding}")
1059 if encoding == "identity":
1060 encoding = None
1062 # te encoding
1063 te_encoding = payload.headers.get(CONTENT_TRANSFER_ENCODING, "").lower()
1064 if te_encoding not in ("", "base64", "quoted-printable", "binary"):
1065 raise RuntimeError(f"unknown content transfer encoding: {te_encoding}")
1066 if te_encoding == "binary":
1067 te_encoding = None
1069 # size
1070 size = payload.size
1071 if size is not None and not (encoding or te_encoding):
1072 payload.headers[CONTENT_LENGTH] = str(size)
1074 self._parts.append((payload, encoding, te_encoding)) # type: ignore[arg-type]
1075 return payload
1077 def append_json(
1078 self, obj: Any, headers: Mapping[str, str] | None = None
1079 ) -> Payload:
1080 """Helper to append JSON part."""
1081 if headers is None:
1082 headers = CIMultiDict()
1084 return self.append_payload(JsonPayload(obj, headers=headers))
1086 def append_form(
1087 self,
1088 obj: Sequence[tuple[str, str]] | Mapping[str, str],
1089 headers: Mapping[str, str] | None = None,
1090 ) -> Payload:
1091 """Helper to append form urlencoded part."""
1092 assert isinstance(obj, (Sequence, Mapping))
1094 if headers is None:
1095 headers = CIMultiDict()
1097 if isinstance(obj, Mapping):
1098 obj = list(obj.items())
1099 data = urlencode(obj, doseq=True)
1101 return self.append_payload(
1102 StringPayload(
1103 data, headers=headers, content_type="application/x-www-form-urlencoded"
1104 )
1105 )
1107 @property
1108 def size(self) -> int | None:
1109 """Size of the payload."""
1110 total = 0
1111 for part, encoding, te_encoding in self._parts:
1112 part_size = part.size
1113 if encoding or te_encoding or part_size is None:
1114 return None
1116 total += int(
1117 2
1118 + len(self._boundary)
1119 + 2
1120 + part_size # b'--'+self._boundary+b'\r\n'
1121 + len(part._binary_headers)
1122 + 2 # b'\r\n'
1123 )
1125 total += 2 + len(self._boundary) + 4 # b'--'+self._boundary+b'--\r\n'
1126 return total
1128 def decode(self, encoding: str = "utf-8", errors: str = "strict") -> str:
1129 """Return string representation of the multipart data.
1131 WARNING: This method may do blocking I/O if parts contain file payloads.
1132 It should not be called in the event loop. Use as_bytes().decode() instead.
1133 """
1134 return "".join(
1135 "--"
1136 + self.boundary
1137 + "\r\n"
1138 + part._binary_headers.decode(encoding, errors)
1139 + part.decode()
1140 for part, _e, _te in self._parts
1141 )
1143 async def as_bytes(self, encoding: str = "utf-8", errors: str = "strict") -> bytes:
1144 """Return bytes representation of the multipart data.
1146 This method is async-safe and calls as_bytes on underlying payloads.
1147 """
1148 parts: list[bytes] = []
1150 # Process each part
1151 for part, _e, _te in self._parts:
1152 # Add boundary
1153 parts.append(b"--" + self._boundary + b"\r\n")
1155 # Add headers
1156 parts.append(part._binary_headers)
1158 # Add payload content using as_bytes for async safety
1159 part_bytes = await part.as_bytes(encoding, errors)
1160 parts.append(part_bytes)
1162 # Add trailing CRLF
1163 parts.append(b"\r\n")
1165 # Add closing boundary
1166 parts.append(b"--" + self._boundary + b"--\r\n")
1168 return b"".join(parts)
1170 async def write(
1171 self, writer: AbstractStreamWriter, close_boundary: bool = True
1172 ) -> None:
1173 """Write body."""
1174 for part, encoding, te_encoding in self._parts:
1175 if self._is_form_data:
1176 # https://datatracker.ietf.org/doc/html/rfc7578#section-4.2
1177 assert CONTENT_DISPOSITION in part.headers
1178 assert "name=" in part.headers[CONTENT_DISPOSITION]
1180 await writer.write(b"--" + self._boundary + b"\r\n")
1181 await writer.write(part._binary_headers)
1183 if encoding or te_encoding:
1184 w = MultipartPayloadWriter(writer)
1185 if encoding:
1186 w.enable_compression(encoding)
1187 if te_encoding:
1188 w.enable_encoding(te_encoding)
1189 await part.write(w) # type: ignore[arg-type]
1190 await w.write_eof()
1191 else:
1192 await part.write(writer)
1194 await writer.write(b"\r\n")
1196 if close_boundary:
1197 await writer.write(b"--" + self._boundary + b"--\r\n")
1199 async def close(self) -> None:
1200 """
1201 Close all part payloads that need explicit closing.
1203 IMPORTANT: This method must not await anything that might not finish
1204 immediately, as it may be called during cleanup/cancellation. Schedule
1205 any long-running operations without awaiting them.
1206 """
1207 if self._consumed:
1208 return
1209 self._consumed = True
1211 # Close all parts that need explicit closing
1212 # We catch and log exceptions to ensure all parts get a chance to close
1213 # we do not use asyncio.gather() here because we are not allowed
1214 # to suspend given we may be called during cleanup
1215 for idx, (part, _, _) in enumerate(self._parts):
1216 if not part.autoclose and not part.consumed:
1217 try:
1218 await part.close()
1219 except Exception as exc:
1220 internal_logger.error(
1221 "Failed to close multipart part %d: %s", idx, exc, exc_info=True
1222 )
1225class MultipartPayloadWriter:
1226 def __init__(self, writer: AbstractStreamWriter) -> None:
1227 self._writer = writer
1228 self._encoding: str | None = None
1229 self._compress: ZLibCompressor | None = None
1230 self._encoding_buffer: bytearray | None = None
1232 def enable_encoding(self, encoding: str) -> None:
1233 if encoding == "base64":
1234 self._encoding = encoding
1235 self._encoding_buffer = bytearray()
1236 elif encoding == "quoted-printable":
1237 self._encoding = "quoted-printable"
1239 def enable_compression(
1240 self, encoding: str = "deflate", strategy: int | None = None
1241 ) -> None:
1242 self._compress = ZLibCompressor(
1243 encoding=encoding,
1244 suppress_deflate_header=True,
1245 strategy=strategy,
1246 )
1248 async def write_eof(self) -> None:
1249 if self._compress is not None:
1250 chunk = self._compress.flush()
1251 if chunk:
1252 self._compress = None
1253 await self.write(chunk)
1255 if self._encoding == "base64":
1256 if self._encoding_buffer:
1257 await self._writer.write(base64.b64encode(self._encoding_buffer))
1259 async def write(self, chunk: bytes) -> None:
1260 if self._compress is not None:
1261 if chunk:
1262 chunk = await self._compress.compress(chunk)
1263 if not chunk:
1264 return
1266 if self._encoding == "base64":
1267 buf = self._encoding_buffer
1268 assert buf is not None
1269 buf.extend(chunk)
1271 if buf:
1272 div, mod = divmod(len(buf), 3)
1273 enc_chunk, self._encoding_buffer = (buf[: div * 3], buf[div * 3 :])
1274 if enc_chunk:
1275 b64chunk = base64.b64encode(enc_chunk)
1276 await self._writer.write(b64chunk)
1277 elif self._encoding == "quoted-printable":
1278 await self._writer.write(binascii.b2a_qp(chunk))
1279 else:
1280 await self._writer.write(chunk)