|
__all__ = [ |
|
"ProtocolError", |
|
"LocalProtocolError", |
|
"RemoteProtocolError", |
|
"validate", |
|
"make_sentinel", |
|
"bytesify", |
|
] |
|
|
|
|
|
class ProtocolError(Exception): |
|
"""Exception indicating a violation of the HTTP/1.1 protocol. |
|
|
|
This as an abstract base class, with two concrete base classes: |
|
:exc:`LocalProtocolError`, which indicates that you tried to do something |
|
that HTTP/1.1 says is illegal, and :exc:`RemoteProtocolError`, which |
|
indicates that the remote peer tried to do something that HTTP/1.1 says is |
|
illegal. See :ref:`error-handling` for details. |
|
|
|
In addition to the normal :exc:`Exception` features, it has one attribute: |
|
|
|
.. attribute:: error_status_hint |
|
|
|
This gives a suggestion as to what status code a server might use if |
|
this error occurred as part of a request. |
|
|
|
For a :exc:`RemoteProtocolError`, this is useful as a suggestion for |
|
how you might want to respond to a misbehaving peer, if you're |
|
implementing a server. |
|
|
|
For a :exc:`LocalProtocolError`, this can be taken as a suggestion for |
|
how your peer might have responded to *you* if h11 had allowed you to |
|
continue. |
|
|
|
The default is 400 Bad Request, a generic catch-all for protocol |
|
violations. |
|
|
|
""" |
|
|
|
def __init__(self, msg, error_status_hint=400): |
|
if type(self) is ProtocolError: |
|
raise TypeError("tried to directly instantiate ProtocolError") |
|
Exception.__init__(self, msg) |
|
self.error_status_hint = error_status_hint |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
class LocalProtocolError(ProtocolError): |
|
def _reraise_as_remote_protocol_error(self): |
|
|
|
|
|
|
|
|
|
|
|
|
|
self.__class__ = RemoteProtocolError |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
raise self |
|
|
|
|
|
class RemoteProtocolError(ProtocolError): |
|
pass |
|
|
|
|
|
def validate(regex, data, msg="malformed data", *format_args): |
|
match = regex.fullmatch(data) |
|
if not match: |
|
if format_args: |
|
msg = msg.format(*format_args) |
|
raise LocalProtocolError(msg) |
|
return match.groupdict() |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
class _SentinelBase(type): |
|
def __repr__(self): |
|
return self.__name__ |
|
|
|
|
|
def make_sentinel(name): |
|
cls = _SentinelBase(name, (_SentinelBase,), {}) |
|
cls.__class__ = cls |
|
return cls |
|
|
|
|
|
|
|
|
|
|
|
def bytesify(s): |
|
|
|
if type(s) is bytes: |
|
return s |
|
if isinstance(s, str): |
|
s = s.encode("ascii") |
|
if isinstance(s, int): |
|
raise TypeError("expected bytes-like object, not int") |
|
return bytes(s) |
|
|