1  import re
   2  from typing import AnyStr, cast, List, overload, Sequence, Tuple, TYPE_CHECKING, Union
   3  
   4  from ._abnf import field_name, field_value
   5  from ._util import bytesify, LocalProtocolError, validate
   6  
   7  if TYPE_CHECKING:
   8      from ._events import Request
   9  
  10  try:
  11      from typing import Literal
  12  except ImportError:
  13      from typing_extensions import Literal  # type: ignore
  14  
  15  
  16  # Facts
  17  # -----
  18  #
  19  # Headers are:
  20  #   keys: case-insensitive ascii
  21  #   values: mixture of ascii and raw bytes
  22  #
  23  # "Historically, HTTP has allowed field content with text in the ISO-8859-1
  24  # charset [ISO-8859-1], supporting other charsets only through use of
  25  # [RFC2047] encoding.  In practice, most HTTP header field values use only a
  26  # subset of the US-ASCII charset [USASCII]. Newly defined header fields SHOULD
  27  # limit their field values to US-ASCII octets.  A recipient SHOULD treat other
  28  # octets in field content (obs-text) as opaque data."
  29  # And it deprecates all non-ascii values
  30  #
  31  # Leading/trailing whitespace in header names is forbidden
  32  #
  33  # Values get leading/trailing whitespace stripped
  34  #
  35  # Content-Disposition actually needs to contain unicode semantically; to
  36  # accomplish this it has a terrifically weird way of encoding the filename
  37  # itself as ascii (and even this still has lots of cross-browser
  38  # incompatibilities)
  39  #
  40  # Order is important:
  41  # "a proxy MUST NOT change the order of these field values when forwarding a
  42  # message"
  43  # (and there are several headers where the order indicates a preference)
  44  #
  45  # Multiple occurences of the same header:
  46  # "A sender MUST NOT generate multiple header fields with the same field name
  47  # in a message unless either the entire field value for that header field is
  48  # defined as a comma-separated list [or the header is Set-Cookie which gets a
  49  # special exception]" - RFC 7230. (cookies are in RFC 6265)
  50  #
  51  # So every header aside from Set-Cookie can be merged by b", ".join if it
  52  # occurs repeatedly. But, of course, they can't necessarily be split by
  53  # .split(b","), because quoting.
  54  #
  55  # Given all this mess (case insensitive, duplicates allowed, order is
  56  # important, ...), there doesn't appear to be any standard way to handle
  57  # headers in Python -- they're almost like dicts, but... actually just
  58  # aren't. For now we punt and just use a super simple representation: headers
  59  # are a list of pairs
  60  #
  61  #   [(name1, value1), (name2, value2), ...]
  62  #
  63  # where all entries are bytestrings, names are lowercase and have no
  64  # leading/trailing whitespace, and values are bytestrings with no
  65  # leading/trailing whitespace. Searching and updating are done via naive O(n)
  66  # methods.
  67  #
  68  # Maybe a dict-of-lists would be better?
  69  
  70  _content_length_re = re.compile(rb"[0-9]+")
  71  _field_name_re = re.compile(field_name.encode("ascii"))
  72  _field_value_re = re.compile(field_value.encode("ascii"))
  73  
  74  
  75  class Headers(Sequence[Tuple[bytes, bytes]]):
  76      """
  77      A list-like interface that allows iterating over headers as byte-pairs
  78      of (lowercased-name, value).
  79  
  80      Internally we actually store the representation as three-tuples,
  81      including both the raw original casing, in order to preserve casing
  82      over-the-wire, and the lowercased name, for case-insensitive comparisions.
  83  
  84      r = Request(
  85          method="GET",
  86          target="/",
  87          headers=[("Host", "example.org"), ("Connection", "keep-alive")],
  88          http_version="1.1",
  89      )
  90      assert r.headers == [
  91          (b"host", b"example.org"),
  92          (b"connection", b"keep-alive")
  93      ]
  94      assert r.headers.raw_items() == [
  95          (b"Host", b"example.org"),
  96          (b"Connection", b"keep-alive")
  97      ]
  98      """
  99  
 100      __slots__ = "_full_items"
 101  
 102      def __init__(self, full_items: List[Tuple[bytes, bytes, bytes]]) -> None:
 103          self._full_items = full_items
 104  
 105      def __bool__(self) -> bool:
 106          return bool(self._full_items)
 107  
 108      def __eq__(self, other: object) -> bool:
 109          return list(self) == list(other)  # type: ignore
 110  
 111      def __len__(self) -> int:
 112          return len(self._full_items)
 113  
 114      def __repr__(self) -> str:
 115          return "<Headers(%s)>" % repr(list(self))
 116  
 117      def __getitem__(self, idx: int) -> Tuple[bytes, bytes]:  # type: ignore[override]
 118          _, name, value = self._full_items[idx]
 119          return (name, value)
 120  
 121      def raw_items(self) -> List[Tuple[bytes, bytes]]:
 122          return [(raw_name, value) for raw_name, _, value in self._full_items]
 123  
 124  
 125  HeaderTypes = Union[
 126      List[Tuple[bytes, bytes]],
 127      List[Tuple[bytes, str]],
 128      List[Tuple[str, bytes]],
 129      List[Tuple[str, str]],
 130  ]
 131  
 132  
 133  @overload
 134  def normalize_and_validate(headers: Headers, _parsed: Literal[True]) -> Headers:
 135      ...
 136  
 137  
 138  @overload
 139  def normalize_and_validate(headers: HeaderTypes, _parsed: Literal[False]) -> Headers:
 140      ...
 141  
 142  
 143  @overload
 144  def normalize_and_validate(
 145      headers: Union[Headers, HeaderTypes], _parsed: bool = False
 146  ) -> Headers:
 147      ...
 148  
 149  
 150  def normalize_and_validate(
 151      headers: Union[Headers, HeaderTypes], _parsed: bool = False
 152  ) -> Headers:
 153      new_headers = []
 154      seen_content_length = None
 155      saw_transfer_encoding = False
 156      for name, value in headers:
 157          # For headers coming out of the parser, we can safely skip some steps,
 158          # because it always returns bytes and has already run these regexes
 159          # over the data:
 160          if not _parsed:
 161              name = bytesify(name)
 162              value = bytesify(value)
 163              validate(_field_name_re, name, "Illegal header name {!r}", name)
 164              validate(_field_value_re, value, "Illegal header value {!r}", value)
 165          assert isinstance(name, bytes)
 166          assert isinstance(value, bytes)
 167  
 168          raw_name = name
 169          name = name.lower()
 170          if name == b"content-length":
 171              lengths = {length.strip() for length in value.split(b",")}
 172              if len(lengths) != 1:
 173                  raise LocalProtocolError("conflicting Content-Length headers")
 174              value = lengths.pop()
 175              validate(_content_length_re, value, "bad Content-Length")
 176              if seen_content_length is None:
 177                  seen_content_length = value
 178                  new_headers.append((raw_name, name, value))
 179              elif seen_content_length != value:
 180                  raise LocalProtocolError("conflicting Content-Length headers")
 181          elif name == b"transfer-encoding":
 182              # "A server that receives a request message with a transfer coding
 183              # it does not understand SHOULD respond with 501 (Not
 184              # Implemented)."
 185              # https://tools.ietf.org/html/rfc7230#section-3.3.1
 186              if saw_transfer_encoding:
 187                  raise LocalProtocolError(
 188                      "multiple Transfer-Encoding headers", error_status_hint=501
 189                  )
 190              # "All transfer-coding names are case-insensitive"
 191              # -- https://tools.ietf.org/html/rfc7230#section-4
 192              value = value.lower()
 193              if value != b"chunked":
 194                  raise LocalProtocolError(
 195                      "Only Transfer-Encoding: chunked is supported",
 196                      error_status_hint=501,
 197                  )
 198              saw_transfer_encoding = True
 199              new_headers.append((raw_name, name, value))
 200          else:
 201              new_headers.append((raw_name, name, value))
 202      return Headers(new_headers)
 203  
 204  
 205  def get_comma_header(headers: Headers, name: bytes) -> List[bytes]:
 206      # Should only be used for headers whose value is a list of
 207      # comma-separated, case-insensitive values.
 208      #
 209      # The header name `name` is expected to be lower-case bytes.
 210      #
 211      # Connection: meets these criteria (including cast insensitivity).
 212      #
 213      # Content-Length: technically is just a single value (1*DIGIT), but the
 214      # standard makes reference to implementations that do multiple values, and
 215      # using this doesn't hurt. Ditto, case insensitivity doesn't things either
 216      # way.
 217      #
 218      # Transfer-Encoding: is more complex (allows for quoted strings), so
 219      # splitting on , is actually wrong. For example, this is legal:
 220      #
 221      #    Transfer-Encoding: foo; options="1,2", chunked
 222      #
 223      # and should be parsed as
 224      #
 225      #    foo; options="1,2"
 226      #    chunked
 227      #
 228      # but this naive function will parse it as
 229      #
 230      #    foo; options="1
 231      #    2"
 232      #    chunked
 233      #
 234      # However, this is okay because the only thing we are going to do with
 235      # any Transfer-Encoding is reject ones that aren't just "chunked", so
 236      # both of these will be treated the same anyway.
 237      #
 238      # Expect: the only legal value is the literal string
 239      # "100-continue". Splitting on commas is harmless. Case insensitive.
 240      #
 241      out: List[bytes] = []
 242      for _, found_name, found_raw_value in headers._full_items:
 243          if found_name == name:
 244              found_raw_value = found_raw_value.lower()
 245              for found_split_value in found_raw_value.split(b","):
 246                  found_split_value = found_split_value.strip()
 247                  if found_split_value:
 248                      out.append(found_split_value)
 249      return out
 250  
 251  
 252  def set_comma_header(headers: Headers, name: bytes, new_values: List[bytes]) -> Headers:
 253      # The header name `name` is expected to be lower-case bytes.
 254      #
 255      # Note that when we store the header we use title casing for the header
 256      # names, in order to match the conventional HTTP header style.
 257      #
 258      # Simply calling `.title()` is a blunt approach, but it's correct
 259      # here given the cases where we're using `set_comma_header`...
 260      #
 261      # Connection, Content-Length, Transfer-Encoding.
 262      new_headers: List[Tuple[bytes, bytes]] = []
 263      for found_raw_name, found_name, found_raw_value in headers._full_items:
 264          if found_name != name:
 265              new_headers.append((found_raw_name, found_raw_value))
 266      for new_value in new_values:
 267          new_headers.append((name.title(), new_value))
 268      return normalize_and_validate(new_headers)
 269  
 270  
 271  def has_expect_100_continue(request: "Request") -> bool:
 272      # https://tools.ietf.org/html/rfc7231#section-5.1.1
 273      # "A server that receives a 100-continue expectation in an HTTP/1.0 request
 274      # MUST ignore that expectation."
 275      if request.http_version < b"1.1":
 276          return False
 277      expect = get_comma_header(request.headers, b"expect")
 278      return b"100-continue" in expect
 279