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, Union, 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
48if sys.version_info >= (3, 11):
49 from typing import Self
50else:
51 from typing import TypeVar
53 Self = TypeVar("Self", bound="BodyPartReader")
55__all__ = (
56 "MultipartReader",
57 "MultipartWriter",
58 "BodyPartReader",
59 "BadContentDispositionHeader",
60 "BadContentDispositionParam",
61 "parse_content_disposition",
62 "content_disposition_filename",
63)
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")
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 parts = []
199 fnparams = sorted(
200 (key, value) for key, value in params.items() if key.startswith(name_suf)
201 )
202 for num, (key, value) in enumerate(fnparams):
203 _, tail = key.split("*", 1)
204 if tail.endswith("*"):
205 tail = tail[:-1]
206 if tail == str(num):
207 parts.append(value)
208 else:
209 break
210 if not parts:
211 return None
212 value = "".join(parts)
213 if "'" in value:
214 encoding, _, value = value.split("'", 2)
215 encoding = encoding or "utf-8"
216 try:
217 return unquote(value, encoding, "strict")
218 except (builtins.LookupError, UnicodeDecodeError):
219 # Both the charset name and the octets are attacker-controlled
220 # here; an unknown encoding raises the builtin LookupError
221 # (shadowed in this module by payload.LookupError) and
222 # undecodable bytes raise UnicodeDecodeError.
223 return None
224 return value
227class MultipartResponseWrapper:
228 """Wrapper around the MultipartReader.
230 It takes care about
231 underlying connection and close it when it needs in.
232 """
234 def __init__(
235 self,
236 resp: "ClientResponse",
237 stream: "MultipartReader",
238 ) -> None:
239 self.resp = resp
240 self.stream = stream
242 def __aiter__(self) -> "MultipartResponseWrapper":
243 return self
245 async def __anext__(
246 self,
247 ) -> Union["MultipartReader", "BodyPartReader"]:
248 part = await self.next()
249 if part is None:
250 raise StopAsyncIteration
251 return part
253 def at_eof(self) -> bool:
254 """Returns True when all response data had been read."""
255 return self.resp.content.at_eof()
257 async def next(
258 self,
259 ) -> Union["MultipartReader", "BodyPartReader"] | None:
260 """Emits next multipart reader object."""
261 item = await self.stream.next()
262 if self.stream.at_eof():
263 await self.release()
264 return item
266 async def release(self) -> None:
267 """Release the connection gracefully.
269 All remaining content is read to the void.
270 """
271 self.resp.release()
274class BodyPartReader:
275 """Multipart reader for single body part."""
277 chunk_size = 8192
279 def __init__(
280 self,
281 boundary: bytes,
282 headers: HeadersDictProxy,
283 content: StreamReader,
284 *,
285 subtype: str = "mixed",
286 default_charset: str | None = None,
287 max_decompress_size: int = DEFAULT_CHUNK_SIZE,
288 client_max_size: int = sys.maxsize,
289 max_size_error_cls: type[Exception] = ValueError,
290 ) -> None:
291 self.headers = headers
292 self._boundary = boundary
293 self._boundary_len = len(boundary) + 2 # Boundary + \r\n
294 self._content = content
295 self._default_charset = default_charset
296 self._at_eof = False
297 self._is_form_data = subtype == "form-data"
298 # https://datatracker.ietf.org/doc/html/rfc7578#section-4.8
299 length = None if self._is_form_data else self.headers.get(CONTENT_LENGTH, None)
300 if length is not None and not (length.isascii() and length.isdigit()):
301 # Reject sign prefixes, underscores, whitespace and non-ASCII
302 # digits that int() would otherwise accept.
303 # https://www.rfc-editor.org/rfc/rfc9110#section-8.6
304 raise ValueError(f"invalid Content-Length: {length!r}")
305 self._length = int(length) if length is not None else None
306 self._read_bytes = 0
307 self._unread: deque[bytes] = deque()
308 self._prev_chunk: bytes | None = None
309 self._content_eof = 0
310 self._cache: dict[str, Any] = {}
311 self._max_decompress_size = max_decompress_size
312 self._client_max_size = client_max_size
313 self._max_size_error_cls = max_size_error_cls
315 def __aiter__(self) -> Self:
316 return self
318 async def __anext__(self) -> bytes:
319 part = await self.next()
320 if part is None:
321 raise StopAsyncIteration
322 return part
324 async def next(self) -> bytes | None:
325 item = await self.read()
326 if not item:
327 return None
328 return item
330 async def read(self, *, decode: bool = False) -> bytes:
331 """Reads body part data.
333 decode: Decodes data following by encoding
334 method from Content-Encoding header. If it missed
335 data remains untouched
336 """
337 if self._at_eof:
338 return b""
339 data = bytearray()
340 while not self._at_eof:
341 data.extend(await self.read_chunk(self.chunk_size))
342 if len(data) > self._client_max_size:
343 raise self._max_size_error_cls(self._client_max_size)
344 # https://github.com/python/mypy/issues/17537
345 if decode: # type: ignore[unreachable]
346 decoded_data = bytearray()
347 async for d in self.decode_iter(data):
348 decoded_data.extend(d)
349 if len(decoded_data) > self._client_max_size:
350 raise self._max_size_error_cls(self._client_max_size)
351 return decoded_data
352 return data
354 async def read_chunk(self, size: int = chunk_size) -> bytes:
355 """Reads body part content chunk of the specified size.
357 size: chunk size
358 """
359 if self._at_eof:
360 return b""
361 if self._length:
362 chunk = await self._read_chunk_from_length(size)
363 else:
364 chunk = await self._read_chunk_from_stream(size)
366 # For the case of base64 data, we must read a fragment of size with a
367 # remainder of 0 by dividing by 4 for string without symbols \n or \r
368 encoding = self.headers.get(CONTENT_TRANSFER_ENCODING)
369 if encoding and encoding.lower() == "base64":
370 stripped_chunk = b"".join(chunk.split())
371 remainder = len(stripped_chunk) % 4
373 while remainder != 0 and not self.at_eof():
374 over_chunk_size = 4 - remainder
375 over_chunk = b""
377 if self._prev_chunk:
378 over_chunk = self._prev_chunk[:over_chunk_size]
379 self._prev_chunk = self._prev_chunk[len(over_chunk) :]
381 if len(over_chunk) != over_chunk_size:
382 over_chunk += await self._content.read(4 - len(over_chunk))
384 if not over_chunk:
385 self._at_eof = True
387 stripped_chunk += b"".join(over_chunk.split())
388 chunk += over_chunk
389 remainder = len(stripped_chunk) % 4
391 self._read_bytes += len(chunk)
392 if self._read_bytes == self._length:
393 self._at_eof = True
394 if self._at_eof and await self._content.readline() != b"\r\n":
395 raise ValueError("Reader did not read all the data or it is malformed")
396 return chunk
398 async def _read_chunk_from_length(self, size: int) -> bytes:
399 # Reads body part content chunk of the specified size.
400 # The body part must has Content-Length header with proper value.
401 assert self._length is not None, "Content-Length required for chunked read"
402 chunk_size = min(size, self._length - self._read_bytes)
403 chunk = await self._content.read(chunk_size)
404 if self._content.at_eof():
405 self._at_eof = True
406 return chunk
408 async def _read_chunk_from_stream(self, size: int) -> bytes:
409 # Reads content chunk of body part with unknown length.
410 # The Content-Length header for body part is not necessary.
411 assert (
412 size >= self._boundary_len
413 ), "Chunk size must be greater or equal than boundary length + 2"
414 first_chunk = self._prev_chunk is None
415 if first_chunk:
416 # We need to re-add the CRLF that got removed from headers parsing.
417 self._prev_chunk = b"\r\n" + await self._content.read(size)
419 chunk = b""
420 # content.read() may return less than size, so we need to loop to ensure
421 # we have enough data to detect the boundary.
422 while len(chunk) < self._boundary_len:
423 chunk += await self._content.read(size)
424 self._content_eof += int(self._content.at_eof())
425 if self._content_eof > 2:
426 raise ValueError("Reading after EOF")
427 if self._content_eof:
428 break
429 if len(chunk) > size:
430 self._content.unread_data(chunk[size:])
431 chunk = chunk[:size]
433 assert self._prev_chunk is not None
434 window = self._prev_chunk + chunk
435 sub = b"\r\n" + self._boundary
436 if first_chunk:
437 idx = window.find(sub)
438 else:
439 idx = window.find(sub, max(0, len(self._prev_chunk) - len(sub)))
440 if idx >= 0:
441 # pushing boundary back to content
442 with warnings.catch_warnings():
443 warnings.filterwarnings("ignore", category=DeprecationWarning)
444 self._content.unread_data(window[idx:])
445 self._prev_chunk = self._prev_chunk[:idx]
446 chunk = window[len(self._prev_chunk) : idx]
447 if not chunk:
448 self._at_eof = True
449 result = self._prev_chunk[2 if first_chunk else 0 :] # Strip initial CRLF
450 self._prev_chunk = chunk
451 return result
453 async def readline(self) -> bytes:
454 """Reads body part by line by line."""
455 if self._at_eof:
456 return b""
458 if self._unread:
459 line = self._unread.popleft()
460 else:
461 line = await self._content.readline()
463 if line.startswith(self._boundary):
464 # the very last boundary may not come with \r\n,
465 # so set single rules for everyone
466 sline = line.rstrip(b"\r\n")
467 boundary = self._boundary
468 last_boundary = self._boundary + b"--"
469 # ensure that we read exactly the boundary, not something alike
470 if sline == boundary or sline == last_boundary:
471 self._at_eof = True
472 self._unread.append(line)
473 return b""
474 else:
475 next_line = await self._content.readline()
476 if next_line.startswith(self._boundary):
477 line = line[:-2] # strip CRLF but only once
478 self._unread.append(next_line)
480 return line
482 async def release(self) -> None:
483 """Like read(), but reads all the data to the void."""
484 if self._at_eof:
485 return
486 while not self._at_eof:
487 await self.read_chunk(self.chunk_size)
489 async def text(self, *, encoding: str | None = None) -> str:
490 """Like read(), but assumes that body part contains text data."""
491 data = await self.read(decode=True)
492 # see https://www.w3.org/TR/html5/forms.html#multipart/form-data-encoding-algorithm
493 # and https://dvcs.w3.org/hg/xhr/raw-file/tip/Overview.html#dom-xmlhttprequest-send
494 encoding = encoding or self.get_charset(default="utf-8")
495 return data.decode(encoding)
497 async def json(self, *, encoding: str | None = None) -> dict[str, Any] | None:
498 """Like read(), but assumes that body parts contains JSON data."""
499 data = await self.read(decode=True)
500 if not data:
501 return None
502 encoding = encoding or self.get_charset(default="utf-8")
503 return cast(dict[str, Any], json.loads(data.decode(encoding)))
505 async def form(self, *, encoding: str | None = None) -> list[tuple[str, str]]:
506 """Like read(), but assumes that body parts contain form urlencoded data."""
507 data = await self.read(decode=True)
508 if not data:
509 return []
510 if encoding is not None:
511 real_encoding = encoding
512 else:
513 real_encoding = self.get_charset(default="utf-8")
514 try:
515 decoded_data = data.rstrip().decode(real_encoding)
516 except UnicodeDecodeError:
517 raise ValueError("data cannot be decoded with %s encoding" % real_encoding)
519 return parse_qsl(
520 decoded_data,
521 keep_blank_values=True,
522 encoding=real_encoding,
523 )
525 def at_eof(self) -> bool:
526 """Returns True if the boundary was reached or False otherwise."""
527 return self._at_eof
529 def _apply_content_transfer_decoding(self, data: bytes) -> bytes:
530 """Apply Content-Transfer-Encoding decoding if header is present."""
531 if CONTENT_TRANSFER_ENCODING in self.headers:
532 return self._decode_content_transfer(data)
533 return data
535 def _needs_content_decoding(self) -> bool:
536 """Check if Content-Encoding decoding should be applied."""
537 # https://datatracker.ietf.org/doc/html/rfc7578#section-4.8
538 return not self._is_form_data and CONTENT_ENCODING in self.headers
540 def decode(self, data: bytes) -> bytes:
541 """Decodes data synchronously.
543 Decodes data according the specified Content-Encoding
544 or Content-Transfer-Encoding headers value.
546 Note: For large payloads, consider using decode_iter() instead
547 to avoid blocking the event loop during decompression.
548 """
549 data = self._apply_content_transfer_decoding(data)
550 if self._needs_content_decoding():
551 return self._decode_content(data)
552 return data
554 async def decode_iter(self, data: bytes) -> AsyncIterator[bytes]:
555 """Async generator that yields decoded data chunks.
557 Decodes data according the specified Content-Encoding
558 or Content-Transfer-Encoding headers value.
560 This method offloads decompression to an executor for large payloads
561 to avoid blocking the event loop.
562 """
563 data = self._apply_content_transfer_decoding(data)
564 if self._needs_content_decoding():
565 async for d in self._decode_content_async(data):
566 yield d
567 else:
568 yield data
570 def _decode_content(self, data: bytes) -> bytes:
571 encoding = self.headers.get(CONTENT_ENCODING, "").lower()
572 if encoding == "identity":
573 return data
574 if encoding in {"deflate", "gzip"}:
575 return ZLibDecompressor(
576 encoding=encoding,
577 suppress_deflate_header=True,
578 ).decompress_sync(data, max_length=self._max_decompress_size)
580 raise RuntimeError(f"unknown content encoding: {encoding}")
582 async def _decode_content_async(self, data: bytes) -> AsyncIterator[bytes]:
583 encoding = self.headers.get(CONTENT_ENCODING, "").lower()
584 if encoding == "identity":
585 yield data
586 elif encoding in {"deflate", "gzip"}:
587 d = ZLibDecompressor(
588 encoding=encoding,
589 suppress_deflate_header=True,
590 )
591 yield await d.decompress(data, max_length=self._max_decompress_size)
592 while d.data_available:
593 yield await d.decompress(b"", max_length=self._max_decompress_size)
594 else:
595 raise RuntimeError(f"unknown content encoding: {encoding}")
597 def _decode_content_transfer(self, data: bytes) -> bytes:
598 encoding = self.headers.get(CONTENT_TRANSFER_ENCODING, "").lower()
600 if encoding == "base64":
601 return base64.b64decode(data)
602 elif encoding == "quoted-printable":
603 return binascii.a2b_qp(data)
604 elif encoding in ("binary", "8bit", "7bit"):
605 return data
606 else:
607 raise RuntimeError(f"unknown content transfer encoding: {encoding}")
609 def get_charset(self, default: str) -> str:
610 """Returns charset parameter from Content-Type header or default."""
611 ctype = self.headers.get(CONTENT_TYPE, "")
612 mimetype = parse_mimetype(ctype)
613 return mimetype.parameters.get("charset", self._default_charset or default)
615 @reify
616 def name(self) -> str | None:
617 """Returns name specified in Content-Disposition header.
619 If the header is missing or malformed, returns None.
620 """
621 _, params = parse_content_disposition(self.headers.get(CONTENT_DISPOSITION))
622 return content_disposition_filename(params, "name")
624 @reify
625 def filename(self) -> str | None:
626 """Returns filename specified in Content-Disposition header.
628 Returns None if the header is missing or malformed.
629 """
630 _, params = parse_content_disposition(self.headers.get(CONTENT_DISPOSITION))
631 return content_disposition_filename(params, "filename")
634@payload_type(BodyPartReader, order=Order.try_first)
635class BodyPartReaderPayload(Payload):
636 _value: BodyPartReader
637 # _autoclose = False (inherited) - Streaming reader that may have resources
639 def __init__(self, value: BodyPartReader, *args: Any, **kwargs: Any) -> None:
640 super().__init__(value, *args, **kwargs)
642 params: dict[str, str] = {}
643 if value.name is not None:
644 params["name"] = value.name
645 if value.filename is not None:
646 params["filename"] = value.filename
648 if params:
649 self.set_content_disposition("attachment", True, **params)
651 def decode(self, encoding: str = "utf-8", errors: str = "strict") -> str:
652 raise TypeError("Unable to decode.")
654 async def as_bytes(self, encoding: str = "utf-8", errors: str = "strict") -> bytes:
655 """Raises TypeError as body parts should be consumed via write().
657 This is intentional: BodyPartReader payloads are designed for streaming
658 large data (potentially gigabytes) and must be consumed only once via
659 the write() method to avoid memory exhaustion. They cannot be buffered
660 in memory for reuse.
661 """
662 raise TypeError("Unable to read body part as bytes. Use write() to consume.")
664 async def write(self, writer: AbstractStreamWriter) -> None:
665 field = self._value
666 while chunk := await field.read_chunk(size=DEFAULT_CHUNK_SIZE):
667 async for d in field.decode_iter(chunk):
668 await writer.write(d)
671class MultipartReader:
672 """Multipart body reader."""
674 #: Response wrapper, used when multipart readers constructs from response.
675 response_wrapper_cls = MultipartResponseWrapper
676 #: Multipart reader class, used to handle multipart/* body parts.
677 #: None points to type(self)
678 multipart_reader_cls: type["MultipartReader"] | None = None
679 #: Body part reader class for non multipart/* content types.
680 part_reader_cls = BodyPartReader
682 def __init__(
683 self,
684 headers: Mapping[str, str],
685 content: StreamReader,
686 *,
687 client_max_size: int = sys.maxsize,
688 max_field_size: int = 8190,
689 max_headers: int = 128,
690 max_size_error_cls: type[Exception] = ValueError,
691 ) -> None:
692 self._mimetype = parse_mimetype(headers[CONTENT_TYPE])
693 assert self._mimetype.type == "multipart", "multipart/* content type expected"
694 if "boundary" not in self._mimetype.parameters:
695 raise ValueError(
696 "boundary missed for Content-Type: %s" % headers[CONTENT_TYPE]
697 )
699 self.headers = headers
700 self._boundary = ("--" + self._get_boundary()).encode()
701 self._client_max_size = client_max_size
702 self._content = content
703 self._default_charset: str | None = None
704 self._last_part: MultipartReader | BodyPartReader | None = None
705 self._max_field_size = max_field_size
706 self._max_headers = max_headers
707 self._max_size_error_cls = max_size_error_cls
708 self._at_eof = False
709 self._at_bof = True
710 self._unread: list[bytes] = []
712 def __aiter__(self) -> Self:
713 return self
715 async def __anext__(
716 self,
717 ) -> Union["MultipartReader", BodyPartReader] | None:
718 part = await self.next()
719 if part is None:
720 raise StopAsyncIteration
721 return part
723 @classmethod
724 def from_response(
725 cls,
726 response: "ClientResponse",
727 ) -> MultipartResponseWrapper:
728 """Constructs reader instance from HTTP response.
730 :param response: :class:`~aiohttp.client.ClientResponse` instance
731 """
732 obj = cls.response_wrapper_cls(
733 response, cls(response.headers, response.content)
734 )
735 return obj
737 def at_eof(self) -> bool:
738 """Returns True if the final boundary was reached, false otherwise."""
739 return self._at_eof
741 async def next(
742 self,
743 ) -> Union["MultipartReader", BodyPartReader] | None:
744 """Emits the next multipart body part."""
745 # So, if we're at BOF, we need to skip till the boundary.
746 if self._at_eof:
747 return None
748 await self._maybe_release_last_part()
749 if self._at_bof:
750 await self._read_until_first_boundary()
751 self._at_bof = False
752 else:
753 await self._read_boundary()
754 if self._at_eof: # we just read the last boundary, nothing to do there
755 # https://github.com/python/mypy/issues/17537
756 return None # type: ignore[unreachable]
758 part = await self.fetch_next_part()
759 # https://datatracker.ietf.org/doc/html/rfc7578#section-4.6
760 if (
761 self._last_part is None
762 and self._mimetype.subtype == "form-data"
763 and isinstance(part, BodyPartReader)
764 ):
765 _, params = parse_content_disposition(part.headers.get(CONTENT_DISPOSITION))
766 if params.get("name") == "_charset_":
767 # Longest encoding in https://encoding.spec.whatwg.org/encodings.json
768 # is 19 characters, so 32 should be more than enough for any valid encoding.
769 charset = await part.read_chunk(32)
770 if len(charset) > 31:
771 raise RuntimeError("Invalid default charset")
772 self._default_charset = charset.strip().decode()
773 part = await self.fetch_next_part()
774 self._last_part = part
775 return self._last_part
777 async def release(self) -> None:
778 """Reads all the body parts to the void till the final boundary."""
779 while not self._at_eof:
780 item = await self.next()
781 if item is None:
782 break
783 await item.release()
785 async def fetch_next_part(
786 self,
787 ) -> Union["MultipartReader", BodyPartReader]:
788 """Returns the next body part reader."""
789 headers = await self._read_headers()
790 return self._get_part_reader(headers)
792 def _get_part_reader(
793 self,
794 headers: HeadersDictProxy,
795 ) -> Union["MultipartReader", BodyPartReader]:
796 """Dispatches the response by the `Content-Type` header.
798 Returns a suitable reader instance.
800 :param dict headers: Response headers
801 """
802 ctype = headers.get(CONTENT_TYPE, "")
803 mimetype = parse_mimetype(ctype)
805 if mimetype.type == "multipart":
806 if self.multipart_reader_cls is None:
807 return type(self)(
808 headers,
809 self._content,
810 client_max_size=self._client_max_size,
811 max_field_size=self._max_field_size,
812 max_headers=self._max_headers,
813 max_size_error_cls=self._max_size_error_cls,
814 )
815 return self.multipart_reader_cls(
816 headers,
817 self._content,
818 client_max_size=self._client_max_size,
819 max_field_size=self._max_field_size,
820 max_headers=self._max_headers,
821 max_size_error_cls=self._max_size_error_cls,
822 )
823 else:
824 return self.part_reader_cls(
825 self._boundary,
826 headers,
827 self._content,
828 subtype=self._mimetype.subtype,
829 default_charset=self._default_charset,
830 client_max_size=self._client_max_size,
831 max_size_error_cls=self._max_size_error_cls,
832 )
834 def _get_boundary(self) -> str:
835 boundary = self._mimetype.parameters["boundary"]
836 if len(boundary) > 70:
837 raise ValueError("boundary %r is too long (70 chars max)" % boundary)
839 return boundary
841 async def _readline(self) -> bytes:
842 if self._unread:
843 return self._unread.pop()
844 return await self._content.readline()
846 async def _read_until_first_boundary(self) -> None:
847 while True:
848 chunk = await self._readline()
849 if chunk == b"":
850 raise ValueError(f"Could not find starting boundary {self._boundary!r}")
851 chunk = chunk.rstrip()
852 if chunk == self._boundary:
853 return
854 elif chunk == self._boundary + b"--":
855 self._at_eof = True
856 return
858 async def _read_boundary(self) -> None:
859 chunk = (await self._readline()).rstrip()
860 if chunk == self._boundary:
861 pass
862 elif chunk == self._boundary + b"--":
863 self._at_eof = True
864 epilogue = await self._readline()
865 next_line = await self._readline()
867 # the epilogue is expected and then either the end of input or the
868 # parent multipart boundary, if the parent boundary is found then
869 # it should be marked as unread and handed to the parent for
870 # processing
871 if next_line[:2] == b"--":
872 self._unread.append(next_line)
873 # otherwise the request is likely missing an epilogue and both
874 # lines should be passed to the parent for processing
875 # (this handles the old behavior gracefully)
876 else:
877 self._unread.extend([next_line, epilogue])
878 else:
879 raise ValueError(f"Invalid boundary {chunk!r}, expected {self._boundary!r}")
881 async def _read_headers(self) -> HeadersDictProxy:
882 lines = []
883 while True:
884 chunk = await self._content.readline(max_line_length=self._max_field_size)
885 chunk = chunk.rstrip(b"\r\n")
886 lines.append(chunk)
887 if not chunk:
888 break
889 if len(lines) > self._max_headers:
890 raise BadHttpMessage("Too many headers received")
891 parser = HeadersParser(max_field_size=self._max_field_size)
892 headers, _ = parser.parse_headers(lines)
893 return headers
895 async def _maybe_release_last_part(self) -> None:
896 """Ensures that the last read body part is read completely."""
897 if self._last_part is not None:
898 if not self._last_part.at_eof():
899 await self._last_part.release()
900 self._unread.extend(self._last_part._unread)
901 self._last_part = None
904_Part = tuple[Payload, str, str]
907class MultipartWriter(Payload):
908 """Multipart body writer."""
910 _value: None
911 # _consumed = False (inherited) - Can be encoded multiple times
912 _autoclose = True # No file handles, just collects parts in memory
914 def __init__(self, subtype: str = "mixed", boundary: str | None = None) -> None:
915 boundary = boundary if boundary is not None else uuid.uuid4().hex
916 # The underlying Payload API demands a str (utf-8), not bytes,
917 # so we need to ensure we don't lose anything during conversion.
918 # As a result, require the boundary to be ASCII only.
919 # In both situations.
921 try:
922 self._boundary = boundary.encode("ascii")
923 except UnicodeEncodeError:
924 raise ValueError("boundary should contain ASCII only chars") from None
926 if len(boundary) > 70:
927 raise ValueError("boundary %r is too long (70 chars max)" % boundary)
929 ctype = f"multipart/{subtype}; boundary={self._boundary_value}"
931 super().__init__(None, content_type=ctype)
933 self._parts: list[_Part] = []
934 self._is_form_data = subtype == "form-data"
936 def __enter__(self) -> "MultipartWriter":
937 return self
939 def __exit__(
940 self,
941 exc_type: type[BaseException] | None,
942 exc_val: BaseException | None,
943 exc_tb: TracebackType | None,
944 ) -> None:
945 pass
947 def __iter__(self) -> Iterator[_Part]:
948 return iter(self._parts)
950 def __len__(self) -> int:
951 return len(self._parts)
953 def __bool__(self) -> bool:
954 return True
956 _valid_tchar_regex = re.compile(rb"\A[!#$%&'*+\-.^_`|~\w]+\Z")
957 _invalid_qdtext_char_regex = re.compile(rb"[\x00-\x08\x0A-\x1F\x7F]")
959 @property
960 def _boundary_value(self) -> str:
961 """Wrap boundary parameter value in quotes, if necessary.
963 Reads self.boundary and returns a unicode string.
964 """
965 # Refer to RFCs 7231, 7230, 5234.
966 #
967 # parameter = token "=" ( token / quoted-string )
968 # token = 1*tchar
969 # quoted-string = DQUOTE *( qdtext / quoted-pair ) DQUOTE
970 # qdtext = HTAB / SP / %x21 / %x23-5B / %x5D-7E / obs-text
971 # obs-text = %x80-FF
972 # quoted-pair = "\" ( HTAB / SP / VCHAR / obs-text )
973 # tchar = "!" / "#" / "$" / "%" / "&" / "'" / "*"
974 # / "+" / "-" / "." / "^" / "_" / "`" / "|" / "~"
975 # / DIGIT / ALPHA
976 # ; any VCHAR, except delimiters
977 # VCHAR = %x21-7E
978 value = self._boundary
979 if re.match(self._valid_tchar_regex, value):
980 return value.decode("ascii") # cannot fail
982 if re.search(self._invalid_qdtext_char_regex, value):
983 raise ValueError("boundary value contains invalid characters")
985 # escape %x5C and %x22
986 quoted_value_content = value.replace(b"\\", b"\\\\")
987 quoted_value_content = quoted_value_content.replace(b'"', b'\\"')
989 return '"' + quoted_value_content.decode("ascii") + '"'
991 @property
992 def boundary(self) -> str:
993 return self._boundary.decode("ascii")
995 def append(self, obj: Any, headers: Mapping[str, str] | None = None) -> Payload:
996 if headers is None:
997 headers = CIMultiDict()
999 if isinstance(obj, Payload):
1000 obj.headers.update(headers)
1001 return self.append_payload(obj)
1002 else:
1003 try:
1004 payload = get_payload(obj, headers=headers)
1005 except LookupError:
1006 raise TypeError("Cannot create payload from %r" % obj)
1007 else:
1008 return self.append_payload(payload)
1010 def append_payload(self, payload: Payload) -> Payload:
1011 """Adds a new body part to multipart writer."""
1012 encoding: str | None = None
1013 te_encoding: str | None = None
1014 if self._is_form_data:
1015 # https://datatracker.ietf.org/doc/html/rfc7578#section-4.7
1016 # https://datatracker.ietf.org/doc/html/rfc7578#section-4.8
1017 assert (
1018 not {CONTENT_ENCODING, CONTENT_LENGTH, CONTENT_TRANSFER_ENCODING}
1019 & payload.headers.keys()
1020 )
1021 # Set default Content-Disposition in case user doesn't create one
1022 if CONTENT_DISPOSITION not in payload.headers:
1023 name = f"section-{len(self._parts)}"
1024 payload.set_content_disposition("form-data", name=name)
1025 else:
1026 # compression
1027 encoding = payload.headers.get(CONTENT_ENCODING, "").lower()
1028 if encoding and encoding not in ("deflate", "gzip", "identity"):
1029 raise RuntimeError(f"unknown content encoding: {encoding}")
1030 if encoding == "identity":
1031 encoding = None
1033 # te encoding
1034 te_encoding = payload.headers.get(CONTENT_TRANSFER_ENCODING, "").lower()
1035 if te_encoding not in ("", "base64", "quoted-printable", "binary"):
1036 raise RuntimeError(f"unknown content transfer encoding: {te_encoding}")
1037 if te_encoding == "binary":
1038 te_encoding = None
1040 # size
1041 size = payload.size
1042 if size is not None and not (encoding or te_encoding):
1043 payload.headers[CONTENT_LENGTH] = str(size)
1045 self._parts.append((payload, encoding, te_encoding)) # type: ignore[arg-type]
1046 return payload
1048 def append_json(
1049 self, obj: Any, headers: Mapping[str, str] | None = None
1050 ) -> Payload:
1051 """Helper to append JSON part."""
1052 if headers is None:
1053 headers = CIMultiDict()
1055 return self.append_payload(JsonPayload(obj, headers=headers))
1057 def append_form(
1058 self,
1059 obj: Sequence[tuple[str, str]] | Mapping[str, str],
1060 headers: Mapping[str, str] | None = None,
1061 ) -> Payload:
1062 """Helper to append form urlencoded part."""
1063 assert isinstance(obj, (Sequence, Mapping))
1065 if headers is None:
1066 headers = CIMultiDict()
1068 if isinstance(obj, Mapping):
1069 obj = list(obj.items())
1070 data = urlencode(obj, doseq=True)
1072 return self.append_payload(
1073 StringPayload(
1074 data, headers=headers, content_type="application/x-www-form-urlencoded"
1075 )
1076 )
1078 @property
1079 def size(self) -> int | None:
1080 """Size of the payload."""
1081 total = 0
1082 for part, encoding, te_encoding in self._parts:
1083 part_size = part.size
1084 if encoding or te_encoding or part_size is None:
1085 return None
1087 total += int(
1088 2
1089 + len(self._boundary)
1090 + 2
1091 + part_size # b'--'+self._boundary+b'\r\n'
1092 + len(part._binary_headers)
1093 + 2 # b'\r\n'
1094 )
1096 total += 2 + len(self._boundary) + 4 # b'--'+self._boundary+b'--\r\n'
1097 return total
1099 def decode(self, encoding: str = "utf-8", errors: str = "strict") -> str:
1100 """Return string representation of the multipart data.
1102 WARNING: This method may do blocking I/O if parts contain file payloads.
1103 It should not be called in the event loop. Use as_bytes().decode() instead.
1104 """
1105 return "".join(
1106 "--"
1107 + self.boundary
1108 + "\r\n"
1109 + part._binary_headers.decode(encoding, errors)
1110 + part.decode()
1111 for part, _e, _te in self._parts
1112 )
1114 async def as_bytes(self, encoding: str = "utf-8", errors: str = "strict") -> bytes:
1115 """Return bytes representation of the multipart data.
1117 This method is async-safe and calls as_bytes on underlying payloads.
1118 """
1119 parts: list[bytes] = []
1121 # Process each part
1122 for part, _e, _te in self._parts:
1123 # Add boundary
1124 parts.append(b"--" + self._boundary + b"\r\n")
1126 # Add headers
1127 parts.append(part._binary_headers)
1129 # Add payload content using as_bytes for async safety
1130 part_bytes = await part.as_bytes(encoding, errors)
1131 parts.append(part_bytes)
1133 # Add trailing CRLF
1134 parts.append(b"\r\n")
1136 # Add closing boundary
1137 parts.append(b"--" + self._boundary + b"--\r\n")
1139 return b"".join(parts)
1141 async def write(
1142 self, writer: AbstractStreamWriter, close_boundary: bool = True
1143 ) -> None:
1144 """Write body."""
1145 for part, encoding, te_encoding in self._parts:
1146 if self._is_form_data:
1147 # https://datatracker.ietf.org/doc/html/rfc7578#section-4.2
1148 assert CONTENT_DISPOSITION in part.headers
1149 assert "name=" in part.headers[CONTENT_DISPOSITION]
1151 await writer.write(b"--" + self._boundary + b"\r\n")
1152 await writer.write(part._binary_headers)
1154 if encoding or te_encoding:
1155 w = MultipartPayloadWriter(writer)
1156 if encoding:
1157 w.enable_compression(encoding)
1158 if te_encoding:
1159 w.enable_encoding(te_encoding)
1160 await part.write(w) # type: ignore[arg-type]
1161 await w.write_eof()
1162 else:
1163 await part.write(writer)
1165 await writer.write(b"\r\n")
1167 if close_boundary:
1168 await writer.write(b"--" + self._boundary + b"--\r\n")
1170 async def close(self) -> None:
1171 """
1172 Close all part payloads that need explicit closing.
1174 IMPORTANT: This method must not await anything that might not finish
1175 immediately, as it may be called during cleanup/cancellation. Schedule
1176 any long-running operations without awaiting them.
1177 """
1178 if self._consumed:
1179 return
1180 self._consumed = True
1182 # Close all parts that need explicit closing
1183 # We catch and log exceptions to ensure all parts get a chance to close
1184 # we do not use asyncio.gather() here because we are not allowed
1185 # to suspend given we may be called during cleanup
1186 for idx, (part, _, _) in enumerate(self._parts):
1187 if not part.autoclose and not part.consumed:
1188 try:
1189 await part.close()
1190 except Exception as exc:
1191 internal_logger.error(
1192 "Failed to close multipart part %d: %s", idx, exc, exc_info=True
1193 )
1196class MultipartPayloadWriter:
1197 def __init__(self, writer: AbstractStreamWriter) -> None:
1198 self._writer = writer
1199 self._encoding: str | None = None
1200 self._compress: ZLibCompressor | None = None
1201 self._encoding_buffer: bytearray | None = None
1203 def enable_encoding(self, encoding: str) -> None:
1204 if encoding == "base64":
1205 self._encoding = encoding
1206 self._encoding_buffer = bytearray()
1207 elif encoding == "quoted-printable":
1208 self._encoding = "quoted-printable"
1210 def enable_compression(
1211 self, encoding: str = "deflate", strategy: int | None = None
1212 ) -> None:
1213 self._compress = ZLibCompressor(
1214 encoding=encoding,
1215 suppress_deflate_header=True,
1216 strategy=strategy,
1217 )
1219 async def write_eof(self) -> None:
1220 if self._compress is not None:
1221 chunk = self._compress.flush()
1222 if chunk:
1223 self._compress = None
1224 await self.write(chunk)
1226 if self._encoding == "base64":
1227 if self._encoding_buffer:
1228 await self._writer.write(base64.b64encode(self._encoding_buffer))
1230 async def write(self, chunk: bytes) -> None:
1231 if self._compress is not None:
1232 if chunk:
1233 chunk = await self._compress.compress(chunk)
1234 if not chunk:
1235 return
1237 if self._encoding == "base64":
1238 buf = self._encoding_buffer
1239 assert buf is not None
1240 buf.extend(chunk)
1242 if buf:
1243 div, mod = divmod(len(buf), 3)
1244 enc_chunk, self._encoding_buffer = (buf[: div * 3], buf[div * 3 :])
1245 if enc_chunk:
1246 b64chunk = base64.b64encode(enc_chunk)
1247 await self._writer.write(b64chunk)
1248 elif self._encoding == "quoted-printable":
1249 await self._writer.write(binascii.b2a_qp(chunk))
1250 else:
1251 await self._writer.write(chunk)