venv/lib/python3.11/site-packages/pip/_vendor/urllib3/util/ssl_.py

   1 from __future__ import absolute_import
   2
   3 import hmac
   4 import os
   5 import sys
   6 import warnings
   7 from binascii import hexlify, unhexlify
   8 from hashlib import md5, sha1, sha256
   9
  10 from ..exceptions import (
  11     InsecurePlatformWarning,
  12     ProxySchemeUnsupported,
  13     SNIMissingWarning,
  14     SSLError,
  15 )
  16 from ..packages import six
  17 from .url import BRACELESS_IPV6_ADDRZ_RE, IPV4_RE
  18
  19 SSLContext = None
  20 SSLTransport = None
  21 HAS_SNI = False
  22 IS_PYOPENSSL = False
  23 IS_SECURETRANSPORT = False
  24 ALPN_PROTOCOLS = ["http/1.1"]
  25
  26 # Maps the length of a digest to a possible hash function producing this digest
  27 HASHFUNC_MAP = {32: md5, 40: sha1, 64: sha256}
  28
  29
  30 def _const_compare_digest_backport(a, b):
  31     """
  32     Compare two digests of equal length in constant time.
  33
  34     The digests must be of type str/bytes.
  35     Returns True if the digests match, and False otherwise.
  36     """
  37     result = abs(len(a) - len(b))
  38     for left, right in zip(bytearray(a), bytearray(b)):
  39         result |= left ^ right
  40     return result == 0
  41
  42
  43 _const_compare_digest = getattr(hmac, "compare_digest", _const_compare_digest_backport)
  44
  45 try:  # Test for SSL features
  46     import ssl
  47     from ssl import CERT_REQUIRED, wrap_socket
  48 except ImportError:
  49     pass
  50
  51 try:
  52     from ssl import HAS_SNI  # Has SNI?
  53 except ImportError:
  54     pass
  55
  56 try:
  57     from .ssltransport import SSLTransport
  58 except ImportError:
  59     pass
  60
  61
  62 try:  # Platform-specific: Python 3.6
  63     from ssl import PROTOCOL_TLS
  64
  65     PROTOCOL_SSLv23 = PROTOCOL_TLS
  66 except ImportError:
  67     try:
  68         from ssl import PROTOCOL_SSLv23 as PROTOCOL_TLS
  69
  70         PROTOCOL_SSLv23 = PROTOCOL_TLS
  71     except ImportError:
  72         PROTOCOL_SSLv23 = PROTOCOL_TLS = 2
  73
  74 try:
  75     from ssl import PROTOCOL_TLS_CLIENT
  76 except ImportError:
  77     PROTOCOL_TLS_CLIENT = PROTOCOL_TLS
  78
  79
  80 try:
  81     from ssl import OP_NO_COMPRESSION, OP_NO_SSLv2, OP_NO_SSLv3
  82 except ImportError:
  83     OP_NO_SSLv2, OP_NO_SSLv3 = 0x1000000, 0x2000000
  84     OP_NO_COMPRESSION = 0x20000
  85
  86
  87 try:  # OP_NO_TICKET was added in Python 3.6
  88     from ssl import OP_NO_TICKET
  89 except ImportError:
  90     OP_NO_TICKET = 0x4000
  91
  92
  93 # A secure default.
  94 # Sources for more information on TLS ciphers:
  95 #
  96 # - https://wiki.mozilla.org/Security/Server_Side_TLS
  97 # - https://www.ssllabs.com/projects/best-practices/index.html
  98 # - https://hynek.me/articles/hardening-your-web-servers-ssl-ciphers/
  99 #
 100 # The general intent is:
 101 # - prefer cipher suites that offer perfect forward secrecy (DHE/ECDHE),
 102 # - prefer ECDHE over DHE for better performance,
 103 # - prefer any AES-GCM and ChaCha20 over any AES-CBC for better performance and
 104 #   security,
 105 # - prefer AES-GCM over ChaCha20 because hardware-accelerated AES is common,
 106 # - disable NULL authentication, MD5 MACs, DSS, and other
 107 #   insecure ciphers for security reasons.
 108 # - NOTE: TLS 1.3 cipher suites are managed through a different interface
 109 #   not exposed by CPython (yet!) and are enabled by default if they're available.
 110 DEFAULT_CIPHERS = ":".join(
 111     [
 112         "ECDHE+AESGCM",
 113         "ECDHE+CHACHA20",
 114         "DHE+AESGCM",
 115         "DHE+CHACHA20",
 116         "ECDH+AESGCM",
 117         "DH+AESGCM",
 118         "ECDH+AES",
 119         "DH+AES",
 120         "RSA+AESGCM",
 121         "RSA+AES",
 122         "!aNULL",
 123         "!eNULL",
 124         "!MD5",
 125         "!DSS",
 126     ]
 127 )
 128
 129 try:
 130     from ssl import SSLContext  # Modern SSL?
 131 except ImportError:
 132
 133     class SSLContext(object):  # Platform-specific: Python 2
 134         def __init__(self, protocol_version):
 135             self.protocol = protocol_version
 136             # Use default values from a real SSLContext
 137             self.check_hostname = False
 138             self.verify_mode = ssl.CERT_NONE
 139             self.ca_certs = None
 140             self.options = 0
 141             self.certfile = None
 142             self.keyfile = None
 143             self.ciphers = None
 144
 145         def load_cert_chain(self, certfile, keyfile):
 146             self.certfile = certfile
 147             self.keyfile = keyfile
 148
 149         def load_verify_locations(self, cafile=None, capath=None, cadata=None):
 150             self.ca_certs = cafile
 151
 152             if capath is not None:
 153                 raise SSLError("CA directories not supported in older Pythons")
 154
 155             if cadata is not None:
 156                 raise SSLError("CA data not supported in older Pythons")
 157
 158         def set_ciphers(self, cipher_suite):
 159             self.ciphers = cipher_suite
 160
 161         def wrap_socket(self, socket, server_hostname=None, server_side=False):
 162             warnings.warn(
 163                 "A true SSLContext object is not available. This prevents "
 164                 "urllib3 from configuring SSL appropriately and may cause "
 165                 "certain SSL connections to fail. You can upgrade to a newer "
 166                 "version of Python to solve this. For more information, see "
 167                 "https://urllib3.readthedocs.io/en/1.26.x/advanced-usage.html"
 168                 "#ssl-warnings",
 169                 InsecurePlatformWarning,
 170             )
 171             kwargs = {
 172                 "keyfile": self.keyfile,
 173                 "certfile": self.certfile,
 174                 "ca_certs": self.ca_certs,
 175                 "cert_reqs": self.verify_mode,
 176                 "ssl_version": self.protocol,
 177                 "server_side": server_side,
 178             }
 179             return wrap_socket(socket, ciphers=self.ciphers, **kwargs)
 180
 181
 182 def assert_fingerprint(cert, fingerprint):
 183     """
 184     Checks if given fingerprint matches the supplied certificate.
 185
 186     :param cert:
 187         Certificate as bytes object.
 188     :param fingerprint:
 189         Fingerprint as string of hexdigits, can be interspersed by colons.
 190     """
 191
 192     fingerprint = fingerprint.replace(":", "").lower()
 193     digest_length = len(fingerprint)
 194     hashfunc = HASHFUNC_MAP.get(digest_length)
 195     if not hashfunc:
 196         raise SSLError("Fingerprint of invalid length: {0}".format(fingerprint))
 197
 198     # We need encode() here for py32; works on py2 and p33.
 199     fingerprint_bytes = unhexlify(fingerprint.encode())
 200
 201     cert_digest = hashfunc(cert).digest()
 202
 203     if not _const_compare_digest(cert_digest, fingerprint_bytes):
 204         raise SSLError(
 205             'Fingerprints did not match. Expected "{0}", got "{1}".'.format(
 206                 fingerprint, hexlify(cert_digest)
 207             )
 208         )
 209
 210
 211 def resolve_cert_reqs(candidate):
 212     """
 213     Resolves the argument to a numeric constant, which can be passed to
 214     the wrap_socket function/method from the ssl module.
 215     Defaults to :data:`ssl.CERT_REQUIRED`.
 216     If given a string it is assumed to be the name of the constant in the
 217     :mod:`ssl` module or its abbreviation.
 218     (So you can specify `REQUIRED` instead of `CERT_REQUIRED`.
 219     If it's neither `None` nor a string we assume it is already the numeric
 220     constant which can directly be passed to wrap_socket.
 221     """
 222     if candidate is None:
 223         return CERT_REQUIRED
 224
 225     if isinstance(candidate, str):
 226         res = getattr(ssl, candidate, None)
 227         if res is None:
 228             res = getattr(ssl, "CERT_" + candidate)
 229         return res
 230
 231     return candidate
 232
 233
 234 def resolve_ssl_version(candidate):
 235     """
 236     like resolve_cert_reqs
 237     """
 238     if candidate is None:
 239         return PROTOCOL_TLS
 240
 241     if isinstance(candidate, str):
 242         res = getattr(ssl, candidate, None)
 243         if res is None:
 244             res = getattr(ssl, "PROTOCOL_" + candidate)
 245         return res
 246
 247     return candidate
 248
 249
 250 def create_urllib3_context(
 251     ssl_version=None, cert_reqs=None, options=None, ciphers=None
 252 ):
 253     """All arguments have the same meaning as ``ssl_wrap_socket``.
 254
 255     By default, this function does a lot of the same work that
 256     ``ssl.create_default_context`` does on Python 3.4+. It:
 257
 258     - Disables SSLv2, SSLv3, and compression
 259     - Sets a restricted set of server ciphers
 260
 261     If you wish to enable SSLv3, you can do::
 262
 263         from pip._vendor.urllib3.util import ssl_
 264         context = ssl_.create_urllib3_context()
 265         context.options &= ~ssl_.OP_NO_SSLv3
 266
 267     You can do the same to enable compression (substituting ``COMPRESSION``
 268     for ``SSLv3`` in the last line above).
 269
 270     :param ssl_version:
 271         The desired protocol version to use. This will default to
 272         PROTOCOL_SSLv23 which will negotiate the highest protocol that both
 273         the server and your installation of OpenSSL support.
 274     :param cert_reqs:
 275         Whether to require the certificate verification. This defaults to
 276         ``ssl.CERT_REQUIRED``.
 277     :param options:
 278         Specific OpenSSL options. These default to ``ssl.OP_NO_SSLv2``,
 279         ``ssl.OP_NO_SSLv3``, ``ssl.OP_NO_COMPRESSION``, and ``ssl.OP_NO_TICKET``.
 280     :param ciphers:
 281         Which cipher suites to allow the server to select.
 282     :returns:
 283         Constructed SSLContext object with specified options
 284     :rtype: SSLContext
 285     """
 286     # PROTOCOL_TLS is deprecated in Python 3.10
 287     if not ssl_version or ssl_version == PROTOCOL_TLS:
 288         ssl_version = PROTOCOL_TLS_CLIENT
 289
 290     context = SSLContext(ssl_version)
 291
 292     context.set_ciphers(ciphers or DEFAULT_CIPHERS)
 293
 294     # Setting the default here, as we may have no ssl module on import
 295     cert_reqs = ssl.CERT_REQUIRED if cert_reqs is None else cert_reqs
 296
 297     if options is None:
 298         options = 0
 299         # SSLv2 is easily broken and is considered harmful and dangerous
 300         options |= OP_NO_SSLv2
 301         # SSLv3 has several problems and is now dangerous
 302         options |= OP_NO_SSLv3
 303         # Disable compression to prevent CRIME attacks for OpenSSL 1.0+
 304         # (issue #309)
 305         options |= OP_NO_COMPRESSION
 306         # TLSv1.2 only. Unless set explicitly, do not request tickets.
 307         # This may save some bandwidth on wire, and although the ticket is encrypted,
 308         # there is a risk associated with it being on wire,
 309         # if the server is not rotating its ticketing keys properly.
 310         options |= OP_NO_TICKET
 311
 312     context.options |= options
 313
 314     # Enable post-handshake authentication for TLS 1.3, see GH #1634. PHA is
 315     # necessary for conditional client cert authentication with TLS 1.3.
 316     # The attribute is None for OpenSSL <= 1.1.0 or does not exist in older
 317     # versions of Python.  We only enable on Python 3.7.4+ or if certificate
 318     # verification is enabled to work around Python issue #37428
 319     # See: https://bugs.python.org/issue37428
 320     if (cert_reqs == ssl.CERT_REQUIRED or sys.version_info >= (3, 7, 4)) and getattr(
 321         context, "post_handshake_auth", None
 322     ) is not None:
 323         context.post_handshake_auth = True
 324
 325     def disable_check_hostname():
 326         if (
 327             getattr(context, "check_hostname", None) is not None
 328         ):  # Platform-specific: Python 3.2
 329             # We do our own verification, including fingerprints and alternative
 330             # hostnames. So disable it here
 331             context.check_hostname = False
 332
 333     # The order of the below lines setting verify_mode and check_hostname
 334     # matter due to safe-guards SSLContext has to prevent an SSLContext with
 335     # check_hostname=True, verify_mode=NONE/OPTIONAL. This is made even more
 336     # complex because we don't know whether PROTOCOL_TLS_CLIENT will be used
 337     # or not so we don't know the initial state of the freshly created SSLContext.
 338     if cert_reqs == ssl.CERT_REQUIRED:
 339         context.verify_mode = cert_reqs
 340         disable_check_hostname()
 341     else:
 342         disable_check_hostname()
 343         context.verify_mode = cert_reqs
 344
 345     # Enable logging of TLS session keys via defacto standard environment variable
 346     # 'SSLKEYLOGFILE', if the feature is available (Python 3.8+). Skip empty values.
 347     if hasattr(context, "keylog_filename"):
 348         sslkeylogfile = os.environ.get("SSLKEYLOGFILE")
 349         if sslkeylogfile:
 350             context.keylog_filename = sslkeylogfile
 351
 352     return context
 353
 354
 355 def ssl_wrap_socket(
 356     sock,
 357     keyfile=None,
 358     certfile=None,
 359     cert_reqs=None,
 360     ca_certs=None,
 361     server_hostname=None,
 362     ssl_version=None,
 363     ciphers=None,
 364     ssl_context=None,
 365     ca_cert_dir=None,
 366     key_password=None,
 367     ca_cert_data=None,
 368     tls_in_tls=False,
 369 ):
 370     """
 371     All arguments except for server_hostname, ssl_context, and ca_cert_dir have
 372     the same meaning as they do when using :func:`ssl.wrap_socket`.
 373
 374     :param server_hostname:
 375         When SNI is supported, the expected hostname of the certificate
 376     :param ssl_context:
 377         A pre-made :class:`SSLContext` object. If none is provided, one will
 378         be created using :func:`create_urllib3_context`.
 379     :param ciphers:
 380         A string of ciphers we wish the client to support.
 381     :param ca_cert_dir:
 382         A directory containing CA certificates in multiple separate files, as
 383         supported by OpenSSL's -CApath flag or the capath argument to
 384         SSLContext.load_verify_locations().
 385     :param key_password:
 386         Optional password if the keyfile is encrypted.
 387     :param ca_cert_data:
 388         Optional string containing CA certificates in PEM format suitable for
 389         passing as the cadata parameter to SSLContext.load_verify_locations()
 390     :param tls_in_tls:
 391         Use SSLTransport to wrap the existing socket.
 392     """
 393     context = ssl_context
 394     if context is None:
 395         # Note: This branch of code and all the variables in it are no longer
 396         # used by urllib3 itself. We should consider deprecating and removing
 397         # this code.
 398         context = create_urllib3_context(ssl_version, cert_reqs, ciphers=ciphers)
 399
 400     if ca_certs or ca_cert_dir or ca_cert_data:
 401         try:
 402             context.load_verify_locations(ca_certs, ca_cert_dir, ca_cert_data)
 403         except (IOError, OSError) as e:
 404             raise SSLError(e)
 405
 406     elif ssl_context is None and hasattr(context, "load_default_certs"):
 407         # try to load OS default certs; works well on Windows (require Python3.4+)
 408         context.load_default_certs()
 409
 410     # Attempt to detect if we get the goofy behavior of the
 411     # keyfile being encrypted and OpenSSL asking for the
 412     # passphrase via the terminal and instead error out.
 413     if keyfile and key_password is None and _is_key_file_encrypted(keyfile):
 414         raise SSLError("Client private key is encrypted, password is required")
 415
 416     if certfile:
 417         if key_password is None:
 418             context.load_cert_chain(certfile, keyfile)
 419         else:
 420             context.load_cert_chain(certfile, keyfile, key_password)
 421
 422     try:
 423         if hasattr(context, "set_alpn_protocols"):
 424             context.set_alpn_protocols(ALPN_PROTOCOLS)
 425     except NotImplementedError:  # Defensive: in CI, we always have set_alpn_protocols
 426         pass
 427
 428     # If we detect server_hostname is an IP address then the SNI
 429     # extension should not be used according to RFC3546 Section 3.1
 430     use_sni_hostname = server_hostname and not is_ipaddress(server_hostname)
 431     # SecureTransport uses server_hostname in certificate verification.
 432     send_sni = (use_sni_hostname and HAS_SNI) or (
 433         IS_SECURETRANSPORT and server_hostname
 434     )
 435     # Do not warn the user if server_hostname is an invalid SNI hostname.
 436     if not HAS_SNI and use_sni_hostname:
 437         warnings.warn(
 438             "An HTTPS request has been made, but the SNI (Server Name "
 439             "Indication) extension to TLS is not available on this platform. "
 440             "This may cause the server to present an incorrect TLS "
 441             "certificate, which can cause validation failures. You can upgrade to "
 442             "a newer version of Python to solve this. For more information, see "
 443             "https://urllib3.readthedocs.io/en/1.26.x/advanced-usage.html"
 444             "#ssl-warnings",
 445             SNIMissingWarning,
 446         )
 447
 448     if send_sni:
 449         ssl_sock = _ssl_wrap_socket_impl(
 450             sock, context, tls_in_tls, server_hostname=server_hostname
 451         )
 452     else:
 453         ssl_sock = _ssl_wrap_socket_impl(sock, context, tls_in_tls)
 454     return ssl_sock
 455
 456
 457 def is_ipaddress(hostname):
 458     """Detects whether the hostname given is an IPv4 or IPv6 address.
 459     Also detects IPv6 addresses with Zone IDs.
 460
 461     :param str hostname: Hostname to examine.
 462     :return: True if the hostname is an IP address, False otherwise.
 463     """
 464     if not six.PY2 and isinstance(hostname, bytes):
 465         # IDN A-label bytes are ASCII compatible.
 466         hostname = hostname.decode("ascii")
 467     return bool(IPV4_RE.match(hostname) or BRACELESS_IPV6_ADDRZ_RE.match(hostname))
 468
 469
 470 def _is_key_file_encrypted(key_file):
 471     """Detects if a key file is encrypted or not."""
 472     with open(key_file, "r") as f:
 473         for line in f:
 474             # Look for Proc-Type: 4,ENCRYPTED
 475             if "ENCRYPTED" in line:
 476                 return True
 477
 478     return False
 479
 480
 481 def _ssl_wrap_socket_impl(sock, ssl_context, tls_in_tls, server_hostname=None):
 482     if tls_in_tls:
 483         if not SSLTransport:
 484             # Import error, ssl is not available.
 485             raise ProxySchemeUnsupported(
 486                 "TLS in TLS requires support for the 'ssl' module"
 487             )
 488
 489         SSLTransport._validate_ssl_context_for_tls_in_tls(ssl_context)
 490         return SSLTransport(sock, ssl_context, server_hostname)
 491
 492     if server_hostname:
 493         return ssl_context.wrap_socket(sock, server_hostname=server_hostname)
 494     else:
 495         return ssl_context.wrap_socket(sock)