mitmproxy.http

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