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