Spaces:
Sleeping
Sleeping
whisper-large-v3
/
venv
/lib
/python3.10
/site-packages
/pip
/_vendor
/urllib3
/contrib
/pyopenssl.py
""" | |
TLS with SNI_-support for Python 2. Follow these instructions if you would | |
like to verify TLS certificates in Python 2. Note, the default libraries do | |
*not* do certificate checking; you need to do additional work to validate | |
certificates yourself. | |
This needs the following packages installed: | |
* `pyOpenSSL`_ (tested with 16.0.0) | |
* `cryptography`_ (minimum 1.3.4, from pyopenssl) | |
* `idna`_ (minimum 2.0, from cryptography) | |
However, pyopenssl depends on cryptography, which depends on idna, so while we | |
use all three directly here we end up having relatively few packages required. | |
You can install them with the following command: | |
.. code-block:: bash | |
$ python -m pip install pyopenssl cryptography idna | |
To activate certificate checking, call | |
:func:`~urllib3.contrib.pyopenssl.inject_into_urllib3` from your Python code | |
before you begin making HTTP requests. This can be done in a ``sitecustomize`` | |
module, or at any other time before your application begins using ``urllib3``, | |
like this: | |
.. code-block:: python | |
try: | |
import pip._vendor.urllib3.contrib.pyopenssl as pyopenssl | |
pyopenssl.inject_into_urllib3() | |
except ImportError: | |
pass | |
Now you can use :mod:`urllib3` as you normally would, and it will support SNI | |
when the required modules are installed. | |
Activating this module also has the positive side effect of disabling SSL/TLS | |
compression in Python 2 (see `CRIME attack`_). | |
.. _sni: https://en.wikipedia.org/wiki/Server_Name_Indication | |
.. _crime attack: https://en.wikipedia.org/wiki/CRIME_(security_exploit) | |
.. _pyopenssl: https://www.pyopenssl.org | |
.. _cryptography: https://cryptography.io | |
.. _idna: https://github.com/kjd/idna | |
""" | |
from __future__ import absolute_import | |
import OpenSSL.SSL | |
from cryptography import x509 | |
from cryptography.hazmat.backends.openssl import backend as openssl_backend | |
from cryptography.hazmat.backends.openssl.x509 import _Certificate | |
try: | |
from cryptography.x509 import UnsupportedExtension | |
except ImportError: | |
# UnsupportedExtension is gone in cryptography >= 2.1.0 | |
class UnsupportedExtension(Exception): | |
pass | |
from io import BytesIO | |
from socket import error as SocketError | |
from socket import timeout | |
try: # Platform-specific: Python 2 | |
from socket import _fileobject | |
except ImportError: # Platform-specific: Python 3 | |
_fileobject = None | |
from ..packages.backports.makefile import backport_makefile | |
import logging | |
import ssl | |
import sys | |
import warnings | |
from .. import util | |
from ..packages import six | |
from ..util.ssl_ import PROTOCOL_TLS_CLIENT | |
warnings.warn( | |
"'urllib3.contrib.pyopenssl' module is deprecated and will be removed " | |
"in a future release of urllib3 2.x. Read more in this issue: " | |
"https://github.com/urllib3/urllib3/issues/2680", | |
category=DeprecationWarning, | |
stacklevel=2, | |
) | |
__all__ = ["inject_into_urllib3", "extract_from_urllib3"] | |
# SNI always works. | |
HAS_SNI = True | |
# Map from urllib3 to PyOpenSSL compatible parameter-values. | |
_openssl_versions = { | |
util.PROTOCOL_TLS: OpenSSL.SSL.SSLv23_METHOD, | |
PROTOCOL_TLS_CLIENT: OpenSSL.SSL.SSLv23_METHOD, | |
ssl.PROTOCOL_TLSv1: OpenSSL.SSL.TLSv1_METHOD, | |
} | |
if hasattr(ssl, "PROTOCOL_SSLv3") and hasattr(OpenSSL.SSL, "SSLv3_METHOD"): | |
_openssl_versions[ssl.PROTOCOL_SSLv3] = OpenSSL.SSL.SSLv3_METHOD | |
if hasattr(ssl, "PROTOCOL_TLSv1_1") and hasattr(OpenSSL.SSL, "TLSv1_1_METHOD"): | |
_openssl_versions[ssl.PROTOCOL_TLSv1_1] = OpenSSL.SSL.TLSv1_1_METHOD | |
if hasattr(ssl, "PROTOCOL_TLSv1_2") and hasattr(OpenSSL.SSL, "TLSv1_2_METHOD"): | |
_openssl_versions[ssl.PROTOCOL_TLSv1_2] = OpenSSL.SSL.TLSv1_2_METHOD | |
_stdlib_to_openssl_verify = { | |
ssl.CERT_NONE: OpenSSL.SSL.VERIFY_NONE, | |
ssl.CERT_OPTIONAL: OpenSSL.SSL.VERIFY_PEER, | |
ssl.CERT_REQUIRED: OpenSSL.SSL.VERIFY_PEER | |
+ OpenSSL.SSL.VERIFY_FAIL_IF_NO_PEER_CERT, | |
} | |
_openssl_to_stdlib_verify = dict((v, k) for k, v in _stdlib_to_openssl_verify.items()) | |
# OpenSSL will only write 16K at a time | |
SSL_WRITE_BLOCKSIZE = 16384 | |
orig_util_HAS_SNI = util.HAS_SNI | |
orig_util_SSLContext = util.ssl_.SSLContext | |
log = logging.getLogger(__name__) | |
def inject_into_urllib3(): | |
"Monkey-patch urllib3 with PyOpenSSL-backed SSL-support." | |
_validate_dependencies_met() | |
util.SSLContext = PyOpenSSLContext | |
util.ssl_.SSLContext = PyOpenSSLContext | |
util.HAS_SNI = HAS_SNI | |
util.ssl_.HAS_SNI = HAS_SNI | |
util.IS_PYOPENSSL = True | |
util.ssl_.IS_PYOPENSSL = True | |
def extract_from_urllib3(): | |
"Undo monkey-patching by :func:`inject_into_urllib3`." | |
util.SSLContext = orig_util_SSLContext | |
util.ssl_.SSLContext = orig_util_SSLContext | |
util.HAS_SNI = orig_util_HAS_SNI | |
util.ssl_.HAS_SNI = orig_util_HAS_SNI | |
util.IS_PYOPENSSL = False | |
util.ssl_.IS_PYOPENSSL = False | |
def _validate_dependencies_met(): | |
""" | |
Verifies that PyOpenSSL's package-level dependencies have been met. | |
Throws `ImportError` if they are not met. | |
""" | |
# Method added in `cryptography==1.1`; not available in older versions | |
from cryptography.x509.extensions import Extensions | |
if getattr(Extensions, "get_extension_for_class", None) is None: | |
raise ImportError( | |
"'cryptography' module missing required functionality. " | |
"Try upgrading to v1.3.4 or newer." | |
) | |
# pyOpenSSL 0.14 and above use cryptography for OpenSSL bindings. The _x509 | |
# attribute is only present on those versions. | |
from OpenSSL.crypto import X509 | |
x509 = X509() | |
if getattr(x509, "_x509", None) is None: | |
raise ImportError( | |
"'pyOpenSSL' module missing required functionality. " | |
"Try upgrading to v0.14 or newer." | |
) | |
def _dnsname_to_stdlib(name): | |
""" | |
Converts a dNSName SubjectAlternativeName field to the form used by the | |
standard library on the given Python version. | |
Cryptography produces a dNSName as a unicode string that was idna-decoded | |
from ASCII bytes. We need to idna-encode that string to get it back, and | |
then on Python 3 we also need to convert to unicode via UTF-8 (the stdlib | |
uses PyUnicode_FromStringAndSize on it, which decodes via UTF-8). | |
If the name cannot be idna-encoded then we return None signalling that | |
the name given should be skipped. | |
""" | |
def idna_encode(name): | |
""" | |
Borrowed wholesale from the Python Cryptography Project. It turns out | |
that we can't just safely call `idna.encode`: it can explode for | |
wildcard names. This avoids that problem. | |
""" | |
from pip._vendor import idna | |
try: | |
for prefix in [u"*.", u"."]: | |
if name.startswith(prefix): | |
name = name[len(prefix) :] | |
return prefix.encode("ascii") + idna.encode(name) | |
return idna.encode(name) | |
except idna.core.IDNAError: | |
return None | |
# Don't send IPv6 addresses through the IDNA encoder. | |
if ":" in name: | |
return name | |
name = idna_encode(name) | |
if name is None: | |
return None | |
elif sys.version_info >= (3, 0): | |
name = name.decode("utf-8") | |
return name | |
def get_subj_alt_name(peer_cert): | |
""" | |
Given an PyOpenSSL certificate, provides all the subject alternative names. | |
""" | |
# Pass the cert to cryptography, which has much better APIs for this. | |
if hasattr(peer_cert, "to_cryptography"): | |
cert = peer_cert.to_cryptography() | |
else: | |
# This is technically using private APIs, but should work across all | |
# relevant versions before PyOpenSSL got a proper API for this. | |
cert = _Certificate(openssl_backend, peer_cert._x509) | |
# We want to find the SAN extension. Ask Cryptography to locate it (it's | |
# faster than looping in Python) | |
try: | |
ext = cert.extensions.get_extension_for_class(x509.SubjectAlternativeName).value | |
except x509.ExtensionNotFound: | |
# No such extension, return the empty list. | |
return [] | |
except ( | |
x509.DuplicateExtension, | |
UnsupportedExtension, | |
x509.UnsupportedGeneralNameType, | |
UnicodeError, | |
) as e: | |
# A problem has been found with the quality of the certificate. Assume | |
# no SAN field is present. | |
log.warning( | |
"A problem was encountered with the certificate that prevented " | |
"urllib3 from finding the SubjectAlternativeName field. This can " | |
"affect certificate validation. The error was %s", | |
e, | |
) | |
return [] | |
# We want to return dNSName and iPAddress fields. We need to cast the IPs | |
# back to strings because the match_hostname function wants them as | |
# strings. | |
# Sadly the DNS names need to be idna encoded and then, on Python 3, UTF-8 | |
# decoded. This is pretty frustrating, but that's what the standard library | |
# does with certificates, and so we need to attempt to do the same. | |
# We also want to skip over names which cannot be idna encoded. | |
names = [ | |
("DNS", name) | |
for name in map(_dnsname_to_stdlib, ext.get_values_for_type(x509.DNSName)) | |
if name is not None | |
] | |
names.extend( | |
("IP Address", str(name)) for name in ext.get_values_for_type(x509.IPAddress) | |
) | |
return names | |
class WrappedSocket(object): | |
"""API-compatibility wrapper for Python OpenSSL's Connection-class. | |
Note: _makefile_refs, _drop() and _reuse() are needed for the garbage | |
collector of pypy. | |
""" | |
def __init__(self, connection, socket, suppress_ragged_eofs=True): | |
self.connection = connection | |
self.socket = socket | |
self.suppress_ragged_eofs = suppress_ragged_eofs | |
self._makefile_refs = 0 | |
self._closed = False | |
def fileno(self): | |
return self.socket.fileno() | |
# Copy-pasted from Python 3.5 source code | |
def _decref_socketios(self): | |
if self._makefile_refs > 0: | |
self._makefile_refs -= 1 | |
if self._closed: | |
self.close() | |
def recv(self, *args, **kwargs): | |
try: | |
data = self.connection.recv(*args, **kwargs) | |
except OpenSSL.SSL.SysCallError as e: | |
if self.suppress_ragged_eofs and e.args == (-1, "Unexpected EOF"): | |
return b"" | |
else: | |
raise SocketError(str(e)) | |
except OpenSSL.SSL.ZeroReturnError: | |
if self.connection.get_shutdown() == OpenSSL.SSL.RECEIVED_SHUTDOWN: | |
return b"" | |
else: | |
raise | |
except OpenSSL.SSL.WantReadError: | |
if not util.wait_for_read(self.socket, self.socket.gettimeout()): | |
raise timeout("The read operation timed out") | |
else: | |
return self.recv(*args, **kwargs) | |
# TLS 1.3 post-handshake authentication | |
except OpenSSL.SSL.Error as e: | |
raise ssl.SSLError("read error: %r" % e) | |
else: | |
return data | |
def recv_into(self, *args, **kwargs): | |
try: | |
return self.connection.recv_into(*args, **kwargs) | |
except OpenSSL.SSL.SysCallError as e: | |
if self.suppress_ragged_eofs and e.args == (-1, "Unexpected EOF"): | |
return 0 | |
else: | |
raise SocketError(str(e)) | |
except OpenSSL.SSL.ZeroReturnError: | |
if self.connection.get_shutdown() == OpenSSL.SSL.RECEIVED_SHUTDOWN: | |
return 0 | |
else: | |
raise | |
except OpenSSL.SSL.WantReadError: | |
if not util.wait_for_read(self.socket, self.socket.gettimeout()): | |
raise timeout("The read operation timed out") | |
else: | |
return self.recv_into(*args, **kwargs) | |
# TLS 1.3 post-handshake authentication | |
except OpenSSL.SSL.Error as e: | |
raise ssl.SSLError("read error: %r" % e) | |
def settimeout(self, timeout): | |
return self.socket.settimeout(timeout) | |
def _send_until_done(self, data): | |
while True: | |
try: | |
return self.connection.send(data) | |
except OpenSSL.SSL.WantWriteError: | |
if not util.wait_for_write(self.socket, self.socket.gettimeout()): | |
raise timeout() | |
continue | |
except OpenSSL.SSL.SysCallError as e: | |
raise SocketError(str(e)) | |
def sendall(self, data): | |
total_sent = 0 | |
while total_sent < len(data): | |
sent = self._send_until_done( | |
data[total_sent : total_sent + SSL_WRITE_BLOCKSIZE] | |
) | |
total_sent += sent | |
def shutdown(self): | |
# FIXME rethrow compatible exceptions should we ever use this | |
self.connection.shutdown() | |
def close(self): | |
if self._makefile_refs < 1: | |
try: | |
self._closed = True | |
return self.connection.close() | |
except OpenSSL.SSL.Error: | |
return | |
else: | |
self._makefile_refs -= 1 | |
def getpeercert(self, binary_form=False): | |
x509 = self.connection.get_peer_certificate() | |
if not x509: | |
return x509 | |
if binary_form: | |
return OpenSSL.crypto.dump_certificate(OpenSSL.crypto.FILETYPE_ASN1, x509) | |
return { | |
"subject": ((("commonName", x509.get_subject().CN),),), | |
"subjectAltName": get_subj_alt_name(x509), | |
} | |
def version(self): | |
return self.connection.get_protocol_version_name() | |
def _reuse(self): | |
self._makefile_refs += 1 | |
def _drop(self): | |
if self._makefile_refs < 1: | |
self.close() | |
else: | |
self._makefile_refs -= 1 | |
if _fileobject: # Platform-specific: Python 2 | |
def makefile(self, mode, bufsize=-1): | |
self._makefile_refs += 1 | |
return _fileobject(self, mode, bufsize, close=True) | |
else: # Platform-specific: Python 3 | |
makefile = backport_makefile | |
WrappedSocket.makefile = makefile | |
class PyOpenSSLContext(object): | |
""" | |
I am a wrapper class for the PyOpenSSL ``Context`` object. I am responsible | |
for translating the interface of the standard library ``SSLContext`` object | |
to calls into PyOpenSSL. | |
""" | |
def __init__(self, protocol): | |
self.protocol = _openssl_versions[protocol] | |
self._ctx = OpenSSL.SSL.Context(self.protocol) | |
self._options = 0 | |
self.check_hostname = False | |
def options(self): | |
return self._options | |
def options(self, value): | |
self._options = value | |
self._ctx.set_options(value) | |
def verify_mode(self): | |
return _openssl_to_stdlib_verify[self._ctx.get_verify_mode()] | |
def verify_mode(self, value): | |
self._ctx.set_verify(_stdlib_to_openssl_verify[value], _verify_callback) | |
def set_default_verify_paths(self): | |
self._ctx.set_default_verify_paths() | |
def set_ciphers(self, ciphers): | |
if isinstance(ciphers, six.text_type): | |
ciphers = ciphers.encode("utf-8") | |
self._ctx.set_cipher_list(ciphers) | |
def load_verify_locations(self, cafile=None, capath=None, cadata=None): | |
if cafile is not None: | |
cafile = cafile.encode("utf-8") | |
if capath is not None: | |
capath = capath.encode("utf-8") | |
try: | |
self._ctx.load_verify_locations(cafile, capath) | |
if cadata is not None: | |
self._ctx.load_verify_locations(BytesIO(cadata)) | |
except OpenSSL.SSL.Error as e: | |
raise ssl.SSLError("unable to load trusted certificates: %r" % e) | |
def load_cert_chain(self, certfile, keyfile=None, password=None): | |
self._ctx.use_certificate_chain_file(certfile) | |
if password is not None: | |
if not isinstance(password, six.binary_type): | |
password = password.encode("utf-8") | |
self._ctx.set_passwd_cb(lambda *_: password) | |
self._ctx.use_privatekey_file(keyfile or certfile) | |
def set_alpn_protocols(self, protocols): | |
protocols = [six.ensure_binary(p) for p in protocols] | |
return self._ctx.set_alpn_protos(protocols) | |
def wrap_socket( | |
self, | |
sock, | |
server_side=False, | |
do_handshake_on_connect=True, | |
suppress_ragged_eofs=True, | |
server_hostname=None, | |
): | |
cnx = OpenSSL.SSL.Connection(self._ctx, sock) | |
if isinstance(server_hostname, six.text_type): # Platform-specific: Python 3 | |
server_hostname = server_hostname.encode("utf-8") | |
if server_hostname is not None: | |
cnx.set_tlsext_host_name(server_hostname) | |
cnx.set_connect_state() | |
while True: | |
try: | |
cnx.do_handshake() | |
except OpenSSL.SSL.WantReadError: | |
if not util.wait_for_read(sock, sock.gettimeout()): | |
raise timeout("select timed out") | |
continue | |
except OpenSSL.SSL.Error as e: | |
raise ssl.SSLError("bad handshake: %r" % e) | |
break | |
return WrappedSocket(cnx, sock) | |
def _verify_callback(cnx, x509, err_no, err_depth, return_code): | |
return err_no == 0 | |