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
|
import collections
import contextlib
import platform
import socket
import ssl
import sys
import threading
import pytest
import trustme
from tornado import ioloop, web
from dummyserver.handlers import TestingApp
from dummyserver.server import HAS_IPV6, run_tornado_app
from dummyserver.testcase import HTTPSDummyServerTestCase
from urllib3.util import ssl_
from .tz_stub import stub_timezone_ctx
# The Python 3.8+ default loop on Windows breaks Tornado
@pytest.fixture(scope="session", autouse=True)
def configure_windows_event_loop():
if sys.version_info >= (3, 8) and platform.system() == "Windows":
import asyncio
asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())
ServerConfig = collections.namedtuple("ServerConfig", ["host", "port", "ca_certs"])
@contextlib.contextmanager
def run_server_in_thread(scheme, host, tmpdir, ca, server_cert):
ca_cert_path = str(tmpdir / "ca.pem")
server_cert_path = str(tmpdir / "server.pem")
server_key_path = str(tmpdir / "server.key")
ca.cert_pem.write_to_path(ca_cert_path)
server_cert.private_key_pem.write_to_path(server_key_path)
server_cert.cert_chain_pems[0].write_to_path(server_cert_path)
server_certs = {"keyfile": server_key_path, "certfile": server_cert_path}
io_loop = ioloop.IOLoop.current()
app = web.Application([(r".*", TestingApp)])
server, port = run_tornado_app(app, io_loop, server_certs, scheme, host)
server_thread = threading.Thread(target=io_loop.start)
server_thread.start()
yield ServerConfig(host, port, ca_cert_path)
io_loop.add_callback(server.stop)
io_loop.add_callback(io_loop.stop)
server_thread.join()
@pytest.fixture
def no_san_server(tmp_path_factory):
tmpdir = tmp_path_factory.mktemp("certs")
ca = trustme.CA()
# only common name, no subject alternative names
server_cert = ca.issue_cert(common_name=u"localhost")
with run_server_in_thread("https", "localhost", tmpdir, ca, server_cert) as cfg:
yield cfg
@pytest.fixture
def no_localhost_san_server(tmp_path_factory):
tmpdir = tmp_path_factory.mktemp("certs")
ca = trustme.CA()
# non localhost common name
server_cert = ca.issue_cert(u"example.com")
with run_server_in_thread("https", "localhost", tmpdir, ca, server_cert) as cfg:
yield cfg
@pytest.fixture
def ip_san_server(tmp_path_factory):
tmpdir = tmp_path_factory.mktemp("certs")
ca = trustme.CA()
# IP address in Subject Alternative Name
server_cert = ca.issue_cert(u"127.0.0.1")
with run_server_in_thread("https", "127.0.0.1", tmpdir, ca, server_cert) as cfg:
yield cfg
@pytest.fixture
def ipv6_addr_server(tmp_path_factory):
if not HAS_IPV6:
pytest.skip("Only runs on IPv6 systems")
tmpdir = tmp_path_factory.mktemp("certs")
ca = trustme.CA()
# IP address in Common Name
server_cert = ca.issue_cert(common_name=u"::1")
with run_server_in_thread("https", "::1", tmpdir, ca, server_cert) as cfg:
yield cfg
@pytest.fixture
def ipv6_san_server(tmp_path_factory):
if not HAS_IPV6:
pytest.skip("Only runs on IPv6 systems")
tmpdir = tmp_path_factory.mktemp("certs")
ca = trustme.CA()
# IP address in Subject Alternative Name
server_cert = ca.issue_cert(u"::1")
with run_server_in_thread("https", "::1", tmpdir, ca, server_cert) as cfg:
yield cfg
@pytest.yield_fixture
def stub_timezone(request):
"""
A pytest fixture that runs the test with a stub timezone.
"""
with stub_timezone_ctx(request.param):
yield
@pytest.fixture(scope="session")
def supported_tls_versions():
# We have to create an actual TLS connection
# to test if the TLS version is not disabled by
# OpenSSL config. Ubuntu 20.04 specifically
# disables TLSv1 and TLSv1.1.
tls_versions = set()
_server = HTTPSDummyServerTestCase()
_server._start_server()
for _ssl_version_name in (
"PROTOCOL_TLSv1",
"PROTOCOL_TLSv1_1",
"PROTOCOL_TLSv1_2",
"PROTOCOL_TLS",
):
_ssl_version = getattr(ssl, _ssl_version_name, 0)
if _ssl_version == 0:
continue
_sock = socket.create_connection((_server.host, _server.port))
try:
_sock = ssl_.ssl_wrap_socket(
_sock, cert_reqs=ssl.CERT_NONE, ssl_version=_ssl_version
)
except ssl.SSLError:
pass
else:
tls_versions.add(_sock.version())
_sock.close()
_server._stop_server()
return tls_versions
@pytest.fixture(scope="function")
def requires_tlsv1(supported_tls_versions):
"""Test requires TLSv1 available"""
if not hasattr(ssl, "PROTOCOL_TLSv1") or "TLSv1" not in supported_tls_versions:
pytest.skip("Test requires TLSv1")
@pytest.fixture(scope="function")
def requires_tlsv1_1(supported_tls_versions):
"""Test requires TLSv1.1 available"""
if not hasattr(ssl, "PROTOCOL_TLSv1_1") or "TLSv1.1" not in supported_tls_versions:
pytest.skip("Test requires TLSv1.1")
@pytest.fixture(scope="function")
def requires_tlsv1_2(supported_tls_versions):
"""Test requires TLSv1.2 available"""
if not hasattr(ssl, "PROTOCOL_TLSv1_2") or "TLSv1.2" not in supported_tls_versions:
pytest.skip("Test requires TLSv1.2")
@pytest.fixture(scope="function")
def requires_tlsv1_3(supported_tls_versions):
"""Test requires TLSv1.3 available"""
if (
not getattr(ssl, "HAS_TLSv1_3", False)
or "TLSv1.3" not in supported_tls_versions
):
pytest.skip("Test requires TLSv1.3")
|
