Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/aiohttp/http_parser.py: 19%
Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1import abc
2import asyncio
3import re
4import string
5import sys
6from contextlib import suppress
7from enum import IntEnum
8from re import Pattern
9from typing import (
10 TYPE_CHECKING,
11 Any,
12 ClassVar,
13 Final,
14 Generic,
15 Literal,
16 NamedTuple,
17 TypeVar,
18)
20from multidict import CIMultiDict, CIMultiDictProxy, istr
21from yarl import URL
23from . import hdrs
24from .base_protocol import BaseProtocol
25from .compression_utils import (
26 HAS_BROTLI,
27 HAS_ZSTD,
28 BrotliDecompressor,
29 ZLibDecompressor,
30 ZSTDDecompressor,
31)
32from .helpers import (
33 _EXC_SENTINEL,
34 DEBUG,
35 DEFAULT_CHUNK_SIZE,
36 EMPTY_BODY_METHODS,
37 EMPTY_BODY_STATUS_CODES,
38 NO_EXTENSIONS,
39 BaseTimerContext,
40 set_exception,
41)
42from .http_exceptions import (
43 BadHttpMessage,
44 BadHttpMethod,
45 BadStatusLine,
46 ContentEncodingError,
47 ContentLengthError,
48 InvalidHeader,
49 InvalidURLError,
50 LineTooLong,
51 TransferEncodingError,
52)
53from .http_writer import HttpVersion, HttpVersion10, HttpVersion11
54from .streams import EMPTY_PAYLOAD, StreamReader
55from .typedefs import RawHeaders
57if TYPE_CHECKING:
58 from .client_proto import ResponseHandler
60__all__ = (
61 "HeadersParser",
62 "HttpParser",
63 "HttpRequestParser",
64 "HttpResponseParser",
65 "RawRequestMessage",
66 "RawResponseMessage",
67)
69_SEP = Literal[b"\r\n", b"\n"]
71ASCIISET: Final[set[str]] = set(string.printable)
73# See https://www.rfc-editor.org/rfc/rfc9110.html#name-overview
74# and https://www.rfc-editor.org/rfc/rfc9110.html#name-tokens
75#
76# method = token
77# tchar = "!" / "#" / "$" / "%" / "&" / "'" / "*" / "+" / "-" / "." /
78# "^" / "_" / "`" / "|" / "~" / DIGIT / ALPHA
79# token = 1*tchar
80_TCHAR_SPECIALS: Final[str] = re.escape("!#$%&'*+-.^_`|~")
81TOKENRE: Final[Pattern[str]] = re.compile(f"[0-9A-Za-z{_TCHAR_SPECIALS}]+")
82VERSRE: Final[Pattern[str]] = re.compile(r"HTTP/(\d)\.(\d)", re.ASCII)
83DIGITS: Final[Pattern[str]] = re.compile(r"\d+", re.ASCII)
84HEXDIGITS: Final[Pattern[bytes]] = re.compile(rb"[0-9a-fA-F]+")
85# https://www.rfc-editor.org/rfc/rfc9110#section-5.5-5
86_FIELD_VALUE_FORBIDDEN_CTL_RE: Final[Pattern[str]] = re.compile(
87 r"[\x00-\x08\x0a-\x1f\x7f]"
88)
90# RFC 9110 singleton headers — duplicates are rejected in strict mode.
91# In lax mode (response parser default), the check is skipped entirely
92# since real-world servers (e.g. Google APIs, Werkzeug) commonly send
93# duplicate headers like Content-Type or Server.
94# Lowercased for case-insensitive matching against wire names.
95SINGLETON_HEADERS: Final[frozenset[str]] = frozenset(
96 {
97 "content-length",
98 "content-location",
99 "content-range",
100 "content-type",
101 "etag",
102 "host",
103 "max-forwards",
104 "server",
105 "transfer-encoding",
106 "user-agent",
107 }
108)
111class RawRequestMessage(NamedTuple):
112 method: str
113 path: str
114 version: HttpVersion
115 headers: "CIMultiDictProxy[str]"
116 raw_headers: RawHeaders
117 should_close: bool
118 compression: str | None
119 upgrade: bool
120 chunked: bool
121 url: URL
124class RawResponseMessage(NamedTuple):
125 version: HttpVersion
126 code: int
127 reason: str
128 headers: CIMultiDictProxy[str]
129 raw_headers: RawHeaders
130 should_close: bool
131 compression: str | None
132 upgrade: bool
133 chunked: bool
136_MsgT = TypeVar("_MsgT", RawRequestMessage, RawResponseMessage)
139class PayloadState(IntEnum):
140 PAYLOAD_COMPLETE = 0
141 PAYLOAD_NEEDS_INPUT = 1
142 PAYLOAD_HAS_PENDING_INPUT = 2
145class ParseState(IntEnum):
147 PARSE_NONE = 0
148 PARSE_LENGTH = 1
149 PARSE_CHUNKED = 2
150 PARSE_UNTIL_EOF = 3
153class ChunkState(IntEnum):
154 PARSE_CHUNKED_SIZE = 0
155 PARSE_CHUNKED_CHUNK = 1
156 PARSE_CHUNKED_CHUNK_EOF = 2
157 PARSE_MAYBE_TRAILERS = 3
158 PARSE_TRAILERS = 4
161class HeadersParser:
162 def __init__(
163 self,
164 max_line_size: int = 8190,
165 max_headers: int = 32768,
166 max_field_size: int = 8190,
167 lax: bool = False,
168 ) -> None:
169 self.max_line_size = max_line_size
170 self.max_headers = max_headers
171 self.max_field_size = max_field_size
172 self._lax = lax
174 def parse_headers(
175 self, lines: list[bytes]
176 ) -> tuple["CIMultiDictProxy[str]", RawHeaders]:
177 headers: CIMultiDict[str] = CIMultiDict()
178 # note: "raw" does not mean inclusion of OWS before/after the field value
179 raw_headers = []
181 lines_idx = 0
182 line = lines[lines_idx]
183 line_count = len(lines)
185 while line:
186 # Parse initial header name : value pair.
187 try:
188 bname, bvalue = line.split(b":", 1)
189 except ValueError:
190 raise InvalidHeader(line) from None
192 if len(bname) == 0:
193 raise InvalidHeader(bname)
195 # https://www.rfc-editor.org/rfc/rfc9112.html#section-5.1-2
196 if {bname[0], bname[-1]} & {32, 9}: # {" ", "\t"}
197 raise InvalidHeader(line)
199 bvalue = bvalue.lstrip(b" \t")
200 name = bname.decode("utf-8", "surrogateescape")
201 if not TOKENRE.fullmatch(name):
202 raise InvalidHeader(bname)
204 # next line
205 lines_idx += 1
206 line = lines[lines_idx]
208 # consume continuation lines
209 continuation = self._lax and line and line[0] in (32, 9) # (' ', '\t')
211 # Deprecated: https://www.rfc-editor.org/rfc/rfc9112.html#name-obsolete-line-folding
212 if continuation:
213 header_length = len(bvalue)
214 bvalue_lst = [bvalue]
215 while continuation:
216 header_length += len(line)
217 if header_length > self.max_field_size:
218 header_line = bname + b": " + b"".join(bvalue_lst)
219 raise LineTooLong(
220 header_line[:100] + b"...", self.max_field_size
221 )
222 bvalue_lst.append(line)
224 # next line
225 lines_idx += 1
226 if lines_idx < line_count:
227 line = lines[lines_idx]
228 if line:
229 continuation = line[0] in (32, 9) # (' ', '\t')
230 else:
231 line = b""
232 break
233 bvalue = b"".join(bvalue_lst)
235 bvalue = bvalue.strip(b" \t")
236 value = bvalue.decode("utf-8", "surrogateescape")
238 # https://www.rfc-editor.org/rfc/rfc9110.html#section-5.5-5
239 if self._lax:
240 if "\n" in value or "\r" in value or "\x00" in value:
241 raise InvalidHeader(bvalue)
242 elif _FIELD_VALUE_FORBIDDEN_CTL_RE.search(value):
243 raise InvalidHeader(bvalue)
245 if not self._lax and name in headers and name.lower() in SINGLETON_HEADERS:
246 raise BadHttpMessage(f"Duplicate '{name}' header found.")
247 headers.add(name, value)
248 raw_headers.append((bname, bvalue))
250 return (CIMultiDictProxy(headers), tuple(raw_headers))
253def _is_supported_upgrade(headers: CIMultiDictProxy[str]) -> bool:
254 """Check if the upgrade header is supported."""
255 u = headers.get(hdrs.UPGRADE, "")
256 # .lower() can transform non-ascii characters.
257 return u.isascii() and u.lower() in {"tcp", "websocket"}
260class HttpParser(abc.ABC, Generic[_MsgT]):
261 lax: ClassVar[bool] = False
263 def __init__(
264 self,
265 protocol: BaseProtocol | None = None,
266 loop: asyncio.AbstractEventLoop | None = None,
267 limit: int = 2**16,
268 max_line_size: int = 8190,
269 max_headers: int = 128,
270 max_field_size: int = 8190,
271 timer: BaseTimerContext | None = None,
272 code: int | None = None,
273 method: str | None = None,
274 payload_exception: type[BaseException] | None = None,
275 response_with_body: bool = True,
276 read_until_eof: bool = False,
277 auto_decompress: bool = True,
278 max_msg_queue_size: int = 0,
279 ) -> None:
280 self.protocol = protocol
281 self.loop = loop
282 self.max_line_size = max_line_size
283 self.max_headers = max_headers
284 self.max_field_size = max_field_size
285 self.max_headers = max_headers
286 self.timer = timer
287 self.code = code
288 self.method = method
289 self.payload_exception = payload_exception
290 self.response_with_body = response_with_body
291 self.read_until_eof = read_until_eof
293 self._lines: list[bytes] = []
294 self._tail = b""
295 self._upgraded = False
296 self._pending_upgrade = False
297 self._payload = None
298 self._payload_parser: HttpPayloadParser | None = None
299 self._payload_has_more_data = False
300 self._auto_decompress = auto_decompress
301 self._limit = limit
302 self._headers_parser = HeadersParser(
303 max_line_size, max_headers, max_field_size, self.lax
304 )
305 # Stop emitting messages once this many are queued unconsumed (0 = off).
306 self._max_msg_queue_size = max_msg_queue_size
307 self._msg_in_flight = 0
309 @abc.abstractmethod
310 def parse_message(self, lines: list[bytes]) -> _MsgT: ...
312 @abc.abstractmethod
313 def _is_chunked_te(self, te: str) -> bool: ...
315 def pause_reading(self) -> None:
316 assert self._payload_parser is not None
317 self._payload_parser.pause_reading()
319 def message_consumed(self) -> None:
320 """Protocol drained a queued message; free a slot for parsing."""
321 if self._msg_in_flight > 0:
322 self._msg_in_flight -= 1
324 def feed_eof(self) -> _MsgT | None:
325 if self._payload_parser is not None:
326 self._payload_parser.feed_eof()
327 if self._payload_parser.done:
328 self._payload_parser = None
329 else:
330 # try to extract partial message
331 if self._tail:
332 self._lines.append(self._tail)
334 if self._lines:
335 if self._lines[-1] != "\r\n":
336 self._lines.append(b"")
337 with suppress(Exception):
338 return self.parse_message(self._lines)
339 return None
341 def feed_data(
342 self,
343 data: bytes,
344 SEP: _SEP = b"\r\n",
345 EMPTY: bytes = b"",
346 CONTENT_LENGTH: istr = hdrs.CONTENT_LENGTH,
347 METH_CONNECT: str = hdrs.METH_CONNECT,
348 SEC_WEBSOCKET_KEY1: istr = hdrs.SEC_WEBSOCKET_KEY1,
349 ) -> tuple[list[tuple[_MsgT, StreamReader]], bool, bytes]:
351 messages = []
353 if self._tail:
354 data, self._tail = self._tail + data, b""
356 data_len = len(data)
357 start_pos = 0
358 loop = self.loop
359 max_line_length = self.max_line_size
361 should_close = False
362 while start_pos < data_len or self._payload_has_more_data:
363 # read HTTP message (request/response line + headers), \r\n\r\n
364 # and split by lines
365 if self._payload_parser is None and not self._upgraded:
366 if (
367 self._max_msg_queue_size
368 and self._msg_in_flight >= self._max_msg_queue_size
369 ):
370 # Queue full: buffer the rest and stop. Safe pause point;
371 # any preceding body is consumed before the next request
372 # line. Resumes via feed_data(b"") when the queue drains.
373 self._tail = data[start_pos:]
374 break
375 pos = data.find(SEP, start_pos)
376 # consume \r\n
377 if pos == start_pos and not self._lines:
378 start_pos = pos + len(SEP)
379 continue
381 if pos >= start_pos:
382 if should_close:
383 raise BadHttpMessage("Data after `Connection: close`")
385 # line found
386 line = data[start_pos:pos]
387 if SEP == b"\n": # For lax response parsing
388 line = line.rstrip(b"\r")
389 if len(line) > max_line_length:
390 raise LineTooLong(line[:100] + b"...", max_line_length)
392 self._lines.append(line)
393 # After processing the status/request line, everything is a header.
394 max_line_length = self.max_field_size
396 if len(self._lines) > self.max_headers:
397 raise BadHttpMessage("Too many headers received")
399 start_pos = pos + len(SEP)
401 # \r\n\r\n found
402 if self._lines[-1] == EMPTY:
403 max_trailers = self.max_headers - len(self._lines)
404 try:
405 msg: _MsgT = self.parse_message(self._lines)
406 finally:
407 self._lines.clear()
409 def get_content_length() -> int | None:
410 # payload length
411 length_hdr = msg.headers.get(CONTENT_LENGTH)
412 if length_hdr is None:
413 return None
415 # Shouldn't allow +/- or other number formats.
416 # https://www.rfc-editor.org/rfc/rfc9110#section-8.6-2
417 # msg.headers is already stripped of leading/trailing wsp
418 if not DIGITS.fullmatch(length_hdr):
419 raise InvalidHeader(CONTENT_LENGTH)
421 return int(length_hdr)
423 length = get_content_length()
424 # do not support old websocket spec
425 if SEC_WEBSOCKET_KEY1 in msg.headers:
426 raise InvalidHeader(SEC_WEBSOCKET_KEY1)
428 upgraded = msg.upgrade and _is_supported_upgrade(msg.headers)
430 method = getattr(msg, "method", self.method)
431 # code is only present on responses
432 code = getattr(msg, "code", 0)
434 assert self.protocol is not None
435 # calculate payload
436 empty_body = code in EMPTY_BODY_STATUS_CODES or bool(
437 method and method in EMPTY_BODY_METHODS
438 )
439 if not empty_body and (
440 (length is not None and length > 0) or msg.chunked
441 ):
442 payload = StreamReader(
443 self.protocol,
444 timer=self.timer,
445 loop=loop,
446 limit=self._limit,
447 )
448 payload_parser = HttpPayloadParser(
449 payload,
450 length=length,
451 chunked=msg.chunked,
452 method=method,
453 compression=msg.compression,
454 code=self.code,
455 response_with_body=self.response_with_body,
456 auto_decompress=self._auto_decompress,
457 lax=self.lax,
458 headers_parser=self._headers_parser,
459 max_line_size=self.max_line_size,
460 max_field_size=self.max_field_size,
461 max_trailers=max_trailers,
462 limit=self._limit,
463 )
464 if not payload_parser.done:
465 self._payload_parser = payload_parser
466 # https://www.rfc-editor.org/info/rfc9110/#section-7.8-15
467 # Defer any requested upgrade until the
468 # complete request has been read.
469 self._pending_upgrade = upgraded
470 elif method == METH_CONNECT:
471 assert isinstance(msg, RawRequestMessage)
472 payload = StreamReader(
473 self.protocol,
474 timer=self.timer,
475 loop=loop,
476 limit=self._limit,
477 )
478 self._upgraded = True
479 self._payload_parser = HttpPayloadParser(
480 payload,
481 method=msg.method,
482 compression=msg.compression,
483 auto_decompress=self._auto_decompress,
484 lax=self.lax,
485 headers_parser=self._headers_parser,
486 max_line_size=self.max_line_size,
487 max_field_size=self.max_field_size,
488 max_trailers=max_trailers,
489 limit=self._limit,
490 )
491 elif not empty_body and length is None and self.read_until_eof:
492 payload = StreamReader(
493 self.protocol,
494 timer=self.timer,
495 loop=loop,
496 limit=self._limit,
497 )
498 payload_parser = HttpPayloadParser(
499 payload,
500 length=length,
501 chunked=msg.chunked,
502 method=method,
503 compression=msg.compression,
504 code=self.code,
505 response_with_body=self.response_with_body,
506 auto_decompress=self._auto_decompress,
507 lax=self.lax,
508 headers_parser=self._headers_parser,
509 max_line_size=self.max_line_size,
510 max_field_size=self.max_field_size,
511 max_trailers=max_trailers,
512 limit=self._limit,
513 )
514 if not payload_parser.done:
515 self._payload_parser = payload_parser
516 elif upgraded:
517 # No body to read, so the connection switches to
518 # the upgraded protocol immediately.
519 self._upgraded = True
520 payload = EMPTY_PAYLOAD
521 else:
522 payload = EMPTY_PAYLOAD
524 messages.append((msg, payload))
525 if self._max_msg_queue_size:
526 self._msg_in_flight += 1
527 should_close = msg.should_close
528 else:
529 self._tail = data[start_pos:]
530 # A bare LF here means CRLF was required:
531 # reject instead of buffering, else a following request's
532 # bytes get appended to this line and leak in the error.
533 if b"\n" in self._tail:
534 raise BadHttpMessage("Bad line ending, expected CRLF")
535 if len(self._tail) > self.max_line_size:
536 raise LineTooLong(self._tail[:100] + b"...", self.max_line_size)
537 data = EMPTY
538 break
540 # no parser, just store
541 elif self._payload_parser is None and self._upgraded:
542 assert not self._lines
543 break
545 # feed payload
546 else:
547 assert not self._lines
548 assert self._payload_parser is not None
549 try:
550 payload_state, data = self._payload_parser.feed_data(
551 data[start_pos:], SEP
552 )
553 except Exception as underlying_exc:
554 reraised_exc: BaseException = underlying_exc
555 if self.payload_exception is not None:
556 reraised_exc = self.payload_exception(str(underlying_exc))
558 set_exception(
559 self._payload_parser.payload,
560 reraised_exc,
561 underlying_exc,
562 )
564 payload_state = PayloadState.PAYLOAD_COMPLETE
565 data = b""
566 if isinstance(
567 underlying_exc, (InvalidHeader, TransferEncodingError)
568 ):
569 raise
571 self._payload_has_more_data = (
572 payload_state == PayloadState.PAYLOAD_HAS_PENDING_INPUT
573 )
575 if payload_state is not PayloadState.PAYLOAD_COMPLETE:
576 # We've either consumed all available data, or we're pausing
577 # until the reader buffer is freed up.
578 break
580 start_pos = 0
581 data_len = len(data)
582 self._payload_parser = None
583 if self._pending_upgrade:
584 # Body fully read: the deferred upgrade takes effect and
585 # the rest of the connection is the upgraded protocol.
586 self._upgraded = True
587 self._pending_upgrade = False
589 if data and start_pos < data_len:
590 data = data[start_pos:]
591 else:
592 data = EMPTY
594 return messages, self._upgraded, data
596 def parse_headers(
597 self, lines: list[bytes]
598 ) -> tuple[
599 "CIMultiDictProxy[str]", RawHeaders, bool | None, str | None, bool, bool
600 ]:
601 """Parses RFC 5322 headers from a stream.
603 Line continuations are supported. Returns list of header name
604 and value pairs. Header name is in upper case.
605 """
606 headers, raw_headers = self._headers_parser.parse_headers(lines)
607 close_conn = None
608 encoding = None
609 upgrade = False
610 chunked = False
612 # keep-alive and protocol switching
613 # RFC 9110 section 7.6.1 defines Connection as a comma-separated list.
614 conn_values = headers.getall(hdrs.CONNECTION, ())
615 if conn_values:
616 conn_tokens = {
617 token.lower()
618 for conn_value in conn_values
619 for token in (part.strip(" \t") for part in conn_value.split(","))
620 if token and token.isascii()
621 }
623 if "close" in conn_tokens:
624 close_conn = True
625 elif "keep-alive" in conn_tokens:
626 close_conn = False
628 # https://www.rfc-editor.org/rfc/rfc9110.html#name-101-switching-protocols
629 if "upgrade" in conn_tokens and headers.get(hdrs.UPGRADE):
630 upgrade = True
632 # encoding
633 enc = headers.get(hdrs.CONTENT_ENCODING, "")
634 if enc.isascii() and enc.lower() in {"gzip", "deflate", "br", "zstd"}:
635 encoding = enc
637 # chunking
638 te = headers.get(hdrs.TRANSFER_ENCODING)
639 if te is not None:
640 if self._is_chunked_te(te):
641 chunked = True
643 if hdrs.CONTENT_LENGTH in headers:
644 raise BadHttpMessage(
645 "Transfer-Encoding can't be present with Content-Length",
646 )
648 return (headers, raw_headers, close_conn, encoding, upgrade, chunked)
650 def set_upgraded(self, val: bool) -> None:
651 """Set connection upgraded (to websocket) mode.
653 :param bool val: new state.
654 """
655 self._upgraded = val
658class HttpRequestParser(HttpParser[RawRequestMessage]):
659 """Read request status line.
661 Exception .http_exceptions.BadStatusLine
662 could be raised in case of any errors in status line.
663 Returns RawRequestMessage.
664 """
666 def parse_message(self, lines: list[bytes]) -> RawRequestMessage:
667 # request line
668 line = lines[0].decode("utf-8", "surrogateescape")
669 try:
670 method, path, version = line.split(" ", maxsplit=2)
671 except ValueError:
672 raise BadHttpMethod(line) from None
674 # method
675 if not TOKENRE.fullmatch(method):
676 raise BadHttpMethod(method)
677 method = method.upper()
679 # version
680 match = VERSRE.fullmatch(version)
681 if match is None:
682 raise BadStatusLine(line)
683 version_o = HttpVersion(int(match.group(1)), int(match.group(2)))
685 if method == "CONNECT":
686 # authority-form,
687 # https://datatracker.ietf.org/doc/html/rfc7230#section-5.3.3
688 url = URL.build(authority=path, encoded=True)
689 elif path.startswith("/"):
690 # origin-form,
691 # https://datatracker.ietf.org/doc/html/rfc7230#section-5.3.1
692 path_part, _hash_separator, url_fragment = path.partition("#")
693 path_part, _question_mark_separator, qs_part = path_part.partition("?")
695 # NOTE: `yarl.URL.build()` is used to mimic what the Cython-based
696 # NOTE: parser does, otherwise it results into the same
697 # NOTE: HTTP Request-Line input producing different
698 # NOTE: `yarl.URL()` objects
699 url = URL.build(
700 path=path_part,
701 query_string=qs_part,
702 fragment=url_fragment,
703 encoded=True,
704 )
705 elif path == "*" and method == "OPTIONS":
706 # asterisk-form,
707 url = URL(path, encoded=True)
708 else:
709 # absolute-form for proxy maybe,
710 # https://datatracker.ietf.org/doc/html/rfc7230#section-5.3.2
711 url = URL(path, encoded=True)
712 if url.scheme == "":
713 # not absolute-form
714 raise InvalidURLError(
715 path.encode(errors="surrogateescape").decode("latin1")
716 )
718 # read headers
719 (
720 headers,
721 raw_headers,
722 close,
723 compression,
724 upgrade,
725 chunked,
726 ) = self.parse_headers(lines[1:])
728 if version_o == HttpVersion11 and hdrs.HOST not in headers:
729 raise BadHttpMessage("Missing 'Host' header in request.")
731 if close is None: # then the headers weren't set in the request
732 if version_o <= HttpVersion10: # HTTP 1.0 must asks to not close
733 close = True
734 else: # HTTP 1.1 must ask to close.
735 close = False
737 return RawRequestMessage(
738 method,
739 path,
740 version_o,
741 headers,
742 raw_headers,
743 close,
744 compression,
745 upgrade,
746 chunked,
747 url,
748 )
750 def _is_chunked_te(self, te: str) -> bool:
751 te = te.rsplit(",", maxsplit=1)[-1].strip(" \t")
752 # .lower() transforms some non-ascii chars, so must check first.
753 if te.isascii() and te.lower() == "chunked":
754 return True
755 # https://www.rfc-editor.org/rfc/rfc9112#section-6.3-2.4.3
756 raise BadHttpMessage("Request has invalid `Transfer-Encoding`")
759class HttpResponseParser(HttpParser[RawResponseMessage]):
760 """Read response status line and headers.
762 BadStatusLine could be raised in case of any errors in status line.
763 Returns RawResponseMessage.
764 """
766 protocol: "ResponseHandler"
768 # Lax mode should only be enabled on response parser.
769 lax = not DEBUG
771 def feed_data(
772 self,
773 data: bytes,
774 SEP: _SEP | None = None,
775 *args: Any,
776 **kwargs: Any,
777 ) -> tuple[list[tuple[RawResponseMessage, StreamReader]], bool, bytes]:
778 if SEP is None:
779 SEP = b"\r\n" if DEBUG else b"\n"
780 return super().feed_data(data, SEP, *args, **kwargs)
782 def parse_message(self, lines: list[bytes]) -> RawResponseMessage:
783 line = lines[0].decode("utf-8", "surrogateescape")
784 try:
785 version, status = line.split(maxsplit=1)
786 except ValueError:
787 raise BadStatusLine(line) from None
789 try:
790 status, reason = status.split(maxsplit=1)
791 except ValueError:
792 status = status.strip()
793 reason = ""
795 # version
796 match = VERSRE.fullmatch(version)
797 if match is None:
798 raise BadStatusLine(line)
799 version_o = HttpVersion(int(match.group(1)), int(match.group(2)))
801 # The status code is a three-digit ASCII number, no padding
802 if len(status) != 3 or not DIGITS.fullmatch(status):
803 raise BadStatusLine(line)
804 status_i = int(status)
806 # read headers
807 (
808 headers,
809 raw_headers,
810 close,
811 compression,
812 upgrade,
813 chunked,
814 ) = self.parse_headers(lines[1:])
816 if close is None:
817 if version_o <= HttpVersion10:
818 close = True
819 # https://www.rfc-editor.org/rfc/rfc9112.html#name-message-body-length
820 elif 100 <= status_i < 200 or status_i in {204, 304}:
821 close = False
822 elif hdrs.CONTENT_LENGTH in headers or hdrs.TRANSFER_ENCODING in headers:
823 close = False
824 else:
825 # https://www.rfc-editor.org/rfc/rfc9112.html#section-6.3-2.8
826 close = True
828 return RawResponseMessage(
829 version_o,
830 status_i,
831 reason.strip(),
832 headers,
833 raw_headers,
834 close,
835 compression,
836 upgrade,
837 chunked,
838 )
840 def _is_chunked_te(self, te: str) -> bool:
841 # https://www.rfc-editor.org/rfc/rfc9112#section-6.3-2.4.2
842 return te.rsplit(",", maxsplit=1)[-1].strip(" \t").lower() == "chunked"
845class HttpPayloadParser:
846 def __init__(
847 self,
848 payload: StreamReader,
849 length: int | None = None,
850 chunked: bool = False,
851 compression: str | None = None,
852 code: int | None = None,
853 method: str | None = None,
854 response_with_body: bool = True,
855 auto_decompress: bool = True,
856 lax: bool = False,
857 *,
858 headers_parser: HeadersParser,
859 max_line_size: int = 8190,
860 max_field_size: int = 8190,
861 max_trailers: int = 128,
862 limit: int = DEFAULT_CHUNK_SIZE,
863 ) -> None:
864 self._length = 0
865 self._paused = False
866 self._type = ParseState.PARSE_UNTIL_EOF
867 self._chunk = ChunkState.PARSE_CHUNKED_SIZE
868 self._chunk_size = 0
869 self._chunk_tail = b""
870 self._auto_decompress = auto_decompress
871 self._lax = lax
872 self._headers_parser = headers_parser
873 self._max_line_size = max_line_size
874 self._max_field_size = max_field_size
875 self._max_trailers = max_trailers
876 self._more_data_available = False
877 self._trailer_lines: list[bytes] = []
878 self.done = False
879 self._eof_pending = False
881 # payload decompression wrapper
882 if response_with_body and compression and self._auto_decompress:
883 real_payload: StreamReader | DeflateBuffer = DeflateBuffer(
884 payload, compression, max_decompress_size=limit
885 )
886 else:
887 real_payload = payload
889 # payload parser
890 if not response_with_body:
891 # don't parse payload if it's not expected to be received
892 self._type = ParseState.PARSE_NONE
893 real_payload.feed_eof()
894 self.done = True
895 elif chunked:
896 self._type = ParseState.PARSE_CHUNKED
897 elif length is not None:
898 self._type = ParseState.PARSE_LENGTH
899 self._length = length
900 self._length_expected = length
901 if self._length == 0:
902 real_payload.feed_eof()
903 self.done = True
905 self.payload = real_payload
907 def pause_reading(self) -> None:
908 self._paused = True
910 def feed_eof(self) -> None:
911 if self._type == ParseState.PARSE_UNTIL_EOF:
912 self._eof_pending = True
913 while self._more_data_available:
914 if self._paused:
915 self._paused = False
916 return # Will resume via feed_data(b"") later
917 self._more_data_available = self.payload.feed_data(b"", 0)
918 self.payload.feed_eof()
919 self.done = True
920 self._eof_pending = False
921 elif self._type == ParseState.PARSE_LENGTH:
922 received = self._length_expected - self._length
923 raise ContentLengthError(
924 f"Not enough data to satisfy content length header "
925 f"(received {received} of {self._length_expected} bytes)."
926 )
927 elif self._type == ParseState.PARSE_CHUNKED:
928 raise TransferEncodingError(
929 "Not enough data to satisfy transfer length header."
930 )
932 def feed_data(
933 self, chunk: bytes, SEP: _SEP = b"\r\n", CHUNK_EXT: bytes = b";"
934 ) -> tuple[PayloadState, bytes]:
935 """Receive a chunk of data to process.
937 Return:
938 PayloadState - The current state of payload processing.
939 This function may be called with empty bytes after returning
940 PAYLOAD_HAS_PENDING_INPUT to continue processing after a pause.
941 bytes - If payload is complete, this is the unconsumed bytes intended for the
942 next message/payload, b"" otherwise.
943 """
944 # Read specified amount of bytes
945 if self._type == ParseState.PARSE_LENGTH:
946 if self._chunk_tail:
947 chunk = self._chunk_tail + chunk
948 self._chunk_tail = b""
950 required = self._length
951 self._length = max(required - len(chunk), 0)
952 self._more_data_available = self.payload.feed_data(
953 chunk[:required], required
954 )
955 while self._more_data_available:
956 if self._paused:
957 self._paused = False
958 self._chunk_tail = chunk[required:]
959 return PayloadState.PAYLOAD_HAS_PENDING_INPUT, b""
960 self._more_data_available = self.payload.feed_data(b"", 0)
962 if self._length == 0:
963 self.payload.feed_eof()
964 return PayloadState.PAYLOAD_COMPLETE, chunk[required:]
965 # Chunked transfer encoding parser
966 elif self._type == ParseState.PARSE_CHUNKED:
967 if self._chunk_tail:
968 # We should check the length is sane when not processing payload body.
969 if self._chunk != ChunkState.PARSE_CHUNKED_CHUNK:
970 max_line_length = self._max_line_size
971 if self._chunk == ChunkState.PARSE_TRAILERS:
972 max_line_length = self._max_field_size
973 if len(self._chunk_tail) > max_line_length:
974 raise LineTooLong(
975 self._chunk_tail[:100] + b"...", max_line_length
976 )
978 chunk = self._chunk_tail + chunk
979 self._chunk_tail = b""
981 while chunk or self._more_data_available:
982 # read next chunk size
983 if self._chunk == ChunkState.PARSE_CHUNKED_SIZE:
984 pos = chunk.find(SEP)
985 if pos >= 0:
986 # Only chunk-size lines reach here; trailers enforce
987 # _max_field_size separately in PARSE_TRAILERS below.
988 if pos > self._max_line_size:
989 raise LineTooLong(chunk[:100] + b"...", self._max_line_size)
990 i = chunk.find(CHUNK_EXT, 0, pos)
991 if i >= 0:
992 size_b = chunk[:i] # strip chunk-extensions
993 # Verify no LF in the chunk-extension
994 if b"\n" in (ext := chunk[i:pos]):
995 exc = TransferEncodingError(
996 f"Unexpected LF in chunk-extension: {ext!r}"
997 )
998 set_exception(self.payload, exc)
999 raise exc
1000 else:
1001 size_b = chunk[:pos]
1003 if self._lax: # Allow whitespace in lax mode.
1004 size_b = size_b.strip()
1006 if not re.fullmatch(HEXDIGITS, size_b):
1007 exc = TransferEncodingError(
1008 chunk[:pos].decode("ascii", "surrogateescape")
1009 )
1010 set_exception(self.payload, exc)
1011 raise exc
1012 size = int(bytes(size_b), 16)
1014 chunk = chunk[pos + len(SEP) :]
1015 if size == 0: # eof marker
1016 self._chunk = ChunkState.PARSE_TRAILERS
1017 if self._lax and chunk.startswith(b"\r"):
1018 chunk = chunk[1:]
1019 else:
1020 self._chunk = ChunkState.PARSE_CHUNKED_CHUNK
1021 self._chunk_size = size
1022 self.payload.begin_http_chunk_receiving()
1023 else:
1024 if b"\n" in chunk:
1025 exc = TransferEncodingError(
1026 "Bad chunk-size line ending, expected CRLF"
1027 )
1028 set_exception(self.payload, exc)
1029 raise exc
1030 self._chunk_tail = chunk
1031 return PayloadState.PAYLOAD_NEEDS_INPUT, b""
1033 # read chunk and feed buffer
1034 if self._chunk == ChunkState.PARSE_CHUNKED_CHUNK:
1035 if self._paused:
1036 self._paused = False
1037 self._chunk_tail = chunk
1038 return PayloadState.PAYLOAD_HAS_PENDING_INPUT, b""
1040 required = self._chunk_size
1041 self._chunk_size = max(required - len(chunk), 0)
1042 self._more_data_available = self.payload.feed_data(
1043 chunk[:required], required
1044 )
1045 chunk = chunk[required:]
1047 if self._more_data_available:
1048 continue
1050 if self._chunk_size:
1051 self._paused = False
1052 return PayloadState.PAYLOAD_NEEDS_INPUT, b""
1053 self._chunk = ChunkState.PARSE_CHUNKED_CHUNK_EOF
1054 self.payload.end_http_chunk_receiving()
1056 # toss the CRLF at the end of the chunk
1057 if self._chunk == ChunkState.PARSE_CHUNKED_CHUNK_EOF:
1058 if self._lax and chunk.startswith(b"\r"):
1059 chunk = chunk[1:]
1060 if chunk[: len(SEP)] == SEP:
1061 chunk = chunk[len(SEP) :]
1062 self._chunk = ChunkState.PARSE_CHUNKED_SIZE
1063 elif len(chunk) >= len(SEP) or chunk != SEP[: len(chunk)]:
1064 exc = TransferEncodingError(
1065 "Chunk size mismatch: expected CRLF after chunk data"
1066 )
1067 set_exception(self.payload, exc)
1068 raise exc
1069 else:
1070 self._chunk_tail = chunk
1071 return PayloadState.PAYLOAD_NEEDS_INPUT, b""
1073 if self._chunk == ChunkState.PARSE_TRAILERS:
1074 pos = chunk.find(SEP)
1075 if pos < 0: # No line found
1076 if b"\n" in chunk:
1077 exc = TransferEncodingError(
1078 "Bad trailer line ending, expected CRLF"
1079 )
1080 set_exception(self.payload, exc)
1081 raise exc
1082 self._chunk_tail = chunk
1083 return PayloadState.PAYLOAD_NEEDS_INPUT, b""
1085 line = chunk[:pos]
1086 chunk = chunk[pos + len(SEP) :]
1087 if SEP == b"\n": # For lax response parsing
1088 line = line.rstrip(b"\r")
1090 if len(line) > self._max_field_size:
1091 raise LineTooLong(line[:100] + b"...", self._max_field_size)
1093 self._trailer_lines.append(line)
1095 if len(self._trailer_lines) > self._max_trailers:
1096 raise BadHttpMessage("Too many trailers received")
1098 # \r\n\r\n found, end of stream
1099 if self._trailer_lines[-1] == b"":
1100 # Headers and trailers are defined the same way,
1101 # so we reuse the HeadersParser here.
1102 try:
1103 trailers, raw_trailers = self._headers_parser.parse_headers(
1104 self._trailer_lines
1105 )
1106 finally:
1107 self._trailer_lines.clear()
1108 self.payload.feed_eof()
1109 return PayloadState.PAYLOAD_COMPLETE, chunk
1111 # Read all bytes until eof
1112 elif self._type == ParseState.PARSE_UNTIL_EOF:
1113 self._more_data_available = self.payload.feed_data(chunk, len(chunk))
1114 while self._more_data_available:
1115 if self._paused:
1116 self._paused = False
1117 return PayloadState.PAYLOAD_HAS_PENDING_INPUT, b""
1118 self._more_data_available = self.payload.feed_data(b"", 0)
1120 if self._eof_pending:
1121 self.payload.feed_eof()
1122 self.done = True
1123 self._eof_pending = False
1124 return PayloadState.PAYLOAD_COMPLETE, b""
1126 return PayloadState.PAYLOAD_NEEDS_INPUT, b""
1129class DeflateBuffer:
1130 """DeflateStream decompress stream and feed data into specified stream."""
1132 decompressor: Any
1134 def __init__(
1135 self,
1136 out: StreamReader,
1137 encoding: str | None,
1138 max_decompress_size: int = DEFAULT_CHUNK_SIZE,
1139 ) -> None:
1140 self.out = out
1141 self.size = 0
1142 out.total_compressed_bytes = self.size
1143 self.encoding = encoding
1144 self._started_decoding = False
1146 self.decompressor: BrotliDecompressor | ZLibDecompressor | ZSTDDecompressor
1147 if encoding == "br":
1148 if not HAS_BROTLI: # pragma: no cover
1149 raise ContentEncodingError(
1150 "Can not decode content-encoding: brotli (br). "
1151 "Please install `Brotli`"
1152 )
1153 self.decompressor = BrotliDecompressor()
1154 elif encoding == "zstd":
1155 if not HAS_ZSTD:
1156 raise ContentEncodingError(
1157 "Can not decode content-encoding: zstandard (zstd). "
1158 "Please install `backports.zstd`"
1159 )
1160 self.decompressor = ZSTDDecompressor()
1161 else:
1162 self.decompressor = ZLibDecompressor(encoding=encoding)
1164 self._max_decompress_size = max_decompress_size
1166 def set_exception(
1167 self,
1168 exc: BaseException,
1169 exc_cause: BaseException = _EXC_SENTINEL,
1170 ) -> None:
1171 set_exception(self.out, exc, exc_cause)
1173 def feed_data(self, chunk: bytes, size: int) -> bool:
1174 self.size += size
1175 self.out.total_compressed_bytes = self.size
1177 # Inspect the first real byte once to choose the decompressor. An empty
1178 # chunk (e.g. a chunk-size line arriving without body bytes) has no
1179 # header to sniff, so skip it and wait for the first data byte.
1180 if not self._started_decoding and chunk:
1181 # RFC1950
1182 # bits 0..3 = CM = 0b1000 = 8 = "deflate"
1183 # bits 4..7 = CINFO = 1..7 = windows size.
1184 if self.encoding == "deflate" and chunk[0] & 0xF != 8:
1185 # Change the decoder to decompress incorrectly compressed data
1186 # Actually we should issue a warning about non-RFC-compliant data.
1187 self.decompressor = ZLibDecompressor(
1188 encoding=self.encoding, suppress_deflate_header=True
1189 )
1190 self._started_decoding = True
1192 low_water = self.out._low_water
1193 max_length = (
1194 0 if low_water >= sys.maxsize else max(self._max_decompress_size, low_water)
1195 )
1196 try:
1197 chunk = self.decompressor.decompress_sync(chunk, max_length=max_length)
1198 except Exception:
1199 raise ContentEncodingError(
1200 "Can not decode content-encoding: %s" % self.encoding
1201 )
1203 if chunk:
1204 self.out.feed_data(chunk, len(chunk))
1205 return self.decompressor.data_available # type: ignore[no-any-return]
1207 def feed_eof(self) -> None:
1208 chunk = self.decompressor.flush()
1209 # This should never contain data as we defer the call until exhausting
1210 # the decompression. If .flush() is returning data, this may indicate a
1211 # zip bomb vulnerability as it will decompress all remaining data at once.
1212 assert not chunk
1214 if self.size > 0:
1215 if self.encoding == "deflate" and not self.decompressor.eof:
1216 raise ContentEncodingError("deflate")
1218 self.out.feed_eof()
1220 def begin_http_chunk_receiving(self) -> None:
1221 self.out.begin_http_chunk_receiving()
1223 def end_http_chunk_receiving(self) -> None:
1224 self.out.end_http_chunk_receiving()
1227HttpRequestParserPy = HttpRequestParser
1228HttpResponseParserPy = HttpResponseParser
1229RawRequestMessagePy = RawRequestMessage
1230RawResponseMessagePy = RawResponseMessage
1232try:
1233 if not NO_EXTENSIONS:
1234 from ._http_parser import ( # type: ignore[import-not-found,no-redef]
1235 HttpRequestParser,
1236 HttpResponseParser,
1237 RawRequestMessage,
1238 RawResponseMessage,
1239 )
1241 HttpRequestParserC = HttpRequestParser
1242 HttpResponseParserC = HttpResponseParser
1243 RawRequestMessageC = RawRequestMessage
1244 RawResponseMessageC = RawResponseMessage
1245except ImportError: # pragma: no cover
1246 pass