| 1 | # SPDX-License-Identifier: MIT |
| 2 | from __future__ import absolute_import |
| 3 | import errno |
| 4 | import warnings |
| 5 | import hmac |
| 6 | |
| 7 | from binascii import hexlify, unhexlify |
| 8 | from hashlib import md5, sha1, sha256 |
| 9 | |
| 10 | from ..exceptions import SSLError, InsecurePlatformWarning, SNIMissingWarning |
| 11 | |
| 12 | |
| 13 | SSLContext = None |
| 14 | HAS_SNI = False |
| 15 | IS_PYOPENSSL = False |
| 16 | IS_SECURETRANSPORT = False |
| 17 | |
| 18 | # Maps the length of a digest to a possible hash function producing this digest |
| 19 | HASHFUNC_MAP = { |
| 20 | 32: md5, |
| 21 | 40: sha1, |
| 22 | 64: sha256, |
| 23 | } |
| 24 | |
| 25 | |
| 26 | def _const_compare_digest_backport(a, b): |
| 27 | """ |
| 28 | Compare two digests of equal length in constant time. |
| 29 | |
| 30 | The digests must be of type str/bytes. |
| 31 | Returns True if the digests match, and False otherwise. |
| 32 | """ |
| 33 | result = abs(len(a) - len(b)) |
| 34 | for l, r in zip(bytearray(a), bytearray(b)): |
| 35 | result |= l ^ r |
| 36 | return result == 0 |
| 37 | |
| 38 | |
| 39 | _const_compare_digest = getattr(hmac, 'compare_digest', |
| 40 | _const_compare_digest_backport) |
| 41 | |
| 42 | |
| 43 | try: # Test for SSL features |
| 44 | import ssl |
| 45 | from ssl import wrap_socket, CERT_NONE, PROTOCOL_SSLv23 |
| 46 | from ssl import HAS_SNI # Has SNI? |
| 47 | except ImportError: |
| 48 | pass |
| 49 | |
| 50 | |
| 51 | try: |
| 52 | from ssl import OP_NO_SSLv2, OP_NO_SSLv3, OP_NO_COMPRESSION |
| 53 | except ImportError: |
| 54 | OP_NO_SSLv2, OP_NO_SSLv3 = 0x1000000, 0x2000000 |
| 55 | OP_NO_COMPRESSION = 0x20000 |
| 56 | |
| 57 | # A secure default. |
| 58 | # Sources for more information on TLS ciphers: |
| 59 | # |
| 60 | # - https://wiki.mozilla.org/Security/Server_Side_TLS |
| 61 | # - https://www.ssllabs.com/projects/best-practices/index.html |
| 62 | # - https://hynek.me/articles/hardening-your-web-servers-ssl-ciphers/ |
| 63 | # |
| 64 | # The general intent is: |
| 65 | # - Prefer cipher suites that offer perfect forward secrecy (DHE/ECDHE), |
| 66 | # - prefer ECDHE over DHE for better performance, |
| 67 | # - prefer any AES-GCM and ChaCha20 over any AES-CBC for better performance and |
| 68 | # security, |
| 69 | # - prefer AES-GCM over ChaCha20 because hardware-accelerated AES is common, |
| 70 | # - disable NULL authentication, MD5 MACs and DSS for security reasons. |
| 71 | DEFAULT_CIPHERS = ':'.join([ |
| 72 | 'ECDH+AESGCM', |
| 73 | 'ECDH+CHACHA20', |
| 74 | 'DH+AESGCM', |
| 75 | 'DH+CHACHA20', |
| 76 | 'ECDH+AES256', |
| 77 | 'DH+AES256', |
| 78 | 'ECDH+AES128', |
| 79 | 'DH+AES', |
| 80 | 'RSA+AESGCM', |
| 81 | 'RSA+AES', |
| 82 | '!aNULL', |
| 83 | '!eNULL', |
| 84 | '!MD5', |
| 85 | ]) |
| 86 | |
| 87 | try: |
| 88 | from ssl import SSLContext # Modern SSL? |
| 89 | except ImportError: |
| 90 | import sys |
| 91 | |
| 92 | class SSLContext(object): # Platform-specific: Python 2 & 3.1 |
| 93 | supports_set_ciphers = ((2, 7) <= sys.version_info < (3,) or |
| 94 | (3, 2) <= sys.version_info) |
| 95 | |
| 96 | def __init__(self, protocol_version): |
| 97 | self.protocol = protocol_version |
| 98 | # Use default values from a real SSLContext |
| 99 | self.check_hostname = False |
| 100 | self.verify_mode = ssl.CERT_NONE |
| 101 | self.ca_certs = None |
| 102 | self.options = 0 |
| 103 | self.certfile = None |
| 104 | self.keyfile = None |
| 105 | self.ciphers = None |
| 106 | |
| 107 | def load_cert_chain(self, certfile, keyfile): |
| 108 | self.certfile = certfile |
| 109 | self.keyfile = keyfile |
| 110 | |
| 111 | def load_verify_locations(self, cafile=None, capath=None): |
| 112 | self.ca_certs = cafile |
| 113 | |
| 114 | if capath is not None: |
| 115 | raise SSLError("CA directories not supported in older Pythons") |
| 116 | |
| 117 | def set_ciphers(self, cipher_suite): |
| 118 | if not self.supports_set_ciphers: |
| 119 | raise TypeError( |
| 120 | 'Your version of Python does not support setting ' |
| 121 | 'a custom cipher suite. Please upgrade to Python ' |
| 122 | '2.7, 3.2, or later if you need this functionality.' |
| 123 | ) |
| 124 | self.ciphers = cipher_suite |
| 125 | |
| 126 | def wrap_socket(self, socket, server_hostname=None, server_side=False): |
| 127 | warnings.warn( |
| 128 | 'A true SSLContext object is not available. This prevents ' |
| 129 | 'urllib3 from configuring SSL appropriately and may cause ' |
| 130 | 'certain SSL connections to fail. You can upgrade to a newer ' |
| 131 | 'version of Python to solve this. For more information, see ' |
| 132 | 'https://urllib3.readthedocs.io/en/latest/advanced-usage.html' |
| 133 | '#ssl-warnings', |
| 134 | InsecurePlatformWarning |
| 135 | ) |
| 136 | kwargs = { |
| 137 | 'keyfile': self.keyfile, |
| 138 | 'certfile': self.certfile, |
| 139 | 'ca_certs': self.ca_certs, |
| 140 | 'cert_reqs': self.verify_mode, |
| 141 | 'ssl_version': self.protocol, |
| 142 | 'server_side': server_side, |
| 143 | } |
| 144 | if self.supports_set_ciphers: # Platform-specific: Python 2.7+ |
| 145 | return wrap_socket(socket, ciphers=self.ciphers, **kwargs) |
| 146 | else: # Platform-specific: Python 2.6 |
| 147 | return wrap_socket(socket, **kwargs) |
| 148 | |
| 149 | |
| 150 | def assert_fingerprint(cert, fingerprint): |
| 151 | """ |
| 152 | Checks if given fingerprint matches the supplied certificate. |
| 153 | |
| 154 | :param cert: |
| 155 | Certificate as bytes object. |
| 156 | :param fingerprint: |
| 157 | Fingerprint as string of hexdigits, can be interspersed by colons. |
| 158 | """ |
| 159 | |
| 160 | fingerprint = fingerprint.replace(':', '').lower() |
| 161 | digest_length = len(fingerprint) |
| 162 | hashfunc = HASHFUNC_MAP.get(digest_length) |
| 163 | if not hashfunc: |
| 164 | raise SSLError( |
| 165 | 'Fingerprint of invalid length: {0}'.format(fingerprint)) |
| 166 | |
| 167 | # We need encode() here for py32; works on py2 and p33. |
| 168 | fingerprint_bytes = unhexlify(fingerprint.encode()) |
| 169 | |
| 170 | cert_digest = hashfunc(cert).digest() |
| 171 | |
| 172 | if not _const_compare_digest(cert_digest, fingerprint_bytes): |
| 173 | raise SSLError('Fingerprints did not match. Expected "{0}", got "{1}".' |
| 174 | .format(fingerprint, hexlify(cert_digest))) |
| 175 | |
| 176 | |
| 177 | def resolve_cert_reqs(candidate): |
| 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_NONE`. |
| 182 | If given a string it is assumed to be the name of the constant in the |
| 183 | :mod:`ssl` module or its abbrevation. |
| 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_NONE |
| 190 | |
| 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 |
| 196 | |
| 197 | return candidate |
| 198 | |
| 199 | |
| 200 | def resolve_ssl_version(candidate): |
| 201 | """ |
| 202 | like resolve_cert_reqs |
| 203 | """ |
| 204 | if candidate is None: |
| 205 | return PROTOCOL_SSLv23 |
| 206 | |
| 207 | if isinstance(candidate, str): |
| 208 | res = getattr(ssl, candidate, None) |
| 209 | if res is None: |
| 210 | res = getattr(ssl, 'PROTOCOL_' + candidate) |
| 211 | return res |
| 212 | |
| 213 | return candidate |
| 214 | |
| 215 | |
| 216 | def create_urllib3_context(ssl_version=None, cert_reqs=None, |
| 217 | options=None, ciphers=None): |
| 218 | """All arguments have the same meaning as ``ssl_wrap_socket``. |
| 219 | |
| 220 | By default, this function does a lot of the same work that |
| 221 | ``ssl.create_default_context`` does on Python 3.4+. It: |
| 222 | |
| 223 | - Disables SSLv2, SSLv3, and compression |
| 224 | - Sets a restricted set of server ciphers |
| 225 | |
| 226 | If you wish to enable SSLv3, you can do:: |
| 227 | |
| 228 | from urllib3.util import ssl_ |
| 229 | context = ssl_.create_urllib3_context() |
| 230 | context.options &= ~ssl_.OP_NO_SSLv3 |
| 231 | |
| 232 | You can do the same to enable compression (substituting ``COMPRESSION`` |
| 233 | for ``SSLv3`` in the last line above). |
| 234 | |
| 235 | :param ssl_version: |
| 236 | The desired protocol version to use. This will default to |
| 237 | PROTOCOL_SSLv23 which will negotiate the highest protocol that both |
| 238 | the server and your installation of OpenSSL support. |
| 239 | :param cert_reqs: |
| 240 | Whether to require the certificate verification. This defaults to |
| 241 | ``ssl.CERT_REQUIRED``. |
| 242 | :param options: |
| 243 | Specific OpenSSL options. These default to ``ssl.OP_NO_SSLv2``, |
| 244 | ``ssl.OP_NO_SSLv3``, ``ssl.OP_NO_COMPRESSION``. |
| 245 | :param ciphers: |
| 246 | Which cipher suites to allow the server to select. |
| 247 | :returns: |
| 248 | Constructed SSLContext object with specified options |
| 249 | :rtype: SSLContext |
| 250 | """ |
| 251 | context = SSLContext(ssl_version or ssl.PROTOCOL_SSLv23) |
| 252 | |
| 253 | # Setting the default here, as we may have no ssl module on import |
| 254 | cert_reqs = ssl.CERT_REQUIRED if cert_reqs is None else cert_reqs |
| 255 | |
| 256 | if options is None: |
| 257 | options = 0 |
| 258 | # SSLv2 is easily broken and is considered harmful and dangerous |
| 259 | options |= OP_NO_SSLv2 |
| 260 | # SSLv3 has several problems and is now dangerous |
| 261 | options |= OP_NO_SSLv3 |
| 262 | # Disable compression to prevent CRIME attacks for OpenSSL 1.0+ |
| 263 | # (issue #309) |
| 264 | options |= OP_NO_COMPRESSION |
| 265 | |
| 266 | context.options |= options |
| 267 | |
| 268 | if getattr(context, 'supports_set_ciphers', True): # Platform-specific: Python 2.6 |
| 269 | context.set_ciphers(ciphers or DEFAULT_CIPHERS) |
| 270 | |
| 271 | context.verify_mode = cert_reqs |
| 272 | if getattr(context, 'check_hostname', None) is not None: # Platform-specific: Python 3.2 |
| 273 | # We do our own verification, including fingerprints and alternative |
| 274 | # hostnames. So disable it here |
| 275 | context.check_hostname = False |
| 276 | return context |
| 277 | |
| 278 | |
| 279 | def ssl_wrap_socket(sock, keyfile=None, certfile=None, cert_reqs=None, |
| 280 | ca_certs=None, server_hostname=None, |
| 281 | ssl_version=None, ciphers=None, ssl_context=None, |
| 282 | ca_cert_dir=None): |
| 283 | """ |
| 284 | All arguments except for server_hostname, ssl_context, and ca_cert_dir have |
| 285 | the same meaning as they do when using :func:`ssl.wrap_socket`. |
| 286 | |
| 287 | :param server_hostname: |
| 288 | When SNI is supported, the expected hostname of the certificate |
| 289 | :param ssl_context: |
| 290 | A pre-made :class:`SSLContext` object. If none is provided, one will |
| 291 | be created using :func:`create_urllib3_context`. |
| 292 | :param ciphers: |
| 293 | A string of ciphers we wish the client to support. This is not |
| 294 | supported on Python 2.6 as the ssl module does not support it. |
| 295 | :param ca_cert_dir: |
| 296 | A directory containing CA certificates in multiple separate files, as |
| 297 | supported by OpenSSL's -CApath flag or the capath argument to |
| 298 | SSLContext.load_verify_locations(). |
| 299 | """ |
| 300 | context = ssl_context |
| 301 | if context is None: |
| 302 | # Note: This branch of code and all the variables in it are no longer |
| 303 | # used by urllib3 itself. We should consider deprecating and removing |
| 304 | # this code. |
| 305 | context = create_urllib3_context(ssl_version, cert_reqs, |
| 306 | ciphers=ciphers) |
| 307 | |
| 308 | if ca_certs or ca_cert_dir: |
| 309 | try: |
| 310 | context.load_verify_locations(ca_certs, ca_cert_dir) |
| 311 | except IOError as e: # Platform-specific: Python 2.6, 2.7, 3.2 |
| 312 | raise SSLError(e) |
| 313 | # Py33 raises FileNotFoundError which subclasses OSError |
| 314 | # These are not equivalent unless we check the errno attribute |
| 315 | except OSError as e: # Platform-specific: Python 3.3 and beyond |
| 316 | if e.errno == errno.ENOENT: |
| 317 | raise SSLError(e) |
| 318 | raise |
| 319 | elif getattr(context, 'load_default_certs', None) is not None: |
| 320 | # try to load OS default certs; works well on Windows (require Python3.4+) |
| 321 | context.load_default_certs() |
| 322 | |
| 323 | if certfile: |
| 324 | context.load_cert_chain(certfile, keyfile) |
| 325 | if HAS_SNI: # Platform-specific: OpenSSL with enabled SNI |
| 326 | return context.wrap_socket(sock, server_hostname=server_hostname) |
| 327 | |
| 328 | warnings.warn( |
| 329 | 'An HTTPS request has been made, but the SNI (Subject Name ' |
| 330 | 'Indication) extension to TLS is not available on this platform. ' |
| 331 | 'This may cause the server to present an incorrect TLS ' |
| 332 | 'certificate, which can cause validation failures. You can upgrade to ' |
| 333 | 'a newer version of Python to solve this. For more information, see ' |
| 334 | 'https://urllib3.readthedocs.io/en/latest/advanced-usage.html' |
| 335 | '#ssl-warnings', |
| 336 | SNIMissingWarning |
| 337 | ) |
| 338 | return context.wrap_socket(sock) |