JezK
Edit File: ssl_.py
from __future__ import absolute_import import hmac import os import sys import warnings from binascii import hexlify, unhexlify from hashlib import md5, sha1, sha256 from ..exceptions import ( InsecurePlatformWarning, ProxySchemeUnsupported, SNIMissingWarning, SSLError, ) from ..packages import six from .url import BRACELESS_IPV6_ADDRZ_RE, IPV4_RE SSLContext = None SSLTransport = None HAS_SNI = False IS_PYOPENSSL = False IS_SECURETRANSPORT = False ALPN_PROTOCOLS = ["http/1.1"] # Maps the length of a digest to a possible hash function producing this digest HASHFUNC_MAP = {32: md5, 40: sha1, 64: sha256} def _const_compare_digest_backport(a, b): """ Compare two digests of equal length in constant time. The digests must be of type str/bytes. Returns True if the digests match, and False otherwise. """ result = abs(len(a) - len(b)) for left, right in zip(bytearray(a), bytearray(b)): result |= left ^ right return result == 0 _const_compare_digest = getattr(hmac, "compare_digest", _const_compare_digest_backport) try: # Test for SSL features import ssl from ssl import CERT_REQUIRED, wrap_socket except ImportError: pass try: from ssl import HAS_SNI # Has SNI? except ImportError: pass try: from .ssltransport import SSLTransport except ImportError: pass try: # Platform-specific: Python 3.6 from ssl import PROTOCOL_TLS PROTOCOL_SSLv23 = PROTOCOL_TLS except ImportError: try: from ssl import PROTOCOL_SSLv23 as PROTOCOL_TLS PROTOCOL_SSLv23 = PROTOCOL_TLS except ImportError: PROTOCOL_SSLv23 = PROTOCOL_TLS = 2 try: from ssl import PROTOCOL_TLS_CLIENT except ImportError: PROTOCOL_TLS_CLIENT = PROTOCOL_TLS try: from ssl import OP_NO_COMPRESSION, OP_NO_SSLv2, OP_NO_SSLv3 except ImportError: OP_NO_SSLv2, OP_NO_SSLv3 = 0x1000000, 0x2000000 OP_NO_COMPRESSION = 0x20000 try: # OP_NO_TICKET was added in Python 3.6 from ssl import OP_NO_TICKET except ImportError: OP_NO_TICKET = 0x4000 # A secure default. # Sources for more information on TLS ciphers: # # - https://wiki.mozilla.org/Security/Server_Side_TLS # - https://www.ssllabs.com/projects/best-practices/index.html # - https://hynek.me/articles/hardening-your-web-servers-ssl-ciphers/ # # The general intent is: # - prefer cipher suites that offer perfect forward secrecy (DHE/ECDHE), # - prefer ECDHE over DHE for better performance, # - prefer any AES-GCM and ChaCha20 over any AES-CBC for better performance and # security, # - prefer AES-GCM over ChaCha20 because hardware-accelerated AES is common, # - disable NULL authentication, MD5 MACs, DSS, and other # insecure ciphers for security reasons. # - NOTE: TLS 1.3 cipher suites are managed through a different interface # not exposed by CPython (yet!) and are enabled by default if they're available. DEFAULT_CIPHERS = ":".join( [ "ECDHE+AESGCM", "ECDHE+CHACHA20", "DHE+AESGCM", "DHE+CHACHA20", "ECDH+AESGCM", "DH+AESGCM", "ECDH+AES", "DH+AES", "RSA+AESGCM", "RSA+AES", "!aNULL", "!eNULL", "!MD5", "!DSS", ] ) try: from ssl import SSLContext # Modern SSL? except ImportError: class SSLContext(object): # Platform-specific: Python 2 def __init__(self, protocol_version): self.protocol = protocol_version # Use default values from a real SSLContext self.check_hostname = False self.verify_mode = ssl.CERT_NONE self.ca_certs = None self.options = 0 self.certfile = None self.keyfile = None self.ciphers = None def load_cert_chain(self, certfile, keyfile): self.certfile = certfile self.keyfile = keyfile def load_verify_locations(self, cafile=None, capath=None, cadata=None): self.ca_certs = cafile if capath is not None: raise SSLError("CA directories not supported in older Pythons") if cadata is not None: raise SSLError("CA data not supported in older Pythons") def set_ciphers(self, cipher_suite): self.ciphers = cipher_suite def wrap_socket(self, socket, server_hostname=None, server_side=False): warnings.warn( "A true SSLContext object is not available. This prevents " "urllib3 from configuring SSL appropriately and may cause " "certain SSL connections to fail. You can upgrade to a newer " "version of Python to solve this. For more information, see " "https://urllib3.readthedocs.io/en/1.26.x/advanced-usage.html" "#ssl-warnings", InsecurePlatformWarning, ) kwargs = { "keyfile": self.keyfile, "certfile": self.certfile, "ca_certs": self.ca_certs, "cert_reqs": self.verify_mode, "ssl_version": self.protocol, "server_side": server_side, } return wrap_socket(socket, ciphers=self.ciphers, **kwargs) def assert_fingerprint(cert, fingerprint): """ Checks if given fingerprint matches the supplied certificate. :param cert: Certificate as bytes object. :param fingerprint: Fingerprint as string of hexdigits, can be interspersed by colons. """ fingerprint = fingerprint.replace(":", "").lower() digest_length = len(fingerprint) hashfunc = HASHFUNC_MAP.get(digest_length) if not hashfunc: raise SSLError("Fingerprint of invalid length: {0}".format(fingerprint)) # We need encode() here for py32; works on py2 and p33. fingerprint_bytes = unhexlify(fingerprint.encode()) cert_digest = hashfunc(cert).digest() if not _const_compare_digest(cert_digest, fingerprint_bytes): raise SSLError( 'Fingerprints did not match. Expected "{0}", got "{1}".'.format( fingerprint, hexlify(cert_digest) ) ) def resolve_cert_reqs(candidate): """ Resolves the argument to a numeric constant, which can be passed to the wrap_socket function/method from the ssl module. Defaults to :data:`ssl.CERT_REQUIRED`. If given a string it is assumed to be the name of the constant in the :mod:`ssl` module or its abbreviation. (So you can specify `REQUIRED` instead of `CERT_REQUIRED`. If it's neither `None` nor a string we assume it is already the numeric constant which can directly be passed to wrap_socket. """ if candidate is None: return CERT_REQUIRED if isinstance(candidate, str): res = getattr(ssl, candidate, None) if res is None: res = getattr(ssl, "CERT_" + candidate) return res return candidate def resolve_ssl_version(candidate): """ like resolve_cert_reqs """ if candidate is None: return PROTOCOL_TLS if isinstance(candidate, str): res = getattr(ssl, candidate, None) if res is None: res = getattr(ssl, "PROTOCOL_" + candidate) return res return candidate def create_urllib3_context( ssl_version=None, cert_reqs=None, options=None, ciphers=None ): """All arguments have the same meaning as ``ssl_wrap_socket``. By default, this function does a lot of the same work that ``ssl.create_default_context`` does on Python 3.4+. It: - Disables SSLv2, SSLv3, and compression - Sets a restricted set of server ciphers If you wish to enable SSLv3, you can do:: from pip._vendor.urllib3.util import ssl_ context = ssl_.create_urllib3_context() context.options &= ~ssl_.OP_NO_SSLv3 You can do the same to enable compression (substituting ``COMPRESSION`` for ``SSLv3`` in the last line above). :param ssl_version: The desired protocol version to use. This will default to PROTOCOL_SSLv23 which will negotiate the highest protocol that both the server and your installation of OpenSSL support. :param cert_reqs: Whether to require the certificate verification. This defaults to ``ssl.CERT_REQUIRED``. :param options: Specific OpenSSL options. These default to ``ssl.OP_NO_SSLv2``, ``ssl.OP_NO_SSLv3``, ``ssl.OP_NO_COMPRESSION``, and ``ssl.OP_NO_TICKET``. :param ciphers: Which cipher suites to allow the server to select. :returns: Constructed SSLContext object with specified options :rtype: SSLContext """ # PROTOCOL_TLS is deprecated in Python 3.10 if not ssl_version or ssl_version == PROTOCOL_TLS: ssl_version = PROTOCOL_TLS_CLIENT context = SSLContext(ssl_version) context.set_ciphers(ciphers or DEFAULT_CIPHERS) # Setting the default here, as we may have no ssl module on import cert_reqs = ssl.CERT_REQUIRED if cert_reqs is None else cert_reqs if options is None: options = 0 # SSLv2 is easily broken and is considered harmful and dangerous options |= OP_NO_SSLv2 # SSLv3 has several problems and is now dangerous options |= OP_NO_SSLv3 # Disable compression to prevent CRIME attacks for OpenSSL 1.0+ # (issue #309) options |= OP_NO_COMPRESSION # TLSv1.2 only. Unless set explicitly, do not request tickets. # This may save some bandwidth on wire, and although the ticket is encrypted, # there is a risk associated with it being on wire, # if the server is not rotating its ticketing keys properly. options |= OP_NO_TICKET context.options |= options # Enable post-handshake authentication for TLS 1.3, see GH #1634. PHA is # necessary for conditional client cert authentication with TLS 1.3. # The attribute is None for OpenSSL <= 1.1.0 or does not exist in older # versions of Python. We only enable on Python 3.7.4+ or if certificate # verification is enabled to work around Python issue #37428 # See: https://bugs.python.org/issue37428 if (cert_reqs == ssl.CERT_REQUIRED or sys.version_info >= (3, 7, 4)) and getattr( context, "post_handshake_auth", None ) is not None: context.post_handshake_auth = True def disable_check_hostname(): if ( getattr(context, "check_hostname", None) is not None ): # Platform-specific: Python 3.2 # We do our own verification, including fingerprints and alternative # hostnames. So disable it here context.check_hostname = False # The order of the below lines setting verify_mode and check_hostname # matter due to safe-guards SSLContext has to prevent an SSLContext with # check_hostname=True, verify_mode=NONE/OPTIONAL. This is made even more # complex because we don't know whether PROTOCOL_TLS_CLIENT will be used # or not so we don't know the initial state of the freshly created SSLContext. if cert_reqs == ssl.CERT_REQUIRED: context.verify_mode = cert_reqs disable_check_hostname() else: disable_check_hostname() context.verify_mode = cert_reqs # Enable logging of TLS session keys via defacto standard environment variable # 'SSLKEYLOGFILE', if the feature is available (Python 3.8+). Skip empty values. if hasattr(context, "keylog_filename"): sslkeylogfile = os.environ.get("SSLKEYLOGFILE") if sslkeylogfile: context.keylog_filename = sslkeylogfile return context def ssl_wrap_socket( sock, keyfile=None, certfile=None, cert_reqs=None, ca_certs=None, server_hostname=None, ssl_version=None, ciphers=None, ssl_context=None, ca_cert_dir=None, key_password=None, ca_cert_data=None, tls_in_tls=False, ): """ All arguments except for server_hostname, ssl_context, and ca_cert_dir have the same meaning as they do when using :func:`ssl.wrap_socket`. :param server_hostname: When SNI is supported, the expected hostname of the certificate :param ssl_context: A pre-made :class:`SSLContext` object. If none is provided, one will be created using :func:`create_urllib3_context`. :param ciphers: A string of ciphers we wish the client to support. :param ca_cert_dir: A directory containing CA certificates in multiple separate files, as supported by OpenSSL's -CApath flag or the capath argument to SSLContext.load_verify_locations(). :param key_password: Optional password if the keyfile is encrypted. :param ca_cert_data: Optional string containing CA certificates in PEM format suitable for passing as the cadata parameter to SSLContext.load_verify_locations() :param tls_in_tls: Use SSLTransport to wrap the existing socket. """ context = ssl_context if context is None: # Note: This branch of code and all the variables in it are no longer # used by urllib3 itself. We should consider deprecating and removing # this code. context = create_urllib3_context(ssl_version, cert_reqs, ciphers=ciphers) if ca_certs or ca_cert_dir or ca_cert_data: try: context.load_verify_locations(ca_certs, ca_cert_dir, ca_cert_data) except (IOError, OSError) as e: raise SSLError(e) elif ssl_context is None and hasattr(context, "load_default_certs"): # try to load OS default certs; works well on Windows (require Python3.4+) context.load_default_certs() # Attempt to detect if we get the goofy behavior of the # keyfile being encrypted and OpenSSL asking for the # passphrase via the terminal and instead error out. if keyfile and key_password is None and _is_key_file_encrypted(keyfile): raise SSLError("Client private key is encrypted, password is required") if certfile: if key_password is None: context.load_cert_chain(certfile, keyfile) else: context.load_cert_chain(certfile, keyfile, key_password) try: if hasattr(context, "set_alpn_protocols"): context.set_alpn_protocols(ALPN_PROTOCOLS) except NotImplementedError: # Defensive: in CI, we always have set_alpn_protocols pass # If we detect server_hostname is an IP address then the SNI # extension should not be used according to RFC3546 Section 3.1 use_sni_hostname = server_hostname and not is_ipaddress(server_hostname) # SecureTransport uses server_hostname in certificate verification. send_sni = (use_sni_hostname and HAS_SNI) or ( IS_SECURETRANSPORT and server_hostname ) # Do not warn the user if server_hostname is an invalid SNI hostname. if not HAS_SNI and use_sni_hostname: warnings.warn( "An HTTPS request has been made, but the SNI (Server Name " "Indication) extension to TLS is not available on this platform. " "This may cause the server to present an incorrect TLS " "certificate, which can cause validation failures. You can upgrade to " "a newer version of Python to solve this. For more information, see " "https://urllib3.readthedocs.io/en/1.26.x/advanced-usage.html" "#ssl-warnings", SNIMissingWarning, ) if send_sni: ssl_sock = _ssl_wrap_socket_impl( sock, context, tls_in_tls, server_hostname=server_hostname ) else: ssl_sock = _ssl_wrap_socket_impl(sock, context, tls_in_tls) return ssl_sock def is_ipaddress(hostname): """Detects whether the hostname given is an IPv4 or IPv6 address. Also detects IPv6 addresses with Zone IDs. :param str hostname: Hostname to examine. :return: True if the hostname is an IP address, False otherwise. """ if not six.PY2 and isinstance(hostname, bytes): # IDN A-label bytes are ASCII compatible. hostname = hostname.decode("ascii") return bool(IPV4_RE.match(hostname) or BRACELESS_IPV6_ADDRZ_RE.match(hostname)) def _is_key_file_encrypted(key_file): """Detects if a key file is encrypted or not.""" with open(key_file, "r") as f: for line in f: # Look for Proc-Type: 4,ENCRYPTED if "ENCRYPTED" in line: return True return False def _ssl_wrap_socket_impl(sock, ssl_context, tls_in_tls, server_hostname=None): if tls_in_tls: if not SSLTransport: # Import error, ssl is not available. raise ProxySchemeUnsupported( "TLS in TLS requires support for the 'ssl' module" ) SSLTransport._validate_ssl_context_for_tls_in_tls(ssl_context) return SSLTransport(sock, ssl_context, server_hostname) if server_hostname: return ssl_context.wrap_socket(sock, server_hostname=server_hostname) else: return ssl_context.wrap_socket(sock)