Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/h11/_util.py: 47%

1from 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 

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 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. 

59class 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 

80class RemoteProtocolError(ProtocolError): 

81 pass 

82 

83 

84def 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 

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: 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. 

127def 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)