1  from typing import Any, Dict, NoReturn, Pattern, Tuple, Type, TypeVar, Union
   2  
   3  __all__ = [
   4      "ProtocolError",
   5      "LocalProtocolError",
   6      "RemoteProtocolError",
   7      "validate",
   8      "bytesify",
   9  ]
  10  
  11  
  12  class ProtocolError(Exception):
  13      """Exception indicating a violation of the HTTP/1.1 protocol.
  14  
  15      This as an abstract base class, with two concrete base classes:
  16      :exc:`LocalProtocolError`, which indicates that you tried to do something
  17      that HTTP/1.1 says is illegal, and :exc:`RemoteProtocolError`, which
  18      indicates that the remote peer tried to do something that HTTP/1.1 says is
  19      illegal. See :ref:`error-handling` for details.
  20  
  21      In addition to the normal :exc:`Exception` features, it has one attribute:
  22  
  23      .. attribute:: error_status_hint
  24  
  25         This gives a suggestion as to what status code a server might use if
  26         this error occurred as part of a request.
  27  
  28         For a :exc:`RemoteProtocolError`, this is useful as a suggestion for
  29         how you might want to respond to a misbehaving peer, if you're
  30         implementing a server.
  31  
  32         For a :exc:`LocalProtocolError`, this can be taken as a suggestion for
  33         how your peer might have responded to *you* if h11 had allowed you to
  34         continue.
  35  
  36         The default is 400 Bad Request, a generic catch-all for protocol
  37         violations.
  38  
  39      """
  40  
  41      def __init__(self, msg: str, error_status_hint: int = 400) -> None:
  42          if type(self) is ProtocolError:
  43              raise TypeError("tried to directly instantiate ProtocolError")
  44          Exception.__init__(self, msg)
  45          self.error_status_hint = error_status_hint
  46  
  47  
  48  # Strategy: there are a number of public APIs where a LocalProtocolError can
  49  # be raised (send(), all the different event constructors, ...), and only one
  50  # public API where RemoteProtocolError can be raised
  51  # (receive_data()). Therefore we always raise LocalProtocolError internally,
  52  # and then receive_data will translate this into a RemoteProtocolError.
  53  #
  54  # Internally:
  55  #   LocalProtocolError is the generic "ProtocolError".
  56  # Externally:
  57  #   LocalProtocolError is for local errors and RemoteProtocolError is for
  58  #   remote errors.
  59  class LocalProtocolError(ProtocolError):
  60      def _reraise_as_remote_protocol_error(self) -> NoReturn:
  61          # After catching a LocalProtocolError, use this method to re-raise it
  62          # as a RemoteProtocolError. This method must be called from inside an
  63          # except: block.
  64          #
  65          # An easy way to get an equivalent RemoteProtocolError is just to
  66          # modify 'self' in place.
  67          self.__class__ = RemoteProtocolError  # type: ignore
  68          # But the re-raising is somewhat non-trivial -- you might think that
  69          # now that we've modified the in-flight exception object, that just
  70          # doing 'raise' to re-raise it would be enough. But it turns out that
  71          # this doesn't work, because Python tracks the exception type
  72          # (exc_info[0]) separately from the exception object (exc_info[1]),
  73          # and we only modified the latter. So we really do need to re-raise
  74          # the new type explicitly.
  75          # On py3, the traceback is part of the exception object, so our
  76          # in-place modification preserved it and we can just re-raise:
  77          raise self
  78  
  79  
  80  class RemoteProtocolError(ProtocolError):
  81      pass
  82  
  83  
  84  def validate(
  85      regex: Pattern[bytes], data: bytes, msg: str = "malformed data", *format_args: Any
  86  ) -> Dict[str, bytes]:
  87      match = regex.fullmatch(data)
  88      if not match:
  89          if format_args:
  90              msg = msg.format(*format_args)
  91          raise LocalProtocolError(msg)
  92      return match.groupdict()
  93  
  94  
  95  # Sentinel values
  96  #
  97  # - Inherit identity-based comparison and hashing from object
  98  # - Have a nice repr
  99  # - Have a *bonus property*: type(sentinel) is sentinel
 100  #
 101  # The bonus property is useful if you want to take the return value from
 102  # next_event() and do some sort of dispatch based on type(event).
 103  
 104  _T_Sentinel = TypeVar("_T_Sentinel", bound="Sentinel")
 105  
 106  
 107  class Sentinel(type):
 108      def __new__(
 109          cls: Type[_T_Sentinel],
 110          name: str,
 111          bases: Tuple[type, ...],
 112          namespace: Dict[str, Any],
 113          **kwds: Any
 114      ) -> _T_Sentinel:
 115          assert bases == (Sentinel,)
 116          v = super().__new__(cls, name, bases, namespace, **kwds)
 117          v.__class__ = v  # type: ignore
 118          return v
 119  
 120      def __repr__(self) -> str:
 121          return self.__name__
 122  
 123  
 124  # Used for methods, request targets, HTTP versions, header names, and header
 125  # values. Accepts ascii-strings, or bytes/bytearray/memoryview/..., and always
 126  # returns bytes.
 127  def bytesify(s: Union[bytes, bytearray, memoryview, int, str]) -> bytes:
 128      # Fast-path:
 129      if type(s) is bytes:
 130          return s
 131      if isinstance(s, str):
 132          s = s.encode("ascii")
 133      if isinstance(s, int):
 134          raise TypeError("expected bytes-like object, not int")
 135      return bytes(s)
 136