1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
|
import unittest
from websockets.exceptions import InvalidHeaderFormat, InvalidHeaderValue
from websockets.headers import *
class HeadersTests(unittest.TestCase):
def test_build_host(self):
for (host, port, secure), (result, result_with_port) in [
(("localhost", 80, False), ("localhost", "localhost:80")),
(("localhost", 8000, False), ("localhost:8000", "localhost:8000")),
(("localhost", 443, True), ("localhost", "localhost:443")),
(("localhost", 8443, True), ("localhost:8443", "localhost:8443")),
(("example.com", 80, False), ("example.com", "example.com:80")),
(("example.com", 8000, False), ("example.com:8000", "example.com:8000")),
(("example.com", 443, True), ("example.com", "example.com:443")),
(("example.com", 8443, True), ("example.com:8443", "example.com:8443")),
(("127.0.0.1", 80, False), ("127.0.0.1", "127.0.0.1:80")),
(("127.0.0.1", 8000, False), ("127.0.0.1:8000", "127.0.0.1:8000")),
(("127.0.0.1", 443, True), ("127.0.0.1", "127.0.0.1:443")),
(("127.0.0.1", 8443, True), ("127.0.0.1:8443", "127.0.0.1:8443")),
(("::1", 80, False), ("[::1]", "[::1]:80")),
(("::1", 8000, False), ("[::1]:8000", "[::1]:8000")),
(("::1", 443, True), ("[::1]", "[::1]:443")),
(("::1", 8443, True), ("[::1]:8443", "[::1]:8443")),
]:
with self.subTest(host=host, port=port, secure=secure):
self.assertEqual(
build_host(host, port, secure),
result,
)
self.assertEqual(
build_host(host, port, secure, always_include_port=True),
result_with_port,
)
def test_parse_connection(self):
for header, parsed in [
# Realistic use cases
("Upgrade", ["Upgrade"]), # Safari, Chrome
("keep-alive, Upgrade", ["keep-alive", "Upgrade"]), # Firefox
# Pathological example
(",,\t, , ,Upgrade ,,", ["Upgrade"]),
]:
with self.subTest(header=header):
self.assertEqual(parse_connection(header), parsed)
def test_parse_connection_invalid_header_format(self):
for header in ["???", "keep-alive; Upgrade"]:
with self.subTest(header=header):
with self.assertRaises(InvalidHeaderFormat):
parse_connection(header)
def test_parse_upgrade(self):
for header, parsed in [
# Realistic use case
("websocket", ["websocket"]),
# Synthetic example
("http/3.0, websocket", ["http/3.0", "websocket"]),
# Pathological example
(",, WebSocket, \t,,", ["WebSocket"]),
]:
with self.subTest(header=header):
self.assertEqual(parse_upgrade(header), parsed)
def test_parse_upgrade_invalid_header_format(self):
for header in ["???", "websocket 2", "http/3.0; websocket"]:
with self.subTest(header=header):
with self.assertRaises(InvalidHeaderFormat):
parse_upgrade(header)
def test_parse_extension(self):
for header, parsed in [
# Synthetic examples
("foo", [("foo", [])]),
("foo, bar", [("foo", []), ("bar", [])]),
(
'foo; name; token=token; quoted-string="quoted-string", '
"bar; quux; quuux",
[
(
"foo",
[
("name", None),
("token", "token"),
("quoted-string", "quoted-string"),
],
),
("bar", [("quux", None), ("quuux", None)]),
],
),
# Pathological example
(
",\t, , ,foo ;bar = 42,, baz,,",
[("foo", [("bar", "42")]), ("baz", [])],
),
# Realistic use cases for permessage-deflate
("permessage-deflate", [("permessage-deflate", [])]),
(
"permessage-deflate; client_max_window_bits",
[("permessage-deflate", [("client_max_window_bits", None)])],
),
(
"permessage-deflate; server_max_window_bits=10",
[("permessage-deflate", [("server_max_window_bits", "10")])],
),
]:
with self.subTest(header=header):
self.assertEqual(parse_extension(header), parsed)
# Also ensure that build_extension round-trips cleanly.
unparsed = build_extension(parsed)
self.assertEqual(parse_extension(unparsed), parsed)
def test_parse_extension_invalid_header_format(self):
for header in [
# Truncated examples
"",
",\t,",
"foo;",
"foo; bar;",
"foo; bar=",
'foo; bar="baz',
# Wrong delimiter
"foo, bar, baz=quux; quuux",
# Value in quoted string parameter that isn't a token
'foo; bar=" "',
]:
with self.subTest(header=header):
with self.assertRaises(InvalidHeaderFormat):
parse_extension(header)
def test_parse_subprotocol(self):
for header, parsed in [
# Synthetic examples
("foo", ["foo"]),
("foo, bar", ["foo", "bar"]),
# Pathological example
(",\t, , ,foo ,, bar,baz,,", ["foo", "bar", "baz"]),
]:
with self.subTest(header=header):
self.assertEqual(parse_subprotocol(header), parsed)
# Also ensure that build_subprotocol round-trips cleanly.
unparsed = build_subprotocol(parsed)
self.assertEqual(parse_subprotocol(unparsed), parsed)
def test_parse_subprotocol_invalid_header(self):
for header in [
# Truncated examples
"",
",\t,",
# Wrong delimiter
"foo; bar",
]:
with self.subTest(header=header):
with self.assertRaises(InvalidHeaderFormat):
parse_subprotocol(header)
def test_validate_subprotocols(self):
for subprotocols in [[], ["sip"], ["v1.usp"], ["sip", "v1.usp"]]:
with self.subTest(subprotocols=subprotocols):
validate_subprotocols(subprotocols)
def test_validate_subprotocols_invalid(self):
for subprotocols, exception in [
({"sip": None}, TypeError),
("sip", TypeError),
([""], ValueError),
]:
with self.subTest(subprotocols=subprotocols):
with self.assertRaises(exception):
validate_subprotocols(subprotocols)
def test_build_www_authenticate_basic(self):
# Test vector from RFC 7617
self.assertEqual(
build_www_authenticate_basic("foo"), 'Basic realm="foo", charset="UTF-8"'
)
def test_build_www_authenticate_basic_invalid_realm(self):
# Realm contains a control character forbidden in quoted-string encoding
with self.assertRaises(ValueError):
build_www_authenticate_basic("\u0007")
def test_build_authorization_basic(self):
# Test vector from RFC 7617
self.assertEqual(
build_authorization_basic("Aladdin", "open sesame"),
"Basic QWxhZGRpbjpvcGVuIHNlc2FtZQ==",
)
def test_build_authorization_basic_utf8(self):
# Test vector from RFC 7617
self.assertEqual(
build_authorization_basic("test", "123£"), "Basic dGVzdDoxMjPCow=="
)
def test_parse_authorization_basic(self):
for header, parsed in [
("Basic QWxhZGRpbjpvcGVuIHNlc2FtZQ==", ("Aladdin", "open sesame")),
# Password contains non-ASCII character
("Basic dGVzdDoxMjPCow==", ("test", "123£")),
# Password contains a colon
("Basic YWxhZGRpbjpvcGVuOnNlc2FtZQ==", ("aladdin", "open:sesame")),
# Scheme name must be case insensitive
("basic QWxhZGRpbjpvcGVuIHNlc2FtZQ==", ("Aladdin", "open sesame")),
]:
with self.subTest(header=header):
self.assertEqual(parse_authorization_basic(header), parsed)
def test_parse_authorization_basic_invalid_header_format(self):
for header in [
"// Basic QWxhZGRpbjpvcGVuIHNlc2FtZQ==",
"Basic\tQWxhZGRpbjpvcGVuIHNlc2FtZQ==",
"Basic ****************************",
"Basic QWxhZGRpbjpvcGVuIHNlc2FtZQ== //",
]:
with self.subTest(header=header):
with self.assertRaises(InvalidHeaderFormat):
parse_authorization_basic(header)
def test_parse_authorization_basic_invalid_header_value(self):
for header in [
"Digest ...",
"Basic QWxhZGRpbjpvcGVuIHNlc2FtZQ",
"Basic QWxhZGNlc2FtZQ==",
]:
with self.subTest(header=header):
with self.assertRaises(InvalidHeaderValue):
parse_authorization_basic(header)
|
