16
Self Samples
28
Cumulative
7
Lines Hit
0.10%
% of Total
5
Max Self
11
Max Total
Intensity:
Cold
Hot
Self Time
Total Time
Show All
Hot Only
Heat
Specialization
Line
Self
Total
Code
1
import re
2
from typing import AnyStr, cast, List, overload, Sequence, Tuple, TYPE_CHECKING, Union
3
4
from ._abnf import field_name, field_value
5
from ._util import bytesify, LocalProtocolError, validate
6
7
if TYPE_CHECKING:
8
from ._events import Request
9
10
try:
11
from typing import Literal
12
except ImportError:
13
from typing_extensions import Literal # type: ignore
14
15
CONTENT_LENGTH_MAX_DIGITS = 20 # allow up to 1 billion TB - 1
16
17
18
# Facts
19
# -----
20
#
21
# Headers are:
22
# keys: case-insensitive ascii
23
# values: mixture of ascii and raw bytes
24
#
25
# "Historically, HTTP has allowed field content with text in the ISO-8859-1
26
# charset [ISO-8859-1], supporting other charsets only through use of
27
# [RFC2047] encoding. In practice, most HTTP header field values use only a
28
# subset of the US-ASCII charset [USASCII]. Newly defined header fields SHOULD
29
# limit their field values to US-ASCII octets. A recipient SHOULD treat other
30
# octets in field content (obs-text) as opaque data."
31
# And it deprecates all non-ascii values
32
#
33
# Leading/trailing whitespace in header names is forbidden
34
#
35
# Values get leading/trailing whitespace stripped
36
#
37
# Content-Disposition actually needs to contain unicode semantically; to
38
# accomplish this it has a terrifically weird way of encoding the filename
39
# itself as ascii (and even this still has lots of cross-browser
40
# incompatibilities)
41
#
42
# Order is important:
43
# "a proxy MUST NOT change the order of these field values when forwarding a
44
# message"
45
# (and there are several headers where the order indicates a preference)
46
#
47
# Multiple occurences of the same header:
48
# "A sender MUST NOT generate multiple header fields with the same field name
49
# in a message unless either the entire field value for that header field is
50
# defined as a comma-separated list [or the header is Set-Cookie which gets a
51
# special exception]" - RFC 7230. (cookies are in RFC 6265)
52
#
53
# So every header aside from Set-Cookie can be merged by b", ".join if it
54
# occurs repeatedly. But, of course, they can't necessarily be split by
55
# .split(b","), because quoting.
56
#
57
# Given all this mess (case insensitive, duplicates allowed, order is
58
# important, ...), there doesn't appear to be any standard way to handle
59
# headers in Python -- they're almost like dicts, but... actually just
60
# aren't. For now we punt and just use a super simple representation: headers
61
# are a list of pairs
62
#
63
# [(name1, value1), (name2, value2), ...]
64
#
65
# where all entries are bytestrings, names are lowercase and have no
66
# leading/trailing whitespace, and values are bytestrings with no
67
# leading/trailing whitespace. Searching and updating are done via naive O(n)
68
# methods.
69
#
70
# Maybe a dict-of-lists would be better?
71
72
_content_length_re = re.compile(rb"[0-9]+")
73
_field_name_re = re.compile(field_name.encode("ascii"))
74
_field_value_re = re.compile(field_value.encode("ascii"))
75
76
77
class Headers(Sequence[Tuple[bytes, bytes]]):
78
"""
79
A list-like interface that allows iterating over headers as byte-pairs
80
of (lowercased-name, value).
81
82
Internally we actually store the representation as three-tuples,
83
including both the raw original casing, in order to preserve casing
84
over-the-wire, and the lowercased name, for case-insensitive comparisions.
85
86
r = Request(
87
method="GET",
88
target="/",
89
headers=[("Host", "example.org"), ("Connection", "keep-alive")],
90
http_version="1.1",
91
)
92
assert r.headers == [
93
(b"host", b"example.org"),
94
(b"connection", b"keep-alive")
95
]
96
assert r.headers.raw_items() == [
97
(b"Host", b"example.org"),
98
(b"Connection", b"keep-alive")
99
]
100
"""
101
102
__slots__ = "_full_items"
103
104
def __init__(self, full_items: List[Tuple[bytes, bytes, bytes]]) -> None:
105
self._full_items = full_items
106
107
def __bool__(self) -> bool:
108
return bool(self._full_items)
109
110
def __eq__(self, other: object) -> bool:
111
return list(self) == list(other) # type: ignore
112
113
def __len__(self) -> int:
114
return len(self._full_items)
115
116
def __repr__(self) -> str:
117
return "<Headers(%s)>" % repr(list(self))
118
119
def __getitem__(self, idx: int) -> Tuple[bytes, bytes]: # type: ignore[override]
120
_, name, value = self._full_items[idx]
121
return (name, value)
122
123
def raw_items(self) -> List[Tuple[bytes, bytes]]:
124
return [(raw_name, value) for raw_name, _, value in self._full_items]
125
126
127
HeaderTypes = Union[
128
List[Tuple[bytes, bytes]],
129
List[Tuple[bytes, str]],
130
List[Tuple[str, bytes]],
131
List[Tuple[str, str]],
132
]
133
134
135
@overload
136
def normalize_and_validate(headers: Headers, _parsed: Literal[True]) -> Headers:
137
...
138
139
140
@overload
141
def normalize_and_validate(headers: HeaderTypes, _parsed: Literal[False]) -> Headers:
142
...
143
144
145
@overload
146
def normalize_and_validate(
147
headers: Union[Headers, HeaderTypes], _parsed: bool = False
148
) -> Headers:
149
...
150
151
152
def normalize_and_validate(
153
headers: Union[Headers, HeaderTypes], _parsed: bool = False
154
) -> Headers:
155
new_headers = []
156
seen_content_length = None
157
saw_transfer_encoding = False
158
1
1
for name, value in headers:
159
# For headers coming out of the parser, we can safely skip some steps,
160
# because it always returns bytes and has already run these regexes
161
# over the data:
162
if not _parsed:
163
name = bytesify(name)
164
value = bytesify(value)
165
3
9
validate(_field_name_re, name, "Illegal header name {!r}", name)
166
5
11
validate(_field_value_re, value, "Illegal header value {!r}", value)
167
assert isinstance(name, bytes)
168
assert isinstance(value, bytes)
169
170
raw_name = name
171
name = name.lower()
172
1
1
if name == b"content-length":
173
1
1
lengths = {length.strip() for length in value.split(b",")}
174
if len(lengths) != 1:
175
raise LocalProtocolError("conflicting Content-Length headers")
176
value = lengths.pop()
177
validate(_content_length_re, value, "bad Content-Length")
178
if len(value) > CONTENT_LENGTH_MAX_DIGITS:
179
raise LocalProtocolError("bad Content-Length")
180
if seen_content_length is None:
181
seen_content_length = value
182
new_headers.append((raw_name, name, value))
183
elif seen_content_length != value:
184
raise LocalProtocolError("conflicting Content-Length headers")
185
elif name == b"transfer-encoding":
186
# "A server that receives a request message with a transfer coding
187
# it does not understand SHOULD respond with 501 (Not
188
# Implemented)."
189
# https://tools.ietf.org/html/rfc7230#section-3.3.1
190
if saw_transfer_encoding:
191
raise LocalProtocolError(
192
"multiple Transfer-Encoding headers", error_status_hint=501
193
)
194
# "All transfer-coding names are case-insensitive"
195
# -- https://tools.ietf.org/html/rfc7230#section-4
196
value = value.lower()
197
if value != b"chunked":
198
raise LocalProtocolError(
199
"Only Transfer-Encoding: chunked is supported",
200
error_status_hint=501,
201
)
202
saw_transfer_encoding = True
203
new_headers.append((raw_name, name, value))
204
else:
205
new_headers.append((raw_name, name, value))
206
3
3
return Headers(new_headers)
207
208
209
def get_comma_header(headers: Headers, name: bytes) -> List[bytes]:
210
# Should only be used for headers whose value is a list of
211
# comma-separated, case-insensitive values.
212
#
213
# The header name `name` is expected to be lower-case bytes.
214
#
215
# Connection: meets these criteria (including cast insensitivity).
216
#
217
# Content-Length: technically is just a single value (1*DIGIT), but the
218
# standard makes reference to implementations that do multiple values, and
219
# using this doesn't hurt. Ditto, case insensitivity doesn't things either
220
# way.
221
#
222
# Transfer-Encoding: is more complex (allows for quoted strings), so
223
# splitting on , is actually wrong. For example, this is legal:
224
#
225
# Transfer-Encoding: foo; options="1,2", chunked
226
#
227
# and should be parsed as
228
#
229
# foo; options="1,2"
230
# chunked
231
#
232
# but this naive function will parse it as
233
#
234
# foo; options="1
235
# 2"
236
# chunked
237
#
238
# However, this is okay because the only thing we are going to do with
239
# any Transfer-Encoding is reject ones that aren't just "chunked", so
240
# both of these will be treated the same anyway.
241
#
242
# Expect: the only legal value is the literal string
243
# "100-continue". Splitting on commas is harmless. Case insensitive.
244
#
245
out: List[bytes] = []
246
for _, found_name, found_raw_value in headers._full_items:
247
if found_name == name:
248
found_raw_value = found_raw_value.lower()
249
for found_split_value in found_raw_value.split(b","):
250
found_split_value = found_split_value.strip()
251
if found_split_value:
252
out.append(found_split_value)
253
return out
254
255
256
def set_comma_header(headers: Headers, name: bytes, new_values: List[bytes]) -> Headers:
257
# The header name `name` is expected to be lower-case bytes.
258
#
259
# Note that when we store the header we use title casing for the header
260
# names, in order to match the conventional HTTP header style.
261
#
262
# Simply calling `.title()` is a blunt approach, but it's correct
263
# here given the cases where we're using `set_comma_header`...
264
#
265
# Connection, Content-Length, Transfer-Encoding.
266
new_headers: List[Tuple[bytes, bytes]] = []
267
for found_raw_name, found_name, found_raw_value in headers._full_items:
268
if found_name != name:
269
new_headers.append((found_raw_name, found_raw_value))
270
for new_value in new_values:
271
new_headers.append((name.title(), new_value))
272
2
2
return normalize_and_validate(new_headers)
273
274
275
def has_expect_100_continue(request: "Request") -> bool:
276
# https://tools.ietf.org/html/rfc7231#section-5.1.1
277
# "A server that receives a 100-continue expectation in an HTTP/1.0 request
278
# MUST ignore that expectation."
279
if request.http_version < b"1.1":
280
return False
281
expect = get_comma_header(request.headers, b"expect")
282
return b"100-continue" in expect