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        return url.unparse(self.scheme, self.host, self.port, self.path)
 795
 796    @url.setter
 797    def url(self, val: str | bytes) -> None:
 798        val = always_str(val, "utf-8", "surrogateescape")
 799        self.scheme, self.host, self.port, self.path = url.parse(val)
 800
 801    @property
 802    def pretty_host(self) -> str:
 803        """
 804        *Read-only:* Like `Request.host`, but using `Request.host_header` header as an additional (preferred) data source.
 805        This is useful in transparent mode where `Request.host` is only an IP address.
 806
 807        *Warning:* When working in adversarial environments, this may not reflect the actual destination
 808        as the Host header could be spoofed.
 809        """
 810        authority = self.host_header
 811        if authority:
 812            return url.parse_authority(authority, check=False)[0]
 813        else:
 814            return self.host
 815
 816    @property
 817    def pretty_url(self) -> str:
 818        """
 819        *Read-only:* Like `Request.url`, but using `Request.pretty_host` instead of `Request.host`.
 820        """
 821        if self.first_line_format == "authority":
 822            return self.authority
 823
 824        host_header = self.host_header
 825        if not host_header:
 826            return self.url
 827
 828        pretty_host, pretty_port = url.parse_authority(host_header, check=False)
 829        pretty_port = pretty_port or url.default_port(self.scheme) or 443
 830
 831        return url.unparse(self.scheme, pretty_host, pretty_port, self.path)
 832
 833    def _get_query(self):
 834        query = urllib.parse.urlparse(self.url).query
 835        return tuple(url.decode(query))
 836
 837    def _set_query(self, query_data):
 838        query = url.encode(query_data)
 839        _, _, path, params, _, fragment = urllib.parse.urlparse(self.url)
 840        self.path = urllib.parse.urlunparse(["", "", path, params, query, fragment])
 841
 842    @property
 843    def query(self) -> multidict.MultiDictView[str, str]:
 844        """
 845        The request query as a mutable mapping view on the request's path.
 846        For the most part, this behaves like a dictionary.
 847        Modifications to the MultiDictView update `Request.path`, and vice versa.
 848        """
 849        return multidict.MultiDictView(self._get_query, self._set_query)
 850
 851    @query.setter
 852    def query(self, value):
 853        self._set_query(value)
 854
 855    def _get_cookies(self):
 856        h = self.headers.get_all("Cookie")
 857        return tuple(cookies.parse_cookie_headers(h))
 858
 859    def _set_cookies(self, value):
 860        self.headers["cookie"] = cookies.format_cookie_header(value)
 861
 862    @property
 863    def cookies(self) -> multidict.MultiDictView[str, str]:
 864        """
 865        The request cookies.
 866        For the most part, this behaves like a dictionary.
 867        Modifications to the MultiDictView update `Request.headers`, and vice versa.
 868        """
 869        return multidict.MultiDictView(self._get_cookies, self._set_cookies)
 870
 871    @cookies.setter
 872    def cookies(self, value):
 873        self._set_cookies(value)
 874
 875    @property
 876    def path_components(self) -> tuple[str, ...]:
 877        """
 878        The URL's path components as a tuple of strings.
 879        Components are unquoted.
 880        """
 881        path = urllib.parse.urlparse(self.url).path
 882        # This needs to be a tuple so that it's immutable.
 883        # Otherwise, this would fail silently:
 884        #   request.path_components.append("foo")
 885        return tuple(url.unquote(i) for i in path.split("/") if i)
 886
 887    @path_components.setter
 888    def path_components(self, components: Iterable[str]):
 889        components = map(lambda x: url.quote(x, safe=""), components)
 890        path = "/" + "/".join(components)
 891        _, _, _, params, query, fragment = urllib.parse.urlparse(self.url)
 892        self.path = urllib.parse.urlunparse(["", "", path, params, query, fragment])
 893
 894    def anticache(self) -> None:
 895        """
 896        Modifies this request to remove headers that might produce a cached response.
 897        """
 898        delheaders = (
 899            "if-modified-since",
 900            "if-none-match",
 901        )
 902        for i in delheaders:
 903            self.headers.pop(i, None)
 904
 905    def anticomp(self) -> None:
 906        """
 907        Modify the Accept-Encoding header to only accept uncompressed responses.
 908        """
 909        self.headers["accept-encoding"] = "identity"
 910
 911    def constrain_encoding(self) -> None:
 912        """
 913        Limits the permissible Accept-Encoding values, based on what we can decode appropriately.
 914        """
 915        accept_encoding = self.headers.get("accept-encoding")
 916        if accept_encoding:
 917            self.headers["accept-encoding"] = ", ".join(
 918                e
 919                for e in {"gzip", "identity", "deflate", "br", "zstd"}
 920                if e in accept_encoding
 921            )
 922
 923    def _get_urlencoded_form(self):
 924        is_valid_content_type = (
 925            "application/x-www-form-urlencoded"
 926            in self.headers.get("content-type", "").lower()
 927        )
 928        if is_valid_content_type:
 929            return tuple(url.decode(self.get_text(strict=False)))
 930        return ()
 931
 932    def _set_urlencoded_form(self, form_data: Sequence[tuple[str, str]]) -> None:
 933        """
 934        Sets the body to the URL-encoded form data, and adds the appropriate content-type header.
 935        This will overwrite the existing content if there is one.
 936        """
 937        self.headers["content-type"] = "application/x-www-form-urlencoded"
 938        self.content = url.encode(form_data, self.get_text(strict=False)).encode()
 939
 940    @property
 941    def urlencoded_form(self) -> multidict.MultiDictView[str, str]:
 942        """
 943        The URL-encoded form data.
 944
 945        If the content-type indicates non-form data or the form could not be parsed, this is set to
 946        an empty `MultiDictView`.
 947
 948        Modifications to the MultiDictView update `Request.content`, and vice versa.
 949        """
 950        return multidict.MultiDictView(
 951            self._get_urlencoded_form, self._set_urlencoded_form
 952        )
 953
 954    @urlencoded_form.setter
 955    def urlencoded_form(self, value):
 956        self._set_urlencoded_form(value)
 957
 958    def _get_multipart_form(self) -> list[tuple[bytes, bytes]]:
 959        is_valid_content_type = (
 960            "multipart/form-data" in self.headers.get("content-type", "").lower()
 961        )
 962        if is_valid_content_type and self.content is not None:
 963            try:
 964                return multipart.decode_multipart(
 965                    self.headers.get("content-type"), self.content
 966                )
 967            except ValueError:
 968                pass
 969        return []
 970
 971    def _set_multipart_form(self, value: list[tuple[bytes, bytes]]) -> None:
 972        ct = self.headers.get("content-type", "")
 973        is_valid_content_type = ct.lower().startswith("multipart/form-data")
 974        if not is_valid_content_type:
 975            """
 976            Generate a random boundary here.
 977
 978            See <https://datatracker.ietf.org/doc/html/rfc2046#section-5.1.1> for specifications
 979            on generating the boundary.
 980            """
 981            boundary = "-" * 20 + binascii.hexlify(os.urandom(16)).decode()
 982            self.headers["content-type"] = ct = (
 983                f"multipart/form-data; boundary={boundary}"
 984            )
 985        self.content = multipart.encode_multipart(ct, value)
 986
 987    @property
 988    def multipart_form(self) -> multidict.MultiDictView[bytes, bytes]:
 989        """
 990        The multipart form data.
 991
 992        If the content-type indicates non-form data or the form could not be parsed, this is set to
 993        an empty `MultiDictView`.
 994
 995        Modifications to the MultiDictView update `Request.content`, and vice versa.
 996        """
 997        return multidict.MultiDictView(
 998            self._get_multipart_form, self._set_multipart_form
 999        )
1000
1001    @multipart_form.setter
1002    def multipart_form(self, value: list[tuple[bytes, bytes]]) -> None:
1003        self._set_multipart_form(value)
1004
1005
1006class Response(Message):
1007    """
1008    An HTTP response.
1009    """
1010
1011    data: ResponseData
1012
1013    def __init__(
1014        self,
1015        http_version: bytes,
1016        status_code: int,
1017        reason: bytes,
1018        headers: Headers | tuple[tuple[bytes, bytes], ...],
1019        content: bytes | None,
1020        trailers: None | Headers | tuple[tuple[bytes, bytes], ...],
1021        timestamp_start: float,
1022        timestamp_end: float | None,
1023    ):
1024        # auto-convert invalid types to retain compatibility with older code.
1025        if isinstance(http_version, str):
1026            http_version = http_version.encode("ascii", "strict")
1027        if isinstance(reason, str):
1028            reason = reason.encode("ascii", "strict")
1029
1030        if isinstance(content, str):
1031            raise ValueError(f"Content must be bytes, not {type(content).__name__}")
1032        if not isinstance(headers, Headers):
1033            headers = Headers(headers)
1034        if trailers is not None and not isinstance(trailers, Headers):
1035            trailers = Headers(trailers)
1036
1037        self.data = ResponseData(
1038            http_version=http_version,
1039            status_code=status_code,
1040            reason=reason,
1041            headers=headers,
1042            content=content,
1043            trailers=trailers,
1044            timestamp_start=timestamp_start,
1045            timestamp_end=timestamp_end,
1046        )
1047
1048    def __repr__(self) -> str:
1049        if self.raw_content:
1050            ct = self.headers.get("content-type", "unknown content type")
1051            size = human.pretty_size(len(self.raw_content))
1052            details = f"{ct}, {size}"
1053        else:
1054            details = "no content"
1055        return f"Response({self.status_code}, {details})"
1056
1057    @classmethod
1058    def make(
1059        cls,
1060        status_code: int = 200,
1061        content: bytes | str = b"",
1062        headers: (
1063            Headers | Mapping[str, str | bytes] | Iterable[tuple[bytes, bytes]]
1064        ) = (),
1065    ) -> "Response":
1066        """
1067        Simplified API for creating response objects.
1068        """
1069        if isinstance(headers, Headers):
1070            headers = headers
1071        elif isinstance(headers, dict):
1072            headers = Headers(
1073                (
1074                    always_bytes(k, "utf-8", "surrogateescape"),  # type: ignore
1075                    always_bytes(v, "utf-8", "surrogateescape"),
1076                )
1077                for k, v in headers.items()
1078            )
1079        elif isinstance(headers, Iterable):
1080            headers = Headers(headers)  # type: ignore
1081        else:
1082            raise TypeError(
1083                "Expected headers to be an iterable or dict, but is {}.".format(
1084                    type(headers).__name__
1085                )
1086            )
1087
1088        resp = cls(
1089            b"HTTP/1.1",
1090            status_code,
1091            status_codes.RESPONSES.get(status_code, "").encode(),
1092            headers,
1093            None,
1094            None,
1095            time.time(),
1096            time.time(),
1097        )
1098
1099        # Assign this manually to update the content-length header.
1100        if isinstance(content, bytes):
1101            resp.content = content
1102        elif isinstance(content, str):
1103            resp.text = content
1104        else:
1105            raise TypeError(
1106                f"Expected content to be str or bytes, but is {type(content).__name__}."
1107            )
1108
1109        return resp
1110
1111    @property
1112    def status_code(self) -> int:
1113        """
1114        HTTP Status Code, e.g. ``200``.
1115        """
1116        return self.data.status_code
1117
1118    @status_code.setter
1119    def status_code(self, status_code: int) -> None:
1120        self.data.status_code = status_code
1121
1122    @property
1123    def reason(self) -> str:
1124        """
1125        HTTP reason phrase, for example "Not Found".
1126
1127        HTTP/2 responses do not contain a reason phrase, an empty string will be returned instead.
1128        """
1129        # Encoding: http://stackoverflow.com/a/16674906/934719
1130        return self.data.reason.decode("ISO-8859-1")
1131
1132    @reason.setter
1133    def reason(self, reason: str | bytes) -> None:
1134        self.data.reason = strutils.always_bytes(reason, "ISO-8859-1")
1135
1136    def _get_cookies(self):
1137        h = self.headers.get_all("set-cookie")
1138        all_cookies = cookies.parse_set_cookie_headers(h)
1139        return tuple((name, (value, attrs)) for name, value, attrs in all_cookies)
1140
1141    def _set_cookies(self, value):
1142        cookie_headers = []
1143        for k, v in value:
1144            header = cookies.format_set_cookie_header([(k, v[0], v[1])])
1145            cookie_headers.append(header)
1146        self.headers.set_all("set-cookie", cookie_headers)
1147
1148    @property
1149    def cookies(
1150        self,
1151    ) -> multidict.MultiDictView[str, tuple[str, multidict.MultiDict[str, str | None]]]:
1152        """
1153        The response cookies. A possibly empty `MultiDictView`, where the keys are cookie
1154        name strings, and values are `(cookie value, attributes)` tuples. Within
1155        attributes, unary attributes (e.g. `HTTPOnly`) are indicated by a `None` value.
1156        Modifications to the MultiDictView update `Response.headers`, and vice versa.
1157
1158        *Warning:* Changes to `attributes` will not be picked up unless you also reassign
1159        the `(cookie value, attributes)` tuple directly in the `MultiDictView`.
1160        """
1161        return multidict.MultiDictView(self._get_cookies, self._set_cookies)
1162
1163    @cookies.setter
1164    def cookies(self, value):
1165        self._set_cookies(value)
1166
1167    def refresh(self, now=None):
1168        """
1169        This fairly complex and heuristic function refreshes a server
1170        response for replay.
1171
1172         - It adjusts date, expires, and last-modified headers.
1173         - It adjusts cookie expiration.
1174        """
1175        if not now:
1176            now = time.time()
1177        delta = now - self.timestamp_start
1178        refresh_headers = [
1179            "date",
1180            "expires",
1181            "last-modified",
1182        ]
1183        for i in refresh_headers:
1184            if i in self.headers:
1185                d = parsedate_tz(self.headers[i])
1186                if d:
1187                    new = mktime_tz(d) + delta
1188                    try:
1189                        self.headers[i] = formatdate(new, usegmt=True)
1190                    except OSError:  # pragma: no cover
1191                        pass  # value out of bounds on Windows only (which is why we exclude it from coverage).
1192        c = []
1193        for set_cookie_header in self.headers.get_all("set-cookie"):
1194            try:
1195                refreshed = cookies.refresh_set_cookie_header(set_cookie_header, delta)
1196            except ValueError:
1197                refreshed = set_cookie_header
1198            c.append(refreshed)
1199        if c:
1200            self.headers.set_all("set-cookie", c)
1201
1202
1203class HTTPFlow(flow.Flow):
1204    """
1205    An HTTPFlow is a collection of objects representing a single HTTP
1206    transaction.
1207    """
1208
1209    request: Request
1210    """The client's HTTP request."""
1211    response: Response | None = None
1212    """The server's HTTP response."""
1213    error: flow.Error | None = None
1214    """
1215    A connection or protocol error affecting this flow.
1216
1217    Note that it's possible for a Flow to have both a response and an error
1218    object. This might happen, for instance, when a response was received
1219    from the server, but there was an error sending it back to the client.
1220    """
1221
1222    websocket: WebSocketData | None = None
1223    """
1224    If this HTTP flow initiated a WebSocket connection, this attribute contains all associated WebSocket data.
1225    """
1226
1227    def get_state(self) -> serializable.State:
1228        return {
1229            **super().get_state(),
1230            "request": self.request.get_state(),
1231            "response": self.response.get_state() if self.response else None,
1232            "websocket": self.websocket.get_state() if self.websocket else None,
1233        }
1234
1235    def set_state(self, state: serializable.State) -> None:
1236        self.request = Request.from_state(state.pop("request"))
1237        self.response = Response.from_state(r) if (r := state.pop("response")) else None
1238        self.websocket = (
1239            WebSocketData.from_state(w) if (w := state.pop("websocket")) else None
1240        )
1241        super().set_state(state)
1242
1243    def __repr__(self):
1244        s = "<HTTPFlow"
1245        for a in (
1246            "request",
1247            "response",
1248            "websocket",
1249            "error",
1250            "client_conn",
1251            "server_conn",
1252        ):
1253            if getattr(self, a, False):
1254                s += f"\r\n  {a} = {{flow.{a}}}"
1255        s += ">"
1256        return s.format(flow=self)
1257
1258    @property
1259    def timestamp_start(self) -> float:
1260        """*Read-only:* An alias for `Request.timestamp_start`."""
1261        return self.request.timestamp_start
1262
1263    @property
1264    def mode(self) -> str:  # pragma: no cover
1265        warnings.warn("HTTPFlow.mode is deprecated.", DeprecationWarning, stacklevel=2)
1266        return getattr(self, "_mode", "regular")
1267
1268    @mode.setter
1269    def mode(self, val: str) -> None:  # pragma: no cover
1270        warnings.warn("HTTPFlow.mode is deprecated.", DeprecationWarning, stacklevel=2)
1271        self._mode = val
1272
1273    def copy(self):
1274        f = super().copy()
1275        if self.request:
1276            f.request = self.request.copy()
1277        if self.response:
1278            f.response = self.response.copy()
1279        return f
1280
1281
1282__all__ = [
1283    "HTTPFlow",
1284    "Message",
1285    "Request",
1286    "Response",
1287    "Headers",
1288]