Create BaseHTTPConnection APIs

10 files changed, 425 insertions, 185 deletions

diff --git a/src/urllib3/__init__.py b/src/urllib3/__init__.py
index ec99d060..66f81506 100644
--- a/src/urllib3/__init__.py
+++ b/src/urllib3/__init__.py
@@ -9,9 +9,9 @@ from logging import NullHandler
 from typing import Any, Mapping, Optional, TextIO, Type, Union
 
 from . import exceptions
+from ._base_connection import _TYPE_BODY
 from ._collections import HTTPHeaderDict
 from ._version import __version__
-from .connection import _TYPE_BODY
 from .connectionpool import HTTPConnectionPool, HTTPSConnectionPool, connection_from_url
 from .filepost import _TYPE_FIELDS, encode_multipart_formdata
 from .poolmanager import PoolManager, ProxyManager, proxy_from_url
diff --git a/src/urllib3/_base_connection.py b/src/urllib3/_base_connection.py
new file mode 100644
index 00000000..7083298a
--- /dev/null
+++ b/src/urllib3/_base_connection.py
@@ -0,0 +1,171 @@
+import typing
+
+from .util.connection import _TYPE_SOCKET_OPTIONS
+from .util.timeout import _DEFAULT_TIMEOUT, _TYPE_TIMEOUT
+from .util.url import Url
+
+_TYPE_BODY = typing.Union[bytes, typing.IO[typing.Any], typing.Iterable[bytes], str]
+
+
+class ProxyConfig(typing.NamedTuple):
+    ssl_context: typing.Optional["ssl.SSLContext"]
+    use_forwarding_for_https: bool
+    assert_hostname: typing.Union[None, str, "Literal[False]"]
+    assert_fingerprint: typing.Optional[str]
+
+
+class _ResponseOptions(typing.NamedTuple):
+    # TODO: Remove this in favor of a better
+    # HTTP request/response lifecycle tracking.
+    request_method: str
+    request_url: str
+    preload_content: bool
+    decode_content: bool
+    enforce_content_length: bool
+
+
+if typing.TYPE_CHECKING:
+    import ssl
+
+    from typing_extensions import Literal, Protocol
+
+    from .response import BaseHTTPResponse
+
+    class BaseHTTPConnection(Protocol):
+        default_port: typing.ClassVar[int]
+        default_socket_options: typing.ClassVar[_TYPE_SOCKET_OPTIONS]
+
+        host: str
+        port: int
+        timeout: typing.Optional[
+            float
+        ]  # Instance doesn't store _DEFAULT_TIMEOUT, must be resolved.
+        blocksize: int
+        source_address: typing.Optional[typing.Tuple[str, int]]
+        socket_options: typing.Optional[_TYPE_SOCKET_OPTIONS]
+
+        proxy: typing.Optional[Url]
+        proxy_config: typing.Optional[ProxyConfig]
+
+        is_verified: bool
+        proxy_is_verified: typing.Optional[bool]
+
+        def __init__(
+            self,
+            host: str,
+            port: typing.Optional[int] = None,
+            *,
+            timeout: _TYPE_TIMEOUT = _DEFAULT_TIMEOUT,
+            source_address: typing.Optional[typing.Tuple[str, int]] = None,
+            blocksize: int = 8192,
+            socket_options: typing.Optional[_TYPE_SOCKET_OPTIONS] = ...,
+            proxy: typing.Optional[Url] = None,
+            proxy_config: typing.Optional[ProxyConfig] = None,
+        ) -> None:
+            ...
+
+        def set_tunnel(
+            self,
+            host: str,
+            port: typing.Optional[int] = None,
+            headers: typing.Optional[typing.Mapping[str, str]] = None,
+            scheme: str = "http",
+        ) -> None:
+            ...
+
+        def connect(self) -> None:
+            ...
+
+        def request(
+            self,
+            method: str,
+            url: str,
+            body: typing.Optional[_TYPE_BODY] = None,
+            headers: typing.Optional[typing.Mapping[str, str]] = None,
+            # We know *at least* botocore is depending on the order of the
+            # first 3 parameters so to be safe we only mark the later ones
+            # as keyword-only to ensure we have space to extend.
+            *,
+            chunked: bool = False,
+            preload_content: bool = True,
+            decode_content: bool = True,
+            enforce_content_length: bool = True,
+        ) -> None:
+            ...
+
+        def getresponse(self) -> "BaseHTTPResponse":
+            ...
+
+        def close(self) -> None:
+            ...
+
+        @property
+        def is_closed(self) -> bool:
+            """Whether the connection either is brand new or has been previously closed.
+            If this property is True then both ``is_connected`` and ``has_connected_to_proxy``
+            properties must be False.
+            """
+
+        @property
+        def is_connected(self) -> bool:
+            """Whether the connection is actively connected to any origin (proxy or target)"""
+
+        @property
+        def has_connected_to_proxy(self) -> bool:
+            """Whether the connection has successfully connected to its proxy.
+            This returns False if no proxy is in use. Used to determine whether
+            errors are coming from the proxy layer or from tunnelling to the target origin.
+            """
+
+    class BaseHTTPSConnection(BaseHTTPConnection, Protocol):
+        default_port: typing.ClassVar[int]
+        default_socket_options: typing.ClassVar[_TYPE_SOCKET_OPTIONS]
+
+        # Certificate verification methods
+        cert_reqs: typing.Optional[typing.Union[int, str]]
+        assert_hostname: typing.Union[None, str, "Literal[False]"]
+        assert_fingerprint: typing.Optional[str]
+        ssl_context: typing.Optional[ssl.SSLContext]
+
+        # Trusted CAs
+        ca_certs: typing.Optional[str]
+        ca_cert_dir: typing.Optional[str]
+        ca_cert_data: typing.Union[None, str, bytes]
+
+        # TLS version
+        ssl_minimum_version: typing.Optional[int]
+        ssl_maximum_version: typing.Optional[int]
+        ssl_version: typing.Optional[typing.Union[int, str]]  # Deprecated
+
+        # Client certificates
+        cert_file: typing.Optional[str]
+        key_file: typing.Optional[str]
+        key_password: typing.Optional[str]
+
+        def __init__(
+            self,
+            host: str,
+            port: typing.Optional[int] = None,
+            *,
+            timeout: _TYPE_TIMEOUT = _DEFAULT_TIMEOUT,
+            source_address: typing.Optional[typing.Tuple[str, int]] = None,
+            blocksize: int = 8192,
+            socket_options: typing.Optional[_TYPE_SOCKET_OPTIONS] = ...,
+            proxy: typing.Optional[Url] = None,
+            proxy_config: typing.Optional[ProxyConfig] = None,
+            cert_reqs: typing.Optional[typing.Union[int, str]] = None,
+            assert_hostname: typing.Union[None, str, "Literal[False]"] = None,
+            assert_fingerprint: typing.Optional[str] = None,
+            server_hostname: typing.Optional[str] = None,
+            ssl_context: typing.Optional["ssl.SSLContext"] = None,
+            ca_certs: typing.Optional[str] = None,
+            ca_cert_dir: typing.Optional[str] = None,
+            ca_cert_data: typing.Union[None, str, bytes] = None,
+            ssl_minimum_version: typing.Optional[int] = None,
+            ssl_maximum_version: typing.Optional[int] = None,
+            ssl_version: typing.Optional[typing.Union[int, str]] = None,  # Deprecated
+            cert_file: typing.Optional[str] = None,
+            key_file: typing.Optional[str] = None,
+            key_password: typing.Optional[str] = None,
+        ) -> None:
+            ...
diff --git a/src/urllib3/_request_methods.py b/src/urllib3/_request_methods.py
index d4bada0d..df65b152 100644
--- a/src/urllib3/_request_methods.py
+++ b/src/urllib3/_request_methods.py
@@ -2,8 +2,8 @@ import json as _json
 from typing import Any, Dict, Mapping, Optional, Sequence, Tuple, Union
 from urllib.parse import urlencode
 
+from ._base_connection import _TYPE_BODY
 from ._collections import HTTPHeaderDict
-from .connection import _TYPE_BODY
 from .filepost import _TYPE_FIELDS, encode_multipart_formdata
 from .response import BaseHTTPResponse
 
diff --git a/src/urllib3/connection.py b/src/urllib3/connection.py
index 679c4514..1eb9b54c 100644
--- a/src/urllib3/connection.py
+++ b/src/urllib3/connection.py
@@ -9,12 +9,8 @@ from http.client import HTTPException as HTTPException  # noqa: F401
 from http.client import ResponseNotReady
 from socket import timeout as SocketTimeout
 from typing import (
-    IO,
     TYPE_CHECKING,
-    Any,
-    Callable,
     ClassVar,
-    Iterable,
     Mapping,
     NamedTuple,
     Optional,
@@ -34,6 +30,7 @@ from ._collections import HTTPHeaderDict
 from .util.response import assert_header_parsing
 from .util.timeout import _DEFAULT_TIMEOUT, _TYPE_TIMEOUT, Timeout
 from .util.util import to_str
+from .util.wait import wait_for_read
 
 try:  # Compiled with SSL?
     import ssl
@@ -46,6 +43,9 @@ except (ImportError, AttributeError):
         pass
 
 
+from ._base_connection import _TYPE_BODY
+from ._base_connection import ProxyConfig as ProxyConfig
+from ._base_connection import _ResponseOptions as _ResponseOptions
 from ._version import __version__
 from .exceptions import (
     ConnectTimeoutError,
@@ -83,25 +83,6 @@ RECENT_DATE = datetime.date(2022, 1, 1)
 
 _CONTAINS_CONTROL_CHAR_RE = re.compile(r"[^-!#$%&'*+.^_`|~0-9a-zA-Z]")
 
-_TYPE_BODY = Union[bytes, IO[Any], Iterable[bytes], str]
-
-
-class ProxyConfig(NamedTuple):
-    ssl_context: Optional["ssl.SSLContext"]
-    use_forwarding_for_https: bool
-    assert_hostname: Union[None, str, "Literal[False]"]
-    assert_fingerprint: Optional[str]
-
-
-class _ResponseOptions(NamedTuple):
-    # TODO: Remove this in favor of a better
-    # HTTP request/response lifecycle tracking.
-    request_method: str
-    request_url: str
-    preload_content: bool
-    decode_content: bool
-    enforce_content_length: bool
-
 
 class HTTPConnection(_HTTPConnection):
     """
@@ -128,33 +109,36 @@ class HTTPConnection(_HTTPConnection):
       Or you may want to disable the defaults by passing an empty list (e.g., ``[]``).
     """
 
-    default_port: int = port_by_scheme["http"]
+    default_port: ClassVar[int] = port_by_scheme["http"]  # type: ignore[misc]
 
     #: Disable Nagle's algorithm by default.
     #: ``[(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)]``
-    default_socket_options: connection._TYPE_SOCKET_OPTIONS = [
+    default_socket_options: ClassVar[connection._TYPE_SOCKET_OPTIONS] = [
         (socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
     ]
 
     #: Whether this connection verifies the host's certificate.
     is_verified: bool = False
 
-    #: Whether this proxy connection (if used) verifies the proxy host's
-    #: certificate. If no HTTPS proxy is being used will be ``None``.
+    #: Whether this proxy connection verified the proxy host's certificate.
+    # If no proxy is currently connected to the value will be ``None``.
     proxy_is_verified: Optional[bool] = None
 
     blocksize: int
     source_address: Optional[Tuple[str, int]]
     socket_options: Optional[connection._TYPE_SOCKET_OPTIONS]
-    _tunnel_host: Optional[str]
-    _tunnel: ClassVar[Callable[["HTTPConnection"], None]]
-    _connecting_to_proxy: bool
+
+    _has_connected_to_proxy: bool
     _response_options: Optional[_ResponseOptions]
+    _tunnel_host: Optional[str]
+    _tunnel_port: Optional[int]
+    _tunnel_scheme: Optional[str]
 
     def __init__(
         self,
         host: str,
         port: Optional[int] = None,
+        *,
         timeout: _TYPE_TIMEOUT = _DEFAULT_TIMEOUT,
         source_address: Optional[Tuple[str, int]] = None,
         blocksize: int = 8192,
@@ -164,15 +148,6 @@ class HTTPConnection(_HTTPConnection):
         proxy: Optional[Url] = None,
         proxy_config: Optional[ProxyConfig] = None,
     ) -> None:
-        # Pre-set source_address.
-        self.source_address = source_address
-
-        self.socket_options = socket_options
-
-        # Proxy options provided by the user.
-        self.proxy = proxy
-        self.proxy_config = proxy_config
-
         super().__init__(
             host=host,
             port=port,
@@ -180,9 +155,15 @@ class HTTPConnection(_HTTPConnection):
             source_address=source_address,
             blocksize=blocksize,
         )
+        self.socket_options = socket_options
+        self.proxy = proxy
+        self.proxy_config = proxy_config
 
-        self._connecting_to_proxy = False
+        self._has_connected_to_proxy = False
         self._response_options = None
+        self._tunnel_host: Optional[str] = None
+        self._tunnel_port: Optional[int] = None
+        self._tunnel_scheme: Optional[str] = None
 
     # https://github.com/python/mypy/issues/4125
     # Mypy treats this as LSP violation, which is considered a bug.
@@ -223,7 +204,6 @@ class HTTPConnection(_HTTPConnection):
 
         :return: New socket connection.
         """
-
         try:
             sock = connection.create_connection(
                 (self._dns_host, self.port),
@@ -246,25 +226,62 @@ class HTTPConnection(_HTTPConnection):
 
         return sock
 
-    def _is_using_tunnel(self) -> Optional[str]:
-        return self._tunnel_host
+    def set_tunnel(
+        self,
+        host: str,
+        port: Optional[int] = None,
+        headers: Optional[Mapping[str, str]] = None,
+        scheme: str = "http",
+    ) -> None:
+        if scheme not in ("http", "https"):
+            raise ValueError(
+                f"Invalid proxy scheme for tunneling: {scheme!r}, must be either 'http' or 'https'"
+            )
+        super().set_tunnel(host, port=port, headers=headers)
+        self._tunnel_scheme = scheme
+
+    def connect(self) -> None:
+        self.sock = self._new_conn()
+        if self._tunnel_host:
+            # If we're tunneling it means we're connected to our proxy.
+            self._has_connected_to_proxy = True
 
-    def _prepare_conn(self, conn: socket.socket) -> None:
-        self.sock = conn
-        if self._is_using_tunnel():
             # TODO: Fix tunnel so it doesn't depend on self.sock state.
-            self._tunnel()
-            self._connecting_to_proxy = False
+            self._tunnel()  # type: ignore[attr-defined]
 
-    def connect(self) -> None:
-        self._connecting_to_proxy = bool(self.proxy)
-        conn = self._new_conn()
-        self._prepare_conn(conn)
-        self._connecting_to_proxy = False
+        # If there's a proxy to be connected to we are fully connected.
+        # This is set twice (once above and here) due to forwarding proxies
+        # not using tunnelling.
+        self._has_connected_to_proxy = bool(self.proxy)
+
+    @property
+    def is_closed(self) -> bool:
+        return self.sock is None
+
+    @property
+    def is_connected(self) -> bool:
+        if self.sock is None:
+            return False
+        return not wait_for_read(self.sock, timeout=0.0)
+
+    @property
+    def has_connected_to_proxy(self) -> bool:
+        return self._has_connected_to_proxy
 
     def close(self) -> None:
-        self._connecting_to_proxy = False
-        super().close()
+        try:
+            super().close()
+        finally:
+            # Reset all stateful properties so connection
+            # can be re-used without leaking prior configs.
+            self.sock = None
+            self.is_verified = False
+            self.proxy_is_verified = None
+            self._has_connected_to_proxy = False
+            self._response_options = None
+            self._tunnel_host = None
+            self._tunnel_port = None
+            self._tunnel_scheme = None
 
     def putrequest(
         self,
@@ -306,6 +323,7 @@ class HTTPConnection(_HTTPConnection):
         url: str,
         body: Optional[_TYPE_BODY] = None,
         headers: Optional[Mapping[str, str]] = None,
+        *,
         chunked: bool = False,
         preload_content: bool = True,
         decode_content: bool = True,
@@ -403,6 +421,12 @@ class HTTPConnection(_HTTPConnection):
         Alternative to the common request method, which sends the
         body with chunked encoding and not as one block
         """
+        warnings.warn(
+            "HTTPConnection.request_chunked() is deprecated and will be removed in a "
+            "future version. Instead use HTTPConnection.request(..., chunked=True).",
+            category=DeprecationWarning,
+            stacklevel=2,
+        )
         self.request(method, url, body=body, headers=headers, chunked=True)
 
     def getresponse(  # type: ignore[override]
@@ -423,6 +447,10 @@ class HTTPConnection(_HTTPConnection):
         resp_options = self._response_options
         self._response_options = None
 
+        # Since the connection's timeout value may have been updated
+        # we need to set the timeout on the socket.
+        self.sock.settimeout(self.timeout)
+
         # This is needed here to avoid circular import errors
         from .response import HTTPResponse
 
@@ -463,7 +491,7 @@ class HTTPSConnection(HTTPConnection):
     socket by means of :py:func:`urllib3.util.ssl_wrap_socket`.
     """
 
-    default_port = port_by_scheme["https"]
+    default_port = port_by_scheme["https"]  # type: ignore[misc]
 
     cert_reqs: Optional[Union[int, str]] = None
     ca_certs: Optional[str] = None
@@ -473,18 +501,13 @@ class HTTPSConnection(HTTPConnection):
     ssl_minimum_version: Optional[int] = None
     ssl_maximum_version: Optional[int] = None
     assert_fingerprint: Optional[str] = None
-    tls_in_tls_required: bool = False
 
     def __init__(
         self,
         host: str,
         port: Optional[int] = None,
-        key_file: Optional[str] = None,
-        cert_file: Optional[str] = None,
-        key_password: Optional[str] = None,
+        *,
         timeout: _TYPE_TIMEOUT = _DEFAULT_TIMEOUT,
-        ssl_context: Optional["ssl.SSLContext"] = None,
-        server_hostname: Optional[str] = None,
         source_address: Optional[Tuple[str, int]] = None,
         blocksize: int = 8192,
         socket_options: Optional[
@@ -492,6 +515,20 @@ class HTTPSConnection(HTTPConnection):
         ] = HTTPConnection.default_socket_options,
         proxy: Optional[Url] = None,
         proxy_config: Optional[ProxyConfig] = None,
+        cert_reqs: Optional[Union[int, str]] = None,
+        assert_hostname: Union[None, str, "Literal[False]"] = None,
+        assert_fingerprint: Optional[str] = None,
+        server_hostname: Optional[str] = None,
+        ssl_context: Optional["ssl.SSLContext"] = None,
+        ca_certs: Optional[str] = None,
+        ca_cert_dir: Optional[str] = None,
+        ca_cert_data: Union[None, str, bytes] = None,
+        ssl_minimum_version: Optional[int] = None,
+        ssl_maximum_version: Optional[int] = None,
+        ssl_version: Optional[Union[int, str]] = None,  # Deprecated
+        cert_file: Optional[str] = None,
+        key_file: Optional[str] = None,
+        key_password: Optional[str] = None,
     ) -> None:
 
         super().__init__(
@@ -510,9 +547,22 @@ class HTTPSConnection(HTTPConnection):
         self.key_password = key_password
         self.ssl_context = ssl_context
         self.server_hostname = server_hostname
-        self.ssl_version = None
-        self.ssl_minimum_version = None
-        self.ssl_maximum_version = None
+        self.assert_hostname = assert_hostname
+        self.assert_fingerprint = assert_fingerprint
+        self.ssl_version = ssl_version
+        self.ssl_minimum_version = ssl_minimum_version
+        self.ssl_maximum_version = ssl_maximum_version
+        self.ca_certs = ca_certs and os.path.expanduser(ca_certs)
+        self.ca_cert_dir = ca_cert_dir and os.path.expanduser(ca_cert_dir)
+        self.ca_cert_data = ca_cert_data
+
+        # cert_reqs depends on ssl_context so calculate last.
+        if cert_reqs is None:
+            if self.ssl_context is not None:
+                cert_reqs = self.ssl_context.verify_mode
+            else:
+                cert_reqs = resolve_cert_reqs(None)
+        self.cert_reqs = cert_reqs
 
     def set_cert(
         self,
@@ -529,6 +579,14 @@ class HTTPSConnection(HTTPConnection):
         """
         This method should only be called once, before the connection is used.
         """
+        warnings.warn(
+            "HTTPSConnection.set_cert() is deprecated and will be removed in a "
+            "future version. Instead provide the parameters to the HTTPSConnection "
+            "constructor.",
+            category=DeprecationWarning,
+            stacklevel=2,
+        )
+
         # If cert_reqs is not provided we'll assume CERT_REQUIRED unless we also
         # have an SSLContext object in which case we'll use its verify_mode.
         if cert_reqs is None:
@@ -548,25 +606,24 @@ class HTTPSConnection(HTTPConnection):
         self.ca_cert_data = ca_cert_data
 
     def connect(self) -> None:
-        self._connecting_to_proxy = bool(self.proxy)
-
         sock: Union[socket.socket, "ssl.SSLSocket"]
         self.sock = sock = self._new_conn()
         server_hostname: str = self.host
         tls_in_tls = False
 
-        if self._is_using_tunnel():
-            if self.tls_in_tls_required:
+        # Do we need to establish a tunnel?
+        if self._tunnel_host is not None:
+            # We're tunneling to an HTTPS origin so need to do TLS-in-TLS.
+            if self._tunnel_scheme == "https":
                 self.sock = sock = self._connect_tls_proxy(self.host, sock)
                 tls_in_tls = True
 
-            self._connecting_to_proxy = False
+            # If we're tunneling it means we're connected to our proxy.
+            self._has_connected_to_proxy = True
 
-            self._tunnel()
+            self._tunnel()  # type: ignore[attr-defined]
             # Override the host with the one we're requesting data from.
-            server_hostname = cast(
-                str, self._tunnel_host
-            )  # self._tunnel_host is not None, because self._is_using_tunnel() returned a truthy value.
+            server_hostname = self._tunnel_host
 
         if self.server_hostname is not None:
             server_hostname = self.server_hostname
@@ -601,15 +658,18 @@ class HTTPSConnection(HTTPConnection):
         )
         self.sock = sock_and_verified.socket
         self.is_verified = sock_and_verified.is_verified
-        self._connecting_to_proxy = False
+
+        # If there's a proxy to be connected to we are fully connected.
+        # This is set twice (once above and here) due to forwarding proxies
+        # not using tunnelling.
+        self._has_connected_to_proxy = bool(self.proxy)
 
     def _connect_tls_proxy(self, hostname: str, sock: socket.socket) -> "ssl.SSLSocket":
         """
         Establish a TLS connection to the proxy using the provided SSL context.
         """
-        proxy_config = cast(
-            ProxyConfig, self.proxy_config
-        )  # `_connect_tls_proxy` is called when self._is_using_tunnel() is truthy.
+        # `_connect_tls_proxy` is called when self._tunnel_host is truthy.
+        proxy_config = cast(ProxyConfig, self.proxy_config)
         ssl_context = proxy_config.ssl_context
         sock_and_verified = _ssl_wrap_socket_and_match_hostname(
             sock,
diff --git a/src/urllib3/connectionpool.py b/src/urllib3/connectionpool.py
index 35c5c23b..9d520661 100644
--- a/src/urllib3/connectionpool.py
+++ b/src/urllib3/connectionpool.py
@@ -8,9 +8,9 @@ from socket import timeout as SocketTimeout
 from types import TracebackType
 from typing import TYPE_CHECKING, Any, Mapping, Optional, Type, TypeVar, Union, overload
 
+from ._base_connection import _TYPE_BODY
 from ._request_methods import RequestMethods
 from .connection import (
-    _TYPE_BODY,
     BaseSSLError,
     BrokenPipeError,
     DummyConnection,
@@ -18,7 +18,6 @@ from .connection import (
     HTTPException,
     HTTPSConnection,
     ProxyConfig,
-    VerifiedHTTPSConnection,
     _wrap_proxy_error,
 )
 from .connection import port_by_scheme as port_by_scheme
@@ -37,7 +36,7 @@ from .exceptions import (
     SSLError,
     TimeoutError,
 )
-from .response import HTTPResponse
+from .response import BaseHTTPResponse
 from .util.connection import is_connection_dropped
 from .util.proxy import connection_requires_http_tunnel
 from .util.request import _TYPE_BODY_POSITION, set_file_position
@@ -54,6 +53,8 @@ if TYPE_CHECKING:
 
     from typing_extensions import Literal
 
+    from ._base_connection import BaseHTTPConnection, BaseHTTPSConnection
+
 log = logging.getLogger(__name__)
 
 _TYPE_TIMEOUT = Union[Timeout, float, _TYPE_DEFAULT]
@@ -81,9 +82,14 @@ class ConnectionPool:
             raise LocationValueError("No host specified.")
 
         self.host = _normalize_host(host, scheme=self.scheme)
-        self._proxy_host = host.lower()
         self.port = port
 
+        # This property uses 'normalize_host()' (not '_normalize_host()')
+        # to avoid removing square braces around IPv6 addresses.
+        # This value is sent to `HTTPConnection.set_tunnel()` if called
+        # because square braces are required for HTTP CONNECT tunneling.
+        self._tunnel_host = normalize_host(host, scheme=self.scheme).lower()
+
     def __str__(self) -> str:
         return f"{type(self).__name__}(host={self.host!r}, port={self.port!r})"
 
@@ -164,8 +170,9 @@ class HTTPConnectionPool(ConnectionPool, RequestMethods):
     """
 
     scheme = "http"
-    ConnectionCls: Type[Union[HTTPConnection, HTTPSConnection]] = HTTPConnection
-    ResponseCls = HTTPResponse
+    ConnectionCls: Union[
+        Type["BaseHTTPConnection"], Type["BaseHTTPSConnection"]
+    ] = HTTPConnection
 
     def __init__(
         self,
@@ -228,7 +235,7 @@ class HTTPConnectionPool(ConnectionPool, RequestMethods):
         # HTTPConnectionPool object is garbage collected.
         weakref.finalize(self, _close_pool_connections, pool)
 
-    def _new_conn(self) -> HTTPConnection:
+    def _new_conn(self) -> "BaseHTTPConnection":
         """
         Return a fresh :class:`HTTPConnection`.
         """
@@ -248,7 +255,7 @@ class HTTPConnectionPool(ConnectionPool, RequestMethods):
         )
         return conn
 
-    def _get_conn(self, timeout: Optional[float] = None) -> HTTPConnection:
+    def _get_conn(self, timeout: Optional[float] = None) -> "BaseHTTPConnection":
         """
         Get a connection. Will return a pooled connection if one is available.
 
@@ -286,7 +293,7 @@ class HTTPConnectionPool(ConnectionPool, RequestMethods):
 
         return conn or self._new_conn()
 
-    def _put_conn(self, conn: Optional[HTTPConnection]) -> None:
+    def _put_conn(self, conn: Optional["BaseHTTPConnection"]) -> None:
         """
         Put a connection back into the pool.
 
@@ -330,13 +337,13 @@ class HTTPConnectionPool(ConnectionPool, RequestMethods):
         if conn:
             conn.close()
 
-    def _validate_conn(self, conn: HTTPConnection) -> None:
+    def _validate_conn(self, conn: "BaseHTTPConnection") -> None:
         """
         Called right before a request is made, after the socket is created.
         """
         pass
 
-    def _prepare_proxy(self, conn: HTTPConnection) -> None:
+    def _prepare_proxy(self, conn: "BaseHTTPConnection") -> None:
         # Nothing to do for HTTP connections.
         pass
 
@@ -373,7 +380,7 @@ class HTTPConnectionPool(ConnectionPool, RequestMethods):
 
     def _make_request(
         self,
-        conn: HTTPConnection,
+        conn: "BaseHTTPConnection",
         method: str,
         url: str,
         body: Optional[_TYPE_BODY] = None,
@@ -381,11 +388,11 @@ class HTTPConnectionPool(ConnectionPool, RequestMethods):
         retries: Optional[Retry] = None,
         timeout: _TYPE_TIMEOUT = _DEFAULT_TIMEOUT,
         chunked: bool = False,
-        response_conn: Optional[HTTPConnection] = None,
+        response_conn: Optional["BaseHTTPConnection"] = None,
         preload_content: bool = True,
         decode_content: bool = True,
         enforce_content_length: bool = True,
-    ) -> HTTPResponse:
+    ) -> BaseHTTPResponse:
         """
         Perform a request on a given urllib connection object taken from our
         pool.
@@ -476,9 +483,11 @@ class HTTPConnectionPool(ConnectionPool, RequestMethods):
             new_e: Exception = e
             if isinstance(e, (BaseSSLError, CertificateError)):
                 new_e = SSLError(e)
+            # If the connection didn't successfully connect to it's proxy
+            # then there
             if isinstance(
                 new_e, (OSError, NewConnectionError, TimeoutError, SSLError)
-            ) and (conn and conn._connecting_to_proxy and conn.proxy):
+            ) and (conn and conn.proxy and not conn.has_connected_to_proxy):
                 new_e = _wrap_proxy_error(new_e, conn.proxy.scheme)
             raise new_e
 
@@ -511,7 +520,7 @@ class HTTPConnectionPool(ConnectionPool, RequestMethods):
         # Reset the timeout for the recv() on the socket
         read_timeout = timeout_obj.read_timeout
 
-        if conn.sock:
+        if not conn.is_closed:
             # In Python 3 socket.py will catch EAGAIN and return None when you
             # try and read into the file pointer created by http.client, which
             # instead raises a BadStatusLine exception. Instead of catching
@@ -521,7 +530,7 @@ class HTTPConnectionPool(ConnectionPool, RequestMethods):
                 raise ReadTimeoutError(
                     self, url, f"Read timed out. (read timeout={read_timeout})"
                 )
-            conn.sock.settimeout(read_timeout)
+            conn.timeout = read_timeout
 
         # Receive the response from the server
         try:
@@ -532,11 +541,11 @@ class HTTPConnectionPool(ConnectionPool, RequestMethods):
 
         # Set properties that are used by the pooling layer.
         response.retries = retries
-        response._connection = response_conn
-        response._pool = self
+        response._connection = response_conn  # type: ignore[attr-defined]
+        response._pool = self  # type: ignore[attr-defined]
 
         log.debug(
-            '%s://%s:%s "%s %s %s" %s %s',
+            '%s://%s:%s "%s %s %s" %s',
             self.scheme,
             self.host,
             self.port,
@@ -545,7 +554,7 @@ class HTTPConnectionPool(ConnectionPool, RequestMethods):
             # HTTP version
             conn._http_vsn_str,  # type: ignore[attr-defined]
             response.status,
-            response.length_remaining,
+            response.length_remaining,  # type: ignore[attr-defined]
         )
 
         return response
@@ -598,8 +607,10 @@ class HTTPConnectionPool(ConnectionPool, RequestMethods):
         release_conn: Optional[bool] = None,
         chunked: bool = False,
         body_pos: Optional[_TYPE_BODY_POSITION] = None,
+        preload_content: bool = True,
+        decode_content: bool = True,
         **response_kw: Any,
-    ) -> HTTPResponse:
+    ) -> BaseHTTPResponse:
         """
         Get a connection from the pool and perform an HTTP request. This is the
         lowest level call for making a request, so you'll need to specify all
@@ -668,6 +679,13 @@ class HTTPConnectionPool(ConnectionPool, RequestMethods):
             block for ``pool_timeout`` seconds and raise EmptyPoolError if no
             connection is available within the time period.
 
+        :param bool preload_content:
+            If True, the response's body will be preloaded into memory.
+
+        :param bool decode_content:
+            If True, will attempt to decode the body based on the
+            'content-encoding' header.
+
         :param release_conn:
             If False, then the urlopen call will not release the connection
             back into the pool once a response is received (but will release if
@@ -675,10 +693,10 @@ class HTTPConnectionPool(ConnectionPool, RequestMethods):
             `preload_content=True`). This is useful if you're not preloading
             the response's content immediately. You will need to call
             ``r.release_conn()`` on the response ``r`` to return the connection
-            back into the pool. If None, it takes the value of
-            ``response_kw.get('preload_content', True)``.
+            back into the pool. If None, it takes the value of ``preload_content``
+            which defaults to ``True``.
 
-        :param chunked:
+        :param bool chunked:
             If True, urllib3 will send the body using chunked transfer
             encoding. Otherwise, urllib3 will send the body using the standard
             content-length form. Defaults to False.
@@ -687,12 +705,7 @@ class HTTPConnectionPool(ConnectionPool, RequestMethods):
             Position to seek to in file-like body in the event of a retry or
             redirect. Typically this won't need to be set because urllib3 will
             auto-populate the value when needed.
-
-        :param \\**response_kw:
-            Additional parameters are passed to
-            :meth:`urllib3.connection.HTTPConnection.getresponse`
         """
-
         parsed_url = parse_url(url)
         destination_scheme = parsed_url.scheme
 
@@ -703,7 +716,7 @@ class HTTPConnectionPool(ConnectionPool, RequestMethods):
             retries = Retry.from_int(retries, redirect=redirect, default=self.retries)
 
         if release_conn is None:
-            release_conn = response_kw.get("preload_content", True)
+            release_conn = preload_content
 
         # Check host
         if assert_same_host and not self.is_same_host(url):
@@ -758,20 +771,15 @@ class HTTPConnectionPool(ConnectionPool, RequestMethods):
 
             conn.timeout = timeout_obj.connect_timeout  # type: ignore[assignment]
 
-            is_new_proxy_conn = self.proxy is not None and not getattr(
-                conn, "sock", None
-            )
-            if is_new_proxy_conn:
-                assert isinstance(self.proxy, Url)
-                conn._connecting_to_proxy = True
-                if http_tunnel_required:
-                    try:
-                        self._prepare_proxy(conn)
-                    except (BaseSSLError, OSError, SocketTimeout) as e:
-                        self._raise_timeout(
-                            err=e, url=self.proxy.url, timeout_value=conn.timeout
-                        )
-                        raise
+            # Is this a closed/new connection that requires CONNECT tunnelling?
+            if self.proxy is not None and http_tunnel_required and conn.is_closed:
+                try:
+                    self._prepare_proxy(conn)
+                except (BaseSSLError, OSError, SocketTimeout) as e:
+                    self._raise_timeout(
+                        err=e, url=self.proxy.url, timeout_value=conn.timeout
+                    )
+                    raise
 
             # If we're going to release the connection in ``finally:``, then
             # the response doesn't need to know about the connection. Otherwise
@@ -790,6 +798,8 @@ class HTTPConnectionPool(ConnectionPool, RequestMethods):
                 chunked=chunked,
                 retries=retries,
                 response_conn=response_conn,
+                preload_content=preload_content,
+                decode_content=decode_content,
                 **response_kw,
             )
 
@@ -827,7 +837,7 @@ class HTTPConnectionPool(ConnectionPool, RequestMethods):
                     SSLError,
                     HTTPException,
                 ),
-            ) and (conn and conn._connecting_to_proxy and conn.proxy):
+            ) and (conn and conn.proxy and not conn.has_connected_to_proxy):
                 new_e = _wrap_proxy_error(new_e, conn.proxy.scheme)
             elif isinstance(new_e, (OSError, HTTPException)):
                 new_e = ProtocolError("Connection aborted.", new_e)
@@ -875,6 +885,8 @@ class HTTPConnectionPool(ConnectionPool, RequestMethods):
                 release_conn=release_conn,
                 chunked=chunked,
                 body_pos=body_pos,
+                preload_content=preload_content,
+                decode_content=decode_content,
                 **response_kw,
             )
 
@@ -908,6 +920,8 @@ class HTTPConnectionPool(ConnectionPool, RequestMethods):
                 release_conn=release_conn,
                 chunked=chunked,
                 body_pos=body_pos,
+                preload_content=preload_content,
+                decode_content=decode_content,
                 **response_kw,
             )
 
@@ -938,6 +952,8 @@ class HTTPConnectionPool(ConnectionPool, RequestMethods):
                 release_conn=release_conn,
                 chunked=chunked,
                 body_pos=body_pos,
+                preload_content=preload_content,
+                decode_content=decode_content,
                 **response_kw,
             )
 
@@ -959,7 +975,7 @@ class HTTPSConnectionPool(HTTPConnectionPool):
     """
 
     scheme = "https"
-    ConnectionCls = HTTPSConnection
+    ConnectionCls: Type["BaseHTTPSConnection"] = HTTPSConnection
 
     def __init__(
         self,
@@ -1011,44 +1027,22 @@ class HTTPSConnectionPool(HTTPConnectionPool):
         self.assert_hostname = assert_hostname
         self.assert_fingerprint = assert_fingerprint
 
-    def _prepare_conn(self, conn: HTTPSConnection) -> HTTPConnection:
-        """
-        Prepare the ``connection`` for :meth:`urllib3.util.ssl_wrap_socket`
-        and establish the tunnel if proxy is used.
-        """
-
-        if isinstance(conn, VerifiedHTTPSConnection):
-            conn.set_cert(
-                key_file=self.key_file,
-                key_password=self.key_password,
-                cert_file=self.cert_file,
-                cert_reqs=self.cert_reqs,
-                ca_certs=self.ca_certs,
-                ca_cert_dir=self.ca_cert_dir,
-                assert_hostname=self.assert_hostname,
-                assert_fingerprint=self.assert_fingerprint,
-            )
-            conn.ssl_version = self.ssl_version
-            conn.ssl_minimum_version = self.ssl_minimum_version
-            conn.ssl_maximum_version = self.ssl_maximum_version
-
-        return conn
-
     def _prepare_proxy(self, conn: HTTPSConnection) -> None:  # type: ignore[override]
-        """
-        Establishes a tunnel connection through HTTP CONNECT.
-
-        Tunnel connection is established early because otherwise httplib would
-        improperly set Host: header to proxy's IP:port.
-        """
-        conn.set_tunnel(self._proxy_host, self.port, self.proxy_headers)
-
+        """Establishes a tunnel connection through HTTP CONNECT."""
         if self.proxy and self.proxy.scheme == "https":
-            conn.tls_in_tls_required = True
+            tunnel_scheme = "https"
+        else:
+            tunnel_scheme = "http"
 
+        conn.set_tunnel(
+            scheme=tunnel_scheme,
+            host=self._tunnel_host,
+            port=self.port,
+            headers=self.proxy_headers,
+        )
         conn.connect()
 
-    def _new_conn(self) -> HTTPConnection:
+    def _new_conn(self) -> "BaseHTTPSConnection":
         """
         Return a fresh :class:`urllib3.connection.HTTPConnection`.
         """
@@ -1060,7 +1054,7 @@ class HTTPSConnectionPool(HTTPConnectionPool):
             self.port or "443",
         )
 
-        if not self.ConnectionCls or self.ConnectionCls is DummyConnection:  # type: ignore[comparison-overlap, truthy-function]
+        if not self.ConnectionCls or self.ConnectionCls is DummyConnection:  # type: ignore[comparison-overlap]
             raise ImportError(
                 "Can't connect to HTTPS URL because the SSL module is not available."
             )
@@ -1071,26 +1065,32 @@ class HTTPSConnectionPool(HTTPConnectionPool):
             actual_host = self.proxy.host
             actual_port = self.proxy.port
 
-        conn = self.ConnectionCls(
+        return self.ConnectionCls(
             host=actual_host,
             port=actual_port,
             timeout=self.timeout.connect_timeout,
             cert_file=self.cert_file,
             key_file=self.key_file,
             key_password=self.key_password,
+            cert_reqs=self.cert_reqs,
+            ca_certs=self.ca_certs,
+            ca_cert_dir=self.ca_cert_dir,
+            assert_hostname=self.assert_hostname,
+            assert_fingerprint=self.assert_fingerprint,
+            ssl_version=self.ssl_version,
+            ssl_minimum_version=self.ssl_minimum_version,
+            ssl_maximum_version=self.ssl_maximum_version,
             **self.conn_kw,
         )
 
-        return self._prepare_conn(conn)
-
-    def _validate_conn(self, conn: HTTPConnection) -> None:
+    def _validate_conn(self, conn: "BaseHTTPConnection") -> None:
         """
         Called right before a request is made, after the socket is created.
         """
         super()._validate_conn(conn)
 
         # Force connect early to allow us to validate the connection.
-        if not conn.sock:
+        if conn.is_closed:
             conn.connect()
 
         if not conn.is_verified:
diff --git a/src/urllib3/response.py b/src/urllib3/response.py
index 501eaad0..a9951c38 100644
--- a/src/urllib3/response.py
+++ b/src/urllib3/response.py
@@ -46,8 +46,9 @@ except (AttributeError, ImportError, ValueError):  # Defensive:
     zstd = None
 
 from . import util
+from ._base_connection import _TYPE_BODY
 from ._collections import HTTPHeaderDict
-from .connection import _TYPE_BODY, BaseSSLError, HTTPConnection, HTTPException
+from .connection import BaseSSLError, HTTPConnection, HTTPException
 from .exceptions import (
     BodyNotHttplibCompatible,
     DecodeError,
diff --git a/src/urllib3/util/connection.py b/src/urllib3/util/connection.py
index 2c5bce75..cc02af32 100644
--- a/src/urllib3/util/connection.py
+++ b/src/urllib3/util/connection.py
@@ -1,25 +1,21 @@
 import socket
-from typing import Optional, Sequence, Tuple, Union
+from typing import TYPE_CHECKING, Optional, Sequence, Tuple, Union
 
 from ..exceptions import LocationParseError
 from .timeout import _DEFAULT_TIMEOUT, _TYPE_TIMEOUT
-from .wait import wait_for_read
 
 _TYPE_SOCKET_OPTIONS = Sequence[Tuple[int, int, Union[int, bytes]]]
 
+if TYPE_CHECKING:
+    from .._base_connection import BaseHTTPConnection
 
-def is_connection_dropped(conn: socket.socket) -> bool:  # Platform-specific
+
+def is_connection_dropped(conn: "BaseHTTPConnection") -> bool:  # Platform-specific
     """
     Returns True if the connection is dropped and should be closed.
-
-    :param conn:
-        :class:`http.client.HTTPConnection` object.
+    :param conn: :class:`urllib3.connection.HTTPConnection` object.
     """
-    sock = getattr(conn, "sock", None)
-    if sock is None:  # Connection already closed (such as by httplib).
-        return True
-    # Returns True if readable, which here means it's been dropped
-    return wait_for_read(sock, timeout=0.0)
+    return not conn.is_connected
 
 
 # This function is copied from socket.py in the Python 2.7 standard
diff --git a/src/urllib3/util/retry.py b/src/urllib3/util/retry.py
index cc5587a7..12b1c7fb 100644
--- a/src/urllib3/util/retry.py
+++ b/src/urllib3/util/retry.py
@@ -28,7 +28,7 @@ from .util import reraise
 
 if TYPE_CHECKING:
     from ..connectionpool import ConnectionPool
-    from ..response import HTTPResponse
+    from ..response import BaseHTTPResponse
 
 log = logging.getLogger(__name__)
 
@@ -315,7 +315,7 @@ class Retry:
 
         return seconds
 
-    def get_retry_after(self, response: "HTTPResponse") -> Optional[float]:
+    def get_retry_after(self, response: "BaseHTTPResponse") -> Optional[float]:
         """Get the value of Retry-After in seconds."""
 
         retry_after = response.getheader("Retry-After")
@@ -325,7 +325,7 @@ class Retry:
 
         return self.parse_retry_after(retry_after)
 
-    def sleep_for_retry(self, response: "HTTPResponse") -> bool:
+    def sleep_for_retry(self, response: "BaseHTTPResponse") -> bool:
         retry_after = self.get_retry_after(response)
         if retry_after:
             time.sleep(retry_after)
@@ -339,7 +339,7 @@ class Retry:
             return
         time.sleep(backoff)
 
-    def sleep(self, response: Optional["HTTPResponse"] = None) -> None:
+    def sleep(self, response: Optional["BaseHTTPResponse"] = None) -> None:
         """Sleep between retry attempts.
 
         This method will respect a server's ``Retry-After`` response header
@@ -422,7 +422,7 @@ class Retry:
         self,
         method: Optional[str] = None,
         url: Optional[str] = None,
-        response: Optional["HTTPResponse"] = None,
+        response: Optional["BaseHTTPResponse"] = None,
         error: Optional[Exception] = None,
         _pool: Optional["ConnectionPool"] = None,
         _stacktrace: Optional[TracebackType] = None,
@@ -431,7 +431,7 @@ class Retry:
 
         :param response: A response object, or None, if the server did not
             return a response.
-        :type response: :class:`~urllib3.response.HTTPResponse`
+        :type response: :class:`~urllib3.response.BaseHTTPResponse`
         :param Exception error: An error encountered during the request, or
             None if the response was received successfully.
 
diff --git a/src/urllib3/util/timeout.py b/src/urllib3/util/timeout.py
index bad91cc9..1798fa9e 100644
--- a/src/urllib3/util/timeout.py
+++ b/src/urllib3/util/timeout.py
@@ -10,7 +10,9 @@ if TYPE_CHECKING:
 
 
 class _TYPE_DEFAULT(Enum):
-    token = 0
+    # This value should never be passed to socket.settimeout() so for safety we use a -1.
+    # socket.settimout() raises a ValueError for negative values.
+    token = -1
 
 
 _DEFAULT_TIMEOUT: "Final[_TYPE_DEFAULT]" = _TYPE_DEFAULT.token
diff --git a/src/urllib3/util/url.py b/src/urllib3/util/url.py
index 71ff6582..c7645f78 100644
--- a/src/urllib3/util/url.py
+++ b/src/urllib3/util/url.py
@@ -290,6 +290,16 @@ def _remove_path_dot_segments(path: str) -> str:
     return "/".join(output)
 
 
+@overload
+def _normalize_host(host: None, scheme: Optional[str]) -> None:
+    ...
+
+
+@overload
+def _normalize_host(host: str, scheme: Optional[str]) -> str:
+    ...
+
+
 def _normalize_host(host: Optional[str], scheme: Optional[str]) -> Optional[str]:
     if host:
         if scheme in _NORMALIZABLE_SCHEMES: