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