Add type hints to urllib3.contrib.pyopenssl
MODIFY
noxfile.py
noxfile.py
@@ -12,6 +12,7 @@ TYPED_FILES = {
"src/urllib3/contrib/_securetransport/bindings.py",
"src/urllib3/contrib/_securetransport/low_level.py",
"src/urllib3/contrib/ntlmpool.py",
+ "src/urllib3/contrib/pyopenssl.py",
"src/urllib3/contrib/securetransport.py",
"src/urllib3/contrib/socks.py",
"src/urllib3/__init__.py",

MODIFY
src/urllib3/contrib/pyopenssl.py
src/urllib3/contrib/pyopenssl.py
@@ -46,7 +46,7 @@ compression in Python 2 (see `CRIME attack`_).
.. _idna: https://github.com/kjd/idna
"""
-import OpenSSL.SSL
+import OpenSSL.SSL # type: ignore
from cryptography import x509
from cryptography.hazmat.backends.openssl import backend as openssl_backend
from cryptography.hazmat.backends.openssl.x509 import _Certificate
@@ -55,7 +55,7 @@ try:
from cryptography.x509 import UnsupportedExtension
except ImportError:
# UnsupportedExtension is gone in cryptography >= 2.1.0
- class UnsupportedExtension(Exception):
+ class UnsupportedExtension(Exception): # type: ignore
pass
@@ -64,9 +64,14 @@ import ssl
from io import BytesIO
from socket import socket as socket_cls
from socket import timeout
+from typing import TYPE_CHECKING, Any, Dict, List, Optional, Tuple, Union
from .. import util
+if TYPE_CHECKING:
+ from OpenSSL.crypto import CRL, X509 # type: ignore
+
+
__all__ = ["inject_into_urllib3", "extract_from_urllib3"]
# SNI always works.
@@ -79,8 +84,8 @@ USE_DEFAULT_SSLCONTEXT_CIPHERS = util.ssl_._is_ge_openssl_v1_1_1(
# Map from urllib3 to PyOpenSSL compatible parameter-values.
_openssl_versions = {
- util.ssl_.PROTOCOL_TLS: OpenSSL.SSL.SSLv23_METHOD,
- util.ssl_.PROTOCOL_TLS_CLIENT: OpenSSL.SSL.SSLv23_METHOD,
+ util.ssl_.PROTOCOL_TLS: OpenSSL.SSL.SSLv23_METHOD, # type: ignore
+ util.ssl_.PROTOCOL_TLS_CLIENT: OpenSSL.SSL.SSLv23_METHOD, # type: ignore
ssl.PROTOCOL_TLSv1: OpenSSL.SSL.TLSv1_METHOD,
}
@@ -113,13 +118,13 @@ orig_util_USE_SYSTEM_SSL_CIPHERS = util.ssl_.USE_DEFAULT_SSLCONTEXT_CIPHERS
log = logging.getLogger(__name__)
-def inject_into_urllib3():
+def inject_into_urllib3() -> None:
"Monkey-patch urllib3 with PyOpenSSL-backed SSL-support."
_validate_dependencies_met()
- util.SSLContext = PyOpenSSLContext
- util.ssl_.SSLContext = PyOpenSSLContext
+ util.SSLContext = PyOpenSSLContext # type: ignore
+ util.ssl_.SSLContext = PyOpenSSLContext # type: ignore
util.HAS_SNI = HAS_SNI
util.ssl_.HAS_SNI = HAS_SNI
util.IS_PYOPENSSL = True
@@ -127,7 +132,7 @@ def inject_into_urllib3():
util.ssl_.USE_DEFAULT_SSLCONTEXT_CIPHERS = USE_DEFAULT_SSLCONTEXT_CIPHERS
-def extract_from_urllib3():
+def extract_from_urllib3() -> None:
"Undo monkey-patching by :func:`inject_into_urllib3`."
util.SSLContext = orig_util_SSLContext
@@ -139,13 +144,13 @@ def extract_from_urllib3():
util.ssl_.USE_DEFAULT_SSLCONTEXT_CIPHERS = orig_util_USE_SYSTEM_SSL_CIPHERS
-def _validate_dependencies_met():
+def _validate_dependencies_met() -> None:
"""
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
+ from cryptography.x509.extensions import Extensions # type: ignore
if getattr(Extensions, "get_extension_for_class", None) is None:
raise ImportError(
@@ -165,7 +170,7 @@ def _validate_dependencies_met():
)
-def _dnsname_to_stdlib(name):
+def _dnsname_to_stdlib(name: str) -> Optional[str]:
"""
Converts a dNSName SubjectAlternativeName field to the form used by the
standard library on the given Python version.
@@ -179,7 +184,7 @@ def _dnsname_to_stdlib(name):
the name given should be skipped.
"""
- def idna_encode(name):
+ def idna_encode(name: str) -> Optional[bytes]:
"""
Borrowed wholesale from the Python Cryptography Project. It turns out
that we can't just safely call `idna.encode`: it can explode for
@@ -200,13 +205,13 @@ def _dnsname_to_stdlib(name):
if ":" in name:
return name
- name = idna_encode(name)
- if name is None:
+ encoded_name = idna_encode(name)
+ if encoded_name is None:
return None
- return name.decode("utf-8")
+ return encoded_name.decode("utf-8")
-def get_subj_alt_name(peer_cert):
+def get_subj_alt_name(peer_cert: "CRL") -> List[Tuple[str, str]]:
"""
Given an PyOpenSSL certificate, provides all the subject alternative names.
"""
@@ -263,24 +268,29 @@ def get_subj_alt_name(peer_cert):
class WrappedSocket:
"""API-compatibility wrapper for Python OpenSSL's Connection-class."""
- def __init__(self, connection, socket, suppress_ragged_eofs=True):
+ def __init__(
+ self,
+ connection: OpenSSL.SSL.Connection,
+ socket: socket_cls,
+ suppress_ragged_eofs: bool = True,
+ ) -> None:
self.connection = connection
self.socket = socket
self.suppress_ragged_eofs = suppress_ragged_eofs
self._io_refs = 0
self._closed = False
- def fileno(self):
+ def fileno(self) -> int:
return self.socket.fileno()
# Copy-pasted from Python 3.5 source code
- def _decref_socketios(self):
+ def _decref_socketios(self) -> None:
if self._io_refs > 0:
self._io_refs -= 1
if self._closed:
self.close()
- def recv(self, *args, **kwargs):
+ def recv(self, *args: Any, **kwargs: Any) -> bytes:
try:
data = self.connection.recv(*args, **kwargs)
except OpenSSL.SSL.SysCallError as e:
@@ -295,7 +305,7 @@ class WrappedSocket:
raise
except OpenSSL.SSL.WantReadError:
if not util.wait_for_read(self.socket, self.socket.gettimeout()):
- raise timeout("The read operation timed out")
+ raise timeout("The read operation timed out") # type: ignore
else:
return self.recv(*args, **kwargs)
@@ -303,11 +313,11 @@ class WrappedSocket:
except OpenSSL.SSL.Error as e:
raise ssl.SSLError(f"read error: {e!r}")
else:
- return data
+ return data # type: ignore
- def recv_into(self, *args, **kwargs):
+ def recv_into(self, *args: Any, **kwargs: Any) -> int:
try:
- return self.connection.recv_into(*args, **kwargs)
+ return self.connection.recv_into(*args, **kwargs) # type: ignore
except OpenSSL.SSL.SysCallError as e:
if self.suppress_ragged_eofs and e.args == (-1, "Unexpected EOF"):
return 0
@@ -320,7 +330,7 @@ class WrappedSocket:
raise
except OpenSSL.SSL.WantReadError:
if not util.wait_for_read(self.socket, self.socket.gettimeout()):
- raise timeout("The read operation timed out")
+ raise timeout("The read operation timed out") # type: ignore
else:
return self.recv_into(*args, **kwargs)
@@ -328,13 +338,13 @@ class WrappedSocket:
except OpenSSL.SSL.Error as e:
raise ssl.SSLError(f"read error: {e!r}")
- def settimeout(self, timeout):
+ def settimeout(self, timeout: float) -> None:
return self.socket.settimeout(timeout)
- def _send_until_done(self, data):
+ def _send_until_done(self, data: bytes) -> int:
while True:
try:
- return self.connection.send(data)
+ return self.connection.send(data) # type: ignore
except OpenSSL.SSL.WantWriteError:
if not util.wait_for_write(self.socket, self.socket.gettimeout()):
raise timeout()
@@ -342,7 +352,7 @@ class WrappedSocket:
except OpenSSL.SSL.SysCallError as e:
raise OSError(e.args[0], str(e))
- def sendall(self, data):
+ def sendall(self, data: bytes) -> None:
total_sent = 0
while total_sent < len(data):
sent = self._send_until_done(
@@ -350,39 +360,39 @@ class WrappedSocket:
)
total_sent += sent
- def shutdown(self):
+ def shutdown(self) -> None:
# FIXME rethrow compatible exceptions should we ever use this
self.connection.shutdown()
- def close(self):
+ def close(self) -> None:
if self._io_refs < 1:
try:
self._closed = True
- return self.connection.close()
+ return self.connection.close() # type: ignore
except OpenSSL.SSL.Error:
return
else:
self._io_refs -= 1
- def getpeercert(self, binary_form=False):
+ def getpeercert(self, binary_form: bool = False) -> Dict[str, List[Any]]:
x509 = self.connection.get_peer_certificate()
if not x509:
- return x509
+ return x509 # type: ignore
if binary_form:
- return OpenSSL.crypto.dump_certificate(OpenSSL.crypto.FILETYPE_ASN1, x509)
+ return OpenSSL.crypto.dump_certificate(OpenSSL.crypto.FILETYPE_ASN1, x509) # type: ignore
return {
- "subject": ((("commonName", x509.get_subject().CN),),),
+ "subject": ((("commonName", x509.get_subject().CN),),), # type: ignore
"subjectAltName": get_subj_alt_name(x509),
}
- def version(self):
- return self.connection.get_protocol_version_name()
+ def version(self) -> str:
+ return self.connection.get_protocol_version_name() # type: ignore
-WrappedSocket.makefile = socket_cls.makefile
+WrappedSocket.makefile = socket_cls.makefile # type: ignore
class PyOpenSSLContext:
@@ -392,42 +402,47 @@ class PyOpenSSLContext:
to calls into PyOpenSSL.
"""
- def __init__(self, protocol):
+ def __init__(self, protocol: int) -> None:
self.protocol = _openssl_versions[protocol]
self._ctx = OpenSSL.SSL.Context(self.protocol)
self._options = 0
self.check_hostname = False
@property
- def options(self):
+ def options(self) -> int:
return self._options
@options.setter
- def options(self, value):
+ def options(self, value: int) -> None:
self._options = value
self._ctx.set_options(value)
@property
- def verify_mode(self):
+ def verify_mode(self) -> int:
return _openssl_to_stdlib_verify[self._ctx.get_verify_mode()]
@verify_mode.setter
- def verify_mode(self, value):
+ def verify_mode(self, value: int) -> None:
self._ctx.set_verify(_stdlib_to_openssl_verify[value], _verify_callback)
- def set_default_verify_paths(self):
+ def set_default_verify_paths(self) -> None:
self._ctx.set_default_verify_paths()
- def set_ciphers(self, ciphers):
+ def set_ciphers(self, ciphers: Union[bytes, str]) -> None:
if isinstance(ciphers, str):
ciphers = ciphers.encode("utf-8")
self._ctx.set_cipher_list(ciphers)
- def load_verify_locations(self, cafile=None, capath=None, cadata=None):
+ def load_verify_locations(
+ self,
+ cafile: Optional[str] = None,
+ capath: Optional[str] = None,
+ cadata: Optional[bytes] = None,
+ ) -> None:
if cafile is not None:
- cafile = cafile.encode("utf-8")
+ cafile = cafile.encode("utf-8") # type: ignore
if capath is not None:
- capath = capath.encode("utf-8")
+ capath = capath.encode("utf-8") # type: ignore
try:
self._ctx.load_verify_locations(cafile, capath)
if cadata is not None:
@@ -435,26 +450,31 @@ class PyOpenSSLContext:
except OpenSSL.SSL.Error as e:
raise ssl.SSLError(f"unable to load trusted certificates: {e!r}")
- def load_cert_chain(self, certfile, keyfile=None, password=None):
+ def load_cert_chain(
+ self,
+ certfile: str,
+ keyfile: Optional[str] = None,
+ password: Optional[str] = None,
+ ) -> None:
self._ctx.use_certificate_chain_file(certfile)
if password is not None:
if not isinstance(password, bytes):
- password = password.encode("utf-8")
+ password = password.encode("utf-8") # type: ignore
self._ctx.set_passwd_cb(lambda *_: password)
self._ctx.use_privatekey_file(keyfile or certfile)
- def set_alpn_protocols(self, protocols):
+ def set_alpn_protocols(self, protocols: List[Union[bytes, str]]) -> None:
protocols = [util.util.to_bytes(p, "ascii") for p in protocols]
- return self._ctx.set_alpn_protos(protocols)
+ return self._ctx.set_alpn_protos(protocols) # type: ignore
def wrap_socket(
self,
- sock,
- server_side=False,
- do_handshake_on_connect=True,
- suppress_ragged_eofs=True,
- server_hostname=None,
- ):
+ sock: socket_cls,
+ server_side: bool = False,
+ do_handshake_on_connect: bool = True,
+ suppress_ragged_eofs: bool = True,
+ server_hostname: Optional[Union[bytes, str]] = None,
+ ) -> WrappedSocket:
cnx = OpenSSL.SSL.Connection(self._ctx, sock)
# If server_hostname is an IP, don't use it for SNI, per RFC6066 Section 3
@@ -470,7 +490,7 @@ class PyOpenSSLContext:
cnx.do_handshake()
except OpenSSL.SSL.WantReadError:
if not util.wait_for_read(sock, sock.gettimeout()):
- raise timeout("select timed out")
+ raise timeout("select timed out") # type: ignore
continue
except OpenSSL.SSL.Error as e:
raise ssl.SSLError(f"bad handshake: {e!r}")
@@ -479,5 +499,11 @@ class PyOpenSSLContext:
return WrappedSocket(cnx, sock)
-def _verify_callback(cnx, x509, err_no, err_depth, return_code):
+def _verify_callback(
+ cnx: OpenSSL.SSL.Connection,
+ x509: "X509",
+ err_no: int,
+ err_depth: int,
+ return_code: int,
+) -> bool:
return err_no == 0

Add type hints to urllib3.contrib.pyopenssl