| 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) |
