urllib3

1
from __future__ import annotations
2

3
import io
4
import logging
5
import socket
6
import ssl
7
import sys
8
import typing
9
import warnings
10
from itertools import chain
11
from test import ImportBlocker, ModuleStash, notBrotli, notZstd, onlyBrotli, onlyZstd
12
from unittest import mock
13
from unittest.mock import MagicMock, Mock, patch
14
from urllib.parse import urlparse
15

16
import pytest
17

18
from urllib3 import add_stderr_logger, disable_warnings
19
from urllib3.connection import ProxyConfig
20
from urllib3.exceptions import (
21
    InsecureRequestWarning,
22
    LocationParseError,
23
    TimeoutStateError,
24
    UnrewindableBodyError,
25
)
26
from urllib3.util import is_fp_closed
27
from urllib3.util.connection import _has_ipv6, allowed_gai_family, create_connection
28
from urllib3.util.proxy import connection_requires_http_tunnel
29
from urllib3.util.request import _FAILEDTELL, make_headers, rewind_body
30
from urllib3.util.response import assert_header_parsing
31
from urllib3.util.ssl_ import (
32
    _TYPE_VERSION_INFO,
33
    _is_has_never_check_common_name_reliable,
34
    resolve_cert_reqs,
35
    resolve_ssl_version,
36
    ssl_wrap_socket,
37
)
38
from urllib3.util.timeout import _DEFAULT_TIMEOUT, Timeout
39
from urllib3.util.url import Url, _encode_invalid_chars, parse_url
40
from urllib3.util.util import to_bytes, to_str
41

42
from . import clear_warnings
43

44
# This number represents a time in seconds, it doesn't mean anything in
45
# isolation. Setting to a high-ish value to avoid conflicts with the smaller
46
# numbers used for timeouts
47
TIMEOUT_EPOCH = 1000
48

49

50
class TestUtil:
51
    url_host_map = [
52
        # Hosts
53
        ("http://google.com/mail", ("http", "google.com", None)),
54
        ("http://google.com/mail/", ("http", "google.com", None)),
55
        ("google.com/mail", ("http", "google.com", None)),
56
        ("http://google.com/", ("http", "google.com", None)),
57
        ("http://google.com", ("http", "google.com", None)),
58
        ("http://www.google.com", ("http", "www.google.com", None)),
59
        ("http://mail.google.com", ("http", "mail.google.com", None)),
60
        ("http://google.com:8000/mail/", ("http", "google.com", 8000)),
61
        ("http://google.com:8000", ("http", "google.com", 8000)),
62
        ("https://google.com", ("https", "google.com", None)),
63
        ("https://google.com:8000", ("https", "google.com", 8000)),
64
        ("http://user:password@127.0.0.1:1234", ("http", "127.0.0.1", 1234)),
65
        ("http://google.com/foo=http://bar:42/baz", ("http", "google.com", None)),
66
        ("http://google.com?foo=http://bar:42/baz", ("http", "google.com", None)),
67
        ("http://google.com#foo=http://bar:42/baz", ("http", "google.com", None)),
68
        # IPv4
69
        ("173.194.35.7", ("http", "173.194.35.7", None)),
70
        ("http://173.194.35.7", ("http", "173.194.35.7", None)),
71
        ("http://173.194.35.7/test", ("http", "173.194.35.7", None)),
72
        ("http://173.194.35.7:80", ("http", "173.194.35.7", 80)),
73
        ("http://173.194.35.7:80/test", ("http", "173.194.35.7", 80)),
74
        # IPv6
75
        ("[2a00:1450:4001:c01::67]", ("http", "[2a00:1450:4001:c01::67]", None)),
76
        ("http://[2a00:1450:4001:c01::67]", ("http", "[2a00:1450:4001:c01::67]", None)),
77
        (
78
            "http://[2a00:1450:4001:c01::67]/test",
79
            ("http", "[2a00:1450:4001:c01::67]", None),
80
        ),
81
        (
82
            "http://[2a00:1450:4001:c01::67]:80",
83
            ("http", "[2a00:1450:4001:c01::67]", 80),
84
        ),
85
        (
86
            "http://[2a00:1450:4001:c01::67]:80/test",
87
            ("http", "[2a00:1450:4001:c01::67]", 80),
88
        ),
89
        # More IPv6 from http://www.ietf.org/rfc/rfc2732.txt
90
        (
91
            "http://[fedc:ba98:7654:3210:fedc:ba98:7654:3210]:8000/index.html",
92
            ("http", "[fedc:ba98:7654:3210:fedc:ba98:7654:3210]", 8000),
93
        ),
94
        (
95
            "http://[1080:0:0:0:8:800:200c:417a]/index.html",
96
            ("http", "[1080:0:0:0:8:800:200c:417a]", None),
97
        ),
98
        ("http://[3ffe:2a00:100:7031::1]", ("http", "[3ffe:2a00:100:7031::1]", None)),
99
        (
100
            "http://[1080::8:800:200c:417a]/foo",
101
            ("http", "[1080::8:800:200c:417a]", None),
102
        ),
103
        ("http://[::192.9.5.5]/ipng", ("http", "[::192.9.5.5]", None)),
104
        (
105
            "http://[::ffff:129.144.52.38]:42/index.html",
106
            ("http", "[::ffff:129.144.52.38]", 42),
107
        ),
108
        (
109
            "http://[2010:836b:4179::836b:4179]",
110
            ("http", "[2010:836b:4179::836b:4179]", None),
111
        ),
112
        # Scoped IPv6 (with ZoneID), both RFC 6874 compliant and not.
113
        ("http://[a::b%25zone]", ("http", "[a::b%zone]", None)),
114
        ("http://[a::b%zone]", ("http", "[a::b%zone]", None)),
115
        # Hosts
116
        ("HTTP://GOOGLE.COM/mail/", ("http", "google.com", None)),
117
        ("GOogle.COM/mail", ("http", "google.com", None)),
118
        ("HTTP://GoOgLe.CoM:8000/mail/", ("http", "google.com", 8000)),
119
        ("HTTP://user:password@EXAMPLE.COM:1234", ("http", "example.com", 1234)),
120
        ("173.194.35.7", ("http", "173.194.35.7", None)),
121
        ("HTTP://173.194.35.7", ("http", "173.194.35.7", None)),
122
        (
123
            "HTTP://[2a00:1450:4001:c01::67]:80/test",
124
            ("http", "[2a00:1450:4001:c01::67]", 80),
125
        ),
126
        (
127
            "HTTP://[FEDC:BA98:7654:3210:FEDC:BA98:7654:3210]:8000/index.html",
128
            ("http", "[fedc:ba98:7654:3210:fedc:ba98:7654:3210]", 8000),
129
        ),
130
        (
131
            "HTTPS://[1080:0:0:0:8:800:200c:417A]/index.html",
132
            ("https", "[1080:0:0:0:8:800:200c:417a]", None),
133
        ),
134
        ("abOut://eXamPlE.com?info=1", ("about", "eXamPlE.com", None)),
135
        (
136
            "http+UNIX://%2fvar%2frun%2fSOCKET/path",
137
            ("http+unix", "%2fvar%2frun%2fSOCKET", None),
138
        ),
139
    ]
140

141
    @pytest.mark.parametrize(["url", "scheme_host_port"], url_host_map)
142
    def test_scheme_host_port(
143
        self, url: str, scheme_host_port: tuple[str, str, int | None]
144
    ) -> None:
145
        parsed_url = parse_url(url)
146
        scheme, host, port = scheme_host_port
147

148
        assert (parsed_url.scheme or "http") == scheme
149
        assert parsed_url.hostname == parsed_url.host == host
150
        assert parsed_url.port == port
151

152
    def test_encode_invalid_chars_none(self) -> None:
153
        assert _encode_invalid_chars(None, set()) is None
154

155
    @pytest.mark.parametrize(
156
        "url",
157
        [
158
            "http://google.com:foo",
159
            "http://::1/",
160
            "http://::1:80/",
161
            "http://google.com:-80",
162
            "http://google.com:65536",
163
            "http://google.com:\xb2\xb2",  # \xb2 = ^2
164
            # Invalid IDNA labels
165
            "http://\uD7FF.com",
166
            "http://❤️",
167
            # Unicode surrogates
168
            "http://\uD800.com",
169
            "http://\uDC00.com",
170
        ],
171
    )
172
    def test_invalid_url(self, url: str) -> None:
173
        with pytest.raises(LocationParseError):
174
            parse_url(url)
175

176
    @pytest.mark.parametrize(
177
        "url, expected_normalized_url",
178
        [
179
            ("HTTP://GOOGLE.COM/MAIL/", "http://google.com/MAIL/"),
180
            (
181
                "http://user@domain.com:password@example.com/~tilde@?@",
182
                "http://user%40domain.com:password@example.com/~tilde@?@",
183
            ),
184
            (
185
                "HTTP://JeremyCline:Hunter2@Example.com:8080/",
186
                "http://JeremyCline:Hunter2@example.com:8080/",
187
            ),
188
            ("HTTPS://Example.Com/?Key=Value", "https://example.com/?Key=Value"),
189
            ("Https://Example.Com/#Fragment", "https://example.com/#Fragment"),
190
            # IPv6 addresses with zone IDs. Both RFC 6874 (%25) as well as
191
            # non-standard (unquoted %) variants.
192
            ("[::1%zone]", "[::1%zone]"),
193
            ("[::1%25zone]", "[::1%zone]"),
194
            ("[::1%25]", "[::1%25]"),
195
            ("[::Ff%etH0%Ff]/%ab%Af", "[::ff%etH0%FF]/%AB%AF"),
196
            (
197
                "http://user:pass@[AaAa::Ff%25etH0%Ff]/%ab%Af",
198
                "http://user:pass@[aaaa::ff%etH0%FF]/%AB%AF",
199
            ),
200
            # Invalid characters for the query/fragment getting encoded
201
            (
202
                'http://google.com/p[]?parameter[]="hello"#fragment#',
203
                "http://google.com/p%5B%5D?parameter%5B%5D=%22hello%22#fragment%23",
204
            ),
205
            # Percent encoding isn't applied twice despite '%' being invalid
206
            # but the percent encoding is still normalized.
207
            (
208
                "http://google.com/p%5B%5d?parameter%5b%5D=%22hello%22#fragment%23",
209
                "http://google.com/p%5B%5D?parameter%5B%5D=%22hello%22#fragment%23",
210
            ),
211
        ],
212
    )
213
    def test_parse_url_normalization(
214
        self, url: str, expected_normalized_url: str
215
    ) -> None:
216
        """Assert parse_url normalizes the scheme/host, and only the scheme/host"""
217
        actual_normalized_url = parse_url(url).url
218
        assert actual_normalized_url == expected_normalized_url
219

220
    @pytest.mark.parametrize("char", [chr(i) for i in range(0x00, 0x21)] + ["\x7F"])
221
    def test_control_characters_are_percent_encoded(self, char: str) -> None:
222
        percent_char = "%" + (hex(ord(char))[2:].zfill(2).upper())
223
        url = parse_url(
224
            f"http://user{char}@example.com/path{char}?query{char}#fragment{char}"
225
        )
226

227
        assert url == Url(
228
            "http",
229
            auth="user" + percent_char,
230
            host="example.com",
231
            path="/path" + percent_char,
232
            query="query" + percent_char,
233
            fragment="fragment" + percent_char,
234
        )
235

236
    parse_url_host_map = [
237
        ("http://google.com/mail", Url("http", host="google.com", path="/mail")),
238
        ("http://google.com/mail/", Url("http", host="google.com", path="/mail/")),
239
        ("http://google.com/mail", Url("http", host="google.com", path="mail")),
240
        ("google.com/mail", Url(host="google.com", path="/mail")),
241
        ("http://google.com/", Url("http", host="google.com", path="/")),
242
        ("http://google.com", Url("http", host="google.com")),
243
        ("http://google.com?foo", Url("http", host="google.com", path="", query="foo")),
244
        # Path/query/fragment
245
        ("", Url()),
246
        ("/", Url(path="/")),
247
        ("#?/!google.com/?foo", Url(path="", fragment="?/!google.com/?foo")),
248
        ("/foo", Url(path="/foo")),
249
        ("/foo?bar=baz", Url(path="/foo", query="bar=baz")),
250
        (
251
            "/foo?bar=baz#banana?apple/orange",
252
            Url(path="/foo", query="bar=baz", fragment="banana?apple/orange"),
253
        ),
254
        (
255
            "/redirect?target=http://localhost:61020/",
256
            Url(path="redirect", query="target=http://localhost:61020/"),
257
        ),
258
        # Port
259
        ("http://google.com/", Url("http", host="google.com", path="/")),
260
        ("http://google.com:80/", Url("http", host="google.com", port=80, path="/")),
261
        ("http://google.com:80", Url("http", host="google.com", port=80)),
262
        # Auth
263
        (
264
            "http://foo:bar@localhost/",
265
            Url("http", auth="foo:bar", host="localhost", path="/"),
266
        ),
267
        ("http://foo@localhost/", Url("http", auth="foo", host="localhost", path="/")),
268
        (
269
            "http://foo:bar@localhost/",
270
            Url("http", auth="foo:bar", host="localhost", path="/"),
271
        ),
272
    ]
273

274
    non_round_tripping_parse_url_host_map = [
275
        # Path/query/fragment
276
        ("?", Url(path="", query="")),
277
        ("#", Url(path="", fragment="")),
278
        # Path normalization
279
        ("/abc/../def", Url(path="/def")),
280
        # Empty Port
281
        ("http://google.com:", Url("http", host="google.com")),
282
        ("http://google.com:/", Url("http", host="google.com", path="/")),
283
        # Uppercase IRI
284
        (
285
            "http://Königsgäßchen.de/straße",
286
            Url("http", host="xn--knigsgchen-b4a3dun.de", path="/stra%C3%9Fe"),
287
        ),
288
        # Percent-encode in userinfo
289
        (
290
            "http://user@email.com:password@example.com/",
291
            Url("http", auth="user%40email.com:password", host="example.com", path="/"),
292
        ),
293
        (
294
            'http://user":quoted@example.com/',
295
            Url("http", auth="user%22:quoted", host="example.com", path="/"),
296
        ),
297
        # Unicode Surrogates
298
        ("http://google.com/\uD800", Url("http", host="google.com", path="%ED%A0%80")),
299
        (
300
            "http://google.com?q=\uDC00",
301
            Url("http", host="google.com", path="", query="q=%ED%B0%80"),
302
        ),
303
        (
304
            "http://google.com#\uDC00",
305
            Url("http", host="google.com", path="", fragment="%ED%B0%80"),
306
        ),
307
    ]
308

309
    @pytest.mark.parametrize(
310
        "url, expected_url",
311
        chain(parse_url_host_map, non_round_tripping_parse_url_host_map),
312
    )
313
    def test_parse_url(self, url: str, expected_url: Url) -> None:
314
        returned_url = parse_url(url)
315
        assert returned_url == expected_url
316
        assert returned_url.hostname == returned_url.host == expected_url.host
317

318
    @pytest.mark.parametrize("url, expected_url", parse_url_host_map)
319
    def test_unparse_url(self, url: str, expected_url: Url) -> None:
320
        assert url == expected_url.url
321

322
    @pytest.mark.parametrize(
323
        ["url", "expected_url"],
324
        [
325
            # RFC 3986 5.2.4
326
            ("/abc/../def", Url(path="/def")),
327
            ("/..", Url(path="/")),
328
            ("/./abc/./def/", Url(path="/abc/def/")),
329
            ("/.", Url(path="/")),
330
            ("/./", Url(path="/")),
331
            ("/abc/./.././d/././e/.././f/./../../ghi", Url(path="/ghi")),
332
        ],
333
    )
334
    def test_parse_and_normalize_url_paths(self, url: str, expected_url: Url) -> None:
335
        actual_url = parse_url(url)
336
        assert actual_url == expected_url
337
        assert actual_url.url == expected_url.url
338

339
    def test_parse_url_invalid_IPv6(self) -> None:
340
        with pytest.raises(LocationParseError):
341
            parse_url("[::1")
342

343
    def test_parse_url_negative_port(self) -> None:
344
        with pytest.raises(LocationParseError):
345
            parse_url("https://www.google.com:-80/")
346

347
    def test_parse_url_remove_leading_zeros(self) -> None:
348
        url = parse_url("https://example.com:0000000000080")
349
        assert url.port == 80
350

351
    def test_parse_url_only_zeros(self) -> None:
352
        url = parse_url("https://example.com:0")
353
        assert url.port == 0
354

355
        url = parse_url("https://example.com:000000000000")
356
        assert url.port == 0
357

358
    def test_Url_str(self) -> None:
359
        U = Url("http", host="google.com")
360
        assert str(U) == U.url
361

362
    request_uri_map = [
363
        ("http://google.com/mail", "/mail"),
364
        ("http://google.com/mail/", "/mail/"),
365
        ("http://google.com/", "/"),
366
        ("http://google.com", "/"),
367
        ("", "/"),
368
        ("/", "/"),
369
        ("?", "/?"),
370
        ("#", "/"),
371
        ("/foo?bar=baz", "/foo?bar=baz"),
372
    ]
373

374
    @pytest.mark.parametrize("url, expected_request_uri", request_uri_map)
375
    def test_request_uri(self, url: str, expected_request_uri: str) -> None:
376
        returned_url = parse_url(url)
377
        assert returned_url.request_uri == expected_request_uri
378

379
    url_authority_map: list[tuple[str, str | None]] = [
380
        ("http://user:pass@google.com/mail", "user:pass@google.com"),
381
        ("http://user:pass@google.com:80/mail", "user:pass@google.com:80"),
382
        ("http://user@google.com:80/mail", "user@google.com:80"),
383
        ("http://user:pass@192.168.1.1/path", "user:pass@192.168.1.1"),
384
        ("http://user:pass@192.168.1.1:80/path", "user:pass@192.168.1.1:80"),
385
        ("http://user@192.168.1.1:80/path", "user@192.168.1.1:80"),
386
        ("http://user:pass@[::1]/path", "user:pass@[::1]"),
387
        ("http://user:pass@[::1]:80/path", "user:pass@[::1]:80"),
388
        ("http://user@[::1]:80/path", "user@[::1]:80"),
389
        ("http://user:pass@localhost/path", "user:pass@localhost"),
390
        ("http://user:pass@localhost:80/path", "user:pass@localhost:80"),
391
        ("http://user@localhost:80/path", "user@localhost:80"),
392
    ]
393

394
    url_netloc_map = [
395
        ("http://google.com/mail", "google.com"),
396
        ("http://google.com:80/mail", "google.com:80"),
397
        ("http://192.168.0.1/path", "192.168.0.1"),
398
        ("http://192.168.0.1:80/path", "192.168.0.1:80"),
399
        ("http://[::1]/path", "[::1]"),
400
        ("http://[::1]:80/path", "[::1]:80"),
401
        ("http://localhost", "localhost"),
402
        ("http://localhost:80", "localhost:80"),
403
        ("google.com/foobar", "google.com"),
404
        ("google.com:12345", "google.com:12345"),
405
        ("/", None),
406
    ]
407

408
    combined_netloc_authority_map = url_authority_map + url_netloc_map
409

410
    # We compose this list due to variances between parse_url
411
    # and urlparse when URIs don't provide a scheme.
412
    url_authority_with_schemes_map = [
413
        u for u in combined_netloc_authority_map if u[0].startswith("http")
414
    ]
415

416
    @pytest.mark.parametrize("url, expected_authority", combined_netloc_authority_map)
417
    def test_authority(self, url: str, expected_authority: str | None) -> None:
418
        assert parse_url(url).authority == expected_authority
419

420
    @pytest.mark.parametrize("url, expected_authority", url_authority_with_schemes_map)
421
    def test_authority_matches_urllib_netloc(
422
        self, url: str, expected_authority: str | None
423
    ) -> None:
424
        """Validate this matches the behavior of urlparse().netloc"""
425
        assert urlparse(url).netloc == expected_authority
426

427
    @pytest.mark.parametrize("url, expected_netloc", url_netloc_map)
428
    def test_netloc(self, url: str, expected_netloc: str | None) -> None:
429
        assert parse_url(url).netloc == expected_netloc
430

431
    url_vulnerabilities = [
432
        # urlparse doesn't follow RFC 3986 Section 3.2
433
        (
434
            "http://google.com#@evil.com/",
435
            Url("http", host="google.com", path="", fragment="@evil.com/"),
436
        ),
437
        # CVE-2016-5699
438
        (
439
            "http://127.0.0.1%0d%0aConnection%3a%20keep-alive",
440
            Url("http", host="127.0.0.1%0d%0aconnection%3a%20keep-alive"),
441
        ),
442
        # NodeJS unicode -> double dot
443
        (
444
            "http://google.com/\uff2e\uff2e/abc",
445
            Url("http", host="google.com", path="/%EF%BC%AE%EF%BC%AE/abc"),
446
        ),
447
        # Scheme without ://
448
        (
449
            "javascript:a='@google.com:12345/';alert(0)",
450
            Url(scheme="javascript", path="a='@google.com:12345/';alert(0)"),
451
        ),
452
        ("//google.com/a/b/c", Url(host="google.com", path="/a/b/c")),
453
        # International URLs
454
        (
455
            "http://ヒ:キ@ヒ.abc.ニ/ヒ?キ#ワ",
456
            Url(
457
                "http",
458
                host="xn--pdk.abc.xn--idk",
459
                auth="%E3%83%92:%E3%82%AD",
460
                path="/%E3%83%92",
461
                query="%E3%82%AD",
462
                fragment="%E3%83%AF",
463
            ),
464
        ),
465
        # Injected headers (CVE-2016-5699, CVE-2019-9740, CVE-2019-9947)
466
        (
467
            "10.251.0.83:7777?a=1 HTTP/1.1\r\nX-injected: header",
468
            Url(
469
                host="10.251.0.83",
470
                port=7777,
471
                path="",
472
                query="a=1%20HTTP/1.1%0D%0AX-injected:%20header",
473
            ),
474
        ),
475
        (
476
            "http://127.0.0.1:6379?\r\nSET test failure12\r\n:8080/test/?test=a",
477
            Url(
478
                scheme="http",
479
                host="127.0.0.1",
480
                port=6379,
481
                path="",
482
                query="%0D%0ASET%20test%20failure12%0D%0A:8080/test/?test=a",
483
            ),
484
        ),
485
        # See https://bugs.xdavidhu.me/google/2020/03/08/the-unexpected-google-wide-domain-check-bypass/
486
        (
487
            "https://user:pass@xdavidhu.me\\test.corp.google.com:8080/path/to/something?param=value#hash",
488
            Url(
489
                scheme="https",
490
                auth="user:pass",
491
                host="xdavidhu.me",
492
                path="/%5Ctest.corp.google.com:8080/path/to/something",
493
                query="param=value",
494
                fragment="hash",
495
            ),
496
        ),
497
        # Tons of '@' causing backtracking
498
        pytest.param(
499
            "https://" + ("@" * 10000) + "[",
500
            False,
501
            id="Tons of '@' causing backtracking 1",
502
        ),
503
        pytest.param(
504
            "https://user:" + ("@" * 10000) + "example.com",
505
            Url(
506
                scheme="https",
507
                auth="user:" + ("%40" * 9999),
508
                host="example.com",
509
            ),
510
            id="Tons of '@' causing backtracking 2",
511
        ),
512
    ]
513

514
    @pytest.mark.parametrize("url, expected_url", url_vulnerabilities)
515
    def test_url_vulnerabilities(
516
        self, url: str, expected_url: typing.Literal[False] | Url
517
    ) -> None:
518
        if expected_url is False:
519
            with pytest.raises(LocationParseError):
520
                parse_url(url)
521
        else:
522
            assert parse_url(url) == expected_url
523

524
    def test_parse_url_bytes_type_error(self) -> None:
525
        with pytest.raises(TypeError):
526
            parse_url(b"https://www.google.com/")  # type: ignore[arg-type]
527

528
    @pytest.mark.parametrize(
529
        "kwargs, expected",
530
        [
531
            pytest.param(
532
                {"accept_encoding": True},
533
                {"accept-encoding": "gzip,deflate,br,zstd"},
534
                marks=[onlyBrotli(), onlyZstd()],  # type: ignore[list-item]
535
            ),
536
            pytest.param(
537
                {"accept_encoding": True},
538
                {"accept-encoding": "gzip,deflate,br"},
539
                marks=[onlyBrotli(), notZstd()],  # type: ignore[list-item]
540
            ),
541
            pytest.param(
542
                {"accept_encoding": True},
543
                {"accept-encoding": "gzip,deflate,zstd"},
544
                marks=[notBrotli(), onlyZstd()],  # type: ignore[list-item]
545
            ),
546
            pytest.param(
547
                {"accept_encoding": True},
548
                {"accept-encoding": "gzip,deflate"},
549
                marks=[notBrotli(), notZstd()],  # type: ignore[list-item]
550
            ),
551
            ({"accept_encoding": "foo,bar"}, {"accept-encoding": "foo,bar"}),
552
            ({"accept_encoding": ["foo", "bar"]}, {"accept-encoding": "foo,bar"}),
553
            pytest.param(
554
                {"accept_encoding": True, "user_agent": "banana"},
555
                {"accept-encoding": "gzip,deflate,br,zstd", "user-agent": "banana"},
556
                marks=[onlyBrotli(), onlyZstd()],  # type: ignore[list-item]
557
            ),
558
            pytest.param(
559
                {"accept_encoding": True, "user_agent": "banana"},
560
                {"accept-encoding": "gzip,deflate,br", "user-agent": "banana"},
561
                marks=[onlyBrotli(), notZstd()],  # type: ignore[list-item]
562
            ),
563
            pytest.param(
564
                {"accept_encoding": True, "user_agent": "banana"},
565
                {"accept-encoding": "gzip,deflate,zstd", "user-agent": "banana"},
566
                marks=[notBrotli(), onlyZstd()],  # type: ignore[list-item]
567
            ),
568
            pytest.param(
569
                {"accept_encoding": True, "user_agent": "banana"},
570
                {"accept-encoding": "gzip,deflate", "user-agent": "banana"},
571
                marks=[notBrotli(), notZstd()],  # type: ignore[list-item]
572
            ),
573
            ({"user_agent": "banana"}, {"user-agent": "banana"}),
574
            ({"keep_alive": True}, {"connection": "keep-alive"}),
575
            ({"basic_auth": "foo:bar"}, {"authorization": "Basic Zm9vOmJhcg=="}),
576
            (
577
                {"proxy_basic_auth": "foo:bar"},
578
                {"proxy-authorization": "Basic Zm9vOmJhcg=="},
579
            ),
580
            ({"disable_cache": True}, {"cache-control": "no-cache"}),
581
        ],
582
    )
583
    def test_make_headers(
584
        self, kwargs: dict[str, bool | str], expected: dict[str, str]
585
    ) -> None:
586
        assert make_headers(**kwargs) == expected  # type: ignore[arg-type]
587

588
    def test_rewind_body(self) -> None:
589
        body = io.BytesIO(b"test data")
590
        assert body.read() == b"test data"
591

592
        # Assert the file object has been consumed
593
        assert body.read() == b""
594

595
        # Rewind it back to just be b'data'
596
        rewind_body(body, 5)
597
        assert body.read() == b"data"
598

599
    def test_rewind_body_failed_tell(self) -> None:
600
        body = io.BytesIO(b"test data")
601
        body.read()  # Consume body
602

603
        # Simulate failed tell()
604
        body_pos = _FAILEDTELL
605
        with pytest.raises(UnrewindableBodyError):
606
            rewind_body(body, body_pos)
607

608
    def test_rewind_body_bad_position(self) -> None:
609
        body = io.BytesIO(b"test data")
610
        body.read()  # Consume body
611

612
        # Pass non-integer position
613
        with pytest.raises(ValueError):
614
            rewind_body(body, body_pos=None)  # type: ignore[arg-type]
615
        with pytest.raises(ValueError):
616
            rewind_body(body, body_pos=object())  # type: ignore[arg-type]
617

618
    def test_rewind_body_failed_seek(self) -> None:
619
        class BadSeek(io.StringIO):
620
            def seek(self, offset: int, whence: int = 0) -> typing.NoReturn:
621
                raise OSError
622

623
        with pytest.raises(UnrewindableBodyError):
624
            rewind_body(BadSeek(), body_pos=2)
625

626
    def test_add_stderr_logger(self) -> None:
627
        handler = add_stderr_logger(level=logging.INFO)  # Don't actually print debug
628
        logger = logging.getLogger("urllib3")
629
        assert handler in logger.handlers
630

631
        logger.debug("Testing add_stderr_logger")
632
        logger.removeHandler(handler)
633

634
    def test_disable_warnings(self) -> None:
635
        with warnings.catch_warnings(record=True) as w:
636
            clear_warnings()
637
            warnings.simplefilter("default", InsecureRequestWarning)
638
            warnings.warn("This is a test.", InsecureRequestWarning)
639
            assert len(w) == 1
640
            disable_warnings()
641
            warnings.warn("This is a test.", InsecureRequestWarning)
642
            assert len(w) == 1
643

644
    def _make_time_pass(
645
        self, seconds: int, timeout: Timeout, time_mock: Mock
646
    ) -> Timeout:
647
        """Make some time pass for the timeout object"""
648
        time_mock.return_value = TIMEOUT_EPOCH
649
        timeout.start_connect()
650
        time_mock.return_value = TIMEOUT_EPOCH + seconds
651
        return timeout
652

653
    @pytest.mark.parametrize(
654
        "kwargs, message",
655
        [
656
            ({"total": -1}, "less than"),
657
            ({"connect": 2, "total": -1}, "less than"),
658
            ({"read": -1}, "less than"),
659
            ({"connect": False}, "cannot be a boolean"),
660
            ({"read": True}, "cannot be a boolean"),
661
            ({"connect": 0}, "less than or equal"),
662
            ({"read": "foo"}, "int, float or None"),
663
            ({"read": "1.0"}, "int, float or None"),
664
        ],
665
    )
666
    def test_invalid_timeouts(
667
        self, kwargs: dict[str, int | bool], message: str
668
    ) -> None:
669
        with pytest.raises(ValueError, match=message):
670
            Timeout(**kwargs)
671

672
    @patch("time.monotonic")
673
    def test_timeout(self, time_monotonic: MagicMock) -> None:
674
        timeout = Timeout(total=3)
675

676
        # make 'no time' elapse
677
        timeout = self._make_time_pass(
678
            seconds=0, timeout=timeout, time_mock=time_monotonic
679
        )
680
        assert timeout.read_timeout == 3
681
        assert timeout.connect_timeout == 3
682

683
        timeout = Timeout(total=3, connect=2)
684
        assert timeout.connect_timeout == 2
685

686
        timeout = Timeout()
687
        assert timeout.connect_timeout == _DEFAULT_TIMEOUT
688

689
        # Connect takes 5 seconds, leaving 5 seconds for read
690
        timeout = Timeout(total=10, read=7)
691
        timeout = self._make_time_pass(
692
            seconds=5, timeout=timeout, time_mock=time_monotonic
693
        )
694
        assert timeout.read_timeout == 5
695

696
        # Connect takes 2 seconds, read timeout still 7 seconds
697
        timeout = Timeout(total=10, read=7)
698
        timeout = self._make_time_pass(
699
            seconds=2, timeout=timeout, time_mock=time_monotonic
700
        )
701
        assert timeout.read_timeout == 7
702

703
        timeout = Timeout(total=10, read=7)
704
        assert timeout.read_timeout == 7
705

706
        timeout = Timeout(total=None, read=None, connect=None)
707
        assert timeout.connect_timeout is None
708
        assert timeout.read_timeout is None
709
        assert timeout.total is None
710

711
        timeout = Timeout(5)
712
        assert timeout.total == 5
713

714
    def test_timeout_default_resolve(self) -> None:
715
        """The timeout default is resolved when read_timeout is accessed."""
716
        timeout = Timeout()
717
        with patch("urllib3.util.timeout.getdefaulttimeout", return_value=2):
718
            assert timeout.read_timeout == 2
719

720
        with patch("urllib3.util.timeout.getdefaulttimeout", return_value=3):
721
            assert timeout.read_timeout == 3
722

723
    def test_timeout_str(self) -> None:
724
        timeout = Timeout(connect=1, read=2, total=3)
725
        assert str(timeout) == "Timeout(connect=1, read=2, total=3)"
726
        timeout = Timeout(connect=1, read=None, total=3)
727
        assert str(timeout) == "Timeout(connect=1, read=None, total=3)"
728

729
    @patch("time.monotonic")
730
    def test_timeout_elapsed(self, time_monotonic: MagicMock) -> None:
731
        time_monotonic.return_value = TIMEOUT_EPOCH
732
        timeout = Timeout(total=3)
733
        with pytest.raises(TimeoutStateError):
734
            timeout.get_connect_duration()
735

736
        timeout.start_connect()
737
        with pytest.raises(TimeoutStateError):
738
            timeout.start_connect()
739

740
        time_monotonic.return_value = TIMEOUT_EPOCH + 2
741
        assert timeout.get_connect_duration() == 2
742
        time_monotonic.return_value = TIMEOUT_EPOCH + 37
743
        assert timeout.get_connect_duration() == 37
744

745
    def test_is_fp_closed_object_supports_closed(self) -> None:
746
        class ClosedFile:
747
            @property
748
            def closed(self) -> typing.Literal[True]:
749
                return True
750

751
        assert is_fp_closed(ClosedFile())
752

753
    def test_is_fp_closed_object_has_none_fp(self) -> None:
754
        class NoneFpFile:
755
            @property
756
            def fp(self) -> None:
757
                return None
758

759
        assert is_fp_closed(NoneFpFile())
760

761
    def test_is_fp_closed_object_has_fp(self) -> None:
762
        class FpFile:
763
            @property
764
            def fp(self) -> typing.Literal[True]:
765
                return True
766

767
        assert not is_fp_closed(FpFile())
768

769
    def test_is_fp_closed_object_has_neither_fp_nor_closed(self) -> None:
770
        class NotReallyAFile:
771
            pass
772

773
        with pytest.raises(ValueError):
774
            is_fp_closed(NotReallyAFile())
775

776
    def test_has_ipv6_disabled_on_compile(self) -> None:
777
        with patch("socket.has_ipv6", False):
778
            assert not _has_ipv6("::1")
779

780
    def test_has_ipv6_enabled_but_fails(self) -> None:
781
        with patch("socket.has_ipv6", True):
782
            with patch("socket.socket") as mock:
783
                instance = mock.return_value
784
                instance.bind = Mock(side_effect=Exception("No IPv6 here!"))
785
                assert not _has_ipv6("::1")
786

787
    def test_has_ipv6_enabled_and_working(self) -> None:
788
        with patch("socket.has_ipv6", True):
789
            with patch("socket.socket") as mock:
790
                instance = mock.return_value
791
                instance.bind.return_value = True
792
                assert _has_ipv6("::1")
793

794
    def test_ip_family_ipv6_enabled(self) -> None:
795
        with patch("urllib3.util.connection.HAS_IPV6", True):
796
            assert allowed_gai_family() == socket.AF_UNSPEC
797

798
    def test_ip_family_ipv6_disabled(self) -> None:
799
        with patch("urllib3.util.connection.HAS_IPV6", False):
800
            assert allowed_gai_family() == socket.AF_INET
801

802
    @pytest.mark.parametrize("headers", [b"foo", None, object])
803
    def test_assert_header_parsing_throws_typeerror_with_non_headers(
804
        self, headers: bytes | object | None
805
    ) -> None:
806
        with pytest.raises(TypeError):
807
            assert_header_parsing(headers)  # type: ignore[arg-type]
808

809
    def test_connection_requires_http_tunnel_no_proxy(self) -> None:
810
        assert not connection_requires_http_tunnel(
811
            proxy_url=None, proxy_config=None, destination_scheme=None
812
        )
813

814
    def test_connection_requires_http_tunnel_http_proxy(self) -> None:
815
        proxy = parse_url("http://proxy:8080")
816
        proxy_config = ProxyConfig(
817
            ssl_context=None,
818
            use_forwarding_for_https=False,
819
            assert_hostname=None,
820
            assert_fingerprint=None,
821
        )
822
        destination_scheme = "http"
823
        assert not connection_requires_http_tunnel(
824
            proxy, proxy_config, destination_scheme
825
        )
826

827
        destination_scheme = "https"
828
        assert connection_requires_http_tunnel(proxy, proxy_config, destination_scheme)
829

830
    def test_connection_requires_http_tunnel_https_proxy(self) -> None:
831
        proxy = parse_url("https://proxy:8443")
832
        proxy_config = ProxyConfig(
833
            ssl_context=None,
834
            use_forwarding_for_https=False,
835
            assert_hostname=None,
836
            assert_fingerprint=None,
837
        )
838
        destination_scheme = "http"
839
        assert not connection_requires_http_tunnel(
840
            proxy, proxy_config, destination_scheme
841
        )
842

843
    def test_assert_header_parsing_no_error_on_multipart(self) -> None:
844
        from http import client
845

846
        header_msg = io.BytesIO()
847
        header_msg.write(
848
            b'Content-Type: multipart/encrypted;protocol="application/'
849
            b'HTTP-SPNEGO-session-encrypted";boundary="Encrypted Boundary"'
850
            b"\nServer: Microsoft-HTTPAPI/2.0\nDate: Fri, 16 Aug 2019 19:28:01 GMT"
851
            b"\nContent-Length: 1895\n\n\n"
852
        )
853
        header_msg.seek(0)
854
        assert_header_parsing(client.parse_headers(header_msg))
855

856
    @pytest.mark.parametrize("host", [".localhost", "...", "t" * 64])
857
    def test_create_connection_with_invalid_idna_labels(self, host: str) -> None:
858
        with pytest.raises(
859
            LocationParseError,
860
            match=f"Failed to parse: '{host}', label empty or too long",
861
        ):
862
            create_connection((host, 80))
863

864
    @pytest.mark.parametrize(
865
        "host",
866
        [
867
            "a.example.com",
868
            "localhost.",
869
            "[dead::beef]",
870
            "[dead::beef%en5]",
871
            "[dead::beef%en5.]",
872
        ],
873
    )
874
    @patch("socket.getaddrinfo")
875
    @patch("socket.socket")
876
    def test_create_connection_with_valid_idna_labels(
877
        self, socket: MagicMock, getaddrinfo: MagicMock, host: str
878
    ) -> None:
879
        getaddrinfo.return_value = [(None, None, None, None, None)]
880
        socket.return_value = Mock()
881
        create_connection((host, 80))
882

883
    @patch("socket.getaddrinfo")
884
    def test_create_connection_error(self, getaddrinfo: MagicMock) -> None:
885
        getaddrinfo.return_value = []
886
        with pytest.raises(OSError, match="getaddrinfo returns an empty list"):
887
            create_connection(("example.com", 80))
888

889
    @patch("socket.getaddrinfo")
890
    def test_dnsresolver_forced_error(self, getaddrinfo: MagicMock) -> None:
891
        getaddrinfo.side_effect = socket.gaierror()
892
        with pytest.raises(socket.gaierror):
893
            # dns is valid but we force the error just for the sake of the test
894
            create_connection(("example.com", 80))
895

896
    def test_dnsresolver_expected_error(self) -> None:
897
        with pytest.raises(socket.gaierror):
898
            # windows: [Errno 11001] getaddrinfo failed in windows
899
            # linux: [Errno -2] Name or service not known
900
            # macos: [Errno 8] nodename nor servname provided, or not known
901
            create_connection(("badhost.invalid", 80))
902

903
    @patch("socket.getaddrinfo")
904
    @patch("socket.socket")
905
    def test_create_connection_with_scoped_ipv6(
906
        self, socket: MagicMock, getaddrinfo: MagicMock
907
    ) -> None:
908
        # Check that providing create_connection with a scoped IPv6 address
909
        # properly propagates the scope to getaddrinfo, and that the returned
910
        # scoped ID makes it to the socket creation call.
911
        fake_scoped_sa6 = ("a::b", 80, 0, 42)
912
        getaddrinfo.return_value = [
913
            (
914
                socket.AF_INET6,
915
                socket.SOCK_STREAM,
916
                socket.IPPROTO_TCP,
917
                "",
918
                fake_scoped_sa6,
919
            )
920
        ]
921
        socket.return_value = fake_sock = MagicMock()
922

923
        create_connection(("a::b%iface", 80))
924
        assert getaddrinfo.call_args[0][0] == "a::b%iface"
925
        fake_sock.connect.assert_called_once_with(fake_scoped_sa6)
926

927
    @pytest.mark.parametrize(
928
        "input,params,expected",
929
        (
930
            ("test", {}, "test"),  # str input
931
            (b"test", {}, "test"),  # bytes input
932
            (b"test", {"encoding": "utf-8"}, "test"),  # bytes input with utf-8
933
            (b"test", {"encoding": "ascii"}, "test"),  # bytes input with ascii
934
        ),
935
    )
936
    def test_to_str(
937
        self, input: bytes | str, params: dict[str, str], expected: str
938
    ) -> None:
939
        assert to_str(input, **params) == expected
940

941
    def test_to_str_error(self) -> None:
942
        with pytest.raises(TypeError, match="not expecting type int"):
943
            to_str(1)  # type: ignore[arg-type]
944

945
    @pytest.mark.parametrize(
946
        "input,params,expected",
947
        (
948
            (b"test", {}, b"test"),  # str input
949
            ("test", {}, b"test"),  # bytes input
950
            ("é", {}, b"\xc3\xa9"),  # bytes input
951
            ("test", {"encoding": "utf-8"}, b"test"),  # bytes input with utf-8
952
            ("test", {"encoding": "ascii"}, b"test"),  # bytes input with ascii
953
        ),
954
    )
955
    def test_to_bytes(
956
        self, input: bytes | str, params: dict[str, str], expected: bytes
957
    ) -> None:
958
        assert to_bytes(input, **params) == expected
959

960
    def test_to_bytes_error(self) -> None:
961
        with pytest.raises(TypeError, match="not expecting type int"):
962
            to_bytes(1)  # type: ignore[arg-type]
963

964

965
class TestUtilSSL:
966
    """Test utils that use an SSL backend."""
967

968
    @pytest.mark.parametrize(
969
        "candidate, requirements",
970
        [
971
            (None, ssl.CERT_REQUIRED),
972
            (ssl.CERT_NONE, ssl.CERT_NONE),
973
            (ssl.CERT_REQUIRED, ssl.CERT_REQUIRED),
974
            ("REQUIRED", ssl.CERT_REQUIRED),
975
            ("CERT_REQUIRED", ssl.CERT_REQUIRED),
976
        ],
977
    )
978
    def test_resolve_cert_reqs(
979
        self, candidate: int | str | None, requirements: int
980
    ) -> None:
981
        assert resolve_cert_reqs(candidate) == requirements
982

983
    @pytest.mark.parametrize(
984
        "candidate, version",
985
        [
986
            (ssl.PROTOCOL_TLSv1, ssl.PROTOCOL_TLSv1),
987
            ("PROTOCOL_TLSv1", ssl.PROTOCOL_TLSv1),
988
            ("TLSv1", ssl.PROTOCOL_TLSv1),
989
            (ssl.PROTOCOL_SSLv23, ssl.PROTOCOL_SSLv23),
990
        ],
991
    )
992
    def test_resolve_ssl_version(self, candidate: int | str, version: int) -> None:
993
        assert resolve_ssl_version(candidate) == version
994

995
    def test_ssl_wrap_socket_loads_the_cert_chain(self) -> None:
996
        socket = Mock()
997
        mock_context = Mock()
998
        ssl_wrap_socket(
999
            ssl_context=mock_context, sock=socket, certfile="/path/to/certfile"
1000
        )
1001

1002
        mock_context.load_cert_chain.assert_called_once_with("/path/to/certfile", None)
1003

1004
    @patch("urllib3.util.ssl_.create_urllib3_context")
1005
    def test_ssl_wrap_socket_creates_new_context(
1006
        self, create_urllib3_context: mock.MagicMock
1007
    ) -> None:
1008
        socket = Mock()
1009
        ssl_wrap_socket(socket, cert_reqs=ssl.CERT_REQUIRED)
1010

1011
        create_urllib3_context.assert_called_once_with(None, 2, ciphers=None)
1012

1013
    def test_ssl_wrap_socket_loads_verify_locations(self) -> None:
1014
        socket = Mock()
1015
        mock_context = Mock()
1016
        ssl_wrap_socket(ssl_context=mock_context, ca_certs="/path/to/pem", sock=socket)
1017
        mock_context.load_verify_locations.assert_called_once_with(
1018
            "/path/to/pem", None, None
1019
        )
1020

1021
    def test_ssl_wrap_socket_loads_certificate_directories(self) -> None:
1022
        socket = Mock()
1023
        mock_context = Mock()
1024
        ssl_wrap_socket(
1025
            ssl_context=mock_context, ca_cert_dir="/path/to/pems", sock=socket
1026
        )
1027
        mock_context.load_verify_locations.assert_called_once_with(
1028
            None, "/path/to/pems", None
1029
        )
1030

1031
    def test_ssl_wrap_socket_loads_certificate_data(self) -> None:
1032
        socket = Mock()
1033
        mock_context = Mock()
1034
        ssl_wrap_socket(
1035
            ssl_context=mock_context, ca_cert_data="TOTALLY PEM DATA", sock=socket
1036
        )
1037
        mock_context.load_verify_locations.assert_called_once_with(
1038
            None, None, "TOTALLY PEM DATA"
1039
        )
1040

1041
    def _wrap_socket_and_mock_warn(
1042
        self, sock: socket.socket, server_hostname: str | None
1043
    ) -> tuple[Mock, MagicMock]:
1044
        mock_context = Mock()
1045
        with patch("warnings.warn") as warn:
1046
            ssl_wrap_socket(
1047
                ssl_context=mock_context,
1048
                sock=sock,
1049
                server_hostname=server_hostname,
1050
            )
1051
        return mock_context, warn
1052

1053
    def test_ssl_wrap_socket_sni_ip_address_no_warn(self) -> None:
1054
        """Test that a warning is not made if server_hostname is an IP address."""
1055
        sock = Mock()
1056
        context, warn = self._wrap_socket_and_mock_warn(sock, "8.8.8.8")
1057
        context.wrap_socket.assert_called_once_with(sock, server_hostname="8.8.8.8")
1058
        warn.assert_not_called()
1059

1060
    def test_ssl_wrap_socket_sni_none_no_warn(self) -> None:
1061
        """Test that a warning is not made if server_hostname is not given."""
1062
        sock = Mock()
1063
        context, warn = self._wrap_socket_and_mock_warn(sock, None)
1064
        context.wrap_socket.assert_called_once_with(sock, server_hostname=None)
1065
        warn.assert_not_called()
1066

1067
    @pytest.mark.parametrize(
1068
        "openssl_version, openssl_version_number, implementation_name, version_info, pypy_version_info, reliable",
1069
        [
1070
            # OpenSSL and Python OK -> reliable
1071
            ("OpenSSL 1.1.1", 0x101010CF, "cpython", (3, 9, 3), None, True),
1072
            # Python OK -> reliable
1073
            ("OpenSSL 1.1.1", 0x10101000, "cpython", (3, 9, 3), None, True),
1074
            # PyPy: depends on the version
1075
            ("OpenSSL 1.1.1", 0x10101000, "pypy", (3, 9, 9), (7, 3, 7), False),
1076
            ("OpenSSL 1.1.1", 0x101010CF, "pypy", (3, 8, 12), (7, 3, 8), True),
1077
            # OpenSSL OK -> reliable
1078
            ("OpenSSL 1.1.1", 0x101010CF, "cpython", (3, 9, 2), None, True),
1079
            # not OpenSSSL -> unreliable
1080
            ("LibreSSL 2.8.3", 0x101010CF, "cpython", (3, 10, 0), None, False),
1081
            # old OpenSSL and old Python, unreliable
1082
            ("OpenSSL 1.1.0", 0x10101000, "cpython", (3, 9, 2), None, False),
1083
        ],
1084
    )
1085
    def test_is_has_never_check_common_name_reliable(
1086
        self,
1087
        openssl_version: str,
1088
        openssl_version_number: int,
1089
        implementation_name: str,
1090
        version_info: _TYPE_VERSION_INFO,
1091
        pypy_version_info: _TYPE_VERSION_INFO | None,
1092
        reliable: bool,
1093
    ) -> None:
1094
        assert (
1095
            _is_has_never_check_common_name_reliable(
1096
                openssl_version,
1097
                openssl_version_number,
1098
                implementation_name,
1099
                version_info,
1100
                pypy_version_info,
1101
            )
1102
            == reliable
1103
        )
1104

1105

1106
idna_blocker = ImportBlocker("idna")
1107
module_stash = ModuleStash("urllib3")
1108

1109

1110
class TestUtilWithoutIdna:
1111
    @classmethod
1112
    def setup_class(cls) -> None:
1113
        sys.modules.pop("idna", None)
1114

1115
        module_stash.stash()
1116
        sys.meta_path.insert(0, idna_blocker)
1117

1118
    @classmethod
1119
    def teardown_class(cls) -> None:
1120
        sys.meta_path.remove(idna_blocker)
1121
        module_stash.pop()
1122

1123
    def test_parse_url_without_idna(self) -> None:
1124
        url = "http://\uD7FF.com"
1125
        with pytest.raises(LocationParseError, match=f"Failed to parse: {url}"):
1126
            parse_url(url)
1127