Add type hints for proxy, ssl_, ssltransport and wait modules
MODIFY
noxfile.py
noxfile.py
@@ -18,9 +18,13 @@ TYPED_FILES = {
"src/urllib3/packages/ssl_match_hostname/__init__.py",
"src/urllib3/packages/ssl_match_hostname/_implementation.py",
"src/urllib3/util/connection.py",
+ "src/urllib3/util/proxy.py",
"src/urllib3/util/queue.py",
"src/urllib3/util/response.py",
+ "src/urllib3/util/ssl_.py",
+ "src/urllib3/util/ssltransport.py",
"src/urllib3/util/url.py",
+ "src/urllib3/util/wait.py",
}
SOURCE_FILES = [
"docs/",

MODIFY
src/urllib3/connection.py
src/urllib3/connection.py
@@ -14,7 +14,6 @@ from typing import (
TYPE_CHECKING,
Any,
Callable,
- Dict,
Iterable,
Mapping,
NamedTuple,
@@ -51,6 +50,7 @@ from .exceptions import (
from .packages.ssl_match_hostname import CertificateError, match_hostname
from .util import SKIP_HEADER, SKIPPABLE_HEADERS, connection, ssl_
from .util.ssl_ import (
+ PeerCertRetType,
assert_fingerprint,
create_urllib3_context,
resolve_cert_reqs,
@@ -327,12 +327,6 @@ class HTTPConnection(_HTTPConnection):
self.send(b"0\r\n\r\n")
-_PCTRTT = Tuple[Tuple[str, str], ...]
-_PCTRTTT = Tuple[_PCTRTT, ...]
-_PeerCertRetDictType = Dict[str, Union[str, _PCTRTTT, _PCTRTT]]
-_PeerCertRetType = Union[_PeerCertRetDictType, bytes, None]
-
-
class HTTPSConnection(HTTPConnection):
"""
Many of the parameters to this constructor are passed to the underlying SSL
@@ -586,7 +580,7 @@ class HTTPSConnection(HTTPConnection):
)
-def _match_hostname(cert: _PeerCertRetType, asserted_hostname: str) -> None:
+def _match_hostname(cert: PeerCertRetType, asserted_hostname: str) -> None:
try:
match_hostname(cert, asserted_hostname)
except CertificateError as e:

MODIFY
src/urllib3/util/proxy.py
src/urllib3/util/proxy.py
from typing import TYPE_CHECKING, Optional, Union
from .ssl_ import create_urllib3_context, resolve_cert_reqs, resolve_ssl_version
+from .url import Url
if TYPE_CHECKING:
import ssl
+ from urllib3.connection import ProxyConfig
+
def connection_requires_http_tunnel(
- proxy_url=None, proxy_config=None, destination_scheme=None
-):
+ proxy_url: Optional[Url] = None,
+ proxy_config: "Optional[ProxyConfig]" = None,
+ destination_scheme: Optional[str] = None,
+) -> bool:
"""
Returns True if the connection requires an HTTP CONNECT through the proxy.

MODIFY
src/urllib3/util/ssl_.py
src/urllib3/util/ssl_.py
@@ -3,9 +3,9 @@ import os
import socket
import sys
import warnings
-from binascii import hexlify, unhexlify
+from binascii import unhexlify
from hashlib import md5, sha1, sha256
-from typing import Optional, Union
+from typing import TYPE_CHECKING, Dict, Optional, Tuple, Union, cast, overload
from ..exceptions import ProxySchemeUnsupported, SNIMissingWarning, SSLError
from .url import _BRACELESS_IPV6_ADDRZ_RE, _IPV4_RE
@@ -35,9 +35,14 @@ def _is_ge_openssl_v1_1_1(
)
+if TYPE_CHECKING:
+ from typing_extensions import Literal
+
+ from .ssltransport import SSLTransport as SSLTransportType
+
try: # Do we have ssl at all?
import ssl
- from ssl import (
+ from ssl import ( # type: ignore
CERT_REQUIRED,
HAS_SNI,
OP_NO_COMPRESSION,
@@ -54,15 +59,21 @@ try: # Do we have ssl at all?
OPENSSL_VERSION, OPENSSL_VERSION_NUMBER
)
PROTOCOL_SSLv23 = PROTOCOL_TLS
- from .ssltransport import SSLTransport
+ from .ssltransport import SSLTransport # type: ignore
except ImportError:
- OP_NO_COMPRESSION = 0x20000
- OP_NO_TICKET = 0x4000
- OP_NO_SSLv2 = 0x1000000
- OP_NO_SSLv3 = 0x2000000
+ OP_NO_COMPRESSION = 0x20000 # type: ignore
+ OP_NO_TICKET = 0x4000 # type: ignore
+ OP_NO_SSLv2 = 0x1000000 # type: ignore
+ OP_NO_SSLv3 = 0x2000000 # type: ignore
PROTOCOL_SSLv23 = PROTOCOL_TLS = 2
+_PCTRTT = Tuple[Tuple[str, str], ...]
+_PCTRTTT = Tuple[_PCTRTT, ...]
+PeerCertRetDictType = Dict[str, Union[str, _PCTRTTT, _PCTRTT]]
+PeerCertRetType = Union[PeerCertRetDictType, bytes, None]
+
+
# A secure default.
# Sources for more information on TLS ciphers:
#
@@ -127,7 +138,7 @@ def assert_fingerprint(cert: Optional[bytes], fingerprint: str) -> None:
if not hmac.compare_digest(cert_digest, fingerprint_bytes):
raise SSLError(
- f'Fingerprints did not match. Expected "{fingerprint}", got "{hexlify(cert_digest)}".'
+ f'Fingerprints did not match. Expected "{fingerprint}", got "{cert_digest.hex()}".'
)
@@ -149,7 +160,7 @@ def resolve_cert_reqs(candidate: Union[None, int, str]) -> int:
res = getattr(ssl, candidate, None)
if res is None:
res = getattr(ssl, "CERT_" + candidate)
- return res
+ return cast(int, res)
return candidate
@@ -165,7 +176,7 @@ def resolve_ssl_version(candidate: Union[None, int, str]) -> int:
res = getattr(ssl, candidate, None)
if res is None:
res = getattr(ssl, "PROTOCOL_" + candidate)
- return res
+ return cast(int, res)
return candidate
@@ -210,8 +221,9 @@ def create_urllib3_context(
Constructed SSLContext object with specified options
:rtype: SSLContext
"""
+ if SSLContext is None:
+ raise TypeError("Can't create an SSLContext object without an ssl module")
context = SSLContext(ssl_version or PROTOCOL_TLS)
-
# Unless we're given ciphers defer to either system ciphers in
# the case of OpenSSL 1.1.1+ or use our own secure default ciphers.
if ciphers is not None or not USE_DEFAULT_SSLCONTEXT_CIPHERS:
@@ -264,6 +276,44 @@ def create_urllib3_context(
return context
+@overload
+def ssl_wrap_socket(
+ sock: socket.socket,
+ keyfile: Optional[str] = ...,
+ certfile: Optional[str] = ...,
+ cert_reqs: Optional[int] = ...,
+ ca_certs: Optional[str] = ...,
+ server_hostname: Optional[str] = ...,
+ ssl_version: Optional[int] = ...,
+ ciphers: Optional[str] = ...,
+ ssl_context: Optional["ssl.SSLContext"] = ...,
+ ca_cert_dir: Optional[str] = ...,
+ key_password: Optional[str] = ...,
+ ca_cert_data: Union[None, str, bytes] = ...,
+ tls_in_tls: "Literal[False]" = ...,
+) -> "ssl.SSLSocket":
+ ...
+
+
+@overload
+def ssl_wrap_socket(
+ sock: socket.socket,
+ keyfile: Optional[str] = ...,
+ certfile: Optional[str] = ...,
+ cert_reqs: Optional[int] = ...,
+ ca_certs: Optional[str] = ...,
+ server_hostname: Optional[str] = ...,
+ ssl_version: Optional[int] = ...,
+ ciphers: Optional[str] = ...,
+ ssl_context: Optional["ssl.SSLContext"] = ...,
+ ca_cert_dir: Optional[str] = ...,
+ key_password: Optional[str] = ...,
+ ca_cert_data: Union[None, str, bytes] = ...,
+ tls_in_tls: bool = ...,
+) -> Union["ssl.SSLSocket", "SSLTransportType"]:
+ ...
+
+
def ssl_wrap_socket(
sock: socket.socket,
keyfile: Optional[str] = None,
@@ -278,7 +328,7 @@ def ssl_wrap_socket(
key_password: Optional[str] = None,
ca_cert_data: Union[None, str, bytes] = None,
tls_in_tls: bool = False,
-) -> "ssl.SSLSocket":
+) -> Union["ssl.SSLSocket", "SSLTransportType"]:
"""
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`.
@@ -352,7 +402,7 @@ def ssl_wrap_socket(
return ssl_sock
-def is_ipaddress(hostname):
+def is_ipaddress(hostname: Union[str, bytes]) -> bool:
"""Detects whether the hostname given is an IPv4 or IPv6 address.
Also detects IPv6 addresses with Zone IDs.
@@ -365,7 +415,7 @@ def is_ipaddress(hostname):
return bool(_IPV4_RE.match(hostname) or _BRACELESS_IPV6_ADDRZ_RE.match(hostname))
-def _is_key_file_encrypted(key_file):
+def _is_key_file_encrypted(key_file: str) -> bool:
"""Detects if a key file is encrypted or not."""
with open(key_file) as f:
for line in f:
@@ -376,7 +426,12 @@ def _is_key_file_encrypted(key_file):
return False
-def _ssl_wrap_socket_impl(sock, ssl_context, tls_in_tls, server_hostname=None):
+def _ssl_wrap_socket_impl(
+ sock: socket.socket,
+ ssl_context: "ssl.SSLContext",
+ tls_in_tls: bool,
+ server_hostname: Optional[str] = None,
+) -> Union["ssl.SSLSocket", "SSLTransportType"]:
if tls_in_tls:
if not SSLTransport:
# Import error, ssl is not available.

MODIFY
src/urllib3/util/ssltransport.py
src/urllib3/util/ssltransport.py
import io
import socket
import ssl
+from typing import (
+ TYPE_CHECKING,
+ Any,
+ BinaryIO,
+ Callable,
+ List,
+ Optional,
+ TextIO,
+ Tuple,
+ TypeVar,
+ Union,
+ cast,
+ overload,
+)
from urllib3.exceptions import ProxySchemeUnsupported
+if TYPE_CHECKING:
+
+ from typing_extensions import Literal
+
+ from .ssl_ import PeerCertRetDictType, PeerCertRetType
+
+
+_SelfT = TypeVar("_SelfT", bound="SSLTransport")
+_WriteBuffer = Union[bytearray, memoryview]
+_ReturnValue = TypeVar("_ReturnValue")
+
SSL_BLOCKSIZE = 16384
@@ -19,7 +44,7 @@ class SSLTransport:
"""
@staticmethod
- def _validate_ssl_context_for_tls_in_tls(ssl_context):
+ def _validate_ssl_context_for_tls_in_tls(ssl_context: "ssl.SSLContext") -> None:
"""
Raises a ProxySchemeUnsupported if the provided ssl_context can't be used
for TLS in TLS.
@@ -35,8 +60,12 @@ class SSLTransport:
)
def __init__(
- self, socket, ssl_context, server_hostname=None, suppress_ragged_eofs=True
- ):
+ self,
+ socket: socket.socket,
+ ssl_context: "ssl.SSLContext",
+ server_hostname: Optional[str] = None,
+ suppress_ragged_eofs: bool = True,
+ ) -> None:
"""
Create an SSLTransport around socket using the provided ssl_context.
"""
@@ -53,24 +82,26 @@ class SSLTransport:
# Perform initial handshake.
self._ssl_io_loop(self.sslobj.do_handshake)
- def __enter__(self):
+ def __enter__(self: _SelfT) -> _SelfT:
return self
- def __exit__(self, *_):
+ def __exit__(self, *_: Any) -> None:
self.close()
- def fileno(self):
+ def fileno(self) -> int:
return self.socket.fileno()
- def read(self, len=1024, buffer=None):
+ def read(self, len: int = 1024, buffer: Optional[Any] = None) -> Union[int, bytes]:
return self._wrap_ssl_read(len, buffer)
- def recv(self, len=1024, flags=0):
+ def recv(self, buflen: int = 1024, flags: int = 0) -> Union[int, bytes]:
if flags != 0:
raise ValueError("non-zero flags not allowed in calls to recv")
- return self._wrap_ssl_read(len)
+ return self._wrap_ssl_read(buflen)
- def recv_into(self, buffer, nbytes=None, flags=0):
+ def recv_into(
+ self, buffer: _WriteBuffer, nbytes: Optional[int] = None, flags: int = 0
+ ) -> Union[None, int, bytes]:
if flags != 0:
raise ValueError("non-zero flags not allowed in calls to recv_into")
if buffer and (nbytes is None):
@@ -79,7 +110,7 @@ class SSLTransport:
nbytes = 1024
return self.read(nbytes, buffer)
- def sendall(self, data, flags=0):
+ def sendall(self, data: bytes, flags: int = 0) -> None:
if flags != 0:
raise ValueError("non-zero flags not allowed in calls to sendall")
count = 0
@@ -89,15 +120,20 @@ class SSLTransport:
v = self.send(byte_view[count:])
count += v
- def send(self, data, flags=0):
+ def send(self, data: bytes, flags: int = 0) -> int:
if flags != 0:
raise ValueError("non-zero flags not allowed in calls to send")
- response = self._ssl_io_loop(self.sslobj.write, data)
- return response
+ return self._ssl_io_loop(self.sslobj.write, data)
def makefile(
- self, mode="r", buffering=None, encoding=None, errors=None, newline=None
- ):
+ self,
+ mode: str,
+ buffering: Optional[int] = None,
+ *,
+ encoding: Optional[str] = None,
+ errors: Optional[str] = None,
+ newline: Optional[str] = None,
+ ) -> Union[BinaryIO, TextIO]:
"""
Python's httpclient uses makefile and buffered io when reading HTTP
messages and we need to support it.
@@ -118,7 +154,7 @@ class SSLTransport:
if writing:
rawmode += "w"
raw = socket.SocketIO(self, rawmode)
- self.socket._io_refs += 1
+ self.socket._io_refs += 1 # type: ignore
if buffering is None:
buffering = -1
if buffering < 0:
@@ -126,9 +162,10 @@ class SSLTransport:
if buffering == 0:
if not binary:
raise ValueError("unbuffered streams must be binary")
- return raw
+ return raw # type: ignore
+ buffer: BinaryIO
if reading and writing:
- buffer = io.BufferedRWPair(raw, raw, buffering)
+ buffer = io.BufferedRWPair(raw, raw, buffering) # type: ignore
elif reading:
buffer = io.BufferedReader(raw, buffering)
else:
@@ -137,46 +174,64 @@ class SSLTransport:
if binary:
return buffer
text = io.TextIOWrapper(buffer, encoding, errors, newline)
- text.mode = mode
+ text.mode = mode # type: ignore
return text
- def unwrap(self):
+ def unwrap(self) -> None:
self._ssl_io_loop(self.sslobj.unwrap)
- def close(self):
+ def close(self) -> None:
self.socket.close()
- def getpeercert(self, binary_form=False):
+ @overload
+ def getpeercert(
+ self, binary_form: "Literal[False]" = ...
+ ) -> Optional["PeerCertRetDictType"]:
+ ...
+
+ @overload
+ def getpeercert(self, binary_form: "Literal[True]") -> Optional[bytes]:
+ ...
+
+ @overload
+ def getpeercert(self, binary_form: bool) -> "PeerCertRetType":
+ ...
+
+ def getpeercert(
+ self, binary_form: bool = False
+ ) -> Union[None, bytes, "PeerCertRetDictType", "PeerCertRetType"]:
return self.sslobj.getpeercert(binary_form)
- def version(self):
+ def version(self) -> Optional[str]:
return self.sslobj.version()
- def cipher(self):
+ def cipher(self) -> Optional[Tuple[str, str, int]]:
return self.sslobj.cipher()
- def selected_alpn_protocol(self):
+ def selected_alpn_protocol(self) -> Optional[str]:
return self.sslobj.selected_alpn_protocol()
- def selected_npn_protocol(self):
+ def selected_npn_protocol(self) -> Optional[str]:
return self.sslobj.selected_npn_protocol()
- def shared_ciphers(self):
+ def shared_ciphers(self) -> Optional[List[Tuple[str, str, int]]]:
return self.sslobj.shared_ciphers()
- def compression(self):
+ def compression(self) -> Optional[str]:
return self.sslobj.compression()
- def settimeout(self, value):
+ def settimeout(self, value: Optional[float]) -> None:
self.socket.settimeout(value)
- def gettimeout(self):
+ def gettimeout(self) -> Optional[float]:
return self.socket.gettimeout()
- def _decref_socketios(self):
- self.socket._decref_socketios()
+ def _decref_socketios(self) -> None:
+ self.socket._decref_socketios() # type: ignore
- def _wrap_ssl_read(self, len, buffer=None):
+ def _wrap_ssl_read(
+ self, len: int, buffer: Optional[bytearray] = None
+ ) -> Union[int, bytes]:
try:
return self._ssl_io_loop(self.sslobj.read, len, buffer)
except ssl.SSLError as e:
@@ -185,7 +240,32 @@ class SSLTransport:
else:
raise
- def _ssl_io_loop(self, func, *args):
+ # func is sslobj.do_handshake or sslobj.unwrap
+ @overload
+ def _ssl_io_loop(self, func: Callable[[], None]) -> None:
+ ...
+
+ # func is sslobj.write, arg1 is data
+ @overload
+ def _ssl_io_loop(self, func: Callable[[bytes], int], arg1: bytes) -> int:
+ ...
+
+ # func is sslobj.read, arg1 is len, arg2 is buffer
+ @overload
+ def _ssl_io_loop(
+ self,
+ func: Callable[[int, Optional[bytearray]], bytes],
+ arg1: int,
+ arg2: Optional[bytearray],
+ ) -> bytes:
+ ...
+
+ def _ssl_io_loop(
+ self,
+ func: Callable[..., _ReturnValue],
+ arg1: Union[None, bytes, int] = None,
+ arg2: Optional[bytearray] = None,
+ ) -> _ReturnValue:
""" Performs an I/O loop between incoming/outgoing and the socket."""
should_loop = True
ret = None
@@ -193,7 +273,12 @@ class SSLTransport:
while should_loop:
errno = None
try:
- ret = func(*args)
+ if arg1 is None and arg2 is None:
+ ret = func()
+ elif arg2 is None:
+ ret = func(arg1)
+ else:
+ ret = func(arg1, arg2)
except ssl.SSLError as e:
if e.errno not in (ssl.SSL_ERROR_WANT_READ, ssl.SSL_ERROR_WANT_WRITE):
# WANT_READ, and WANT_WRITE are expected, others are not.
@@ -211,4 +296,4 @@ class SSLTransport:
self.incoming.write(buf)
else:
self.incoming.write_eof()
- return ret
+ return cast(_ReturnValue, ret)

MODIFY
src/urllib3/util/wait.py
src/urllib3/util/wait.py
import select
import socket
from functools import partial
-from typing import Optional
+from typing import List, Optional, Tuple
__all__ = ["wait_for_read", "wait_for_write"]
@@ -29,7 +29,12 @@ __all__ = ["wait_for_read", "wait_for_write"]
# fall back to select() in case poll() is somehow broken or missing.
-def select_wait_for_socket(sock, read=False, write=False, timeout=None):
+def select_wait_for_socket(
+ sock: socket.socket,
+ read: bool = False,
+ write: bool = False,
+ timeout: Optional[float] = None,
+) -> bool:
if not read and not write:
raise RuntimeError("must specify at least one of read=True, write=True")
rcheck = []
@@ -48,7 +53,12 @@ def select_wait_for_socket(sock, read=False, write=False, timeout=None):
return bool(rready or wready or xready)
-def poll_wait_for_socket(sock, read=False, write=False, timeout=None):
+def poll_wait_for_socket(
+ sock: socket.socket,
+ read: bool = False,
+ write: bool = False,
+ timeout: Optional[float] = None,
+) -> bool:
if not read and not write:
raise RuntimeError("must specify at least one of read=True, write=True")
mask = 0
@@ -60,7 +70,7 @@ def poll_wait_for_socket(sock, read=False, write=False, timeout=None):
poll_obj.register(sock, mask)
# For some reason, poll() takes timeout in milliseconds
- def do_poll(t):
+ def do_poll(t: Optional[float]) -> List[Tuple[int, int]]:
if t is not None:
t *= 1000
return poll_obj.poll(t)
@@ -68,7 +78,7 @@ def poll_wait_for_socket(sock, read=False, write=False, timeout=None):
return bool(do_poll(timeout))
-def _have_working_poll():
+def _have_working_poll() -> bool:
# Apparently some systems have a select.poll that fails as soon as you try
# to use it, either due to strange configuration or broken monkeypatching
# from libraries like eventlet/greenlet.
@@ -81,7 +91,12 @@ def _have_working_poll():
return True
-def wait_for_socket(*args, **kwargs):
+def wait_for_socket(
+ sock: socket.socket,
+ read: bool = False,
+ write: bool = False,
+ timeout: Optional[float] = None,
+) -> bool:
# We delay choosing which implementation to use until the first time we're
# called. We could do it at import time, but then we might make the wrong
# decision if someone goes wild with monkeypatching select.poll after
@@ -91,7 +106,7 @@ def wait_for_socket(*args, **kwargs):
wait_for_socket = poll_wait_for_socket
elif hasattr(select, "select"):
wait_for_socket = select_wait_for_socket
- return wait_for_socket(*args, **kwargs)
+ return wait_for_socket(sock, read, write, timeout)
def wait_for_read(sock: socket.socket, timeout: Optional[float] = None) -> bool:
@@ -101,7 +116,7 @@ def wait_for_read(sock: socket.socket, timeout: Optional[float] = None) -> bool:
return wait_for_socket(sock, read=True, timeout=timeout)
-def wait_for_write(sock, timeout=None):
+def wait_for_write(sock: socket.socket, timeout: Optional[float] = None) -> bool:
"""Waits for writing to be available on a given socket.
Returns True if the socket is readable, or False if the timeout expired.
"""

MODIFY
test/test_ssltransport.py
test/test_ssltransport.py
@@ -18,20 +18,10 @@ PER_TEST_TIMEOUT = 60
def server_client_ssl_contexts():
if hasattr(ssl, "PROTOCOL_TLS_SERVER"):
server_context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
- else:
- # python 3.5 workaround.
- # PROTOCOL_TLS_SERVER was added in 3.6
- server_context = ssl.SSLContext(ssl.PROTOCOL_TLS)
server_context.load_cert_chain(DEFAULT_CERTS["certfile"], DEFAULT_CERTS["keyfile"])
if hasattr(ssl, "PROTOCOL_TLS_CLIENT"):
client_context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
- else:
- # python 3.5 workaround.
- # PROTOCOL_TLS_SERVER was added in 3.6
- client_context = ssl.SSLContext(ssl.PROTOCOL_TLS)
- client_context.verify_mode = ssl.CERT_REQUIRED
- client_context.check_hostname = True
client_context.load_verify_locations(DEFAULT_CA)
return server_context, client_context

Add type hints for proxy, ssl_, ssltransport and wait modules