mitmproxy.http

   1import binascii
   2import json
   3import os
   4import time
   5import urllib.parse
   6import warnings
   7from collections.abc import Callable
   8from collections.abc import Iterable
   9from collections.abc import Iterator
  10from collections.abc import Mapping
  11from collections.abc import Sequence
  12from dataclasses import dataclass
  13from dataclasses import fields
  14from email.utils import formatdate
  15from email.utils import mktime_tz
  16from email.utils import parsedate_tz
  17from typing import Any
  18from typing import cast
  19
  20from mitmproxy import flow
  21from mitmproxy.coretypes import multidict
  22from mitmproxy.coretypes import serializable
  23from mitmproxy.net import encoding
  24from mitmproxy.net.http import cookies
  25from mitmproxy.net.http import multipart
  26from mitmproxy.net.http import status_codes
  27from mitmproxy.net.http import url
  28from mitmproxy.net.http.headers import assemble_content_type
  29from mitmproxy.net.http.headers import infer_content_encoding
  30from mitmproxy.net.http.headers import parse_content_type
  31from mitmproxy.utils import human
  32from mitmproxy.utils import strutils
  33from mitmproxy.utils import typecheck
  34from mitmproxy.utils.strutils import always_bytes
  35from mitmproxy.utils.strutils import always_str
  36from mitmproxy.websocket import WebSocketData
  37
  38
  39# While headers _should_ be ASCII, it's not uncommon for certain headers to be utf-8 encoded.
  40def _native(x: bytes) -> str:
  41    return x.decode("utf-8", "surrogateescape")
  42
  43
  44def _always_bytes(x: str | bytes) -> bytes:
  45    return strutils.always_bytes(x, "utf-8", "surrogateescape")
  46
  47
  48# This cannot be easily typed with mypy yet, so we just specify MultiDict without concrete types.
  49class Headers(multidict.MultiDict):  # type: ignore
  50    """
  51    Header class which allows both convenient access to individual headers as well as
  52    direct access to the underlying raw data. Provides a full dictionary interface.
  53
  54    Create headers with keyword arguments:
  55    >>> h = Headers(host="example.com", content_type="application/xml")
  56
  57    Headers mostly behave like a normal dict:
  58    >>> h["Host"]
  59    "example.com"
  60
  61    Headers are case insensitive:
  62    >>> h["host"]
  63    "example.com"
  64
  65    Headers can also be created from a list of raw (header_name, header_value) byte tuples:
  66    >>> h = Headers([
  67        (b"Host",b"example.com"),
  68        (b"Accept",b"text/html"),
  69        (b"accept",b"application/xml")
  70    ])
  71
  72    Multiple headers are folded into a single header as per RFC 7230:
  73    >>> h["Accept"]
  74    "text/html, application/xml"
  75
  76    Setting a header removes all existing headers with the same name:
  77    >>> h["Accept"] = "application/text"
  78    >>> h["Accept"]
  79    "application/text"
  80
  81    `bytes(h)` returns an HTTP/1 header block:
  82    >>> print(bytes(h))
  83    Host: example.com
  84    Accept: application/text
  85
  86    For full control, the raw header fields can be accessed:
  87    >>> h.fields
  88
  89    Caveats:
  90     - For use with the "Set-Cookie" and "Cookie" headers, either use `Response.cookies` or see `Headers.get_all`.
  91    """
  92
  93    def __init__(self, fields: Iterable[tuple[bytes, bytes]] = (), **headers):
  94        """
  95        *Args:*
  96         - *fields:* (optional) list of ``(name, value)`` header byte tuples,
  97           e.g. ``[(b"Host", b"example.com")]``. All names and values must be bytes.
  98         - *\\*\\*headers:* Additional headers to set. Will overwrite existing values from `fields`.
  99           For convenience, underscores in header names will be transformed to dashes -
 100           this behaviour does not extend to other methods.
 101
 102        If ``**headers`` contains multiple keys that have equal ``.lower()`` representations,
 103        the behavior is undefined.
 104        """
 105        super().__init__(fields)
 106
 107        for key, value in self.fields:
 108            if not isinstance(key, bytes) or not isinstance(value, bytes):
 109                raise TypeError("Header fields must be bytes.")
 110
 111        # content_type -> content-type
 112        self.update(
 113            {
 114                _always_bytes(name).replace(b"_", b"-"): _always_bytes(value)
 115                for name, value in headers.items()
 116            }
 117        )
 118
 119    fields: tuple[tuple[bytes, bytes], ...]
 120
 121    @staticmethod
 122    def _reduce_values(values) -> str:
 123        # Headers can be folded
 124        return ", ".join(values)
 125
 126    @staticmethod
 127    def _kconv(key) -> str:
 128        # Headers are case-insensitive
 129        return key.lower()
 130
 131    def __bytes__(self) -> bytes:
 132        if self.fields:
 133            return b"\r\n".join(b": ".join(field) for field in self.fields) + b"\r\n"
 134        else:
 135            return b""
 136
 137    def __delitem__(self, key: str | bytes) -> None:
 138        key = _always_bytes(key)
 139        super().__delitem__(key)
 140
 141    def __iter__(self) -> Iterator[str]:
 142        for x in super().__iter__():
 143            yield _native(x)
 144
 145    def get_all(self, name: str | bytes) -> list[str]:
 146        """
 147        Like `Headers.get`, but does not fold multiple headers into a single one.
 148        This is useful for Set-Cookie and Cookie headers, which do not support folding.
 149
 150        *See also:*
 151         - <https://tools.ietf.org/html/rfc7230#section-3.2.2>
 152         - <https://datatracker.ietf.org/doc/html/rfc6265#section-5.4>
 153         - <https://datatracker.ietf.org/doc/html/rfc7540#section-8.1.2.5>
 154        """
 155        name = _always_bytes(name)
 156        return [_native(x) for x in super().get_all(name)]
 157
 158    def set_all(self, name: str | bytes, values: Iterable[str | bytes]):
 159        """
 160        Explicitly set multiple headers for the given key.
 161        See `Headers.get_all`.
 162        """
 163        name = _always_bytes(name)
 164        values = [_always_bytes(x) for x in values]
 165        return super().set_all(name, values)
 166
 167    def insert(self, index: int, key: str | bytes, value: str | bytes):
 168        key = _always_bytes(key)
 169        value = _always_bytes(value)
 170        super().insert(index, key, value)
 171
 172    def items(self, multi=False):
 173        if multi:
 174            return ((_native(k), _native(v)) for k, v in self.fields)
 175        else:
 176            return super().items()
 177
 178
 179@dataclass
 180class MessageData(serializable.Serializable):
 181    http_version: bytes
 182    headers: Headers
 183    content: bytes | None
 184    trailers: Headers | None
 185    timestamp_start: float
 186    timestamp_end: float | None
 187
 188    # noinspection PyUnreachableCode
 189    if __debug__:
 190
 191        def __post_init__(self):
 192            for field in fields(self):
 193                val = getattr(self, field.name)
 194                typecheck.check_option_type(field.name, val, field.type)
 195
 196    def set_state(self, state):
 197        for k, v in state.items():
 198            if k in ("headers", "trailers") and v is not None:
 199                v = Headers.from_state(v)
 200            setattr(self, k, v)
 201
 202    def get_state(self):
 203        state = vars(self).copy()
 204        state["headers"] = state["headers"].get_state()
 205        if state["trailers"] is not None:
 206            state["trailers"] = state["trailers"].get_state()
 207        return state
 208
 209    @classmethod
 210    def from_state(cls, state):
 211        state["headers"] = Headers.from_state(state["headers"])
 212        if state["trailers"] is not None:
 213            state["trailers"] = Headers.from_state(state["trailers"])
 214        return cls(**state)
 215
 216
 217@dataclass
 218class RequestData(MessageData):
 219    host: str
 220    port: int
 221    method: bytes
 222    scheme: bytes
 223    authority: bytes
 224    path: bytes
 225
 226
 227@dataclass
 228class ResponseData(MessageData):
 229    status_code: int
 230    reason: bytes
 231
 232
 233class Message(serializable.Serializable):
 234    """Base class for `Request` and `Response`."""
 235
 236    @classmethod
 237    def from_state(cls, state):
 238        return cls(**state)
 239
 240    def get_state(self):
 241        return self.data.get_state()
 242
 243    def set_state(self, state):
 244        self.data.set_state(state)
 245
 246    data: MessageData
 247    stream: Callable[[bytes], Iterable[bytes] | bytes] | bool = False
 248    """
 249    This attribute controls if the message body should be streamed.
 250
 251    If `False`, mitmproxy will buffer the entire body before forwarding it to the destination.
 252    This makes it possible to perform string replacements on the entire body.
 253    If `True`, the message body will not be buffered on the proxy
 254    but immediately forwarded instead.
 255    Alternatively, a transformation function can be specified, which will be called for each chunk of data.
 256    Please note that packet boundaries generally should not be relied upon.
 257
 258    This attribute must be set in the `requestheaders` or `responseheaders` hook.
 259    Setting it in `request` or  `response` is already too late, mitmproxy has buffered the message body already.
 260    """
 261
 262    @property
 263    def http_version(self) -> str:
 264        """
 265        HTTP version string, for example `HTTP/1.1`.
 266        """
 267        return self.data.http_version.decode("utf-8", "surrogateescape")
 268
 269    @http_version.setter
 270    def http_version(self, http_version: str | bytes) -> None:
 271        self.data.http_version = strutils.always_bytes(
 272            http_version, "utf-8", "surrogateescape"
 273        )
 274
 275    @property
 276    def is_http10(self) -> bool:
 277        return self.data.http_version == b"HTTP/1.0"
 278
 279    @property
 280    def is_http11(self) -> bool:
 281        return self.data.http_version == b"HTTP/1.1"
 282
 283    @property
 284    def is_http2(self) -> bool:
 285        return self.data.http_version == b"HTTP/2.0"
 286
 287    @property
 288    def is_http3(self) -> bool:
 289        return self.data.http_version == b"HTTP/3"
 290
 291    @property
 292    def headers(self) -> Headers:
 293        """
 294        The HTTP headers.
 295        """
 296        return self.data.headers
 297
 298    @headers.setter
 299    def headers(self, h: Headers) -> None:
 300        self.data.headers = h
 301
 302    @property
 303    def trailers(self) -> Headers | None:
 304        """
 305        The [HTTP trailers](https://developer.mozilla.org/en-US/docs/Web/HTTP/Headers/Trailer).
 306        """
 307        return self.data.trailers
 308
 309    @trailers.setter
 310    def trailers(self, h: Headers | None) -> None:
 311        self.data.trailers = h
 312
 313    @property
 314    def raw_content(self) -> bytes | None:
 315        """
 316        The raw (potentially compressed) HTTP message body.
 317
 318        In contrast to `Message.content` and `Message.text`, accessing this property never raises.
 319
 320        *See also:* `Message.content`, `Message.text`
 321        """
 322        return self.data.content
 323
 324    @raw_content.setter
 325    def raw_content(self, content: bytes | None) -> None:
 326        self.data.content = content
 327
 328    @property
 329    def content(self) -> bytes | None:
 330        """
 331        The uncompressed HTTP message body as bytes.
 332
 333        Accessing this attribute may raise a `ValueError` when the HTTP content-encoding is invalid.
 334
 335        *See also:* `Message.raw_content`, `Message.text`
 336        """
 337        return self.get_content()
 338
 339    @content.setter
 340    def content(self, value: bytes | None) -> None:
 341        self.set_content(value)
 342
 343    @property
 344    def text(self) -> str | None:
 345        """
 346        The uncompressed and decoded HTTP message body as text.
 347
 348        Accessing this attribute may raise a `ValueError` when either content-encoding or charset is invalid.
 349
 350        *See also:* `Message.raw_content`, `Message.content`
 351        """
 352        return self.get_text()
 353
 354    @text.setter
 355    def text(self, value: str | None) -> None:
 356        self.set_text(value)
 357
 358    def set_content(self, value: bytes | None) -> None:
 359        if value is None:
 360            self.raw_content = None
 361            return
 362        if not isinstance(value, bytes):
 363            raise TypeError(
 364                f"Message content must be bytes, not {type(value).__name__}. "
 365                "Please use .text if you want to assign a str."
 366            )
 367        ce = self.headers.get("content-encoding")
 368        try:
 369            self.raw_content = encoding.encode(value, ce or "identity")
 370        except ValueError:
 371            # So we have an invalid content-encoding?
 372            # Let's remove it!
 373            del self.headers["content-encoding"]
 374            self.raw_content = value
 375
 376        if "transfer-encoding" in self.headers:
 377            # https://httpwg.org/specs/rfc7230.html#header.content-length
 378            # don't set content-length if a transfer-encoding is provided
 379            pass
 380        else:
 381            self.headers["content-length"] = str(len(self.raw_content))
 382
 383    def get_content(self, strict: bool = True) -> bytes | None:
 384        """
 385        Similar to `Message.content`, but does not raise if `strict` is `False`.
 386        Instead, the compressed message body is returned as-is.
 387        """
 388        if self.raw_content is None:
 389            return None
 390        ce = self.headers.get("content-encoding")
 391        if ce:
 392            try:
 393                content = encoding.decode(self.raw_content, ce)
 394                # A client may illegally specify a byte -> str encoding here (e.g. utf8)
 395                if isinstance(content, str):
 396                    raise ValueError(f"Invalid Content-Encoding: {ce}")
 397                return content
 398            except ValueError:
 399                if strict:
 400                    raise
 401                return self.raw_content
 402        else:
 403            return self.raw_content
 404
 405    def set_text(self, text: str | None) -> None:
 406        if text is None:
 407            self.content = None
 408            return
 409        enc = infer_content_encoding(self.headers.get("content-type", ""))
 410
 411        try:
 412            self.content = cast(bytes, encoding.encode(text, enc))
 413        except ValueError:
 414            # Fall back to UTF-8 and update the content-type header.
 415            ct = parse_content_type(self.headers.get("content-type", "")) or (
 416                "text",
 417                "plain",
 418                {},
 419            )
 420            ct[2]["charset"] = "utf-8"
 421            self.headers["content-type"] = assemble_content_type(*ct)
 422            enc = "utf8"
 423            self.content = text.encode(enc, "surrogateescape")
 424
 425    def get_text(self, strict: bool = True) -> str | None:
 426        """
 427        Similar to `Message.text`, but does not raise if `strict` is `False`.
 428        Instead, the message body is returned as surrogate-escaped UTF-8.
 429        """
 430        content = self.get_content(strict)
 431        if content is None:
 432            return None
 433        enc = infer_content_encoding(self.headers.get("content-type", ""), content)
 434        try:
 435            return cast(str, encoding.decode(content, enc))
 436        except ValueError:
 437            if strict:
 438                raise
 439            return content.decode("utf8", "surrogateescape")
 440
 441    @property
 442    def timestamp_start(self) -> float:
 443        """
 444        *Timestamp:* Headers received.
 445        """
 446        return self.data.timestamp_start
 447
 448    @timestamp_start.setter
 449    def timestamp_start(self, timestamp_start: float) -> None:
 450        self.data.timestamp_start = timestamp_start
 451
 452    @property
 453    def timestamp_end(self) -> float | None:
 454        """
 455        *Timestamp:* Last byte received.
 456        """
 457        return self.data.timestamp_end
 458
 459    @timestamp_end.setter
 460    def timestamp_end(self, timestamp_end: float | None):
 461        self.data.timestamp_end = timestamp_end
 462
 463    def decode(self, strict: bool = True) -> None:
 464        """
 465        Decodes body based on the current Content-Encoding header, then
 466        removes the header. If there is no Content-Encoding header, no
 467        action is taken.
 468
 469        *Raises:*
 470         - `ValueError`, when the content-encoding is invalid and strict is True.
 471        """
 472        decoded = self.get_content(strict)
 473        self.headers.pop("content-encoding", None)
 474        self.content = decoded
 475
 476    def encode(self, encoding: str) -> None:
 477        """
 478        Encodes body with the given encoding, where e is "gzip", "deflate", "identity", "br", or "zstd".
 479        Any existing content-encodings are overwritten, the content is not decoded beforehand.
 480
 481        *Raises:*
 482         - `ValueError`, when the specified content-encoding is invalid.
 483        """
 484        self.headers["content-encoding"] = encoding
 485        self.content = self.raw_content
 486        if "content-encoding" not in self.headers:
 487            raise ValueError(f"Invalid content encoding {repr(encoding)}")
 488
 489    def json(self, **kwargs: Any) -> Any:
 490        """
 491        Returns the JSON encoded content of the response, if any.
 492        `**kwargs` are optional arguments that will be
 493        passed to `json.loads()`.
 494
 495        Will raise if the content can not be decoded and then parsed as JSON.
 496
 497        *Raises:*
 498         - `json.decoder.JSONDecodeError` if content is not valid JSON.
 499         - `TypeError` if the content is not available, for example because the response
 500            has been streamed.
 501        """
 502        content = self.get_content(strict=False)
 503        if content is None:
 504            raise TypeError("Message content is not available.")
 505        else:
 506            return json.loads(content, **kwargs)
 507
 508
 509class Request(Message):
 510    """
 511    An HTTP request.
 512    """
 513
 514    data: RequestData
 515
 516    def __init__(
 517        self,
 518        host: str,
 519        port: int,
 520        method: bytes,
 521        scheme: bytes,
 522        authority: bytes,
 523        path: bytes,
 524        http_version: bytes,
 525        headers: Headers | tuple[tuple[bytes, bytes], ...],
 526        content: bytes | None,
 527        trailers: Headers | tuple[tuple[bytes, bytes], ...] | None,
 528        timestamp_start: float,
 529        timestamp_end: float | None,
 530    ):
 531        # auto-convert invalid types to retain compatibility with older code.
 532        if isinstance(host, bytes):
 533            host = host.decode("idna", "strict")
 534        if isinstance(method, str):
 535            method = method.encode("ascii", "strict")
 536        if isinstance(scheme, str):
 537            scheme = scheme.encode("ascii", "strict")
 538        if isinstance(authority, str):
 539            authority = authority.encode("ascii", "strict")
 540        if isinstance(path, str):
 541            path = path.encode("ascii", "strict")
 542        if isinstance(http_version, str):
 543            http_version = http_version.encode("ascii", "strict")
 544
 545        if isinstance(content, str):
 546            raise ValueError(f"Content must be bytes, not {type(content).__name__}")
 547        if not isinstance(headers, Headers):
 548            headers = Headers(headers)
 549        if trailers is not None and not isinstance(trailers, Headers):
 550            trailers = Headers(trailers)
 551
 552        self.data = RequestData(
 553            host=host,
 554            port=port,
 555            method=method,
 556            scheme=scheme,
 557            authority=authority,
 558            path=path,
 559            http_version=http_version,
 560            headers=headers,
 561            content=content,
 562            trailers=trailers,
 563            timestamp_start=timestamp_start,
 564            timestamp_end=timestamp_end,
 565        )
 566
 567    def __repr__(self) -> str:
 568        if self.host and self.port:
 569            hostport = f"{self.host}:{self.port}"
 570        else:
 571            hostport = ""
 572        path = self.path or ""
 573        return f"Request({self.method} {hostport}{path})"
 574
 575    @classmethod
 576    def make(
 577        cls,
 578        method: str,
 579        url: str,
 580        content: bytes | str = "",
 581        headers: (
 582            Headers | dict[str | bytes, str | bytes] | Iterable[tuple[bytes, bytes]]
 583        ) = (),
 584    ) -> "Request":
 585        """
 586        Simplified API for creating request objects.
 587        """
 588        # Headers can be list or dict, we differentiate here.
 589        if isinstance(headers, Headers):
 590            pass
 591        elif isinstance(headers, dict):
 592            headers = Headers(
 593                (
 594                    always_bytes(k, "utf-8", "surrogateescape"),
 595                    always_bytes(v, "utf-8", "surrogateescape"),
 596                )
 597                for k, v in headers.items()
 598            )
 599        elif isinstance(headers, Iterable):
 600            headers = Headers(headers)  # type: ignore
 601        else:
 602            raise TypeError(
 603                "Expected headers to be an iterable or dict, but is {}.".format(
 604                    type(headers).__name__
 605                )
 606            )
 607
 608        req = cls(
 609            "",
 610            0,
 611            method.encode("utf-8", "surrogateescape"),
 612            b"",
 613            b"",
 614            b"",
 615            b"HTTP/1.1",
 616            headers,
 617            b"",
 618            None,
 619            time.time(),
 620            time.time(),
 621        )
 622
 623        req.url = url
 624        # Assign this manually to update the content-length header.
 625        if isinstance(content, bytes):
 626            req.content = content
 627        elif isinstance(content, str):
 628            req.text = content
 629        else:
 630            raise TypeError(
 631                f"Expected content to be str or bytes, but is {type(content).__name__}."
 632            )
 633
 634        return req
 635
 636    @property
 637    def first_line_format(self) -> str:
 638        """
 639        *Read-only:* HTTP request form as defined in [RFC 7230](https://tools.ietf.org/html/rfc7230#section-5.3).
 640
 641        origin-form and asterisk-form are subsumed as "relative".
 642        """
 643        if self.method == "CONNECT":
 644            return "authority"
 645        elif self.authority:
 646            return "absolute"
 647        else:
 648            return "relative"
 649
 650    @property
 651    def method(self) -> str:
 652        """
 653        HTTP request method, e.g. "GET".
 654        """
 655        return self.data.method.decode("utf-8", "surrogateescape").upper()
 656
 657    @method.setter
 658    def method(self, val: str | bytes) -> None:
 659        self.data.method = always_bytes(val, "utf-8", "surrogateescape")
 660
 661    @property
 662    def scheme(self) -> str:
 663        """
 664        HTTP request scheme, which should be "http" or "https".
 665        """
 666        return self.data.scheme.decode("utf-8", "surrogateescape")
 667
 668    @scheme.setter
 669    def scheme(self, val: str | bytes) -> None:
 670        self.data.scheme = always_bytes(val, "utf-8", "surrogateescape")
 671
 672    @property
 673    def authority(self) -> str:
 674        """
 675        HTTP request authority.
 676
 677        For HTTP/1, this is the authority portion of the request target
 678        (in either absolute-form or authority-form).
 679        For origin-form and asterisk-form requests, this property is set to an empty string.
 680
 681        For HTTP/2, this is the :authority pseudo header.
 682
 683        *See also:* `Request.host`, `Request.host_header`, `Request.pretty_host`
 684        """
 685        try:
 686            return self.data.authority.decode("idna")
 687        except UnicodeError:
 688            return self.data.authority.decode("utf8", "surrogateescape")
 689
 690    @authority.setter
 691    def authority(self, val: str | bytes) -> None:
 692        if isinstance(val, str):
 693            try:
 694                val = val.encode("idna", "strict")
 695            except UnicodeError:
 696                val = val.encode("utf8", "surrogateescape")  # type: ignore
 697        self.data.authority = val
 698
 699    @property
 700    def host(self) -> str:
 701        """
 702        Target server for this request. This may be parsed from the raw request
 703        (e.g. from a ``GET http://example.com/ HTTP/1.1`` request line)
 704        or inferred from the proxy mode (e.g. an IP in transparent mode).
 705
 706        Setting the host attribute also updates the host header and authority information, if present.
 707
 708        *See also:* `Request.authority`, `Request.host_header`, `Request.pretty_host`
 709        """
 710        return self.data.host
 711
 712    @host.setter
 713    def host(self, val: str | bytes) -> None:
 714        self.data.host = always_str(val, "idna", "strict")
 715        self._update_host_and_authority()
 716
 717    @property
 718    def host_header(self) -> str | None:
 719        """
 720        The request's host/authority header.
 721
 722        This property maps to either ``request.headers["Host"]`` or
 723        ``request.authority``, depending on whether it's HTTP/1.x or HTTP/2.0.
 724
 725        *See also:* `Request.authority`,`Request.host`, `Request.pretty_host`
 726        """
 727        if self.is_http2 or self.is_http3:
 728            return self.authority or self.data.headers.get("Host", None)
 729        else:
 730            return self.data.headers.get("Host", None)
 731
 732    @host_header.setter
 733    def host_header(self, val: None | str | bytes) -> None:
 734        if val is None:
 735            if self.is_http2 or self.is_http3:
 736                self.data.authority = b""
 737            self.headers.pop("Host", None)
 738        else:
 739            if self.is_http2 or self.is_http3:
 740                self.authority = val  # type: ignore
 741            if not (self.is_http2 or self.is_http3) or "Host" in self.headers:
 742                # For h2, we only overwrite, but not create, as :authority is the h2 host header.
 743                self.headers["Host"] = val
 744
 745    @property
 746    def port(self) -> int:
 747        """
 748        Target port.
 749        """
 750        return self.data.port
 751
 752    @port.setter
 753    def port(self, port: int) -> None:
 754        if not isinstance(port, int):
 755            raise ValueError(f"Port must be an integer, not {port!r}.")
 756
 757        self.data.port = port
 758        self._update_host_and_authority()
 759
 760    def _update_host_and_authority(self) -> None:
 761        val = url.hostport(self.scheme, self.host, self.port)
 762
 763        # Update host header
 764        if "Host" in self.data.headers:
 765            self.data.headers["Host"] = val
 766        # Update authority
 767        if self.data.authority:
 768            self.authority = val
 769
 770    @property
 771    def path(self) -> str:
 772        """
 773        HTTP request path, e.g. "/index.html" or "/index.html?a=b".
 774        Usually starts with a slash, except for OPTIONS requests, which may just be "*".
 775
 776        This attribute includes both path and query parts of the target URI
 777        (see Sections 3.3 and 3.4 of [RFC3986](https://datatracker.ietf.org/doc/html/rfc3986)).
 778        """
 779        return self.data.path.decode("utf-8", "surrogateescape")
 780
 781    @path.setter
 782    def path(self, val: str | bytes) -> None:
 783        self.data.path = always_bytes(val, "utf-8", "surrogateescape")
 784
 785    @property
 786    def url(self) -> str:
 787        """
 788        The full URL string, constructed from `Request.scheme`, `Request.host`, `Request.port` and `Request.path`.
 789
 790        Settings this property updates these attributes as well.
 791        """
 792        if self.first_line_format == "authority":
 793            return f"{self.host}:{self.port}"
 794        path = self.path if self.path != "*" else ""
 795        return url.unparse(self.scheme, self.host, self.port, path)
 796
 797    @url.setter
 798    def url(self, val: str | bytes) -> None:
 799        val = always_str(val, "utf-8", "surrogateescape")
 800        self.scheme, self.host, self.port, self.path = url.parse(val)  # type: ignore
 801
 802    @property
 803    def pretty_host(self) -> str:
 804        """
 805        *Read-only:* Like `Request.host`, but using `Request.host_header` header as an additional (preferred) data source.
 806        This is useful in transparent mode where `Request.host` is only an IP address.
 807
 808        *Warning:* When working in adversarial environments, this may not reflect the actual destination
 809        as the Host header could be spoofed.
 810        """
 811        authority = self.host_header
 812        if authority:
 813            return url.parse_authority(authority, check=False)[0]
 814        else:
 815            return self.host
 816
 817    @property
 818    def pretty_url(self) -> str:
 819        """
 820        *Read-only:* Like `Request.url`, but using `Request.pretty_host` instead of `Request.host`.
 821        """
 822        if self.first_line_format == "authority":
 823            return self.authority
 824
 825        host_header = self.host_header
 826        if not host_header:
 827            return self.url
 828
 829        pretty_host, pretty_port = url.parse_authority(host_header, check=False)
 830        pretty_port = pretty_port or url.default_port(self.scheme) or 443
 831        path = self.path if self.path != "*" else ""
 832
 833        return url.unparse(self.scheme, pretty_host, pretty_port, path)
 834
 835    def _get_query(self):
 836        query = urllib.parse.urlparse(self.url).query
 837        return tuple(url.decode(query))
 838
 839    def _set_query(self, query_data):
 840        query = url.encode(query_data)
 841        _, _, path, params, _, fragment = urllib.parse.urlparse(self.url)
 842        self.path = urllib.parse.urlunparse(["", "", path, params, query, fragment])
 843
 844    @property
 845    def query(self) -> multidict.MultiDictView[str, str]:
 846        """
 847        The request query as a mutable mapping view on the request's path.
 848        For the most part, this behaves like a dictionary.
 849        Modifications to the MultiDictView update `Request.path`, and vice versa.
 850        """
 851        return multidict.MultiDictView(self._get_query, self._set_query)
 852
 853    @query.setter
 854    def query(self, value):
 855        self._set_query(value)
 856
 857    def _get_cookies(self):
 858        h = self.headers.get_all("Cookie")
 859        return tuple(cookies.parse_cookie_headers(h))
 860
 861    def _set_cookies(self, value):
 862        self.headers["cookie"] = cookies.format_cookie_header(value)
 863
 864    @property
 865    def cookies(self) -> multidict.MultiDictView[str, str]:
 866        """
 867        The request cookies.
 868        For the most part, this behaves like a dictionary.
 869        Modifications to the MultiDictView update `Request.headers`, and vice versa.
 870        """
 871        return multidict.MultiDictView(self._get_cookies, self._set_cookies)
 872
 873    @cookies.setter
 874    def cookies(self, value):
 875        self._set_cookies(value)
 876
 877    @property
 878    def path_components(self) -> tuple[str, ...]:
 879        """
 880        The URL's path components as a tuple of strings.
 881        Components are unquoted.
 882        """
 883        path = urllib.parse.urlparse(self.url).path
 884        # This needs to be a tuple so that it's immutable.
 885        # Otherwise, this would fail silently:
 886        #   request.path_components.append("foo")
 887        return tuple(url.unquote(i) for i in path.split("/") if i)
 888
 889    @path_components.setter
 890    def path_components(self, components: Iterable[str]):
 891        components = map(lambda x: url.quote(x, safe=""), components)
 892        path = "/" + "/".join(components)
 893        _, _, _, params, query, fragment = urllib.parse.urlparse(self.url)
 894        self.path = urllib.parse.urlunparse(["", "", path, params, query, fragment])
 895
 896    def anticache(self) -> None:
 897        """
 898        Modifies this request to remove headers that might produce a cached response.
 899        """
 900        delheaders = (
 901            "if-modified-since",
 902            "if-none-match",
 903        )
 904        for i in delheaders:
 905            self.headers.pop(i, None)
 906
 907    def anticomp(self) -> None:
 908        """
 909        Modify the Accept-Encoding header to only accept uncompressed responses.
 910        """
 911        self.headers["accept-encoding"] = "identity"
 912
 913    def constrain_encoding(self) -> None:
 914        """
 915        Limits the permissible Accept-Encoding values, based on what we can decode appropriately.
 916        """
 917        accept_encoding = self.headers.get("accept-encoding")
 918        if accept_encoding:
 919            self.headers["accept-encoding"] = ", ".join(
 920                e
 921                for e in {"gzip", "identity", "deflate", "br", "zstd"}
 922                if e in accept_encoding
 923            )
 924
 925    def _get_urlencoded_form(self):
 926        is_valid_content_type = (
 927            "application/x-www-form-urlencoded"
 928            in self.headers.get("content-type", "").lower()
 929        )
 930        if is_valid_content_type:
 931            return tuple(url.decode(self.get_text(strict=False)))
 932        return ()
 933
 934    def _set_urlencoded_form(self, form_data: Sequence[tuple[str, str]]) -> None:
 935        """
 936        Sets the body to the URL-encoded form data, and adds the appropriate content-type header.
 937        This will overwrite the existing content if there is one.
 938        """
 939        self.headers["content-type"] = "application/x-www-form-urlencoded"
 940        self.content = url.encode(form_data, self.get_text(strict=False)).encode()
 941
 942    @property
 943    def urlencoded_form(self) -> multidict.MultiDictView[str, str]:
 944        """
 945        The URL-encoded form data.
 946
 947        If the content-type indicates non-form data or the form could not be parsed, this is set to
 948        an empty `MultiDictView`.
 949
 950        Modifications to the MultiDictView update `Request.content`, and vice versa.
 951        """
 952        return multidict.MultiDictView(
 953            self._get_urlencoded_form, self._set_urlencoded_form
 954        )
 955
 956    @urlencoded_form.setter
 957    def urlencoded_form(self, value):
 958        self._set_urlencoded_form(value)
 959
 960    def _get_multipart_form(self) -> list[tuple[bytes, bytes]]:
 961        is_valid_content_type = (
 962            "multipart/form-data" in self.headers.get("content-type", "").lower()
 963        )
 964        if is_valid_content_type and self.content is not None:
 965            try:
 966                return multipart.decode_multipart(
 967                    self.headers.get("content-type"), self.content
 968                )
 969            except ValueError:
 970                pass
 971        return []
 972
 973    def _set_multipart_form(self, value: list[tuple[bytes, bytes]]) -> None:
 974        ct = self.headers.get("content-type", "")
 975        is_valid_content_type = ct.lower().startswith("multipart/form-data")
 976        if not is_valid_content_type:
 977            """
 978            Generate a random boundary here.
 979
 980            See <https://datatracker.ietf.org/doc/html/rfc2046#section-5.1.1> for specifications
 981            on generating the boundary.
 982            """
 983            boundary = "-" * 20 + binascii.hexlify(os.urandom(16)).decode()
 984            self.headers["content-type"] = ct = (
 985                f"multipart/form-data; boundary={boundary}"
 986            )
 987        self.content = multipart.encode_multipart(ct, value)
 988
 989    @property
 990    def multipart_form(self) -> multidict.MultiDictView[bytes, bytes]:
 991        """
 992        The multipart form data.
 993
 994        If the content-type indicates non-form data or the form could not be parsed, this is set to
 995        an empty `MultiDictView`.
 996
 997        Modifications to the MultiDictView update `Request.content`, and vice versa.
 998        """
 999        return multidict.MultiDictView(
1000            self._get_multipart_form, self._set_multipart_form
1001        )
1002
1003    @multipart_form.setter
1004    def multipart_form(self, value: list[tuple[bytes, bytes]]) -> None:
1005        self._set_multipart_form(value)
1006
1007
1008class Response(Message):
1009    """
1010    An HTTP response.
1011    """
1012
1013    data: ResponseData
1014
1015    def __init__(
1016        self,
1017        http_version: bytes,
1018        status_code: int,
1019        reason: bytes,
1020        headers: Headers | tuple[tuple[bytes, bytes], ...],
1021        content: bytes | None,
1022        trailers: None | Headers | tuple[tuple[bytes, bytes], ...],
1023        timestamp_start: float,
1024        timestamp_end: float | None,
1025    ):
1026        # auto-convert invalid types to retain compatibility with older code.
1027        if isinstance(http_version, str):
1028            http_version = http_version.encode("ascii", "strict")
1029        if isinstance(reason, str):
1030            reason = reason.encode("ascii", "strict")
1031
1032        if isinstance(content, str):
1033            raise ValueError(f"Content must be bytes, not {type(content).__name__}")
1034        if not isinstance(headers, Headers):
1035            headers = Headers(headers)
1036        if trailers is not None and not isinstance(trailers, Headers):
1037            trailers = Headers(trailers)
1038
1039        self.data = ResponseData(
1040            http_version=http_version,
1041            status_code=status_code,
1042            reason=reason,
1043            headers=headers,
1044            content=content,
1045            trailers=trailers,
1046            timestamp_start=timestamp_start,
1047            timestamp_end=timestamp_end,
1048        )
1049
1050    def __repr__(self) -> str:
1051        if self.raw_content:
1052            ct = self.headers.get("content-type", "unknown content type")
1053            size = human.pretty_size(len(self.raw_content))
1054            details = f"{ct}, {size}"
1055        else:
1056            details = "no content"
1057        return f"Response({self.status_code}, {details})"
1058
1059    @classmethod
1060    def make(
1061        cls,
1062        status_code: int = 200,
1063        content: bytes | str = b"",
1064        headers: (
1065            Headers | Mapping[str, str | bytes] | Iterable[tuple[bytes, bytes]]
1066        ) = (),
1067    ) -> "Response":
1068        """
1069        Simplified API for creating response objects.
1070        """
1071        if isinstance(headers, Headers):
1072            headers = headers
1073        elif isinstance(headers, dict):
1074            headers = Headers(
1075                (
1076                    always_bytes(k, "utf-8", "surrogateescape"),  # type: ignore
1077                    always_bytes(v, "utf-8", "surrogateescape"),
1078                )
1079                for k, v in headers.items()
1080            )
1081        elif isinstance(headers, Iterable):
1082            headers = Headers(headers)  # type: ignore
1083        else:
1084            raise TypeError(
1085                "Expected headers to be an iterable or dict, but is {}.".format(
1086                    type(headers).__name__
1087                )
1088            )
1089
1090        resp = cls(
1091            b"HTTP/1.1",
1092            status_code,
1093            status_codes.RESPONSES.get(status_code, "").encode(),
1094            headers,
1095            None,
1096            None,
1097            time.time(),
1098            time.time(),
1099        )
1100
1101        # Assign this manually to update the content-length header.
1102        if isinstance(content, bytes):
1103            resp.content = content
1104        elif isinstance(content, str):
1105            resp.text = content
1106        else:
1107            raise TypeError(
1108                f"Expected content to be str or bytes, but is {type(content).__name__}."
1109            )
1110
1111        return resp
1112
1113    @property
1114    def status_code(self) -> int:
1115        """
1116        HTTP Status Code, e.g. ``200``.
1117        """
1118        return self.data.status_code
1119
1120    @status_code.setter
1121    def status_code(self, status_code: int) -> None:
1122        self.data.status_code = status_code
1123
1124    @property
1125    def reason(self) -> str:
1126        """
1127        HTTP reason phrase, for example "Not Found".
1128
1129        HTTP/2 responses do not contain a reason phrase, an empty string will be returned instead.
1130        """
1131        # Encoding: http://stackoverflow.com/a/16674906/934719
1132        return self.data.reason.decode("ISO-8859-1")
1133
1134    @reason.setter
1135    def reason(self, reason: str | bytes) -> None:
1136        self.data.reason = strutils.always_bytes(reason, "ISO-8859-1")
1137
1138    def _get_cookies(self):
1139        h = self.headers.get_all("set-cookie")
1140        all_cookies = cookies.parse_set_cookie_headers(h)
1141        return tuple((name, (value, attrs)) for name, value, attrs in all_cookies)
1142
1143    def _set_cookies(self, value):
1144        cookie_headers = []
1145        for k, v in value:
1146            header = cookies.format_set_cookie_header([(k, v[0], v[1])])
1147            cookie_headers.append(header)
1148        self.headers.set_all("set-cookie", cookie_headers)
1149
1150    @property
1151    def cookies(
1152        self,
1153    ) -> multidict.MultiDictView[str, tuple[str, multidict.MultiDict[str, str | None]]]:
1154        """
1155        The response cookies. A possibly empty `MultiDictView`, where the keys are cookie
1156        name strings, and values are `(cookie value, attributes)` tuples. Within
1157        attributes, unary attributes (e.g. `HTTPOnly`) are indicated by a `None` value.
1158        Modifications to the MultiDictView update `Response.headers`, and vice versa.
1159
1160        *Warning:* Changes to `attributes` will not be picked up unless you also reassign
1161        the `(cookie value, attributes)` tuple directly in the `MultiDictView`.
1162        """
1163        return multidict.MultiDictView(self._get_cookies, self._set_cookies)
1164
1165    @cookies.setter
1166    def cookies(self, value):
1167        self._set_cookies(value)
1168
1169    def refresh(self, now=None):
1170        """
1171        This fairly complex and heuristic function refreshes a server
1172        response for replay.
1173
1174         - It adjusts date, expires, and last-modified headers.
1175         - It adjusts cookie expiration.
1176        """
1177        if not now:
1178            now = time.time()
1179        delta = now - self.timestamp_start
1180        refresh_headers = [
1181            "date",
1182            "expires",
1183            "last-modified",
1184        ]
1185        for i in refresh_headers:
1186            if i in self.headers:
1187                d = parsedate_tz(self.headers[i])
1188                if d:
1189                    new = mktime_tz(d) + delta
1190                    try:
1191                        self.headers[i] = formatdate(new, usegmt=True)
1192                    except OSError:  # pragma: no cover
1193                        pass  # value out of bounds on Windows only (which is why we exclude it from coverage).
1194        c = []
1195        for set_cookie_header in self.headers.get_all("set-cookie"):
1196            try:
1197                refreshed = cookies.refresh_set_cookie_header(set_cookie_header, delta)
1198            except ValueError:
1199                refreshed = set_cookie_header
1200            c.append(refreshed)
1201        if c:
1202            self.headers.set_all("set-cookie", c)
1203
1204
1205class HTTPFlow(flow.Flow):
1206    """
1207    An HTTPFlow is a collection of objects representing a single HTTP
1208    transaction.
1209    """
1210
1211    request: Request
1212    """The client's HTTP request."""
1213    response: Response | None = None
1214    """The server's HTTP response."""
1215    error: flow.Error | None = None
1216    """
1217    A connection or protocol error affecting this flow.
1218
1219    Note that it's possible for a Flow to have both a response and an error
1220    object. This might happen, for instance, when a response was received
1221    from the server, but there was an error sending it back to the client.
1222    """
1223
1224    websocket: WebSocketData | None = None
1225    """
1226    If this HTTP flow initiated a WebSocket connection, this attribute contains all associated WebSocket data.
1227    """
1228
1229    def get_state(self) -> serializable.State:
1230        return {
1231            **super().get_state(),
1232            "request": self.request.get_state(),
1233            "response": self.response.get_state() if self.response else None,
1234            "websocket": self.websocket.get_state() if self.websocket else None,
1235        }
1236
1237    def set_state(self, state: serializable.State) -> None:
1238        self.request = Request.from_state(state.pop("request"))
1239        self.response = Response.from_state(r) if (r := state.pop("response")) else None
1240        self.websocket = (
1241            WebSocketData.from_state(w) if (w := state.pop("websocket")) else None
1242        )
1243        super().set_state(state)
1244
1245    def __repr__(self):
1246        s = "<HTTPFlow"
1247        for a in (
1248            "request",
1249            "response",
1250            "websocket",
1251            "error",
1252            "client_conn",
1253            "server_conn",
1254        ):
1255            if getattr(self, a, False):
1256                s += f"\r\n  {a} = {{flow.{a}}}"
1257        s += ">"
1258        return s.format(flow=self)
1259
1260    @property
1261    def timestamp_start(self) -> float:
1262        """*Read-only:* An alias for `Request.timestamp_start`."""
1263        return self.request.timestamp_start
1264
1265    @property
1266    def mode(self) -> str:  # pragma: no cover
1267        warnings.warn("HTTPFlow.mode is deprecated.", DeprecationWarning, stacklevel=2)
1268        return getattr(self, "_mode", "regular")
1269
1270    @mode.setter
1271    def mode(self, val: str) -> None:  # pragma: no cover
1272        warnings.warn("HTTPFlow.mode is deprecated.", DeprecationWarning, stacklevel=2)
1273        self._mode = val
1274
1275    def copy(self):
1276        f = super().copy()
1277        if self.request:
1278            f.request = self.request.copy()
1279        if self.response:
1280            f.response = self.response.copy()
1281        return f
1282
1283
1284__all__ = [
1285    "HTTPFlow",
1286    "Message",
1287    "Request",
1288    "Response",
1289    "Headers",
1290]