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