Coverage for /pythoncovmergedfiles/medio/medio/src/aiohttp/aiohttp/http_parser.py: 80%
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, 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 HeadersDictProxy,
41 set_exception,
42)
43from .http_exceptions import (
44 BadHttpMessage,
45 BadHttpMethod,
46 BadStatusLine,
47 ContentEncodingError,
48 ContentLengthError,
49 InvalidHeader,
50 InvalidURLError,
51 LineTooLong,
52 TransferEncodingError,
53)
54from .http_writer import HttpVersion, HttpVersion10, HttpVersion11
55from .streams import EMPTY_PAYLOAD, StreamReader
56from .typedefs import RawHeaders
58if TYPE_CHECKING:
59 from .client_proto import ResponseHandler
61__all__ = (
62 "HeadersParser",
63 "HttpParser",
64 "HttpRequestParser",
65 "HttpResponseParser",
66 "RawRequestMessage",
67 "RawResponseMessage",
68)
70_T = TypeVar("_T")
72_SEP = Literal[b"\r\n", b"\n"]
74ASCIISET: Final[set[str]] = set(string.printable)
76# See https://www.rfc-editor.org/rfc/rfc9110.html#name-overview
77# and https://www.rfc-editor.org/rfc/rfc9110.html#name-tokens
78#
79# method = token
80# tchar = "!" / "#" / "$" / "%" / "&" / "'" / "*" / "+" / "-" / "." /
81# "^" / "_" / "`" / "|" / "~" / DIGIT / ALPHA
82# token = 1*tchar
83_TCHAR_SPECIALS: Final[str] = re.escape("!#$%&'*+-.^_`|~")
84TOKENRE: Final[Pattern[str]] = re.compile(f"[0-9A-Za-z{_TCHAR_SPECIALS}]+")
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)
89_TARGET_FORBIDDEN_CTL_RE: Final[Pattern[str]] = re.compile(r"[\x00-\x1f\x7f]")
90VERSRE: Final[Pattern[str]] = re.compile(r"HTTP/(\d)\.(\d)", re.ASCII)
91DIGITS: Final[Pattern[str]] = re.compile(r"\d+", re.ASCII)
92HEXDIGITS: Final[Pattern[bytes]] = re.compile(rb"[0-9a-fA-F]+")
94# RFC 9110 singleton headers — duplicates are rejected in strict mode.
95# In lax mode (response parser default), the check is skipped entirely
96# since real-world servers (e.g. Google APIs, Werkzeug) commonly send
97# duplicate headers like Content-Type or Server.
98# Lowercased for case-insensitive matching against wire names.
99SINGLETON_HEADERS: Final[frozenset[str]] = frozenset(
100 {
101 "content-length",
102 "content-location",
103 "content-range",
104 "content-type",
105 "etag",
106 "host",
107 "max-forwards",
108 "server",
109 "transfer-encoding",
110 "user-agent",
111 }
112)
115class RawRequestMessage(NamedTuple):
116 method: str
117 path: str
118 version: HttpVersion
119 headers: HeadersDictProxy
120 raw_headers: RawHeaders
121 should_close: bool
122 compression: str | None
123 upgrade: bool
124 chunked: bool
125 url: URL
128class RawResponseMessage(NamedTuple):
129 version: HttpVersion
130 code: int
131 reason: str
132 headers: HeadersDictProxy
133 raw_headers: RawHeaders
134 should_close: bool
135 compression: str | None
136 upgrade: bool
137 chunked: bool
140_MsgT = TypeVar("_MsgT", RawRequestMessage, RawResponseMessage)
143class PayloadState(IntEnum):
144 PAYLOAD_COMPLETE = 0
145 PAYLOAD_NEEDS_INPUT = 1
146 PAYLOAD_HAS_PENDING_INPUT = 2
149class ParseState(IntEnum):
150 PARSE_NONE = 0
151 PARSE_LENGTH = 1
152 PARSE_CHUNKED = 2
153 PARSE_UNTIL_EOF = 3
156class ChunkState(IntEnum):
157 PARSE_CHUNKED_SIZE = 0
158 PARSE_CHUNKED_CHUNK = 1
159 PARSE_CHUNKED_CHUNK_EOF = 2
160 PARSE_TRAILERS = 4
163class HeadersParser:
164 def __init__(self, max_field_size: int = 8190, lax: bool = False) -> None:
165 self.max_field_size = max_field_size
166 self._lax = lax
168 def parse_headers(self, lines: list[bytes]) -> tuple[HeadersDictProxy, RawHeaders]:
169 headers: CIMultiDict[str] = CIMultiDict()
170 # note: "raw" does not mean inclusion of OWS before/after the field value
171 raw_headers = []
173 lines_idx = 0
174 line = lines[lines_idx]
175 line_count = len(lines)
177 while line:
178 # Parse initial header name : value pair.
179 try:
180 bname, bvalue = line.split(b":", 1)
181 except ValueError:
182 raise InvalidHeader(line) from None
184 if len(bname) == 0:
185 raise InvalidHeader(bname)
187 # https://www.rfc-editor.org/rfc/rfc9112.html#section-5.1-2
188 if {bname[0], bname[-1]} & {32, 9}: # {" ", "\t"}
189 raise InvalidHeader(line)
191 bvalue = bvalue.lstrip(b" \t")
192 name = bname.decode("utf-8", "surrogateescape")
193 if not TOKENRE.fullmatch(name):
194 raise InvalidHeader(bname)
196 # next line
197 lines_idx += 1
198 line = lines[lines_idx]
200 # consume continuation lines
201 continuation = self._lax and line and line[0] in (32, 9) # (' ', '\t')
203 # Deprecated: https://www.rfc-editor.org/rfc/rfc9112.html#name-obsolete-line-folding
204 if continuation:
205 header_length = len(bvalue)
206 bvalue_lst = [bvalue]
207 while continuation:
208 header_length += len(line)
209 if header_length > self.max_field_size:
210 header_line = bname + b": " + b"".join(bvalue_lst)
211 raise LineTooLong(
212 header_line[:100] + b"...", self.max_field_size
213 )
214 bvalue_lst.append(line)
216 # next line
217 lines_idx += 1
218 if lines_idx < line_count:
219 line = lines[lines_idx]
220 if line:
221 continuation = line[0] in (32, 9) # (' ', '\t')
222 else:
223 line = b""
224 break
225 bvalue = b"".join(bvalue_lst)
227 bvalue = bvalue.strip(b" \t")
228 value = bvalue.decode("utf-8", "surrogateescape")
230 # https://www.rfc-editor.org/rfc/rfc9110.html#section-5.5-5
231 if self._lax:
232 if "\n" in value or "\r" in value or "\x00" in value:
233 raise InvalidHeader(bvalue)
234 elif _FIELD_VALUE_FORBIDDEN_CTL_RE.search(value):
235 raise InvalidHeader(bvalue)
237 if not self._lax and name in headers and name.lower() in SINGLETON_HEADERS:
238 raise BadHttpMessage(f"Duplicate '{name}' header found.")
239 headers.add(name, value)
240 raw_headers.append((bname, bvalue))
242 return (HeadersDictProxy(headers), tuple(raw_headers))
245def _is_supported_upgrade(headers: HeadersDictProxy) -> bool:
246 """Check if the upgrade header is supported."""
247 u = headers.get(hdrs.UPGRADE, "")
248 # .lower() can transform non-ascii characters.
249 return u.isascii() and u.lower() in {"tcp", "websocket"}
252class HttpParser(abc.ABC, Generic[_MsgT]):
253 lax: ClassVar[bool] = False
255 def __init__(
256 self,
257 protocol: BaseProtocol,
258 loop: asyncio.AbstractEventLoop,
259 limit: int,
260 max_line_size: int = 8190,
261 max_headers: int = 128,
262 max_field_size: int = 8190,
263 timer: BaseTimerContext | None = None,
264 code: int | None = None,
265 method: str | None = None,
266 payload_exception: type[BaseException] | None = None,
267 response_with_body: bool = True,
268 read_until_eof: bool = False,
269 auto_decompress: bool = True,
270 max_msg_queue_size: int = 0,
271 ) -> None:
272 self.protocol = protocol
273 self.loop = loop
274 self.max_line_size = max_line_size
275 self.max_field_size = max_field_size
276 self.max_headers = max_headers
277 self.timer = timer
278 self.code = code
279 self.method = method
280 self.payload_exception = payload_exception
281 self.response_with_body = response_with_body
282 self.read_until_eof = read_until_eof
284 self._lines: list[bytes] = []
285 self._tail = b""
286 self._upgraded = False
287 self._pending_upgrade = False
288 self._payload = None
289 self._payload_parser: HttpPayloadParser | None = None
290 self._payload_has_more_data = False
291 self._auto_decompress = auto_decompress
292 self._limit = limit
293 self._headers_parser = HeadersParser(max_field_size, self.lax)
294 # Stop emitting messages once this many are queued unconsumed (0 = off).
295 self._max_msg_queue_size = max_msg_queue_size
296 self._msg_in_flight = 0
298 @abc.abstractmethod
299 def parse_message(self, lines: list[bytes]) -> _MsgT: ...
301 @abc.abstractmethod
302 def _is_chunked_te(self, te: str) -> bool: ...
304 def pause_reading(self) -> None:
305 assert self._payload_parser is not None
306 self._payload_parser.pause_reading()
308 def message_consumed(self) -> None:
309 """Protocol drained a queued message; free a slot for parsing."""
310 if self._msg_in_flight > 0:
311 self._msg_in_flight -= 1
313 def feed_eof(self) -> _MsgT | None:
314 if self._payload_parser is not None:
315 self._payload_parser.feed_eof()
316 if self._payload_parser.done:
317 self._payload_parser = None
318 else:
319 # try to extract partial message
320 if self._tail:
321 self._lines.append(self._tail)
323 if self._lines:
324 if self._lines[-1] != "\r\n":
325 self._lines.append(b"")
326 with suppress(Exception):
327 return self.parse_message(self._lines)
328 return None
330 def feed_data(
331 self,
332 data: bytes,
333 SEP: _SEP = b"\r\n",
334 EMPTY: bytes = b"",
335 CONTENT_LENGTH: istr = hdrs.CONTENT_LENGTH,
336 METH_CONNECT: str = hdrs.METH_CONNECT,
337 SEC_WEBSOCKET_KEY1: istr = hdrs.SEC_WEBSOCKET_KEY1,
338 ) -> tuple[list[tuple[_MsgT, StreamReader]], bool, bytes]:
339 messages = []
341 if self._tail:
342 data, self._tail = self._tail + data, b""
344 data_len = len(data)
345 start_pos = 0
346 loop = self.loop
347 max_line_length = self.max_line_size
349 should_close = False
350 while start_pos < data_len or self._payload_has_more_data:
351 # read HTTP message (request/response line + headers), \r\n\r\n
352 # and split by lines
353 if self._payload_parser is None and not self._upgraded:
354 if (
355 self._max_msg_queue_size
356 and self._msg_in_flight >= self._max_msg_queue_size
357 ):
358 # Queue full: buffer the rest and stop. Safe pause point;
359 # any preceding body is consumed before the next request
360 # line. Resumes via feed_data(b"") when the queue drains.
361 self._tail = data[start_pos:]
362 # The remainder now lives in self._tail only. Don't return it.
363 data = EMPTY
364 break
365 pos = data.find(SEP, start_pos)
366 # consume \r\n
367 if pos == start_pos and not self._lines:
368 start_pos = pos + len(SEP)
369 continue
371 if pos >= start_pos:
372 if should_close:
373 raise BadHttpMessage("Data after `Connection: close`")
375 # line found
376 line = data[start_pos:pos]
377 if SEP == b"\n": # For lax response parsing
378 line = line.rstrip(b"\r")
379 if len(line) > max_line_length:
380 raise LineTooLong(line[:100] + b"...", max_line_length)
382 self._lines.append(line)
383 # After processing the status/request line, everything is a header.
384 max_line_length = self.max_field_size
386 if len(self._lines) > self.max_headers:
387 raise BadHttpMessage("Too many headers received")
389 start_pos = pos + len(SEP)
391 # \r\n\r\n found
392 if self._lines[-1] == EMPTY:
393 max_trailers = self.max_headers - len(self._lines)
394 try:
395 msg: _MsgT = self.parse_message(self._lines)
396 finally:
397 self._lines.clear()
399 def get_content_length() -> int | None:
400 # payload length
401 length_hdr = msg.headers.get(CONTENT_LENGTH)
402 if length_hdr is None:
403 return None
405 # Shouldn't allow +/- or other number formats.
406 # https://www.rfc-editor.org/rfc/rfc9110#section-8.6-2
407 # msg.headers is already stripped of leading/trailing wsp
408 if not DIGITS.fullmatch(length_hdr):
409 raise InvalidHeader(CONTENT_LENGTH)
411 return int(length_hdr)
413 length = get_content_length()
414 # do not support old websocket spec
415 if SEC_WEBSOCKET_KEY1 in msg.headers:
416 raise InvalidHeader(SEC_WEBSOCKET_KEY1)
418 upgraded = msg.upgrade and _is_supported_upgrade(msg.headers)
420 method = getattr(msg, "method", self.method)
421 # code is only present on responses
422 code = getattr(msg, "code", 0)
424 assert self.protocol is not None
425 # calculate payload
426 # https://www.rfc-editor.org/info/rfc9112/#name-message-body-length
427 # https://www.rfc-editor.org/info/rfc9110/#section-9.3.1-6
428 # EMPTY_BODY_METHODS should only apply to responses.
429 # self.method is None on request parser.
430 empty_body = code in EMPTY_BODY_STATUS_CODES or bool(
431 self.method and self.method in EMPTY_BODY_METHODS
432 )
433 if not empty_body and (
434 (length is not None and length > 0) or msg.chunked
435 ):
436 payload = StreamReader(
437 self.protocol,
438 timer=self.timer,
439 loop=loop,
440 limit=self._limit,
441 )
442 payload_parser = HttpPayloadParser(
443 payload,
444 length=length,
445 chunked=msg.chunked,
446 method=method,
447 compression=msg.compression,
448 code=self.code,
449 response_with_body=self.response_with_body,
450 auto_decompress=self._auto_decompress,
451 lax=self.lax,
452 headers_parser=self._headers_parser,
453 max_line_size=self.max_line_size,
454 max_field_size=self.max_field_size,
455 max_trailers=max_trailers,
456 limit=self._limit,
457 )
458 if not payload_parser.done:
459 self._payload_parser = payload_parser
460 # https://www.rfc-editor.org/info/rfc9110/#section-7.8-15
461 # Defer any requested upgrade until the
462 # complete request has been read.
463 self._pending_upgrade = upgraded
464 elif method == METH_CONNECT:
465 assert isinstance(msg, RawRequestMessage)
466 payload = StreamReader(
467 self.protocol,
468 timer=self.timer,
469 loop=loop,
470 limit=self._limit,
471 )
472 self._upgraded = True
473 self._payload_parser = HttpPayloadParser(
474 payload,
475 method=msg.method,
476 compression=msg.compression,
477 auto_decompress=self._auto_decompress,
478 lax=self.lax,
479 headers_parser=self._headers_parser,
480 max_line_size=self.max_line_size,
481 max_field_size=self.max_field_size,
482 max_trailers=max_trailers,
483 limit=self._limit,
484 )
485 elif not empty_body and length is None and self.read_until_eof:
486 payload = StreamReader(
487 self.protocol,
488 timer=self.timer,
489 loop=loop,
490 limit=self._limit,
491 )
492 payload_parser = HttpPayloadParser(
493 payload,
494 length=length,
495 chunked=msg.chunked,
496 method=method,
497 compression=msg.compression,
498 code=self.code,
499 response_with_body=self.response_with_body,
500 auto_decompress=self._auto_decompress,
501 lax=self.lax,
502 headers_parser=self._headers_parser,
503 max_line_size=self.max_line_size,
504 max_field_size=self.max_field_size,
505 max_trailers=max_trailers,
506 limit=self._limit,
507 )
508 if not payload_parser.done:
509 self._payload_parser = payload_parser
510 elif upgraded:
511 # No body to read, so the connection switches to
512 # the upgraded protocol immediately.
513 self._upgraded = True
514 payload = EMPTY_PAYLOAD
515 else:
516 payload = EMPTY_PAYLOAD
518 messages.append((msg, payload))
519 if self._max_msg_queue_size:
520 self._msg_in_flight += 1
521 should_close = msg.should_close
522 else:
523 self._tail = data[start_pos:]
524 # A bare LF here means CRLF was required:
525 # reject instead of buffering, else a following request's
526 # bytes get appended to this line and leak in the error.
527 if b"\n" in self._tail:
528 raise BadHttpMessage("Bad line ending, expected CRLF")
529 if len(self._tail) > self.max_line_size:
530 raise LineTooLong(self._tail[:100] + b"...", self.max_line_size)
531 data = EMPTY
532 break
534 # no parser, just store
535 elif self._payload_parser is None and self._upgraded:
536 assert not self._lines
537 break
539 # feed payload
540 else:
541 assert not self._lines
542 assert self._payload_parser is not None
543 try:
544 payload_state, data = self._payload_parser.feed_data(
545 data[start_pos:], SEP
546 )
547 except Exception as underlying_exc:
548 reraised_exc: BaseException = underlying_exc
549 if self.payload_exception is not None:
550 reraised_exc = self.payload_exception(str(underlying_exc))
552 set_exception(
553 self._payload_parser.payload,
554 reraised_exc,
555 underlying_exc,
556 )
558 payload_state = PayloadState.PAYLOAD_COMPLETE
559 data = b""
560 if isinstance(
561 underlying_exc, (InvalidHeader, TransferEncodingError)
562 ):
563 raise
565 self._payload_has_more_data = (
566 payload_state == PayloadState.PAYLOAD_HAS_PENDING_INPUT
567 )
569 if payload_state is not PayloadState.PAYLOAD_COMPLETE:
570 # We've either consumed all available data, or we're pausing
571 # until the reader buffer is freed up.
572 break
574 start_pos = 0
575 data_len = len(data)
576 self._payload_parser = None
577 if self._pending_upgrade:
578 # Body fully read: the deferred upgrade takes effect and
579 # the rest of the connection is the upgraded protocol.
580 self._upgraded = True
581 self._pending_upgrade = False
583 if data and start_pos < data_len:
584 data = data[start_pos:]
585 else:
586 data = EMPTY
588 return messages, self._upgraded, data
590 def parse_headers(
591 self, lines: list[bytes]
592 ) -> tuple[HeadersDictProxy, RawHeaders, bool | None, str | None, bool, bool]:
593 """Parses RFC 5322 headers from a stream.
595 Line continuations are supported. Returns list of header name
596 and value pairs. Header name is in upper case.
597 """
598 headers, raw_headers = self._headers_parser.parse_headers(lines)
599 close_conn = None
600 encoding = None
601 upgrade = False
602 chunked = False
604 # keep-alive and protocol switching
605 # RFC 9110 section 7.6.1 defines Connection as a comma-separated list.
606 # We use a simple comma split here rather than getall() for performance,
607 # as the target tokens (close, keep-alive, upgrade) are simple ASCII
608 # values that never contain commas.
609 conn_values = headers.get(hdrs.CONNECTION)
610 if conn_values:
611 conn_tokens = {
612 token.lower()
613 for token in (part.strip(" \t") for part in conn_values.split(","))
614 if token and token.isascii()
615 }
617 if "close" in conn_tokens:
618 close_conn = True
619 elif "keep-alive" in conn_tokens:
620 close_conn = False
622 # https://www.rfc-editor.org/rfc/rfc9110.html#name-101-switching-protocols
623 if "upgrade" in conn_tokens and headers.get(hdrs.UPGRADE):
624 upgrade = True
626 # encoding
627 enc = headers.get(hdrs.CONTENT_ENCODING, "")
628 if enc.isascii() and enc.lower() in {"gzip", "deflate", "br", "zstd"}:
629 encoding = enc
631 # chunking
632 te = headers.get(hdrs.TRANSFER_ENCODING)
633 if te is not None:
634 if self._is_chunked_te(te):
635 chunked = True
637 if hdrs.CONTENT_LENGTH in headers:
638 raise BadHttpMessage(
639 "Transfer-Encoding can't be present with Content-Length",
640 )
642 return (headers, raw_headers, close_conn, encoding, upgrade, chunked)
644 def set_upgraded(self, val: bool) -> None:
645 """Set connection upgraded (to websocket) mode.
647 :param bool val: new state.
648 """
649 self._upgraded = val
652class HttpRequestParser(HttpParser[RawRequestMessage]):
653 """Read request status line.
655 Exception .http_exceptions.BadStatusLine
656 could be raised in case of any errors in status line.
657 Returns RawRequestMessage.
658 """
660 def parse_message(self, lines: list[bytes]) -> RawRequestMessage:
661 # request line
662 line = lines[0].decode("utf-8", "surrogateescape")
663 try:
664 method, path, version = line.split(" ", maxsplit=2)
665 except ValueError:
666 raise BadHttpMethod(line) from None
668 # method
669 if not TOKENRE.fullmatch(method):
670 raise BadHttpMethod(method)
671 method = method.upper()
673 # https://www.rfc-editor.org/rfc/rfc9112#section-3.2-4
674 if _TARGET_FORBIDDEN_CTL_RE.search(path):
675 raise InvalidURLError(
676 path.encode(errors="surrogateescape").decode("latin1")
677 )
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 not url.absolute:
713 # authority-form is only allowed with CONNECT
714 # https://www.rfc-editor.org/info/rfc9112/#section-3.2.3-1
715 raise InvalidURLError(
716 path.encode(errors="surrogateescape").decode("latin1")
717 )
719 # read headers
720 (
721 headers,
722 raw_headers,
723 close,
724 compression,
725 upgrade,
726 chunked,
727 ) = self.parse_headers(lines[1:])
729 if version_o == HttpVersion11 and hdrs.HOST not in headers:
730 raise BadHttpMessage("Missing 'Host' header in request.")
732 if close is None: # then the headers weren't set in the request
733 if version_o <= HttpVersion10: # HTTP 1.0 must asks to not close
734 close = True
735 else: # HTTP 1.1 must ask to close.
736 close = False
738 return RawRequestMessage(
739 method,
740 path,
741 version_o,
742 headers,
743 raw_headers,
744 close,
745 compression,
746 upgrade,
747 chunked,
748 url,
749 )
751 def _is_chunked_te(self, te: str) -> bool:
752 # https://www.rfc-editor.org/rfc/rfc9112#section-7.1-3
753 # "A sender MUST NOT apply the chunked transfer coding more
754 # than once to a message body"
755 parts = [p.strip(" \t") for p in te.split(",")]
756 chunked_count = sum(1 for p in parts if p.isascii() and p.lower() == "chunked")
757 if chunked_count > 1:
758 raise BadHttpMessage("Request has duplicate `chunked` Transfer-Encoding")
759 last = parts[-1]
760 # .lower() transforms some non-ascii chars, so must check first.
761 if last.isascii() and last.lower() == "chunked":
762 return True
763 # https://www.rfc-editor.org/rfc/rfc9112#section-6.3-2.4.3
764 raise BadHttpMessage("Request has invalid `Transfer-Encoding`")
767class HttpResponseParser(HttpParser[RawResponseMessage]):
768 """Read response status line and headers.
770 BadStatusLine could be raised in case of any errors in status line.
771 Returns RawResponseMessage.
772 """
774 protocol: "ResponseHandler"
776 # Lax mode should only be enabled on response parser.
777 lax = not DEBUG
779 def feed_data(
780 self,
781 data: bytes,
782 SEP: _SEP | None = None,
783 *args: Any,
784 **kwargs: Any,
785 ) -> tuple[list[tuple[RawResponseMessage, StreamReader]], bool, bytes]:
786 if SEP is None:
787 SEP = b"\r\n" if DEBUG else b"\n"
788 return super().feed_data(data, SEP, *args, **kwargs)
790 def parse_message(self, lines: list[bytes]) -> RawResponseMessage:
791 line = lines[0].decode("utf-8", "surrogateescape")
792 try:
793 version, status = line.split(maxsplit=1)
794 except ValueError:
795 raise BadStatusLine(line) from None
797 try:
798 status, reason = status.split(maxsplit=1)
799 except ValueError:
800 status = status.strip()
801 reason = ""
803 # version
804 match = VERSRE.fullmatch(version)
805 if match is None:
806 raise BadStatusLine(line)
807 version_o = HttpVersion(int(match.group(1)), int(match.group(2)))
809 # The status code is a three-digit ASCII number, no padding
810 if len(status) != 3 or not DIGITS.fullmatch(status):
811 raise BadStatusLine(line)
812 status_i = int(status)
814 # read headers
815 (
816 headers,
817 raw_headers,
818 close,
819 compression,
820 upgrade,
821 chunked,
822 ) = self.parse_headers(lines[1:])
824 if close is None:
825 if version_o <= HttpVersion10:
826 close = True
827 # https://www.rfc-editor.org/rfc/rfc9112.html#name-message-body-length
828 elif 100 <= status_i < 200 or status_i in {204, 304}:
829 close = False
830 elif hdrs.CONTENT_LENGTH in headers or hdrs.TRANSFER_ENCODING in headers:
831 close = False
832 else:
833 # https://www.rfc-editor.org/rfc/rfc9112.html#section-6.3-2.8
834 close = True
836 return RawResponseMessage(
837 version_o,
838 status_i,
839 reason.strip(),
840 headers,
841 raw_headers,
842 close,
843 compression,
844 upgrade,
845 chunked,
846 )
848 def _is_chunked_te(self, te: str) -> bool:
849 # https://www.rfc-editor.org/rfc/rfc9112#section-6.3-2.4.2
850 return te.rsplit(",", maxsplit=1)[-1].strip(" \t").lower() == "chunked"
853class HttpPayloadParser:
854 def __init__(
855 self,
856 payload: StreamReader,
857 length: int | None = None,
858 chunked: bool = False,
859 compression: str | None = None,
860 code: int | None = None,
861 method: str | None = None,
862 response_with_body: bool = True,
863 auto_decompress: bool = True,
864 lax: bool = False,
865 *,
866 headers_parser: HeadersParser,
867 max_line_size: int = 8190,
868 max_field_size: int = 8190,
869 max_trailers: int = 128,
870 limit: int = DEFAULT_CHUNK_SIZE,
871 ) -> None:
872 self._length = 0
873 self._paused = False
874 self._type = ParseState.PARSE_UNTIL_EOF
875 self._chunk = ChunkState.PARSE_CHUNKED_SIZE
876 self._chunk_size = 0
877 self._chunk_tail = b""
878 self._auto_decompress = auto_decompress
879 self._lax = lax
880 self._headers_parser = headers_parser
881 self._max_line_size = max_line_size
882 self._max_field_size = max_field_size
883 self._max_trailers = max_trailers
884 self._more_data_available = False
885 self._trailer_lines: list[bytes] = []
886 self.done = False
887 self._eof_pending = False
889 # payload decompression wrapper
890 if response_with_body and compression and self._auto_decompress:
891 real_payload: StreamReader | DeflateBuffer = DeflateBuffer(
892 payload, compression, max_decompress_size=limit
893 )
894 else:
895 real_payload = payload
897 # payload parser
898 if not response_with_body:
899 # don't parse payload if it's not expected to be received
900 self._type = ParseState.PARSE_NONE
901 real_payload.feed_eof()
902 self.done = True
903 elif chunked:
904 self._type = ParseState.PARSE_CHUNKED
905 elif length is not None:
906 self._type = ParseState.PARSE_LENGTH
907 self._length = length
908 self._length_expected = length
909 if self._length == 0:
910 real_payload.feed_eof()
911 self.done = True
913 self.payload = real_payload
915 def pause_reading(self) -> None:
916 self._paused = True
918 def feed_eof(self) -> None:
919 if self._type == ParseState.PARSE_UNTIL_EOF:
920 self._eof_pending = True
921 while self._more_data_available:
922 if self._paused:
923 self._paused = False
924 return # Will resume via feed_data(b"") later
925 self._more_data_available = self.payload.feed_data(b"")
926 self.payload.feed_eof()
927 self.done = True
928 self._eof_pending = False
929 elif self._type == ParseState.PARSE_LENGTH:
930 if self._length:
931 received = self._length_expected - self._length
932 raise ContentLengthError(
933 f"Not enough data to satisfy content length header "
934 f"(received {received} of {self._length_expected} bytes)."
935 )
936 # Body has already been received, but parser paused.
937 while self._more_data_available:
938 if self._paused:
939 self._paused = False
940 return # Will resume via feed_data(b"") later
941 self._more_data_available = self.payload.feed_data(b"")
942 self.payload.feed_eof()
943 self.done = True
944 elif self._type == ParseState.PARSE_CHUNKED:
945 raise TransferEncodingError(
946 "Not enough data to satisfy transfer length header."
947 )
949 def feed_data(
950 self, chunk: bytes, SEP: _SEP = b"\r\n", CHUNK_EXT: bytes = b";"
951 ) -> tuple[PayloadState, bytes]:
952 """Receive a chunk of data to process.
954 Return:
955 PayloadState - The current state of payload processing.
956 This function may be called with empty bytes after returning
957 PAYLOAD_HAS_PENDING_INPUT to continue processing after a pause.
958 bytes - If payload is complete, this is the unconsumed bytes intended for the
959 next message/payload, b"" otherwise.
960 """
961 # Read specified amount of bytes
962 if self._type == ParseState.PARSE_LENGTH:
963 if self._chunk_tail:
964 chunk = self._chunk_tail + chunk
965 self._chunk_tail = b""
967 required = self._length
968 self._length = max(required - len(chunk), 0)
969 self._more_data_available = self.payload.feed_data(chunk[:required])
970 while self._more_data_available:
971 if self._paused:
972 self._paused = False
973 self._chunk_tail = chunk[required:]
974 return PayloadState.PAYLOAD_HAS_PENDING_INPUT, b""
975 self._more_data_available = self.payload.feed_data(b"")
977 if self._length == 0:
978 self.payload.feed_eof()
979 return PayloadState.PAYLOAD_COMPLETE, chunk[required:]
980 # Chunked transfer encoding parser
981 elif self._type == ParseState.PARSE_CHUNKED:
982 if self._chunk_tail:
983 # We should check the length is sane when not processing payload body.
984 if self._chunk != ChunkState.PARSE_CHUNKED_CHUNK:
985 max_line_length = self._max_line_size
986 if self._chunk == ChunkState.PARSE_TRAILERS:
987 max_line_length = self._max_field_size
988 if len(self._chunk_tail) > max_line_length:
989 raise LineTooLong(
990 self._chunk_tail[:100] + b"...", max_line_length
991 )
993 chunk = self._chunk_tail + chunk
994 self._chunk_tail = b""
996 while chunk or self._more_data_available:
997 # read next chunk size
998 if self._chunk == ChunkState.PARSE_CHUNKED_SIZE:
999 pos = chunk.find(SEP)
1000 if pos >= 0:
1001 # Only chunk-size lines reach here; trailers enforce
1002 # _max_field_size separately in PARSE_TRAILERS below.
1003 if pos > self._max_line_size:
1004 raise LineTooLong(chunk[:100] + b"...", self._max_line_size)
1005 i = chunk.find(CHUNK_EXT, 0, pos)
1006 if i >= 0:
1007 size_b = chunk[:i] # strip chunk-extensions
1008 # Verify no LF in the chunk-extension
1009 if b"\n" in (ext := chunk[i:pos]):
1010 exc = TransferEncodingError(
1011 f"Unexpected LF in chunk-extension: {ext!r}"
1012 )
1013 set_exception(self.payload, exc)
1014 raise exc
1015 else:
1016 size_b = chunk[:pos]
1018 if self._lax: # Allow whitespace in lax mode.
1019 size_b = size_b.strip()
1021 if not re.fullmatch(HEXDIGITS, size_b):
1022 exc = TransferEncodingError(
1023 chunk[:pos].decode("ascii", "surrogateescape")
1024 )
1025 set_exception(self.payload, exc)
1026 raise exc
1027 size = int(bytes(size_b), 16)
1029 chunk = chunk[pos + len(SEP) :]
1030 if size == 0: # eof marker
1031 self._chunk = ChunkState.PARSE_TRAILERS
1032 if self._lax and chunk.startswith(b"\r"):
1033 chunk = chunk[1:]
1034 else:
1035 self._chunk = ChunkState.PARSE_CHUNKED_CHUNK
1036 self._chunk_size = size
1037 self.payload.begin_http_chunk_receiving()
1038 else:
1039 if b"\n" in chunk:
1040 exc = TransferEncodingError(
1041 "Bad chunk-size line ending, expected CRLF"
1042 )
1043 set_exception(self.payload, exc)
1044 raise exc
1045 self._chunk_tail = chunk
1046 return PayloadState.PAYLOAD_NEEDS_INPUT, b""
1048 # read chunk and feed buffer
1049 if self._chunk == ChunkState.PARSE_CHUNKED_CHUNK:
1050 if self._paused:
1051 self._paused = False
1052 self._chunk_tail = chunk
1053 return PayloadState.PAYLOAD_HAS_PENDING_INPUT, b""
1055 required = self._chunk_size
1056 self._chunk_size = max(required - len(chunk), 0)
1057 self._more_data_available = self.payload.feed_data(chunk[:required])
1058 chunk = chunk[required:]
1060 if self._more_data_available:
1061 continue
1063 if self._chunk_size:
1064 self._paused = False
1065 return PayloadState.PAYLOAD_NEEDS_INPUT, b""
1066 self._chunk = ChunkState.PARSE_CHUNKED_CHUNK_EOF
1067 self.payload.end_http_chunk_receiving()
1069 # toss the CRLF at the end of the chunk
1070 if self._chunk == ChunkState.PARSE_CHUNKED_CHUNK_EOF:
1071 if self._lax and chunk.startswith(b"\r"):
1072 chunk = chunk[1:]
1073 if chunk[: len(SEP)] == SEP:
1074 chunk = chunk[len(SEP) :]
1075 self._chunk = ChunkState.PARSE_CHUNKED_SIZE
1076 elif len(chunk) >= len(SEP) or chunk != SEP[: len(chunk)]:
1077 exc = TransferEncodingError(
1078 "Chunk size mismatch: expected CRLF after chunk data"
1079 )
1080 set_exception(self.payload, exc)
1081 raise exc
1082 else:
1083 self._chunk_tail = chunk
1084 return PayloadState.PAYLOAD_NEEDS_INPUT, b""
1086 if self._chunk == ChunkState.PARSE_TRAILERS:
1087 pos = chunk.find(SEP)
1088 if pos < 0: # No line found
1089 if b"\n" in chunk:
1090 exc = TransferEncodingError(
1091 "Bad trailer line ending, expected CRLF"
1092 )
1093 set_exception(self.payload, exc)
1094 raise exc
1095 self._chunk_tail = chunk
1096 return PayloadState.PAYLOAD_NEEDS_INPUT, b""
1098 line = chunk[:pos]
1099 chunk = chunk[pos + len(SEP) :]
1100 if SEP == b"\n": # For lax response parsing
1101 line = line.rstrip(b"\r")
1103 if len(line) > self._max_field_size:
1104 raise LineTooLong(line[:100] + b"...", self._max_field_size)
1106 self._trailer_lines.append(line)
1108 if len(self._trailer_lines) > self._max_trailers:
1109 raise BadHttpMessage("Too many trailers received")
1111 # \r\n\r\n found, end of stream
1112 if self._trailer_lines[-1] == b"":
1113 # Headers and trailers are defined the same way,
1114 # so we reuse the HeadersParser here.
1115 try:
1116 trailers, raw_trailers = self._headers_parser.parse_headers(
1117 self._trailer_lines
1118 )
1119 finally:
1120 self._trailer_lines.clear()
1121 self.payload.feed_eof()
1122 return PayloadState.PAYLOAD_COMPLETE, chunk
1124 # Read all bytes until eof
1125 elif self._type == ParseState.PARSE_UNTIL_EOF:
1126 self._more_data_available = self.payload.feed_data(chunk)
1127 while self._more_data_available:
1128 if self._paused:
1129 self._paused = False
1130 return PayloadState.PAYLOAD_HAS_PENDING_INPUT, b""
1131 self._more_data_available = self.payload.feed_data(b"")
1133 if self._eof_pending:
1134 self.payload.feed_eof()
1135 self.done = True
1136 self._eof_pending = False
1137 return PayloadState.PAYLOAD_COMPLETE, b""
1139 return PayloadState.PAYLOAD_NEEDS_INPUT, b""
1142class DeflateBuffer:
1143 """DeflateStream decompress stream and feed data into specified stream."""
1145 def __init__(
1146 self,
1147 out: StreamReader,
1148 encoding: str | None,
1149 max_decompress_size: int = DEFAULT_CHUNK_SIZE,
1150 ) -> None:
1151 self.out = out
1152 self.size = 0
1153 out.total_compressed_bytes = self.size
1154 self.encoding = encoding
1155 self._started_decoding = False
1157 self.decompressor: BrotliDecompressor | ZLibDecompressor | ZSTDDecompressor
1158 if encoding == "br":
1159 if not HAS_BROTLI:
1160 raise ContentEncodingError(
1161 "Can not decode content-encoding: brotli (br). "
1162 "Please install `Brotli`"
1163 )
1164 self.decompressor = BrotliDecompressor()
1165 elif encoding == "zstd":
1166 if not HAS_ZSTD:
1167 raise ContentEncodingError(
1168 "Can not decode content-encoding: zstandard (zstd). "
1169 "Please install `backports.zstd`"
1170 )
1171 self.decompressor = ZSTDDecompressor()
1172 else:
1173 self.decompressor = ZLibDecompressor(encoding=encoding)
1175 self._max_decompress_size = max_decompress_size
1177 def set_exception(
1178 self,
1179 exc: type[BaseException] | BaseException,
1180 exc_cause: BaseException = _EXC_SENTINEL,
1181 ) -> None:
1182 set_exception(self.out, exc, exc_cause)
1184 def feed_data(self, chunk: bytes) -> bool:
1185 """Return True if more data is available and this method should be called again with b""."""
1186 self.size += len(chunk)
1187 self.out.total_compressed_bytes = self.size
1189 # Inspect the first real byte once to choose the decompressor. An empty
1190 # chunk (e.g. a chunk-size line arriving without body bytes) has no
1191 # header to sniff, so skip it and wait for the first data byte.
1192 if not self._started_decoding and chunk:
1193 # RFC1950
1194 # bits 0..3 = CM = 0b1000 = 8 = "deflate"
1195 # bits 4..7 = CINFO = 1..7 = windows size.
1196 if self.encoding == "deflate" and chunk[0] & 0xF != 8:
1197 # Change the decoder to decompress incorrectly compressed data
1198 # Actually we should issue a warning about non-RFC-compliant data.
1199 self.decompressor = ZLibDecompressor(
1200 encoding=self.encoding, suppress_deflate_header=True
1201 )
1202 self._started_decoding = True
1204 low_water = self.out._low_water
1205 max_length = (
1206 0 if low_water >= sys.maxsize else max(self._max_decompress_size, low_water)
1207 )
1208 try:
1209 chunk = self.decompressor.decompress_sync(chunk, max_length=max_length)
1210 except Exception:
1211 raise ContentEncodingError(
1212 "Can not decode content-encoding: %s" % self.encoding
1213 )
1215 if chunk:
1216 self.out.feed_data(chunk)
1217 return self.decompressor.data_available
1219 def feed_eof(self) -> None:
1220 chunk = self.decompressor.flush()
1221 # This should never contain data as we defer the call until exhausting
1222 # the decompression. If .flush() is returning data, this may indicate a
1223 # zip bomb vulnerability as it will decompress all remaining data at once.
1224 assert not chunk
1226 if self.size > 0:
1227 # decompressor is not brotli unless encoding is "br"
1228 if self.encoding == "deflate" and not self.decompressor.eof: # type: ignore[union-attr]
1229 raise ContentEncodingError("deflate")
1231 self.out.feed_eof()
1233 def begin_http_chunk_receiving(self) -> None:
1234 self.out.begin_http_chunk_receiving()
1236 def end_http_chunk_receiving(self) -> None:
1237 self.out.end_http_chunk_receiving()
1240HttpRequestParserPy = HttpRequestParser
1241HttpResponseParserPy = HttpResponseParser
1242RawRequestMessagePy = RawRequestMessage
1243RawResponseMessagePy = RawResponseMessage
1245with suppress(ImportError):
1246 if not NO_EXTENSIONS:
1247 from ._http_parser import ( # type: ignore[import-not-found,no-redef]
1248 HttpRequestParser,
1249 HttpResponseParser,
1250 RawRequestMessage,
1251 RawResponseMessage,
1252 )
1254 HttpRequestParserC = HttpRequestParser
1255 HttpResponseParserC = HttpResponseParser
1256 RawRequestMessageC = RawRequestMessage
1257 RawResponseMessageC = RawResponseMessage