Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/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, 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)
64if TYPE_CHECKING:
65 from .client_reqrep import ClientResponse
68class BadContentDispositionHeader(RuntimeWarning):
69 pass
72class BadContentDispositionParam(RuntimeWarning):
73 pass
76def parse_content_disposition(
77 header: str | None,
78) -> tuple[str | None, dict[str, str]]:
79 def is_token(string: str) -> bool:
80 return bool(string) and TOKEN >= set(string)
82 def is_quoted(string: str) -> bool:
83 return len(string) >= 2 and string[0] == string[-1] == '"'
85 def is_rfc5987(string: str) -> bool:
86 return is_token(string) and string.count("'") == 2
88 def is_extended_param(string: str) -> bool:
89 return string.endswith("*")
91 def is_continuous_param(string: str) -> bool:
92 pos = string.find("*") + 1
93 if not pos:
94 return False
95 substring = string[pos:-1] if string.endswith("*") else string[pos:]
96 return substring.isdigit()
98 def unescape(text: str, *, chars: str = "".join(map(re.escape, CHAR))) -> str:
99 return re.sub(f"\\\\([{chars}])", "\\1", text)
101 if not header:
102 return None, {}
104 # https://www.rfc-editor.org/info/rfc9110/#section-5.6.6-2
105 disptype, *parts = header.split(";")
106 disptype = disptype.strip()
107 if not is_token(disptype):
108 warnings.warn(BadContentDispositionHeader(header))
109 return None, {}
111 params: dict[str, str] = {}
112 while parts:
113 item = parts.pop(0)
115 if not item: # To handle trailing semicolons
116 warnings.warn(BadContentDispositionHeader(header))
117 continue
119 if "=" not in item:
120 warnings.warn(BadContentDispositionHeader(header))
121 return None, {}
123 key, value = item.split("=", 1)
124 key = key.lower().strip()
125 value = value.lstrip()
127 if key in params:
128 warnings.warn(BadContentDispositionHeader(header))
129 return None, {}
131 if not is_token(key):
132 warnings.warn(BadContentDispositionParam(item))
133 continue
135 elif is_continuous_param(key):
136 if is_quoted(value):
137 value = unescape(value[1:-1])
138 elif not is_token(value):
139 warnings.warn(BadContentDispositionParam(item))
140 continue
142 elif is_extended_param(key):
143 if is_rfc5987(value):
144 encoding, _, value = value.split("'", 2)
145 encoding = encoding or "utf-8"
146 else:
147 warnings.warn(BadContentDispositionParam(item))
148 continue
150 try:
151 value = unquote(value, encoding, "strict")
152 except (builtins.LookupError, UnicodeDecodeError):
153 # The charset is attacker-controlled here; an unknown name
154 # raises the builtin LookupError (the bare name is shadowed in
155 # this module by payload.LookupError).
156 warnings.warn(BadContentDispositionParam(item))
157 continue
159 else:
160 failed = True
161 rstripped = value.rstrip()
162 if is_quoted(rstripped):
163 failed = False
164 value = unescape(rstripped[1:-1].lstrip("\\/"))
165 elif is_token(value):
166 failed = False
167 elif parts:
168 # maybe just ; in filename, in any case this is just
169 # one case fix, for proper fix we need to redesign parser
170 _value = f"{value};{parts[0]}"
171 if is_quoted(_value):
172 parts.pop(0)
173 value = unescape(_value[1:-1].lstrip("\\/"))
174 failed = False
176 if failed:
177 warnings.warn(BadContentDispositionHeader(header))
178 return None, {}
180 params[key] = value
182 return disptype.lower(), params
185def content_disposition_filename(
186 params: Mapping[str, str], name: str = "filename"
187) -> str | None:
188 name_suf = "%s*" % name
189 if not params:
190 return None
191 elif name_suf in params:
192 return params[name_suf]
193 elif name in params:
194 return params[name]
195 else:
196 parts = []
197 fnparams = sorted(
198 (key, value) for key, value in params.items() if key.startswith(name_suf)
199 )
200 for num, (key, value) in enumerate(fnparams):
201 _, tail = key.split("*", 1)
202 if tail.endswith("*"):
203 tail = tail[:-1]
204 if tail == str(num):
205 parts.append(value)
206 else:
207 break
208 if not parts:
209 return None
210 value = "".join(parts)
211 if "'" in value:
212 encoding, _, value = value.split("'", 2)
213 encoding = encoding or "utf-8"
214 try:
215 return unquote(value, encoding, "strict")
216 except (builtins.LookupError, UnicodeDecodeError):
217 # Both the charset name and the octets are attacker-controlled
218 # here; an unknown encoding raises the builtin LookupError
219 # (shadowed in this module by payload.LookupError) and
220 # undecodable bytes raise UnicodeDecodeError.
221 return None
222 return value
225class MultipartResponseWrapper:
226 """Wrapper around the MultipartReader.
228 It takes care about
229 underlying connection and close it when it needs in.
230 """
232 def __init__(
233 self,
234 resp: "ClientResponse",
235 stream: "MultipartReader",
236 ) -> None:
237 self.resp = resp
238 self.stream = stream
240 def __aiter__(self) -> "MultipartResponseWrapper":
241 return self
243 async def __anext__(
244 self,
245 ) -> Union["MultipartReader", "BodyPartReader"]:
246 part = await self.next()
247 if part is None:
248 raise StopAsyncIteration
249 return part
251 def at_eof(self) -> bool:
252 """Returns True when all response data had been read."""
253 return self.resp.content.at_eof()
255 async def next(
256 self,
257 ) -> Union["MultipartReader", "BodyPartReader"] | None:
258 """Emits next multipart reader object."""
259 item = await self.stream.next()
260 if self.stream.at_eof():
261 await self.release()
262 return item
264 async def release(self) -> None:
265 """Release the connection gracefully.
267 All remaining content is read to the void.
268 """
269 await self.resp.release()
272class BodyPartReader:
273 """Multipart reader for single body part."""
275 chunk_size = 8192
277 def __init__(
278 self,
279 boundary: bytes,
280 headers: "CIMultiDictProxy[str]",
281 content: StreamReader,
282 *,
283 subtype: str = "mixed",
284 default_charset: str | None = None,
285 max_decompress_size: int = DEFAULT_CHUNK_SIZE,
286 client_max_size: int = sys.maxsize,
287 max_size_error_cls: type[Exception] = ValueError,
288 ) -> None:
289 self.headers = headers
290 self._boundary = boundary
291 self._boundary_len = len(boundary) + 2 # Boundary + \r\n
292 self._content = content
293 self._default_charset = default_charset
294 self._at_eof = False
295 self._is_form_data = subtype == "form-data"
296 # https://datatracker.ietf.org/doc/html/rfc7578#section-4.8
297 length = None if self._is_form_data else self.headers.get(CONTENT_LENGTH, None)
298 if length is not None and not (length.isascii() and length.isdigit()):
299 # Reject sign prefixes, underscores, whitespace and non-ASCII
300 # digits that int() would otherwise accept.
301 # https://www.rfc-editor.org/rfc/rfc9110#section-8.6
302 raise ValueError(f"invalid Content-Length: {length!r}")
303 self._length = int(length) if length is not None else None
304 self._read_bytes = 0
305 self._unread: deque[bytes] = deque()
306 self._prev_chunk: bytes | None = None
307 self._content_eof = 0
308 self._cache: dict[str, Any] = {}
309 self._max_decompress_size = max_decompress_size
310 self._client_max_size = client_max_size
311 self._max_size_error_cls = max_size_error_cls
313 def __aiter__(self: Self) -> Self:
314 return self
316 async def __anext__(self) -> bytes:
317 part = await self.next()
318 if part is None:
319 raise StopAsyncIteration
320 return part
322 async def next(self) -> bytes | None:
323 item = await self.read()
324 if not item:
325 return None
326 return item
328 async def read(self, *, decode: bool = False) -> bytes:
329 """Reads body part data.
331 decode: Decodes data following by encoding
332 method from Content-Encoding header. If it missed
333 data remains untouched
334 """
335 if self._at_eof:
336 return b""
337 data = bytearray()
338 while not self._at_eof:
339 data.extend(await self.read_chunk(self.chunk_size))
340 if len(data) > self._client_max_size:
341 raise self._max_size_error_cls(self._client_max_size)
342 if decode:
343 decoded_data = bytearray()
344 async for d in self.decode_iter(data):
345 decoded_data.extend(d)
346 if len(decoded_data) > self._client_max_size:
347 raise self._max_size_error_cls(self._client_max_size)
348 return decoded_data
349 return data
351 async def read_chunk(self, size: int = chunk_size) -> bytes:
352 """Reads body part content chunk of the specified size.
354 size: chunk size
355 """
356 if self._at_eof:
357 return b""
358 if self._length:
359 chunk = await self._read_chunk_from_length(size)
360 else:
361 chunk = await self._read_chunk_from_stream(size)
363 # For the case of base64 data, we must read a fragment of size with a
364 # remainder of 0 by dividing by 4 for string without symbols \n or \r
365 encoding = self.headers.get(CONTENT_TRANSFER_ENCODING)
366 if encoding and encoding.lower() == "base64":
367 stripped_chunk = b"".join(chunk.split())
368 remainder = len(stripped_chunk) % 4
370 while remainder != 0 and not self.at_eof():
371 over_chunk_size = 4 - remainder
372 over_chunk = b""
374 if self._prev_chunk:
375 over_chunk = self._prev_chunk[:over_chunk_size]
376 self._prev_chunk = self._prev_chunk[len(over_chunk) :]
378 if len(over_chunk) != over_chunk_size:
379 over_chunk += await self._content.read(4 - len(over_chunk))
381 if not over_chunk:
382 self._at_eof = True
384 stripped_chunk += b"".join(over_chunk.split())
385 chunk += over_chunk
386 remainder = len(stripped_chunk) % 4
388 self._read_bytes += len(chunk)
389 if self._read_bytes == self._length:
390 self._at_eof = True
391 if self._at_eof and await self._content.readline() != b"\r\n":
392 raise ValueError("Reader did not read all the data or it is malformed")
393 return chunk
395 async def _read_chunk_from_length(self, size: int) -> bytes:
396 # Reads body part content chunk of the specified size.
397 # The body part must has Content-Length header with proper value.
398 assert self._length is not None, "Content-Length required for chunked read"
399 chunk_size = min(size, self._length - self._read_bytes)
400 chunk = await self._content.read(chunk_size)
401 if self._content.at_eof():
402 self._at_eof = True
403 return chunk
405 async def _read_chunk_from_stream(self, size: int) -> bytes:
406 # Reads content chunk of body part with unknown length.
407 # The Content-Length header for body part is not necessary.
408 assert (
409 size >= self._boundary_len
410 ), "Chunk size must be greater or equal than boundary length + 2"
411 first_chunk = self._prev_chunk is None
412 if first_chunk:
413 # We need to re-add the CRLF that got removed from headers parsing.
414 self._prev_chunk = b"\r\n" + await self._content.read(size)
416 chunk = b""
417 # content.read() may return less than size, so we need to loop to ensure
418 # we have enough data to detect the boundary.
419 while len(chunk) < self._boundary_len:
420 chunk += await self._content.read(size)
421 self._content_eof += int(self._content.at_eof())
422 if self._content_eof > 2:
423 raise ValueError("Reading after EOF")
424 if self._content_eof:
425 break
426 if len(chunk) > size:
427 self._content.unread_data(chunk[size:])
428 chunk = chunk[:size]
430 assert self._prev_chunk is not None
431 window = self._prev_chunk + chunk
432 sub = b"\r\n" + self._boundary
433 if first_chunk:
434 idx = window.find(sub)
435 else:
436 idx = window.find(sub, max(0, len(self._prev_chunk) - len(sub)))
437 if idx >= 0:
438 # pushing boundary back to content
439 with warnings.catch_warnings():
440 warnings.filterwarnings("ignore", category=DeprecationWarning)
441 self._content.unread_data(window[idx:])
442 self._prev_chunk = self._prev_chunk[:idx]
443 chunk = window[len(self._prev_chunk) : idx]
444 if not chunk:
445 self._at_eof = True
446 result = self._prev_chunk[2 if first_chunk else 0 :] # Strip initial CRLF
447 self._prev_chunk = chunk
448 return result
450 async def readline(self) -> bytes:
451 """Reads body part by line by line."""
452 if self._at_eof:
453 return b""
455 if self._unread:
456 line = self._unread.popleft()
457 else:
458 line = await self._content.readline()
460 if line.startswith(self._boundary):
461 # the very last boundary may not come with \r\n,
462 # so set single rules for everyone
463 sline = line.rstrip(b"\r\n")
464 boundary = self._boundary
465 last_boundary = self._boundary + b"--"
466 # ensure that we read exactly the boundary, not something alike
467 if sline == boundary or sline == last_boundary:
468 self._at_eof = True
469 self._unread.append(line)
470 return b""
471 else:
472 next_line = await self._content.readline()
473 if next_line.startswith(self._boundary):
474 line = line[:-2] # strip CRLF but only once
475 self._unread.append(next_line)
477 return line
479 async def release(self) -> None:
480 """Like read(), but reads all the data to the void."""
481 if self._at_eof:
482 return
483 while not self._at_eof:
484 await self.read_chunk(self.chunk_size)
486 async def text(self, *, encoding: str | None = None) -> str:
487 """Like read(), but assumes that body part contains text data."""
488 data = await self.read(decode=True)
489 # see https://www.w3.org/TR/html5/forms.html#multipart/form-data-encoding-algorithm
490 # and https://dvcs.w3.org/hg/xhr/raw-file/tip/Overview.html#dom-xmlhttprequest-send
491 encoding = encoding or self.get_charset(default="utf-8")
492 return data.decode(encoding)
494 async def json(self, *, encoding: str | None = None) -> dict[str, Any] | None:
495 """Like read(), but assumes that body parts contains JSON data."""
496 data = await self.read(decode=True)
497 if not data:
498 return None
499 encoding = encoding or self.get_charset(default="utf-8")
500 return cast(dict[str, Any], json.loads(data.decode(encoding)))
502 async def form(self, *, encoding: str | None = None) -> list[tuple[str, str]]:
503 """Like read(), but assumes that body parts contain form urlencoded data."""
504 data = await self.read(decode=True)
505 if not data:
506 return []
507 if encoding is not None:
508 real_encoding = encoding
509 else:
510 real_encoding = self.get_charset(default="utf-8")
511 try:
512 decoded_data = data.rstrip().decode(real_encoding)
513 except UnicodeDecodeError:
514 raise ValueError("data cannot be decoded with %s encoding" % real_encoding)
516 return parse_qsl(
517 decoded_data,
518 keep_blank_values=True,
519 encoding=real_encoding,
520 )
522 def at_eof(self) -> bool:
523 """Returns True if the boundary was reached or False otherwise."""
524 return self._at_eof
526 def _apply_content_transfer_decoding(self, data: _Buffer) -> _Buffer | bytes:
527 """Apply Content-Transfer-Encoding decoding if header is present."""
528 if CONTENT_TRANSFER_ENCODING in self.headers:
529 return self._decode_content_transfer(data)
530 return data
532 def _needs_content_decoding(self) -> bool:
533 """Check if Content-Encoding decoding should be applied."""
534 # https://datatracker.ietf.org/doc/html/rfc7578#section-4.8
535 return not self._is_form_data and CONTENT_ENCODING in self.headers
537 def decode(self, data: _Buffer) -> _Buffer | bytes:
538 """Decodes data synchronously.
540 Decodes data according the specified Content-Encoding
541 or Content-Transfer-Encoding headers value.
543 Note: For large payloads, consider using decode_iter() instead
544 to avoid blocking the event loop during decompression.
545 """
546 decoded = self._apply_content_transfer_decoding(data)
547 if self._needs_content_decoding():
548 return self._decode_content(decoded)
549 return decoded
551 async def decode_iter(self, data: _Buffer) -> AsyncIterator[_Buffer | bytes]:
552 """Async generator that yields decoded data chunks.
554 Decodes data according the specified Content-Encoding
555 or Content-Transfer-Encoding headers value.
557 This method offloads decompression to an executor for large payloads
558 to avoid blocking the event loop.
559 """
560 decoded = self._apply_content_transfer_decoding(data)
561 if self._needs_content_decoding():
562 async for d in self._decode_content_async(decoded):
563 yield d
564 else:
565 yield decoded
567 def _decode_content(self, data: _Buffer) -> _Buffer | bytes:
568 encoding = self.headers.get(CONTENT_ENCODING, "").lower()
569 if encoding == "identity":
570 return data
571 if encoding in {"deflate", "gzip"}:
572 return ZLibDecompressor(
573 encoding=encoding,
574 suppress_deflate_header=True,
575 ).decompress_sync(data, max_length=self._max_decompress_size)
577 raise RuntimeError(f"unknown content encoding: {encoding}")
579 async def _decode_content_async(
580 self, data: _Buffer
581 ) -> AsyncIterator[_Buffer | bytes]:
582 encoding = self.headers.get(CONTENT_ENCODING, "").lower()
583 if encoding == "identity":
584 yield data
585 elif encoding in {"deflate", "gzip"}:
586 d = ZLibDecompressor(
587 encoding=encoding,
588 suppress_deflate_header=True,
589 )
590 yield await d.decompress(data, max_length=self._max_decompress_size)
591 while d.data_available:
592 yield await d.decompress(b"", max_length=self._max_decompress_size)
593 else:
594 raise RuntimeError(f"unknown content encoding: {encoding}")
596 def _decode_content_transfer(self, data: _Buffer) -> _Buffer | bytes:
597 encoding = self.headers.get(CONTENT_TRANSFER_ENCODING, "").lower()
599 if encoding == "base64":
600 return base64.b64decode(data)
601 elif encoding == "quoted-printable":
602 return binascii.a2b_qp(data)
603 elif encoding in ("binary", "8bit", "7bit"):
604 return data
605 else:
606 raise RuntimeError(f"unknown content transfer encoding: {encoding}")
608 def get_charset(self, default: str) -> str:
609 """Returns charset parameter from Content-Type header or default."""
610 ctype = self.headers.get(CONTENT_TYPE, "")
611 mimetype = parse_mimetype(ctype)
612 return mimetype.parameters.get("charset", self._default_charset or default)
614 @reify
615 def name(self) -> str | None:
616 """Returns name specified in Content-Disposition header.
618 If the header is missing or malformed, returns None.
619 """
620 _, params = parse_content_disposition(self.headers.get(CONTENT_DISPOSITION))
621 return content_disposition_filename(params, "name")
623 @reify
624 def filename(self) -> str | None:
625 """Returns filename specified in Content-Disposition header.
627 Returns None if the header is missing or malformed.
628 """
629 _, params = parse_content_disposition(self.headers.get(CONTENT_DISPOSITION))
630 return content_disposition_filename(params, "filename")
633@payload_type(BodyPartReader, order=Order.try_first)
634class BodyPartReaderPayload(Payload):
635 _value: BodyPartReader
636 # _autoclose = False (inherited) - Streaming reader that may have resources
638 def __init__(self, value: BodyPartReader, *args: Any, **kwargs: Any) -> None:
639 super().__init__(value, *args, **kwargs)
641 params: dict[str, str] = {}
642 if value.name is not None:
643 params["name"] = value.name
644 if value.filename is not None:
645 params["filename"] = value.filename
647 if params:
648 self.set_content_disposition("attachment", True, **params)
650 def decode(self, encoding: str = "utf-8", errors: str = "strict") -> str:
651 raise TypeError("Unable to decode.")
653 async def as_bytes(self, encoding: str = "utf-8", errors: str = "strict") -> bytes:
654 """Raises TypeError as body parts should be consumed via write().
656 This is intentional: BodyPartReader payloads are designed for streaming
657 large data (potentially gigabytes) and must be consumed only once via
658 the write() method to avoid memory exhaustion. They cannot be buffered
659 in memory for reuse.
660 """
661 raise TypeError("Unable to read body part as bytes. Use write() to consume.")
663 async def write(self, writer: AbstractStreamWriter) -> None:
664 field = self._value
665 while chunk := await field.read_chunk(size=DEFAULT_CHUNK_SIZE):
666 async for d in field.decode_iter(chunk):
667 await writer.write(d)
670class MultipartReader:
671 """Multipart body reader."""
673 #: Response wrapper, used when multipart readers constructs from response.
674 response_wrapper_cls = MultipartResponseWrapper
675 #: Multipart reader class, used to handle multipart/* body parts.
676 #: None points to type(self)
677 multipart_reader_cls: type["MultipartReader"] | None = None
678 #: Body part reader class for non multipart/* content types.
679 part_reader_cls = BodyPartReader
681 def __init__(
682 self,
683 headers: Mapping[str, str],
684 content: StreamReader,
685 *,
686 client_max_size: int = sys.maxsize,
687 max_field_size: int = 8190,
688 max_headers: int = 128,
689 max_size_error_cls: type[Exception] = ValueError,
690 ) -> None:
691 self._mimetype = parse_mimetype(headers[CONTENT_TYPE])
692 assert self._mimetype.type == "multipart", "multipart/* content type expected"
693 if "boundary" not in self._mimetype.parameters:
694 raise ValueError(
695 "boundary missed for Content-Type: %s" % headers[CONTENT_TYPE]
696 )
698 self.headers = headers
699 self._boundary = ("--" + self._get_boundary()).encode()
700 self._client_max_size = client_max_size
701 self._content = content
702 self._default_charset: str | None = None
703 self._last_part: MultipartReader | BodyPartReader | None = None
704 self._max_field_size = max_field_size
705 self._max_headers = max_headers
706 self._max_size_error_cls = max_size_error_cls
707 self._at_eof = False
708 self._at_bof = True
709 self._unread: list[bytes] = []
711 def __aiter__(self: Self) -> Self:
712 return self
714 async def __anext__(
715 self,
716 ) -> Union["MultipartReader", BodyPartReader] | None:
717 part = await self.next()
718 if part is None:
719 raise StopAsyncIteration
720 return part
722 @classmethod
723 def from_response(
724 cls,
725 response: "ClientResponse",
726 ) -> MultipartResponseWrapper:
727 """Constructs reader instance from HTTP response.
729 :param response: :class:`~aiohttp.client.ClientResponse` instance
730 """
731 obj = cls.response_wrapper_cls(
732 response, cls(response.headers, response.content)
733 )
734 return obj
736 def at_eof(self) -> bool:
737 """Returns True if the final boundary was reached, false otherwise."""
738 return self._at_eof
740 async def next(
741 self,
742 ) -> Union["MultipartReader", BodyPartReader] | None:
743 """Emits the next multipart body part."""
744 # So, if we're at BOF, we need to skip till the boundary.
745 if self._at_eof:
746 return None
747 await self._maybe_release_last_part()
748 if self._at_bof:
749 await self._read_until_first_boundary()
750 self._at_bof = False
751 else:
752 await self._read_boundary()
753 if self._at_eof: # we just read the last boundary, nothing to do there
754 return None
756 part = await self.fetch_next_part()
757 # https://datatracker.ietf.org/doc/html/rfc7578#section-4.6
758 if (
759 self._last_part is None
760 and self._mimetype.subtype == "form-data"
761 and isinstance(part, BodyPartReader)
762 ):
763 _, params = parse_content_disposition(part.headers.get(CONTENT_DISPOSITION))
764 if params.get("name") == "_charset_":
765 # Longest encoding in https://encoding.spec.whatwg.org/encodings.json
766 # is 19 characters, so 32 should be more than enough for any valid encoding.
767 charset = await part.read_chunk(32)
768 if len(charset) > 31:
769 raise RuntimeError("Invalid default charset")
770 self._default_charset = charset.strip().decode()
771 part = await self.fetch_next_part()
772 self._last_part = part
773 return self._last_part
775 async def release(self) -> None:
776 """Reads all the body parts to the void till the final boundary."""
777 while not self._at_eof:
778 item = await self.next()
779 if item is None:
780 break
781 await item.release()
783 async def fetch_next_part(
784 self,
785 ) -> Union["MultipartReader", BodyPartReader]:
786 """Returns the next body part reader."""
787 headers = await self._read_headers()
788 return self._get_part_reader(headers)
790 def _get_part_reader(
791 self,
792 headers: "CIMultiDictProxy[str]",
793 ) -> Union["MultipartReader", BodyPartReader]:
794 """Dispatches the response by the `Content-Type` header.
796 Returns a suitable reader instance.
798 :param dict headers: Response headers
799 """
800 ctype = headers.get(CONTENT_TYPE, "")
801 mimetype = parse_mimetype(ctype)
803 if mimetype.type == "multipart":
804 if self.multipart_reader_cls is None:
805 return type(self)(
806 headers,
807 self._content,
808 client_max_size=self._client_max_size,
809 max_field_size=self._max_field_size,
810 max_headers=self._max_headers,
811 max_size_error_cls=self._max_size_error_cls,
812 )
813 return self.multipart_reader_cls(
814 headers,
815 self._content,
816 client_max_size=self._client_max_size,
817 max_field_size=self._max_field_size,
818 max_headers=self._max_headers,
819 max_size_error_cls=self._max_size_error_cls,
820 )
821 else:
822 return self.part_reader_cls(
823 self._boundary,
824 headers,
825 self._content,
826 subtype=self._mimetype.subtype,
827 default_charset=self._default_charset,
828 client_max_size=self._client_max_size,
829 max_size_error_cls=self._max_size_error_cls,
830 )
832 def _get_boundary(self) -> str:
833 boundary = self._mimetype.parameters["boundary"]
834 if len(boundary) > 70:
835 raise ValueError("boundary %r is too long (70 chars max)" % boundary)
837 return boundary
839 async def _readline(self) -> bytes:
840 if self._unread:
841 return self._unread.pop()
842 return await self._content.readline()
844 async def _read_until_first_boundary(self) -> None:
845 while True:
846 chunk = await self._readline()
847 if chunk == b"":
848 raise ValueError(
849 "Could not find starting boundary %r" % (self._boundary)
850 )
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) -> "CIMultiDictProxy[str]":
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, raw_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
925 ctype = f"multipart/{subtype}; boundary={self._boundary_value}"
927 super().__init__(None, content_type=ctype)
929 self._parts: list[_Part] = []
930 self._is_form_data = subtype == "form-data"
932 def __enter__(self) -> "MultipartWriter":
933 return self
935 def __exit__(
936 self,
937 exc_type: type[BaseException] | None,
938 exc_val: BaseException | None,
939 exc_tb: TracebackType | None,
940 ) -> None:
941 pass
943 def __iter__(self) -> Iterator[_Part]:
944 return iter(self._parts)
946 def __len__(self) -> int:
947 return len(self._parts)
949 def __bool__(self) -> bool:
950 return True
952 _valid_tchar_regex = re.compile(rb"\A[!#$%&'*+\-.^_`|~\w]+\Z")
953 _invalid_qdtext_char_regex = re.compile(rb"[\x00-\x08\x0A-\x1F\x7F]")
955 @property
956 def _boundary_value(self) -> str:
957 """Wrap boundary parameter value in quotes, if necessary.
959 Reads self.boundary and returns a unicode string.
960 """
961 # Refer to RFCs 7231, 7230, 5234.
962 #
963 # parameter = token "=" ( token / quoted-string )
964 # token = 1*tchar
965 # quoted-string = DQUOTE *( qdtext / quoted-pair ) DQUOTE
966 # qdtext = HTAB / SP / %x21 / %x23-5B / %x5D-7E / obs-text
967 # obs-text = %x80-FF
968 # quoted-pair = "\" ( HTAB / SP / VCHAR / obs-text )
969 # tchar = "!" / "#" / "$" / "%" / "&" / "'" / "*"
970 # / "+" / "-" / "." / "^" / "_" / "`" / "|" / "~"
971 # / DIGIT / ALPHA
972 # ; any VCHAR, except delimiters
973 # VCHAR = %x21-7E
974 value = self._boundary
975 if re.match(self._valid_tchar_regex, value):
976 return value.decode("ascii") # cannot fail
978 if re.search(self._invalid_qdtext_char_regex, value):
979 raise ValueError("boundary value contains invalid characters")
981 # escape %x5C and %x22
982 quoted_value_content = value.replace(b"\\", b"\\\\")
983 quoted_value_content = quoted_value_content.replace(b'"', b'\\"')
985 return '"' + quoted_value_content.decode("ascii") + '"'
987 @property
988 def boundary(self) -> str:
989 return self._boundary.decode("ascii")
991 def append(self, obj: Any, headers: Mapping[str, str] | None = None) -> Payload:
992 if headers is None:
993 headers = CIMultiDict()
995 if isinstance(obj, Payload):
996 obj.headers.update(headers)
997 return self.append_payload(obj)
998 else:
999 try:
1000 payload = get_payload(obj, headers=headers)
1001 except LookupError:
1002 raise TypeError("Cannot create payload from %r" % obj)
1003 else:
1004 return self.append_payload(payload)
1006 def append_payload(self, payload: Payload) -> Payload:
1007 """Adds a new body part to multipart writer."""
1008 encoding: str | None = None
1009 te_encoding: str | None = None
1010 if self._is_form_data:
1011 # https://datatracker.ietf.org/doc/html/rfc7578#section-4.7
1012 # https://datatracker.ietf.org/doc/html/rfc7578#section-4.8
1013 assert (
1014 not {CONTENT_ENCODING, CONTENT_LENGTH, CONTENT_TRANSFER_ENCODING}
1015 & payload.headers.keys()
1016 )
1017 # Set default Content-Disposition in case user doesn't create one
1018 if CONTENT_DISPOSITION not in payload.headers:
1019 name = f"section-{len(self._parts)}"
1020 payload.set_content_disposition("form-data", name=name)
1021 else:
1022 # compression
1023 encoding = payload.headers.get(CONTENT_ENCODING, "").lower()
1024 if encoding and encoding not in ("deflate", "gzip", "identity"):
1025 raise RuntimeError(f"unknown content encoding: {encoding}")
1026 if encoding == "identity":
1027 encoding = None
1029 # te encoding
1030 te_encoding = payload.headers.get(CONTENT_TRANSFER_ENCODING, "").lower()
1031 if te_encoding not in ("", "base64", "quoted-printable", "binary"):
1032 raise RuntimeError(f"unknown content transfer encoding: {te_encoding}")
1033 if te_encoding == "binary":
1034 te_encoding = None
1036 # size
1037 size = payload.size
1038 if size is not None and not (encoding or te_encoding):
1039 payload.headers[CONTENT_LENGTH] = str(size)
1041 self._parts.append((payload, encoding, te_encoding)) # type: ignore[arg-type]
1042 return payload
1044 def append_json(
1045 self, obj: Any, headers: Mapping[str, str] | None = None
1046 ) -> Payload:
1047 """Helper to append JSON part."""
1048 if headers is None:
1049 headers = CIMultiDict()
1051 return self.append_payload(JsonPayload(obj, headers=headers))
1053 def append_form(
1054 self,
1055 obj: Sequence[tuple[str, str]] | Mapping[str, str],
1056 headers: Mapping[str, str] | None = None,
1057 ) -> Payload:
1058 """Helper to append form urlencoded part."""
1059 assert isinstance(obj, (Sequence, Mapping))
1061 if headers is None:
1062 headers = CIMultiDict()
1064 if isinstance(obj, Mapping):
1065 obj = list(obj.items())
1066 data = urlencode(obj, doseq=True)
1068 return self.append_payload(
1069 StringPayload(
1070 data, headers=headers, content_type="application/x-www-form-urlencoded"
1071 )
1072 )
1074 @property
1075 def size(self) -> int | None:
1076 """Size of the payload."""
1077 total = 0
1078 for part, encoding, te_encoding in self._parts:
1079 part_size = part.size
1080 if encoding or te_encoding or part_size is None:
1081 return None
1083 total += int(
1084 2
1085 + len(self._boundary)
1086 + 2
1087 + part_size # b'--'+self._boundary+b'\r\n'
1088 + len(part._binary_headers)
1089 + 2 # b'\r\n'
1090 )
1092 total += 2 + len(self._boundary) + 4 # b'--'+self._boundary+b'--\r\n'
1093 return total
1095 def decode(self, encoding: str = "utf-8", errors: str = "strict") -> str:
1096 """Return string representation of the multipart data.
1098 WARNING: This method may do blocking I/O if parts contain file payloads.
1099 It should not be called in the event loop. Use as_bytes().decode() instead.
1100 """
1101 return "".join(
1102 "--"
1103 + self.boundary
1104 + "\r\n"
1105 + part._binary_headers.decode(encoding, errors)
1106 + part.decode()
1107 for part, _e, _te in self._parts
1108 )
1110 async def as_bytes(self, encoding: str = "utf-8", errors: str = "strict") -> bytes:
1111 """Return bytes representation of the multipart data.
1113 This method is async-safe and calls as_bytes on underlying payloads.
1114 """
1115 parts: list[bytes] = []
1117 # Process each part
1118 for part, _e, _te in self._parts:
1119 # Add boundary
1120 parts.append(b"--" + self._boundary + b"\r\n")
1122 # Add headers
1123 parts.append(part._binary_headers)
1125 # Add payload content using as_bytes for async safety
1126 part_bytes = await part.as_bytes(encoding, errors)
1127 parts.append(part_bytes)
1129 # Add trailing CRLF
1130 parts.append(b"\r\n")
1132 # Add closing boundary
1133 parts.append(b"--" + self._boundary + b"--\r\n")
1135 return b"".join(parts)
1137 async def write(
1138 self, writer: AbstractStreamWriter, close_boundary: bool = True
1139 ) -> None:
1140 """Write body."""
1141 for part, encoding, te_encoding in self._parts:
1142 if self._is_form_data:
1143 # https://datatracker.ietf.org/doc/html/rfc7578#section-4.2
1144 assert CONTENT_DISPOSITION in part.headers
1145 assert "name=" in part.headers[CONTENT_DISPOSITION]
1147 await writer.write(b"--" + self._boundary + b"\r\n")
1148 await writer.write(part._binary_headers)
1150 if encoding or te_encoding:
1151 w = MultipartPayloadWriter(writer)
1152 if encoding:
1153 w.enable_compression(encoding)
1154 if te_encoding:
1155 w.enable_encoding(te_encoding)
1156 await part.write(w) # type: ignore[arg-type]
1157 await w.write_eof()
1158 else:
1159 await part.write(writer)
1161 await writer.write(b"\r\n")
1163 if close_boundary:
1164 await writer.write(b"--" + self._boundary + b"--\r\n")
1166 async def close(self) -> None:
1167 """
1168 Close all part payloads that need explicit closing.
1170 IMPORTANT: This method must not await anything that might not finish
1171 immediately, as it may be called during cleanup/cancellation. Schedule
1172 any long-running operations without awaiting them.
1173 """
1174 if self._consumed:
1175 return
1176 self._consumed = True
1178 # Close all parts that need explicit closing
1179 # We catch and log exceptions to ensure all parts get a chance to close
1180 # we do not use asyncio.gather() here because we are not allowed
1181 # to suspend given we may be called during cleanup
1182 for idx, (part, _, _) in enumerate(self._parts):
1183 if not part.autoclose and not part.consumed:
1184 try:
1185 await part.close()
1186 except Exception as exc:
1187 internal_logger.error(
1188 "Failed to close multipart part %d: %s", idx, exc, exc_info=True
1189 )
1192class MultipartPayloadWriter:
1193 def __init__(self, writer: AbstractStreamWriter) -> None:
1194 self._writer = writer
1195 self._encoding: str | None = None
1196 self._compress: ZLibCompressor | None = None
1197 self._encoding_buffer: bytearray | None = None
1199 def enable_encoding(self, encoding: str) -> None:
1200 if encoding == "base64":
1201 self._encoding = encoding
1202 self._encoding_buffer = bytearray()
1203 elif encoding == "quoted-printable":
1204 self._encoding = "quoted-printable"
1206 def enable_compression(
1207 self, encoding: str = "deflate", strategy: int | None = None
1208 ) -> None:
1209 self._compress = ZLibCompressor(
1210 encoding=encoding,
1211 suppress_deflate_header=True,
1212 strategy=strategy,
1213 )
1215 async def write_eof(self) -> None:
1216 if self._compress is not None:
1217 chunk = self._compress.flush()
1218 if chunk:
1219 self._compress = None
1220 await self.write(chunk)
1222 if self._encoding == "base64":
1223 if self._encoding_buffer:
1224 await self._writer.write(base64.b64encode(self._encoding_buffer))
1226 async def write(self, chunk: bytes) -> None:
1227 if self._compress is not None:
1228 if chunk:
1229 chunk = await self._compress.compress(chunk)
1230 if not chunk:
1231 return
1233 if self._encoding == "base64":
1234 buf = self._encoding_buffer
1235 assert buf is not None
1236 buf.extend(chunk)
1238 if buf:
1239 div, mod = divmod(len(buf), 3)
1240 enc_chunk, self._encoding_buffer = (buf[: div * 3], buf[div * 3 :])
1241 if enc_chunk:
1242 b64chunk = base64.b64encode(enc_chunk)
1243 await self._writer.write(b64chunk)
1244 elif self._encoding == "quoted-printable":
1245 await self._writer.write(binascii.b2a_qp(chunk))
1246 else:
1247 await self._writer.write(chunk)