Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/websockets/datastructures.py: 59%

1from __future__ import annotations 

2 

3import re 

4from collections.abc import Iterable, Iterator, Mapping, MutableMapping 

5from typing import Any, Protocol 

6 

7 

8__all__ = [ 

9 "Headers", 

10 "HeadersLike", 

11 "MultipleValuesError", 

12] 

13 

14 

15class MultipleValuesError(LookupError): 

16 """ 

17 Exception raised when :class:`Headers` has multiple values for a key. 

18 

19 """ 

20 

21 def __str__(self) -> str: 

22 # Implement the same logic as KeyError_str in Objects/exceptions.c. 

23 if len(self.args) == 1: 

24 return repr(self.args[0]) 

25 return super().__str__() 

26 

27 

28# Same regex as http11._value_re, but for matching str rather than bytes. 

29is_valid_header_value = re.compile(r"[\x09\x20-\x7e\x80-\xff]*").fullmatch 

30 

31 

32class Headers(MutableMapping[str, str]): 

33 """ 

34 Efficient data structure for manipulating HTTP headers. 

35 

36 A :class:`list` of ``(name, values)`` is inefficient for lookups. 

37 

38 A :class:`dict` doesn't suffice because header names are case-insensitive 

39 and multiple occurrences of headers with the same name are possible. 

40 

41 :class:`Headers` stores HTTP headers in a hybrid data structure to provide 

42 efficient insertions and lookups while preserving the original data. 

43 

44 In order to account for multiple values with minimal hassle, 

45 :class:`Headers` follows this logic: 

46 

47 - When getting a header with ``headers[name]``: 

48 - if there's no value, :exc:`KeyError` is raised; 

49 - if there's exactly one value, it's returned; 

50 - if there's more than one value, :exc:`MultipleValuesError` is raised. 

51 

52 - When setting a header with ``headers[name] = value``, the value is 

53 appended to the list of values for that header. 

54 

55 - When deleting a header with ``del headers[name]``, all values for that 

56 header are removed (this is slow). 

57 

58 Other methods for manipulating headers are consistent with this logic. 

59 

60 As long as no header occurs multiple times, :class:`Headers` behaves like 

61 :class:`dict`, except keys are lower-cased to provide case-insensitivity. 

62 

63 Two methods support manipulating multiple values explicitly: 

64 

65 - :meth:`get_all` returns a list of all values for a header; 

66 - :meth:`raw_items` returns an iterator of ``(name, values)`` pairs. 

67 

68 Header names and values are expected to contain only ASCII text. However, 

69 non-ASCII values happen in practice, even though there is no standard for 

70 transmitting non-ASCII data in HTTP headers. :class:`Headers` supports it 

71 by treating it as ISO-8859-1 data. This is a safe and reversible encoding 

72 to represent arbitrary data in a :class:`str`. 

73 

74 When reading headers from the network, if the actual encoding isn't 

75 ISO-8859-1, you must re-encode and decode, e.g.:: 

76 

77 value = headers[key].encode("iso-8859-1").decode("utf-8") 

78 

79 Conversely, when sending headers to the network, if you need to use a 

80 different encoding, you can encode and decode, e.g.:: 

81 

82 headers[key] = value.encode("utf-8").decode("iso-8859-1") 

83 

84 When assigning a value to a header, as a security hardening measure, the 

85 value is checked for unsafe characters. The name isn't checked because it's 

86 usually a constant in code, unlikely to be tainted by user input. 

87 

88 """ 

89 

90 __slots__ = ["_dict", "_list"] 

91 

92 # Like dict, Headers accepts an optional "mapping or iterable" argument. 

93 def __init__(self, *args: HeadersLike, **kwargs: str) -> None: 

94 self._dict: dict[str, list[str]] = {} 

95 self._list: list[tuple[str, str]] = [] 

96 self.update(*args, **kwargs) 

97 

98 def __str__(self) -> str: 

99 return "".join(f"{key}: {value}\r\n" for key, value in self._list) + "\r\n" 

100 

101 def __repr__(self) -> str: 

102 return f"{self.__class__.__name__}({self._list!r})" 

103 

104 def copy(self) -> Headers: 

105 copy = self.__class__() 

106 copy._dict = self._dict.copy() 

107 copy._list = self._list.copy() 

108 return copy 

109 

110 def serialize(self) -> bytes: 

111 # parse_headers() supports non-ASCII header values. It decodes them as 

112 # ISO-8859-1. Encode back in ISO-8859-1 in order to round-trip cleanly. 

113 return str(self).encode("iso-8859-1") 

114 

115 # Collection methods 

116 

117 def __contains__(self, key: object) -> bool: 

118 return isinstance(key, str) and key.lower() in self._dict 

119 

120 def __iter__(self) -> Iterator[str]: 

121 return iter(self._dict) 

122 

123 def __len__(self) -> int: 

124 return len(self._dict) 

125 

126 # MutableMapping methods 

127 

128 def __getitem__(self, key: str) -> str: 

129 value = self._dict[key.lower()] 

130 if len(value) == 1: 

131 return value[0] 

132 else: 

133 raise MultipleValuesError(key) 

134 

135 def __setitem__(self, key: str, value: str) -> None: 

136 if not is_valid_header_value(str(value)): 

137 raise InvalidHeaderValue(key, value) 

138 self._dict.setdefault(key.lower(), []).append(value) 

139 self._list.append((key, value)) 

140 

141 def __delitem__(self, key: str) -> None: 

142 key_lower = key.lower() 

143 self._dict.__delitem__(key_lower) 

144 # This is inefficient. Fortunately deleting HTTP headers is uncommon. 

145 self._list = [(k, v) for k, v in self._list if k.lower() != key_lower] 

146 

147 def __eq__(self, other: Any) -> bool: 

148 if not isinstance(other, Headers): 

149 return NotImplemented 

150 return self._dict == other._dict 

151 

152 def clear(self) -> None: 

153 """ 

154 Remove all headers. 

155 

156 """ 

157 self._dict = {} 

158 self._list = [] 

159 

160 def update(self, *args: HeadersLike, **kwargs: str) -> None: 

161 """ 

162 Update from a :class:`Headers` instance and/or keyword arguments. 

163 

164 """ 

165 args = tuple( 

166 arg.raw_items() if isinstance(arg, Headers) else arg for arg in args 

167 ) 

168 super().update(*args, **kwargs) 

169 

170 # Methods for handling multiple values 

171 

172 def get_all(self, key: str) -> list[str]: 

173 """ 

174 Return the (possibly empty) list of all values for a header. 

175 

176 Args: 

177 key: Header name. 

178 

179 """ 

180 return self._dict.get(key.lower(), []) 

181 

182 def raw_items(self) -> Iterator[tuple[str, str]]: 

183 """ 

184 Return an iterator of all values as ``(name, value)`` pairs. 

185 

186 """ 

187 return iter(self._list) 

188 

189 # Internal methods 

190 

191 def set_insecure(self, key: str, value: str) -> None: 

192 """ 

193 Set a header without validating its value. 

194 

195 """ 

196 self._dict.setdefault(key.lower(), []).append(value) 

197 self._list.append((key, value)) 

198 

199 

200# copy of _typeshed.SupportsKeysAndGetItem. 

201class SupportsKeysAndGetItem(Protocol): 

202 """ 

203 Dict-like types with ``keys() -> str`` and ``__getitem__(key: str) -> str`` methods. 

204 

205 """ 

206 

207 def keys(self) -> Iterable[str]: ... # pragma: no branch 

208 

209 def __getitem__(self, key: str) -> str: ... # pragma: no branch 

210 

211 

212HeadersLike = ( 

213 Headers | Mapping[str, str] | Iterable[tuple[str, str]] | SupportsKeysAndGetItem 

214) 

215""" 

216Types accepted where :class:`Headers` is expected. 

217 

218In addition to :class:`Headers` itself, this includes dict-like types where both 

219keys and values are :class:`str`. 

220 

221""" 

222 

223 

224# At the bottom to break an import cycle. 

225from .exceptions import InvalidHeaderValue # noqa: E402