codekingpro/portable-devtools

1from typing import Any, Dict, NoReturn, Pattern, Tuple, Type, TypeVar, Union2 3__all__ = [4    "ProtocolError",5    "LocalProtocolError",6    "RemoteProtocolError",7    "validate",8    "bytesify",9]10 11 12class 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 something17    that HTTP/1.1 says is illegal, and :exc:`RemoteProtocolError`, which18    indicates that the remote peer tried to do something that HTTP/1.1 says is19    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_hint24 25       This gives a suggestion as to what status code a server might use if26       this error occurred as part of a request.27 28       For a :exc:`RemoteProtocolError`, this is useful as a suggestion for29       how you might want to respond to a misbehaving peer, if you're30       implementing a server.31 32       For a :exc:`LocalProtocolError`, this can be taken as a suggestion for33       how your peer might have responded to *you* if h11 had allowed you to34       continue.35 36       The default is 400 Bad Request, a generic catch-all for protocol37       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_hint46 47 48# Strategy: there are a number of public APIs where a LocalProtocolError can49# be raised (send(), all the different event constructors, ...), and only one50# public API where RemoteProtocolError can be raised51# (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 for58#   remote errors.59class LocalProtocolError(ProtocolError):60    def _reraise_as_remote_protocol_error(self) -> NoReturn:61        # After catching a LocalProtocolError, use this method to re-raise it62        # as a RemoteProtocolError. This method must be called from inside an63        # except: block.64        #65        # An easy way to get an equivalent RemoteProtocolError is just to66        # modify 'self' in place.67        self.__class__ = RemoteProtocolError  # type: ignore68        # But the re-raising is somewhat non-trivial -- you might think that69        # now that we've modified the in-flight exception object, that just70        # doing 'raise' to re-raise it would be enough. But it turns out that71        # this doesn't work, because Python tracks the exception type72        # (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-raise74        # the new type explicitly.75        # On py3, the traceback is part of the exception object, so our76        # in-place modification preserved it and we can just re-raise:77        raise self78 79 80class RemoteProtocolError(ProtocolError):81    pass82 83 84def validate(85    regex: Pattern[bytes], data: bytes, msg: str = "malformed data", *format_args: Any86) -> 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 values96#97# - Inherit identity-based comparison and hashing from object98# - Have a nice repr99# - Have a *bonus property*: type(sentinel) is sentinel100#101# The bonus property is useful if you want to take the return value from102# next_event() and do some sort of dispatch based on type(event).103 104_T_Sentinel = TypeVar("_T_Sentinel", bound="Sentinel")105 106 107class Sentinel(type):108    def __new__(109        cls: Type[_T_Sentinel],110        name: str,111        bases: Tuple[type, ...],112        namespace: Dict[str, Any],113        **kwds: Any114    ) -> _T_Sentinel:115        assert bases == (Sentinel,)116        v = super().__new__(cls, name, bases, namespace, **kwds)117        v.__class__ = v  # type: ignore118        return v119 120    def __repr__(self) -> str:121        return self.__name__122 123 124# Used for methods, request targets, HTTP versions, header names, and header125# values. Accepts ascii-strings, or bytes/bytearray/memoryview/..., and always126# returns bytes.127def bytesify(s: Union[bytes, bytearray, memoryview, int, str]) -> bytes:128    # Fast-path:129    if type(s) is bytes:130        return s131    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