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