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
|
import asyncio
import pathlib
import ssl
import sys
import httpx
import pytest
import urllib3
import truststore
from tests.conftest import Server, successful_hosts
@pytest.fixture(scope="function")
def inject_truststore():
truststore.inject_into_ssl()
try:
yield
finally:
truststore.extract_from_ssl()
def test_inject_and_extract():
assert ssl.SSLContext is not truststore.SSLContext
try:
original_SSLContext = ssl.SSLContext
ctx = truststore.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
assert isinstance(ctx._ctx, original_SSLContext)
truststore.inject_into_ssl()
assert ssl.SSLContext is truststore.SSLContext
assert urllib3.util.ssl_.SSLContext is truststore.SSLContext
ctx = truststore.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
assert isinstance(ctx._ctx, original_SSLContext)
ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
assert isinstance(ctx, truststore.SSLContext)
ctx = ssl.create_default_context()
assert isinstance(ctx, truststore.SSLContext)
truststore.extract_from_ssl()
assert ssl.SSLContext is original_SSLContext
assert urllib3.util.ssl_.SSLContext is original_SSLContext
ctx = truststore.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
assert isinstance(ctx._ctx, original_SSLContext)
ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
assert isinstance(ctx, original_SSLContext)
ctx = ssl.create_default_context()
assert isinstance(ctx, original_SSLContext)
finally:
truststore.extract_from_ssl()
@successful_hosts
@pytest.mark.usefixtures("inject_truststore")
def test_success_with_inject(host):
with urllib3.PoolManager() as http:
resp = http.request("GET", f"https://{host}")
assert resp.status == 200
@pytest.mark.usefixtures("inject_truststore")
def test_inject_set_values():
ctx = ssl.create_default_context()
assert isinstance(ctx, truststore.SSLContext)
ctx.check_hostname = False
ctx.verify_mode = ssl.CERT_NONE
assert ctx.check_hostname is False
assert ctx.verify_mode == ssl.CERT_NONE
@pytest.mark.asyncio
@pytest.mark.usefixtures("inject_truststore")
async def test_urllib3_works_with_inject(server: Server) -> None:
def test_urllib3():
with urllib3.PoolManager() as client:
resp = client.request("GET", server.base_url)
assert resp.status == 200
thread = asyncio.to_thread(test_urllib3)
await thread
@pytest.mark.asyncio
async def test_aiohttp_works_with_inject(server: Server) -> None:
# We completely isolate the requests module because
# pytest or some other part of our test infra is messing
# with the order it's loaded into modules.
script = str(pathlib.Path(__file__).parent / "aiohttp_with_inject.py")
proc = await asyncio.create_subprocess_exec(sys.executable, script)
await proc.wait()
assert proc.returncode == 200
@pytest.mark.asyncio
async def test_requests_works_with_inject(server: Server) -> None:
# We completely isolate the requests module because
# pytest or some other part of our test infra is messing
# with the order it's loaded into modules.
script = str(pathlib.Path(__file__).parent / "requests_with_inject.py")
proc = await asyncio.create_subprocess_exec(sys.executable, script)
await proc.wait()
assert proc.returncode == 200
@pytest.mark.asyncio
@pytest.mark.usefixtures("inject_truststore")
async def test_sync_httpx_works_with_inject(server: Server) -> None:
def test_httpx():
with httpx.Client() as client:
resp = client.request("GET", server.base_url)
assert resp.status_code == 200
thread = asyncio.to_thread(test_httpx)
await thread
@pytest.mark.usefixtures("inject_truststore")
@pytest.mark.asyncio
async def test_async_httpx_works_with_inject(server: Server) -> None:
async with httpx.AsyncClient() as client:
resp = await client.request("GET", server.base_url)
assert resp.status_code == 200
|
