|
jpayne@7
|
1 from __future__ import annotations
|
|
jpayne@7
|
2
|
|
jpayne@7
|
3 import hmac
|
|
jpayne@7
|
4 import os
|
|
jpayne@7
|
5 import socket
|
|
jpayne@7
|
6 import sys
|
|
jpayne@7
|
7 import typing
|
|
jpayne@7
|
8 import warnings
|
|
jpayne@7
|
9 from binascii import unhexlify
|
|
jpayne@7
|
10 from hashlib import md5, sha1, sha256
|
|
jpayne@7
|
11
|
|
jpayne@7
|
12 from ..exceptions import ProxySchemeUnsupported, SSLError
|
|
jpayne@7
|
13 from .url import _BRACELESS_IPV6_ADDRZ_RE, _IPV4_RE
|
|
jpayne@7
|
14
|
|
jpayne@7
|
15 SSLContext = None
|
|
jpayne@7
|
16 SSLTransport = None
|
|
jpayne@7
|
17 HAS_NEVER_CHECK_COMMON_NAME = False
|
|
jpayne@7
|
18 IS_PYOPENSSL = False
|
|
jpayne@7
|
19 ALPN_PROTOCOLS = ["http/1.1"]
|
|
jpayne@7
|
20
|
|
jpayne@7
|
21 _TYPE_VERSION_INFO = typing.Tuple[int, int, int, str, int]
|
|
jpayne@7
|
22
|
|
jpayne@7
|
23 # Maps the length of a digest to a possible hash function producing this digest
|
|
jpayne@7
|
24 HASHFUNC_MAP = {32: md5, 40: sha1, 64: sha256}
|
|
jpayne@7
|
25
|
|
jpayne@7
|
26
|
|
jpayne@7
|
27 def _is_bpo_43522_fixed(
|
|
jpayne@7
|
28 implementation_name: str,
|
|
jpayne@7
|
29 version_info: _TYPE_VERSION_INFO,
|
|
jpayne@7
|
30 pypy_version_info: _TYPE_VERSION_INFO | None,
|
|
jpayne@7
|
31 ) -> bool:
|
|
jpayne@7
|
32 """Return True for CPython 3.8.9+, 3.9.3+ or 3.10+ and PyPy 7.3.8+ where
|
|
jpayne@7
|
33 setting SSLContext.hostname_checks_common_name to False works.
|
|
jpayne@7
|
34
|
|
jpayne@7
|
35 Outside of CPython and PyPy we don't know which implementations work
|
|
jpayne@7
|
36 or not so we conservatively use our hostname matching as we know that works
|
|
jpayne@7
|
37 on all implementations.
|
|
jpayne@7
|
38
|
|
jpayne@7
|
39 https://github.com/urllib3/urllib3/issues/2192#issuecomment-821832963
|
|
jpayne@7
|
40 https://foss.heptapod.net/pypy/pypy/-/issues/3539
|
|
jpayne@7
|
41 """
|
|
jpayne@7
|
42 if implementation_name == "pypy":
|
|
jpayne@7
|
43 # https://foss.heptapod.net/pypy/pypy/-/issues/3129
|
|
jpayne@7
|
44 return pypy_version_info >= (7, 3, 8) # type: ignore[operator]
|
|
jpayne@7
|
45 elif implementation_name == "cpython":
|
|
jpayne@7
|
46 major_minor = version_info[:2]
|
|
jpayne@7
|
47 micro = version_info[2]
|
|
jpayne@7
|
48 return (
|
|
jpayne@7
|
49 (major_minor == (3, 8) and micro >= 9)
|
|
jpayne@7
|
50 or (major_minor == (3, 9) and micro >= 3)
|
|
jpayne@7
|
51 or major_minor >= (3, 10)
|
|
jpayne@7
|
52 )
|
|
jpayne@7
|
53 else: # Defensive:
|
|
jpayne@7
|
54 return False
|
|
jpayne@7
|
55
|
|
jpayne@7
|
56
|
|
jpayne@7
|
57 def _is_has_never_check_common_name_reliable(
|
|
jpayne@7
|
58 openssl_version: str,
|
|
jpayne@7
|
59 openssl_version_number: int,
|
|
jpayne@7
|
60 implementation_name: str,
|
|
jpayne@7
|
61 version_info: _TYPE_VERSION_INFO,
|
|
jpayne@7
|
62 pypy_version_info: _TYPE_VERSION_INFO | None,
|
|
jpayne@7
|
63 ) -> bool:
|
|
jpayne@7
|
64 # As of May 2023, all released versions of LibreSSL fail to reject certificates with
|
|
jpayne@7
|
65 # only common names, see https://github.com/urllib3/urllib3/pull/3024
|
|
jpayne@7
|
66 is_openssl = openssl_version.startswith("OpenSSL ")
|
|
jpayne@7
|
67 # Before fixing OpenSSL issue #14579, the SSL_new() API was not copying hostflags
|
|
jpayne@7
|
68 # like X509_CHECK_FLAG_NEVER_CHECK_SUBJECT, which tripped up CPython.
|
|
jpayne@7
|
69 # https://github.com/openssl/openssl/issues/14579
|
|
jpayne@7
|
70 # This was released in OpenSSL 1.1.1l+ (>=0x101010cf)
|
|
jpayne@7
|
71 is_openssl_issue_14579_fixed = openssl_version_number >= 0x101010CF
|
|
jpayne@7
|
72
|
|
jpayne@7
|
73 return is_openssl and (
|
|
jpayne@7
|
74 is_openssl_issue_14579_fixed
|
|
jpayne@7
|
75 or _is_bpo_43522_fixed(implementation_name, version_info, pypy_version_info)
|
|
jpayne@7
|
76 )
|
|
jpayne@7
|
77
|
|
jpayne@7
|
78
|
|
jpayne@7
|
79 if typing.TYPE_CHECKING:
|
|
jpayne@7
|
80 from ssl import VerifyMode
|
|
jpayne@7
|
81 from typing import Literal, TypedDict
|
|
jpayne@7
|
82
|
|
jpayne@7
|
83 from .ssltransport import SSLTransport as SSLTransportType
|
|
jpayne@7
|
84
|
|
jpayne@7
|
85 class _TYPE_PEER_CERT_RET_DICT(TypedDict, total=False):
|
|
jpayne@7
|
86 subjectAltName: tuple[tuple[str, str], ...]
|
|
jpayne@7
|
87 subject: tuple[tuple[tuple[str, str], ...], ...]
|
|
jpayne@7
|
88 serialNumber: str
|
|
jpayne@7
|
89
|
|
jpayne@7
|
90
|
|
jpayne@7
|
91 # Mapping from 'ssl.PROTOCOL_TLSX' to 'TLSVersion.X'
|
|
jpayne@7
|
92 _SSL_VERSION_TO_TLS_VERSION: dict[int, int] = {}
|
|
jpayne@7
|
93
|
|
jpayne@7
|
94 try: # Do we have ssl at all?
|
|
jpayne@7
|
95 import ssl
|
|
jpayne@7
|
96 from ssl import ( # type: ignore[assignment]
|
|
jpayne@7
|
97 CERT_REQUIRED,
|
|
jpayne@7
|
98 HAS_NEVER_CHECK_COMMON_NAME,
|
|
jpayne@7
|
99 OP_NO_COMPRESSION,
|
|
jpayne@7
|
100 OP_NO_TICKET,
|
|
jpayne@7
|
101 OPENSSL_VERSION,
|
|
jpayne@7
|
102 OPENSSL_VERSION_NUMBER,
|
|
jpayne@7
|
103 PROTOCOL_TLS,
|
|
jpayne@7
|
104 PROTOCOL_TLS_CLIENT,
|
|
jpayne@7
|
105 OP_NO_SSLv2,
|
|
jpayne@7
|
106 OP_NO_SSLv3,
|
|
jpayne@7
|
107 SSLContext,
|
|
jpayne@7
|
108 TLSVersion,
|
|
jpayne@7
|
109 )
|
|
jpayne@7
|
110
|
|
jpayne@7
|
111 PROTOCOL_SSLv23 = PROTOCOL_TLS
|
|
jpayne@7
|
112
|
|
jpayne@7
|
113 # Setting SSLContext.hostname_checks_common_name = False didn't work before CPython
|
|
jpayne@7
|
114 # 3.8.9, 3.9.3, and 3.10 (but OK on PyPy) or OpenSSL 1.1.1l+
|
|
jpayne@7
|
115 if HAS_NEVER_CHECK_COMMON_NAME and not _is_has_never_check_common_name_reliable(
|
|
jpayne@7
|
116 OPENSSL_VERSION,
|
|
jpayne@7
|
117 OPENSSL_VERSION_NUMBER,
|
|
jpayne@7
|
118 sys.implementation.name,
|
|
jpayne@7
|
119 sys.version_info,
|
|
jpayne@7
|
120 sys.pypy_version_info if sys.implementation.name == "pypy" else None, # type: ignore[attr-defined]
|
|
jpayne@7
|
121 ):
|
|
jpayne@7
|
122 HAS_NEVER_CHECK_COMMON_NAME = False
|
|
jpayne@7
|
123
|
|
jpayne@7
|
124 # Need to be careful here in case old TLS versions get
|
|
jpayne@7
|
125 # removed in future 'ssl' module implementations.
|
|
jpayne@7
|
126 for attr in ("TLSv1", "TLSv1_1", "TLSv1_2"):
|
|
jpayne@7
|
127 try:
|
|
jpayne@7
|
128 _SSL_VERSION_TO_TLS_VERSION[getattr(ssl, f"PROTOCOL_{attr}")] = getattr(
|
|
jpayne@7
|
129 TLSVersion, attr
|
|
jpayne@7
|
130 )
|
|
jpayne@7
|
131 except AttributeError: # Defensive:
|
|
jpayne@7
|
132 continue
|
|
jpayne@7
|
133
|
|
jpayne@7
|
134 from .ssltransport import SSLTransport # type: ignore[assignment]
|
|
jpayne@7
|
135 except ImportError:
|
|
jpayne@7
|
136 OP_NO_COMPRESSION = 0x20000 # type: ignore[assignment]
|
|
jpayne@7
|
137 OP_NO_TICKET = 0x4000 # type: ignore[assignment]
|
|
jpayne@7
|
138 OP_NO_SSLv2 = 0x1000000 # type: ignore[assignment]
|
|
jpayne@7
|
139 OP_NO_SSLv3 = 0x2000000 # type: ignore[assignment]
|
|
jpayne@7
|
140 PROTOCOL_SSLv23 = PROTOCOL_TLS = 2 # type: ignore[assignment]
|
|
jpayne@7
|
141 PROTOCOL_TLS_CLIENT = 16 # type: ignore[assignment]
|
|
jpayne@7
|
142
|
|
jpayne@7
|
143
|
|
jpayne@7
|
144 _TYPE_PEER_CERT_RET = typing.Union["_TYPE_PEER_CERT_RET_DICT", bytes, None]
|
|
jpayne@7
|
145
|
|
jpayne@7
|
146
|
|
jpayne@7
|
147 def assert_fingerprint(cert: bytes | None, fingerprint: str) -> None:
|
|
jpayne@7
|
148 """
|
|
jpayne@7
|
149 Checks if given fingerprint matches the supplied certificate.
|
|
jpayne@7
|
150
|
|
jpayne@7
|
151 :param cert:
|
|
jpayne@7
|
152 Certificate as bytes object.
|
|
jpayne@7
|
153 :param fingerprint:
|
|
jpayne@7
|
154 Fingerprint as string of hexdigits, can be interspersed by colons.
|
|
jpayne@7
|
155 """
|
|
jpayne@7
|
156
|
|
jpayne@7
|
157 if cert is None:
|
|
jpayne@7
|
158 raise SSLError("No certificate for the peer.")
|
|
jpayne@7
|
159
|
|
jpayne@7
|
160 fingerprint = fingerprint.replace(":", "").lower()
|
|
jpayne@7
|
161 digest_length = len(fingerprint)
|
|
jpayne@7
|
162 hashfunc = HASHFUNC_MAP.get(digest_length)
|
|
jpayne@7
|
163 if not hashfunc:
|
|
jpayne@7
|
164 raise SSLError(f"Fingerprint of invalid length: {fingerprint}")
|
|
jpayne@7
|
165
|
|
jpayne@7
|
166 # We need encode() here for py32; works on py2 and p33.
|
|
jpayne@7
|
167 fingerprint_bytes = unhexlify(fingerprint.encode())
|
|
jpayne@7
|
168
|
|
jpayne@7
|
169 cert_digest = hashfunc(cert).digest()
|
|
jpayne@7
|
170
|
|
jpayne@7
|
171 if not hmac.compare_digest(cert_digest, fingerprint_bytes):
|
|
jpayne@7
|
172 raise SSLError(
|
|
jpayne@7
|
173 f'Fingerprints did not match. Expected "{fingerprint}", got "{cert_digest.hex()}"'
|
|
jpayne@7
|
174 )
|
|
jpayne@7
|
175
|
|
jpayne@7
|
176
|
|
jpayne@7
|
177 def resolve_cert_reqs(candidate: None | int | str) -> VerifyMode:
|
|
jpayne@7
|
178 """
|
|
jpayne@7
|
179 Resolves the argument to a numeric constant, which can be passed to
|
|
jpayne@7
|
180 the wrap_socket function/method from the ssl module.
|
|
jpayne@7
|
181 Defaults to :data:`ssl.CERT_REQUIRED`.
|
|
jpayne@7
|
182 If given a string it is assumed to be the name of the constant in the
|
|
jpayne@7
|
183 :mod:`ssl` module or its abbreviation.
|
|
jpayne@7
|
184 (So you can specify `REQUIRED` instead of `CERT_REQUIRED`.
|
|
jpayne@7
|
185 If it's neither `None` nor a string we assume it is already the numeric
|
|
jpayne@7
|
186 constant which can directly be passed to wrap_socket.
|
|
jpayne@7
|
187 """
|
|
jpayne@7
|
188 if candidate is None:
|
|
jpayne@7
|
189 return CERT_REQUIRED
|
|
jpayne@7
|
190
|
|
jpayne@7
|
191 if isinstance(candidate, str):
|
|
jpayne@7
|
192 res = getattr(ssl, candidate, None)
|
|
jpayne@7
|
193 if res is None:
|
|
jpayne@7
|
194 res = getattr(ssl, "CERT_" + candidate)
|
|
jpayne@7
|
195 return res # type: ignore[no-any-return]
|
|
jpayne@7
|
196
|
|
jpayne@7
|
197 return candidate # type: ignore[return-value]
|
|
jpayne@7
|
198
|
|
jpayne@7
|
199
|
|
jpayne@7
|
200 def resolve_ssl_version(candidate: None | int | str) -> int:
|
|
jpayne@7
|
201 """
|
|
jpayne@7
|
202 like resolve_cert_reqs
|
|
jpayne@7
|
203 """
|
|
jpayne@7
|
204 if candidate is None:
|
|
jpayne@7
|
205 return PROTOCOL_TLS
|
|
jpayne@7
|
206
|
|
jpayne@7
|
207 if isinstance(candidate, str):
|
|
jpayne@7
|
208 res = getattr(ssl, candidate, None)
|
|
jpayne@7
|
209 if res is None:
|
|
jpayne@7
|
210 res = getattr(ssl, "PROTOCOL_" + candidate)
|
|
jpayne@7
|
211 return typing.cast(int, res)
|
|
jpayne@7
|
212
|
|
jpayne@7
|
213 return candidate
|
|
jpayne@7
|
214
|
|
jpayne@7
|
215
|
|
jpayne@7
|
216 def create_urllib3_context(
|
|
jpayne@7
|
217 ssl_version: int | None = None,
|
|
jpayne@7
|
218 cert_reqs: int | None = None,
|
|
jpayne@7
|
219 options: int | None = None,
|
|
jpayne@7
|
220 ciphers: str | None = None,
|
|
jpayne@7
|
221 ssl_minimum_version: int | None = None,
|
|
jpayne@7
|
222 ssl_maximum_version: int | None = None,
|
|
jpayne@7
|
223 ) -> ssl.SSLContext:
|
|
jpayne@7
|
224 """Creates and configures an :class:`ssl.SSLContext` instance for use with urllib3.
|
|
jpayne@7
|
225
|
|
jpayne@7
|
226 :param ssl_version:
|
|
jpayne@7
|
227 The desired protocol version to use. This will default to
|
|
jpayne@7
|
228 PROTOCOL_SSLv23 which will negotiate the highest protocol that both
|
|
jpayne@7
|
229 the server and your installation of OpenSSL support.
|
|
jpayne@7
|
230
|
|
jpayne@7
|
231 This parameter is deprecated instead use 'ssl_minimum_version'.
|
|
jpayne@7
|
232 :param ssl_minimum_version:
|
|
jpayne@7
|
233 The minimum version of TLS to be used. Use the 'ssl.TLSVersion' enum for specifying the value.
|
|
jpayne@7
|
234 :param ssl_maximum_version:
|
|
jpayne@7
|
235 The maximum version of TLS to be used. Use the 'ssl.TLSVersion' enum for specifying the value.
|
|
jpayne@7
|
236 Not recommended to set to anything other than 'ssl.TLSVersion.MAXIMUM_SUPPORTED' which is the
|
|
jpayne@7
|
237 default value.
|
|
jpayne@7
|
238 :param cert_reqs:
|
|
jpayne@7
|
239 Whether to require the certificate verification. This defaults to
|
|
jpayne@7
|
240 ``ssl.CERT_REQUIRED``.
|
|
jpayne@7
|
241 :param options:
|
|
jpayne@7
|
242 Specific OpenSSL options. These default to ``ssl.OP_NO_SSLv2``,
|
|
jpayne@7
|
243 ``ssl.OP_NO_SSLv3``, ``ssl.OP_NO_COMPRESSION``, and ``ssl.OP_NO_TICKET``.
|
|
jpayne@7
|
244 :param ciphers:
|
|
jpayne@7
|
245 Which cipher suites to allow the server to select. Defaults to either system configured
|
|
jpayne@7
|
246 ciphers if OpenSSL 1.1.1+, otherwise uses a secure default set of ciphers.
|
|
jpayne@7
|
247 :returns:
|
|
jpayne@7
|
248 Constructed SSLContext object with specified options
|
|
jpayne@7
|
249 :rtype: SSLContext
|
|
jpayne@7
|
250 """
|
|
jpayne@7
|
251 if SSLContext is None:
|
|
jpayne@7
|
252 raise TypeError("Can't create an SSLContext object without an ssl module")
|
|
jpayne@7
|
253
|
|
jpayne@7
|
254 # This means 'ssl_version' was specified as an exact value.
|
|
jpayne@7
|
255 if ssl_version not in (None, PROTOCOL_TLS, PROTOCOL_TLS_CLIENT):
|
|
jpayne@7
|
256 # Disallow setting 'ssl_version' and 'ssl_minimum|maximum_version'
|
|
jpayne@7
|
257 # to avoid conflicts.
|
|
jpayne@7
|
258 if ssl_minimum_version is not None or ssl_maximum_version is not None:
|
|
jpayne@7
|
259 raise ValueError(
|
|
jpayne@7
|
260 "Can't specify both 'ssl_version' and either "
|
|
jpayne@7
|
261 "'ssl_minimum_version' or 'ssl_maximum_version'"
|
|
jpayne@7
|
262 )
|
|
jpayne@7
|
263
|
|
jpayne@7
|
264 # 'ssl_version' is deprecated and will be removed in the future.
|
|
jpayne@7
|
265 else:
|
|
jpayne@7
|
266 # Use 'ssl_minimum_version' and 'ssl_maximum_version' instead.
|
|
jpayne@7
|
267 ssl_minimum_version = _SSL_VERSION_TO_TLS_VERSION.get(
|
|
jpayne@7
|
268 ssl_version, TLSVersion.MINIMUM_SUPPORTED
|
|
jpayne@7
|
269 )
|
|
jpayne@7
|
270 ssl_maximum_version = _SSL_VERSION_TO_TLS_VERSION.get(
|
|
jpayne@7
|
271 ssl_version, TLSVersion.MAXIMUM_SUPPORTED
|
|
jpayne@7
|
272 )
|
|
jpayne@7
|
273
|
|
jpayne@7
|
274 # This warning message is pushing users to use 'ssl_minimum_version'
|
|
jpayne@7
|
275 # instead of both min/max. Best practice is to only set the minimum version and
|
|
jpayne@7
|
276 # keep the maximum version to be it's default value: 'TLSVersion.MAXIMUM_SUPPORTED'
|
|
jpayne@7
|
277 warnings.warn(
|
|
jpayne@7
|
278 "'ssl_version' option is deprecated and will be "
|
|
jpayne@7
|
279 "removed in urllib3 v2.1.0. Instead use 'ssl_minimum_version'",
|
|
jpayne@7
|
280 category=DeprecationWarning,
|
|
jpayne@7
|
281 stacklevel=2,
|
|
jpayne@7
|
282 )
|
|
jpayne@7
|
283
|
|
jpayne@7
|
284 # PROTOCOL_TLS is deprecated in Python 3.10 so we always use PROTOCOL_TLS_CLIENT
|
|
jpayne@7
|
285 context = SSLContext(PROTOCOL_TLS_CLIENT)
|
|
jpayne@7
|
286
|
|
jpayne@7
|
287 if ssl_minimum_version is not None:
|
|
jpayne@7
|
288 context.minimum_version = ssl_minimum_version
|
|
jpayne@7
|
289 else: # Python <3.10 defaults to 'MINIMUM_SUPPORTED' so explicitly set TLSv1.2 here
|
|
jpayne@7
|
290 context.minimum_version = TLSVersion.TLSv1_2
|
|
jpayne@7
|
291
|
|
jpayne@7
|
292 if ssl_maximum_version is not None:
|
|
jpayne@7
|
293 context.maximum_version = ssl_maximum_version
|
|
jpayne@7
|
294
|
|
jpayne@7
|
295 # Unless we're given ciphers defer to either system ciphers in
|
|
jpayne@7
|
296 # the case of OpenSSL 1.1.1+ or use our own secure default ciphers.
|
|
jpayne@7
|
297 if ciphers:
|
|
jpayne@7
|
298 context.set_ciphers(ciphers)
|
|
jpayne@7
|
299
|
|
jpayne@7
|
300 # Setting the default here, as we may have no ssl module on import
|
|
jpayne@7
|
301 cert_reqs = ssl.CERT_REQUIRED if cert_reqs is None else cert_reqs
|
|
jpayne@7
|
302
|
|
jpayne@7
|
303 if options is None:
|
|
jpayne@7
|
304 options = 0
|
|
jpayne@7
|
305 # SSLv2 is easily broken and is considered harmful and dangerous
|
|
jpayne@7
|
306 options |= OP_NO_SSLv2
|
|
jpayne@7
|
307 # SSLv3 has several problems and is now dangerous
|
|
jpayne@7
|
308 options |= OP_NO_SSLv3
|
|
jpayne@7
|
309 # Disable compression to prevent CRIME attacks for OpenSSL 1.0+
|
|
jpayne@7
|
310 # (issue #309)
|
|
jpayne@7
|
311 options |= OP_NO_COMPRESSION
|
|
jpayne@7
|
312 # TLSv1.2 only. Unless set explicitly, do not request tickets.
|
|
jpayne@7
|
313 # This may save some bandwidth on wire, and although the ticket is encrypted,
|
|
jpayne@7
|
314 # there is a risk associated with it being on wire,
|
|
jpayne@7
|
315 # if the server is not rotating its ticketing keys properly.
|
|
jpayne@7
|
316 options |= OP_NO_TICKET
|
|
jpayne@7
|
317
|
|
jpayne@7
|
318 context.options |= options
|
|
jpayne@7
|
319
|
|
jpayne@7
|
320 # Enable post-handshake authentication for TLS 1.3, see GH #1634. PHA is
|
|
jpayne@7
|
321 # necessary for conditional client cert authentication with TLS 1.3.
|
|
jpayne@7
|
322 # The attribute is None for OpenSSL <= 1.1.0 or does not exist when using
|
|
jpayne@7
|
323 # an SSLContext created by pyOpenSSL.
|
|
jpayne@7
|
324 if getattr(context, "post_handshake_auth", None) is not None:
|
|
jpayne@7
|
325 context.post_handshake_auth = True
|
|
jpayne@7
|
326
|
|
jpayne@7
|
327 # The order of the below lines setting verify_mode and check_hostname
|
|
jpayne@7
|
328 # matter due to safe-guards SSLContext has to prevent an SSLContext with
|
|
jpayne@7
|
329 # check_hostname=True, verify_mode=NONE/OPTIONAL.
|
|
jpayne@7
|
330 # We always set 'check_hostname=False' for pyOpenSSL so we rely on our own
|
|
jpayne@7
|
331 # 'ssl.match_hostname()' implementation.
|
|
jpayne@7
|
332 if cert_reqs == ssl.CERT_REQUIRED and not IS_PYOPENSSL:
|
|
jpayne@7
|
333 context.verify_mode = cert_reqs
|
|
jpayne@7
|
334 context.check_hostname = True
|
|
jpayne@7
|
335 else:
|
|
jpayne@7
|
336 context.check_hostname = False
|
|
jpayne@7
|
337 context.verify_mode = cert_reqs
|
|
jpayne@7
|
338
|
|
jpayne@7
|
339 try:
|
|
jpayne@7
|
340 context.hostname_checks_common_name = False
|
|
jpayne@7
|
341 except AttributeError: # Defensive: for CPython < 3.8.9 and 3.9.3; for PyPy < 7.3.8
|
|
jpayne@7
|
342 pass
|
|
jpayne@7
|
343
|
|
jpayne@7
|
344 # Enable logging of TLS session keys via defacto standard environment variable
|
|
jpayne@7
|
345 # 'SSLKEYLOGFILE', if the feature is available (Python 3.8+). Skip empty values.
|
|
jpayne@7
|
346 if hasattr(context, "keylog_filename"):
|
|
jpayne@7
|
347 sslkeylogfile = os.environ.get("SSLKEYLOGFILE")
|
|
jpayne@7
|
348 if sslkeylogfile:
|
|
jpayne@7
|
349 context.keylog_filename = sslkeylogfile
|
|
jpayne@7
|
350
|
|
jpayne@7
|
351 return context
|
|
jpayne@7
|
352
|
|
jpayne@7
|
353
|
|
jpayne@7
|
354 @typing.overload
|
|
jpayne@7
|
355 def ssl_wrap_socket(
|
|
jpayne@7
|
356 sock: socket.socket,
|
|
jpayne@7
|
357 keyfile: str | None = ...,
|
|
jpayne@7
|
358 certfile: str | None = ...,
|
|
jpayne@7
|
359 cert_reqs: int | None = ...,
|
|
jpayne@7
|
360 ca_certs: str | None = ...,
|
|
jpayne@7
|
361 server_hostname: str | None = ...,
|
|
jpayne@7
|
362 ssl_version: int | None = ...,
|
|
jpayne@7
|
363 ciphers: str | None = ...,
|
|
jpayne@7
|
364 ssl_context: ssl.SSLContext | None = ...,
|
|
jpayne@7
|
365 ca_cert_dir: str | None = ...,
|
|
jpayne@7
|
366 key_password: str | None = ...,
|
|
jpayne@7
|
367 ca_cert_data: None | str | bytes = ...,
|
|
jpayne@7
|
368 tls_in_tls: Literal[False] = ...,
|
|
jpayne@7
|
369 ) -> ssl.SSLSocket:
|
|
jpayne@7
|
370 ...
|
|
jpayne@7
|
371
|
|
jpayne@7
|
372
|
|
jpayne@7
|
373 @typing.overload
|
|
jpayne@7
|
374 def ssl_wrap_socket(
|
|
jpayne@7
|
375 sock: socket.socket,
|
|
jpayne@7
|
376 keyfile: str | None = ...,
|
|
jpayne@7
|
377 certfile: str | None = ...,
|
|
jpayne@7
|
378 cert_reqs: int | None = ...,
|
|
jpayne@7
|
379 ca_certs: str | None = ...,
|
|
jpayne@7
|
380 server_hostname: str | None = ...,
|
|
jpayne@7
|
381 ssl_version: int | None = ...,
|
|
jpayne@7
|
382 ciphers: str | None = ...,
|
|
jpayne@7
|
383 ssl_context: ssl.SSLContext | None = ...,
|
|
jpayne@7
|
384 ca_cert_dir: str | None = ...,
|
|
jpayne@7
|
385 key_password: str | None = ...,
|
|
jpayne@7
|
386 ca_cert_data: None | str | bytes = ...,
|
|
jpayne@7
|
387 tls_in_tls: bool = ...,
|
|
jpayne@7
|
388 ) -> ssl.SSLSocket | SSLTransportType:
|
|
jpayne@7
|
389 ...
|
|
jpayne@7
|
390
|
|
jpayne@7
|
391
|
|
jpayne@7
|
392 def ssl_wrap_socket(
|
|
jpayne@7
|
393 sock: socket.socket,
|
|
jpayne@7
|
394 keyfile: str | None = None,
|
|
jpayne@7
|
395 certfile: str | None = None,
|
|
jpayne@7
|
396 cert_reqs: int | None = None,
|
|
jpayne@7
|
397 ca_certs: str | None = None,
|
|
jpayne@7
|
398 server_hostname: str | None = None,
|
|
jpayne@7
|
399 ssl_version: int | None = None,
|
|
jpayne@7
|
400 ciphers: str | None = None,
|
|
jpayne@7
|
401 ssl_context: ssl.SSLContext | None = None,
|
|
jpayne@7
|
402 ca_cert_dir: str | None = None,
|
|
jpayne@7
|
403 key_password: str | None = None,
|
|
jpayne@7
|
404 ca_cert_data: None | str | bytes = None,
|
|
jpayne@7
|
405 tls_in_tls: bool = False,
|
|
jpayne@7
|
406 ) -> ssl.SSLSocket | SSLTransportType:
|
|
jpayne@7
|
407 """
|
|
jpayne@7
|
408 All arguments except for server_hostname, ssl_context, tls_in_tls, ca_cert_data and
|
|
jpayne@7
|
409 ca_cert_dir have the same meaning as they do when using
|
|
jpayne@7
|
410 :func:`ssl.create_default_context`, :meth:`ssl.SSLContext.load_cert_chain`,
|
|
jpayne@7
|
411 :meth:`ssl.SSLContext.set_ciphers` and :meth:`ssl.SSLContext.wrap_socket`.
|
|
jpayne@7
|
412
|
|
jpayne@7
|
413 :param server_hostname:
|
|
jpayne@7
|
414 When SNI is supported, the expected hostname of the certificate
|
|
jpayne@7
|
415 :param ssl_context:
|
|
jpayne@7
|
416 A pre-made :class:`SSLContext` object. If none is provided, one will
|
|
jpayne@7
|
417 be created using :func:`create_urllib3_context`.
|
|
jpayne@7
|
418 :param ciphers:
|
|
jpayne@7
|
419 A string of ciphers we wish the client to support.
|
|
jpayne@7
|
420 :param ca_cert_dir:
|
|
jpayne@7
|
421 A directory containing CA certificates in multiple separate files, as
|
|
jpayne@7
|
422 supported by OpenSSL's -CApath flag or the capath argument to
|
|
jpayne@7
|
423 SSLContext.load_verify_locations().
|
|
jpayne@7
|
424 :param key_password:
|
|
jpayne@7
|
425 Optional password if the keyfile is encrypted.
|
|
jpayne@7
|
426 :param ca_cert_data:
|
|
jpayne@7
|
427 Optional string containing CA certificates in PEM format suitable for
|
|
jpayne@7
|
428 passing as the cadata parameter to SSLContext.load_verify_locations()
|
|
jpayne@7
|
429 :param tls_in_tls:
|
|
jpayne@7
|
430 Use SSLTransport to wrap the existing socket.
|
|
jpayne@7
|
431 """
|
|
jpayne@7
|
432 context = ssl_context
|
|
jpayne@7
|
433 if context is None:
|
|
jpayne@7
|
434 # Note: This branch of code and all the variables in it are only used in tests.
|
|
jpayne@7
|
435 # We should consider deprecating and removing this code.
|
|
jpayne@7
|
436 context = create_urllib3_context(ssl_version, cert_reqs, ciphers=ciphers)
|
|
jpayne@7
|
437
|
|
jpayne@7
|
438 if ca_certs or ca_cert_dir or ca_cert_data:
|
|
jpayne@7
|
439 try:
|
|
jpayne@7
|
440 context.load_verify_locations(ca_certs, ca_cert_dir, ca_cert_data)
|
|
jpayne@7
|
441 except OSError as e:
|
|
jpayne@7
|
442 raise SSLError(e) from e
|
|
jpayne@7
|
443
|
|
jpayne@7
|
444 elif ssl_context is None and hasattr(context, "load_default_certs"):
|
|
jpayne@7
|
445 # try to load OS default certs; works well on Windows.
|
|
jpayne@7
|
446 context.load_default_certs()
|
|
jpayne@7
|
447
|
|
jpayne@7
|
448 # Attempt to detect if we get the goofy behavior of the
|
|
jpayne@7
|
449 # keyfile being encrypted and OpenSSL asking for the
|
|
jpayne@7
|
450 # passphrase via the terminal and instead error out.
|
|
jpayne@7
|
451 if keyfile and key_password is None and _is_key_file_encrypted(keyfile):
|
|
jpayne@7
|
452 raise SSLError("Client private key is encrypted, password is required")
|
|
jpayne@7
|
453
|
|
jpayne@7
|
454 if certfile:
|
|
jpayne@7
|
455 if key_password is None:
|
|
jpayne@7
|
456 context.load_cert_chain(certfile, keyfile)
|
|
jpayne@7
|
457 else:
|
|
jpayne@7
|
458 context.load_cert_chain(certfile, keyfile, key_password)
|
|
jpayne@7
|
459
|
|
jpayne@7
|
460 try:
|
|
jpayne@7
|
461 context.set_alpn_protocols(ALPN_PROTOCOLS)
|
|
jpayne@7
|
462 except NotImplementedError: # Defensive: in CI, we always have set_alpn_protocols
|
|
jpayne@7
|
463 pass
|
|
jpayne@7
|
464
|
|
jpayne@7
|
465 ssl_sock = _ssl_wrap_socket_impl(sock, context, tls_in_tls, server_hostname)
|
|
jpayne@7
|
466 return ssl_sock
|
|
jpayne@7
|
467
|
|
jpayne@7
|
468
|
|
jpayne@7
|
469 def is_ipaddress(hostname: str | bytes) -> bool:
|
|
jpayne@7
|
470 """Detects whether the hostname given is an IPv4 or IPv6 address.
|
|
jpayne@7
|
471 Also detects IPv6 addresses with Zone IDs.
|
|
jpayne@7
|
472
|
|
jpayne@7
|
473 :param str hostname: Hostname to examine.
|
|
jpayne@7
|
474 :return: True if the hostname is an IP address, False otherwise.
|
|
jpayne@7
|
475 """
|
|
jpayne@7
|
476 if isinstance(hostname, bytes):
|
|
jpayne@7
|
477 # IDN A-label bytes are ASCII compatible.
|
|
jpayne@7
|
478 hostname = hostname.decode("ascii")
|
|
jpayne@7
|
479 return bool(_IPV4_RE.match(hostname) or _BRACELESS_IPV6_ADDRZ_RE.match(hostname))
|
|
jpayne@7
|
480
|
|
jpayne@7
|
481
|
|
jpayne@7
|
482 def _is_key_file_encrypted(key_file: str) -> bool:
|
|
jpayne@7
|
483 """Detects if a key file is encrypted or not."""
|
|
jpayne@7
|
484 with open(key_file) as f:
|
|
jpayne@7
|
485 for line in f:
|
|
jpayne@7
|
486 # Look for Proc-Type: 4,ENCRYPTED
|
|
jpayne@7
|
487 if "ENCRYPTED" in line:
|
|
jpayne@7
|
488 return True
|
|
jpayne@7
|
489
|
|
jpayne@7
|
490 return False
|
|
jpayne@7
|
491
|
|
jpayne@7
|
492
|
|
jpayne@7
|
493 def _ssl_wrap_socket_impl(
|
|
jpayne@7
|
494 sock: socket.socket,
|
|
jpayne@7
|
495 ssl_context: ssl.SSLContext,
|
|
jpayne@7
|
496 tls_in_tls: bool,
|
|
jpayne@7
|
497 server_hostname: str | None = None,
|
|
jpayne@7
|
498 ) -> ssl.SSLSocket | SSLTransportType:
|
|
jpayne@7
|
499 if tls_in_tls:
|
|
jpayne@7
|
500 if not SSLTransport:
|
|
jpayne@7
|
501 # Import error, ssl is not available.
|
|
jpayne@7
|
502 raise ProxySchemeUnsupported(
|
|
jpayne@7
|
503 "TLS in TLS requires support for the 'ssl' module"
|
|
jpayne@7
|
504 )
|
|
jpayne@7
|
505
|
|
jpayne@7
|
506 SSLTransport._validate_ssl_context_for_tls_in_tls(ssl_context)
|
|
jpayne@7
|
507 return SSLTransport(sock, ssl_context, server_hostname)
|
|
jpayne@7
|
508
|
|
jpayne@7
|
509 return ssl_context.wrap_socket(sock, server_hostname=server_hostname)
|
