jpayne@69: from __future__ import annotations
jpayne@69: 
jpayne@69: import functools
jpayne@69: import logging
jpayne@69: import typing
jpayne@69: import warnings
jpayne@69: from types import TracebackType
jpayne@69: from urllib.parse import urljoin
jpayne@69: 
jpayne@69: from ._collections import HTTPHeaderDict, RecentlyUsedContainer
jpayne@69: from ._request_methods import RequestMethods
jpayne@69: from .connection import ProxyConfig
jpayne@69: from .connectionpool import HTTPConnectionPool, HTTPSConnectionPool, port_by_scheme
jpayne@69: from .exceptions import (
jpayne@69:     LocationValueError,
jpayne@69:     MaxRetryError,
jpayne@69:     ProxySchemeUnknown,
jpayne@69:     URLSchemeUnknown,
jpayne@69: )
jpayne@69: from .response import BaseHTTPResponse
jpayne@69: from .util.connection import _TYPE_SOCKET_OPTIONS
jpayne@69: from .util.proxy import connection_requires_http_tunnel
jpayne@69: from .util.retry import Retry
jpayne@69: from .util.timeout import Timeout
jpayne@69: from .util.url import Url, parse_url
jpayne@69: 
jpayne@69: if typing.TYPE_CHECKING:
jpayne@69:     import ssl
jpayne@69: 
jpayne@69:     from typing_extensions import Self
jpayne@69: 
jpayne@69: __all__ = ["PoolManager", "ProxyManager", "proxy_from_url"]
jpayne@69: 
jpayne@69: 
jpayne@69: log = logging.getLogger(__name__)
jpayne@69: 
jpayne@69: SSL_KEYWORDS = (
jpayne@69:     "key_file",
jpayne@69:     "cert_file",
jpayne@69:     "cert_reqs",
jpayne@69:     "ca_certs",
jpayne@69:     "ca_cert_data",
jpayne@69:     "ssl_version",
jpayne@69:     "ssl_minimum_version",
jpayne@69:     "ssl_maximum_version",
jpayne@69:     "ca_cert_dir",
jpayne@69:     "ssl_context",
jpayne@69:     "key_password",
jpayne@69:     "server_hostname",
jpayne@69: )
jpayne@69: # Default value for `blocksize` - a new parameter introduced to
jpayne@69: # http.client.HTTPConnection & http.client.HTTPSConnection in Python 3.7
jpayne@69: _DEFAULT_BLOCKSIZE = 16384
jpayne@69: 
jpayne@69: 
jpayne@69: class PoolKey(typing.NamedTuple):
jpayne@69:     """
jpayne@69:     All known keyword arguments that could be provided to the pool manager, its
jpayne@69:     pools, or the underlying connections.
jpayne@69: 
jpayne@69:     All custom key schemes should include the fields in this key at a minimum.
jpayne@69:     """
jpayne@69: 
jpayne@69:     key_scheme: str
jpayne@69:     key_host: str
jpayne@69:     key_port: int | None
jpayne@69:     key_timeout: Timeout | float | int | None
jpayne@69:     key_retries: Retry | bool | int | None
jpayne@69:     key_block: bool | None
jpayne@69:     key_source_address: tuple[str, int] | None
jpayne@69:     key_key_file: str | None
jpayne@69:     key_key_password: str | None
jpayne@69:     key_cert_file: str | None
jpayne@69:     key_cert_reqs: str | None
jpayne@69:     key_ca_certs: str | None
jpayne@69:     key_ca_cert_data: str | bytes | None
jpayne@69:     key_ssl_version: int | str | None
jpayne@69:     key_ssl_minimum_version: ssl.TLSVersion | None
jpayne@69:     key_ssl_maximum_version: ssl.TLSVersion | None
jpayne@69:     key_ca_cert_dir: str | None
jpayne@69:     key_ssl_context: ssl.SSLContext | None
jpayne@69:     key_maxsize: int | None
jpayne@69:     key_headers: frozenset[tuple[str, str]] | None
jpayne@69:     key__proxy: Url | None
jpayne@69:     key__proxy_headers: frozenset[tuple[str, str]] | None
jpayne@69:     key__proxy_config: ProxyConfig | None
jpayne@69:     key_socket_options: _TYPE_SOCKET_OPTIONS | None
jpayne@69:     key__socks_options: frozenset[tuple[str, str]] | None
jpayne@69:     key_assert_hostname: bool | str | None
jpayne@69:     key_assert_fingerprint: str | None
jpayne@69:     key_server_hostname: str | None
jpayne@69:     key_blocksize: int | None
jpayne@69: 
jpayne@69: 
jpayne@69: def _default_key_normalizer(
jpayne@69:     key_class: type[PoolKey], request_context: dict[str, typing.Any]
jpayne@69: ) -> PoolKey:
jpayne@69:     """
jpayne@69:     Create a pool key out of a request context dictionary.
jpayne@69: 
jpayne@69:     According to RFC 3986, both the scheme and host are case-insensitive.
jpayne@69:     Therefore, this function normalizes both before constructing the pool
jpayne@69:     key for an HTTPS request. If you wish to change this behaviour, provide
jpayne@69:     alternate callables to ``key_fn_by_scheme``.
jpayne@69: 
jpayne@69:     :param key_class:
jpayne@69:         The class to use when constructing the key. This should be a namedtuple
jpayne@69:         with the ``scheme`` and ``host`` keys at a minimum.
jpayne@69:     :type  key_class: namedtuple
jpayne@69:     :param request_context:
jpayne@69:         A dictionary-like object that contain the context for a request.
jpayne@69:     :type  request_context: dict
jpayne@69: 
jpayne@69:     :return: A namedtuple that can be used as a connection pool key.
jpayne@69:     :rtype:  PoolKey
jpayne@69:     """
jpayne@69:     # Since we mutate the dictionary, make a copy first
jpayne@69:     context = request_context.copy()
jpayne@69:     context["scheme"] = context["scheme"].lower()
jpayne@69:     context["host"] = context["host"].lower()
jpayne@69: 
jpayne@69:     # These are both dictionaries and need to be transformed into frozensets
jpayne@69:     for key in ("headers", "_proxy_headers", "_socks_options"):
jpayne@69:         if key in context and context[key] is not None:
jpayne@69:             context[key] = frozenset(context[key].items())
jpayne@69: 
jpayne@69:     # The socket_options key may be a list and needs to be transformed into a
jpayne@69:     # tuple.
jpayne@69:     socket_opts = context.get("socket_options")
jpayne@69:     if socket_opts is not None:
jpayne@69:         context["socket_options"] = tuple(socket_opts)
jpayne@69: 
jpayne@69:     # Map the kwargs to the names in the namedtuple - this is necessary since
jpayne@69:     # namedtuples can't have fields starting with '_'.
jpayne@69:     for key in list(context.keys()):
jpayne@69:         context["key_" + key] = context.pop(key)
jpayne@69: 
jpayne@69:     # Default to ``None`` for keys missing from the context
jpayne@69:     for field in key_class._fields:
jpayne@69:         if field not in context:
jpayne@69:             context[field] = None
jpayne@69: 
jpayne@69:     # Default key_blocksize to _DEFAULT_BLOCKSIZE if missing from the context
jpayne@69:     if context.get("key_blocksize") is None:
jpayne@69:         context["key_blocksize"] = _DEFAULT_BLOCKSIZE
jpayne@69: 
jpayne@69:     return key_class(**context)
jpayne@69: 
jpayne@69: 
jpayne@69: #: A dictionary that maps a scheme to a callable that creates a pool key.
jpayne@69: #: This can be used to alter the way pool keys are constructed, if desired.
jpayne@69: #: Each PoolManager makes a copy of this dictionary so they can be configured
jpayne@69: #: globally here, or individually on the instance.
jpayne@69: key_fn_by_scheme = {
jpayne@69:     "http": functools.partial(_default_key_normalizer, PoolKey),
jpayne@69:     "https": functools.partial(_default_key_normalizer, PoolKey),
jpayne@69: }
jpayne@69: 
jpayne@69: pool_classes_by_scheme = {"http": HTTPConnectionPool, "https": HTTPSConnectionPool}
jpayne@69: 
jpayne@69: 
jpayne@69: class PoolManager(RequestMethods):
jpayne@69:     """
jpayne@69:     Allows for arbitrary requests while transparently keeping track of
jpayne@69:     necessary connection pools for you.
jpayne@69: 
jpayne@69:     :param num_pools:
jpayne@69:         Number of connection pools to cache before discarding the least
jpayne@69:         recently used pool.
jpayne@69: 
jpayne@69:     :param headers:
jpayne@69:         Headers to include with all requests, unless other headers are given
jpayne@69:         explicitly.
jpayne@69: 
jpayne@69:     :param \\**connection_pool_kw:
jpayne@69:         Additional parameters are used to create fresh
jpayne@69:         :class:`urllib3.connectionpool.ConnectionPool` instances.
jpayne@69: 
jpayne@69:     Example:
jpayne@69: 
jpayne@69:     .. code-block:: python
jpayne@69: 
jpayne@69:         import urllib3
jpayne@69: 
jpayne@69:         http = urllib3.PoolManager(num_pools=2)
jpayne@69: 
jpayne@69:         resp1 = http.request("GET", "https://google.com/")
jpayne@69:         resp2 = http.request("GET", "https://google.com/mail")
jpayne@69:         resp3 = http.request("GET", "https://yahoo.com/")
jpayne@69: 
jpayne@69:         print(len(http.pools))
jpayne@69:         # 2
jpayne@69: 
jpayne@69:     """
jpayne@69: 
jpayne@69:     proxy: Url | None = None
jpayne@69:     proxy_config: ProxyConfig | None = None
jpayne@69: 
jpayne@69:     def __init__(
jpayne@69:         self,
jpayne@69:         num_pools: int = 10,
jpayne@69:         headers: typing.Mapping[str, str] | None = None,
jpayne@69:         **connection_pool_kw: typing.Any,
jpayne@69:     ) -> None:
jpayne@69:         super().__init__(headers)
jpayne@69:         self.connection_pool_kw = connection_pool_kw
jpayne@69: 
jpayne@69:         self.pools: RecentlyUsedContainer[PoolKey, HTTPConnectionPool]
jpayne@69:         self.pools = RecentlyUsedContainer(num_pools)
jpayne@69: 
jpayne@69:         # Locally set the pool classes and keys so other PoolManagers can
jpayne@69:         # override them.
jpayne@69:         self.pool_classes_by_scheme = pool_classes_by_scheme
jpayne@69:         self.key_fn_by_scheme = key_fn_by_scheme.copy()
jpayne@69: 
jpayne@69:     def __enter__(self) -> Self:
jpayne@69:         return self
jpayne@69: 
jpayne@69:     def __exit__(
jpayne@69:         self,
jpayne@69:         exc_type: type[BaseException] | None,
jpayne@69:         exc_val: BaseException | None,
jpayne@69:         exc_tb: TracebackType | None,
jpayne@69:     ) -> typing.Literal[False]:
jpayne@69:         self.clear()
jpayne@69:         # Return False to re-raise any potential exceptions
jpayne@69:         return False
jpayne@69: 
jpayne@69:     def _new_pool(
jpayne@69:         self,
jpayne@69:         scheme: str,
jpayne@69:         host: str,
jpayne@69:         port: int,
jpayne@69:         request_context: dict[str, typing.Any] | None = None,
jpayne@69:     ) -> HTTPConnectionPool:
jpayne@69:         """
jpayne@69:         Create a new :class:`urllib3.connectionpool.ConnectionPool` based on host, port, scheme, and
jpayne@69:         any additional pool keyword arguments.
jpayne@69: 
jpayne@69:         If ``request_context`` is provided, it is provided as keyword arguments
jpayne@69:         to the pool class used. This method is used to actually create the
jpayne@69:         connection pools handed out by :meth:`connection_from_url` and
jpayne@69:         companion methods. It is intended to be overridden for customization.
jpayne@69:         """
jpayne@69:         pool_cls: type[HTTPConnectionPool] = self.pool_classes_by_scheme[scheme]
jpayne@69:         if request_context is None:
jpayne@69:             request_context = self.connection_pool_kw.copy()
jpayne@69: 
jpayne@69:         # Default blocksize to _DEFAULT_BLOCKSIZE if missing or explicitly
jpayne@69:         # set to 'None' in the request_context.
jpayne@69:         if request_context.get("blocksize") is None:
jpayne@69:             request_context["blocksize"] = _DEFAULT_BLOCKSIZE
jpayne@69: 
jpayne@69:         # Although the context has everything necessary to create the pool,
jpayne@69:         # this function has historically only used the scheme, host, and port
jpayne@69:         # in the positional args. When an API change is acceptable these can
jpayne@69:         # be removed.
jpayne@69:         for key in ("scheme", "host", "port"):
jpayne@69:             request_context.pop(key, None)
jpayne@69: 
jpayne@69:         if scheme == "http":
jpayne@69:             for kw in SSL_KEYWORDS:
jpayne@69:                 request_context.pop(kw, None)
jpayne@69: 
jpayne@69:         return pool_cls(host, port, **request_context)
jpayne@69: 
jpayne@69:     def clear(self) -> None:
jpayne@69:         """
jpayne@69:         Empty our store of pools and direct them all to close.
jpayne@69: 
jpayne@69:         This will not affect in-flight connections, but they will not be
jpayne@69:         re-used after completion.
jpayne@69:         """
jpayne@69:         self.pools.clear()
jpayne@69: 
jpayne@69:     def connection_from_host(
jpayne@69:         self,
jpayne@69:         host: str | None,
jpayne@69:         port: int | None = None,
jpayne@69:         scheme: str | None = "http",
jpayne@69:         pool_kwargs: dict[str, typing.Any] | None = None,
jpayne@69:     ) -> HTTPConnectionPool:
jpayne@69:         """
jpayne@69:         Get a :class:`urllib3.connectionpool.ConnectionPool` based on the host, port, and scheme.
jpayne@69: 
jpayne@69:         If ``port`` isn't given, it will be derived from the ``scheme`` using
jpayne@69:         ``urllib3.connectionpool.port_by_scheme``. If ``pool_kwargs`` is
jpayne@69:         provided, it is merged with the instance's ``connection_pool_kw``
jpayne@69:         variable and used to create the new connection pool, if one is
jpayne@69:         needed.
jpayne@69:         """
jpayne@69: 
jpayne@69:         if not host:
jpayne@69:             raise LocationValueError("No host specified.")
jpayne@69: 
jpayne@69:         request_context = self._merge_pool_kwargs(pool_kwargs)
jpayne@69:         request_context["scheme"] = scheme or "http"
jpayne@69:         if not port:
jpayne@69:             port = port_by_scheme.get(request_context["scheme"].lower(), 80)
jpayne@69:         request_context["port"] = port
jpayne@69:         request_context["host"] = host
jpayne@69: 
jpayne@69:         return self.connection_from_context(request_context)
jpayne@69: 
jpayne@69:     def connection_from_context(
jpayne@69:         self, request_context: dict[str, typing.Any]
jpayne@69:     ) -> HTTPConnectionPool:
jpayne@69:         """
jpayne@69:         Get a :class:`urllib3.connectionpool.ConnectionPool` based on the request context.
jpayne@69: 
jpayne@69:         ``request_context`` must at least contain the ``scheme`` key and its
jpayne@69:         value must be a key in ``key_fn_by_scheme`` instance variable.
jpayne@69:         """
jpayne@69:         if "strict" in request_context:
jpayne@69:             warnings.warn(
jpayne@69:                 "The 'strict' parameter is no longer needed on Python 3+. "
jpayne@69:                 "This will raise an error in urllib3 v2.1.0.",
jpayne@69:                 DeprecationWarning,
jpayne@69:             )
jpayne@69:             request_context.pop("strict")
jpayne@69: 
jpayne@69:         scheme = request_context["scheme"].lower()
jpayne@69:         pool_key_constructor = self.key_fn_by_scheme.get(scheme)
jpayne@69:         if not pool_key_constructor:
jpayne@69:             raise URLSchemeUnknown(scheme)
jpayne@69:         pool_key = pool_key_constructor(request_context)
jpayne@69: 
jpayne@69:         return self.connection_from_pool_key(pool_key, request_context=request_context)
jpayne@69: 
jpayne@69:     def connection_from_pool_key(
jpayne@69:         self, pool_key: PoolKey, request_context: dict[str, typing.Any]
jpayne@69:     ) -> HTTPConnectionPool:
jpayne@69:         """
jpayne@69:         Get a :class:`urllib3.connectionpool.ConnectionPool` based on the provided pool key.
jpayne@69: 
jpayne@69:         ``pool_key`` should be a namedtuple that only contains immutable
jpayne@69:         objects. At a minimum it must have the ``scheme``, ``host``, and
jpayne@69:         ``port`` fields.
jpayne@69:         """
jpayne@69:         with self.pools.lock:
jpayne@69:             # If the scheme, host, or port doesn't match existing open
jpayne@69:             # connections, open a new ConnectionPool.
jpayne@69:             pool = self.pools.get(pool_key)
jpayne@69:             if pool:
jpayne@69:                 return pool
jpayne@69: 
jpayne@69:             # Make a fresh ConnectionPool of the desired type
jpayne@69:             scheme = request_context["scheme"]
jpayne@69:             host = request_context["host"]
jpayne@69:             port = request_context["port"]
jpayne@69:             pool = self._new_pool(scheme, host, port, request_context=request_context)
jpayne@69:             self.pools[pool_key] = pool
jpayne@69: 
jpayne@69:         return pool
jpayne@69: 
jpayne@69:     def connection_from_url(
jpayne@69:         self, url: str, pool_kwargs: dict[str, typing.Any] | None = None
jpayne@69:     ) -> HTTPConnectionPool:
jpayne@69:         """
jpayne@69:         Similar to :func:`urllib3.connectionpool.connection_from_url`.
jpayne@69: 
jpayne@69:         If ``pool_kwargs`` is not provided and a new pool needs to be
jpayne@69:         constructed, ``self.connection_pool_kw`` is used to initialize
jpayne@69:         the :class:`urllib3.connectionpool.ConnectionPool`. If ``pool_kwargs``
jpayne@69:         is provided, it is used instead. Note that if a new pool does not
jpayne@69:         need to be created for the request, the provided ``pool_kwargs`` are
jpayne@69:         not used.
jpayne@69:         """
jpayne@69:         u = parse_url(url)
jpayne@69:         return self.connection_from_host(
jpayne@69:             u.host, port=u.port, scheme=u.scheme, pool_kwargs=pool_kwargs
jpayne@69:         )
jpayne@69: 
jpayne@69:     def _merge_pool_kwargs(
jpayne@69:         self, override: dict[str, typing.Any] | None
jpayne@69:     ) -> dict[str, typing.Any]:
jpayne@69:         """
jpayne@69:         Merge a dictionary of override values for self.connection_pool_kw.
jpayne@69: 
jpayne@69:         This does not modify self.connection_pool_kw and returns a new dict.
jpayne@69:         Any keys in the override dictionary with a value of ``None`` are
jpayne@69:         removed from the merged dictionary.
jpayne@69:         """
jpayne@69:         base_pool_kwargs = self.connection_pool_kw.copy()
jpayne@69:         if override:
jpayne@69:             for key, value in override.items():
jpayne@69:                 if value is None:
jpayne@69:                     try:
jpayne@69:                         del base_pool_kwargs[key]
jpayne@69:                     except KeyError:
jpayne@69:                         pass
jpayne@69:                 else:
jpayne@69:                     base_pool_kwargs[key] = value
jpayne@69:         return base_pool_kwargs
jpayne@69: 
jpayne@69:     def _proxy_requires_url_absolute_form(self, parsed_url: Url) -> bool:
jpayne@69:         """
jpayne@69:         Indicates if the proxy requires the complete destination URL in the
jpayne@69:         request.  Normally this is only needed when not using an HTTP CONNECT
jpayne@69:         tunnel.
jpayne@69:         """
jpayne@69:         if self.proxy is None:
jpayne@69:             return False
jpayne@69: 
jpayne@69:         return not connection_requires_http_tunnel(
jpayne@69:             self.proxy, self.proxy_config, parsed_url.scheme
jpayne@69:         )
jpayne@69: 
jpayne@69:     def urlopen(  # type: ignore[override]
jpayne@69:         self, method: str, url: str, redirect: bool = True, **kw: typing.Any
jpayne@69:     ) -> BaseHTTPResponse:
jpayne@69:         """
jpayne@69:         Same as :meth:`urllib3.HTTPConnectionPool.urlopen`
jpayne@69:         with custom cross-host redirect logic and only sends the request-uri
jpayne@69:         portion of the ``url``.
jpayne@69: 
jpayne@69:         The given ``url`` parameter must be absolute, such that an appropriate
jpayne@69:         :class:`urllib3.connectionpool.ConnectionPool` can be chosen for it.
jpayne@69:         """
jpayne@69:         u = parse_url(url)
jpayne@69: 
jpayne@69:         if u.scheme is None:
jpayne@69:             warnings.warn(
jpayne@69:                 "URLs without a scheme (ie 'https://') are deprecated and will raise an error "
jpayne@69:                 "in a future version of urllib3. To avoid this DeprecationWarning ensure all URLs "
jpayne@69:                 "start with 'https://' or 'http://'. Read more in this issue: "
jpayne@69:                 "https://github.com/urllib3/urllib3/issues/2920",
jpayne@69:                 category=DeprecationWarning,
jpayne@69:                 stacklevel=2,
jpayne@69:             )
jpayne@69: 
jpayne@69:         conn = self.connection_from_host(u.host, port=u.port, scheme=u.scheme)
jpayne@69: 
jpayne@69:         kw["assert_same_host"] = False
jpayne@69:         kw["redirect"] = False
jpayne@69: 
jpayne@69:         if "headers" not in kw:
jpayne@69:             kw["headers"] = self.headers
jpayne@69: 
jpayne@69:         if self._proxy_requires_url_absolute_form(u):
jpayne@69:             response = conn.urlopen(method, url, **kw)
jpayne@69:         else:
jpayne@69:             response = conn.urlopen(method, u.request_uri, **kw)
jpayne@69: 
jpayne@69:         redirect_location = redirect and response.get_redirect_location()
jpayne@69:         if not redirect_location:
jpayne@69:             return response
jpayne@69: 
jpayne@69:         # Support relative URLs for redirecting.
jpayne@69:         redirect_location = urljoin(url, redirect_location)
jpayne@69: 
jpayne@69:         if response.status == 303:
jpayne@69:             # Change the method according to RFC 9110, Section 15.4.4.
jpayne@69:             method = "GET"
jpayne@69:             # And lose the body not to transfer anything sensitive.
jpayne@69:             kw["body"] = None
jpayne@69:             kw["headers"] = HTTPHeaderDict(kw["headers"])._prepare_for_method_change()
jpayne@69: 
jpayne@69:         retries = kw.get("retries")
jpayne@69:         if not isinstance(retries, Retry):
jpayne@69:             retries = Retry.from_int(retries, redirect=redirect)
jpayne@69: 
jpayne@69:         # Strip headers marked as unsafe to forward to the redirected location.
jpayne@69:         # Check remove_headers_on_redirect to avoid a potential network call within
jpayne@69:         # conn.is_same_host() which may use socket.gethostbyname() in the future.
jpayne@69:         if retries.remove_headers_on_redirect and not conn.is_same_host(
jpayne@69:             redirect_location
jpayne@69:         ):
jpayne@69:             new_headers = kw["headers"].copy()
jpayne@69:             for header in kw["headers"]:
jpayne@69:                 if header.lower() in retries.remove_headers_on_redirect:
jpayne@69:                     new_headers.pop(header, None)
jpayne@69:             kw["headers"] = new_headers
jpayne@69: 
jpayne@69:         try:
jpayne@69:             retries = retries.increment(method, url, response=response, _pool=conn)
jpayne@69:         except MaxRetryError:
jpayne@69:             if retries.raise_on_redirect:
jpayne@69:                 response.drain_conn()
jpayne@69:                 raise
jpayne@69:             return response
jpayne@69: 
jpayne@69:         kw["retries"] = retries
jpayne@69:         kw["redirect"] = redirect
jpayne@69: 
jpayne@69:         log.info("Redirecting %s -> %s", url, redirect_location)
jpayne@69: 
jpayne@69:         response.drain_conn()
jpayne@69:         return self.urlopen(method, redirect_location, **kw)
jpayne@69: 
jpayne@69: 
jpayne@69: class ProxyManager(PoolManager):
jpayne@69:     """
jpayne@69:     Behaves just like :class:`PoolManager`, but sends all requests through
jpayne@69:     the defined proxy, using the CONNECT method for HTTPS URLs.
jpayne@69: 
jpayne@69:     :param proxy_url:
jpayne@69:         The URL of the proxy to be used.
jpayne@69: 
jpayne@69:     :param proxy_headers:
jpayne@69:         A dictionary containing headers that will be sent to the proxy. In case
jpayne@69:         of HTTP they are being sent with each request, while in the
jpayne@69:         HTTPS/CONNECT case they are sent only once. Could be used for proxy
jpayne@69:         authentication.
jpayne@69: 
jpayne@69:     :param proxy_ssl_context:
jpayne@69:         The proxy SSL context is used to establish the TLS connection to the
jpayne@69:         proxy when using HTTPS proxies.
jpayne@69: 
jpayne@69:     :param use_forwarding_for_https:
jpayne@69:         (Defaults to False) If set to True will forward requests to the HTTPS
jpayne@69:         proxy to be made on behalf of the client instead of creating a TLS
jpayne@69:         tunnel via the CONNECT method. **Enabling this flag means that request
jpayne@69:         and response headers and content will be visible from the HTTPS proxy**
jpayne@69:         whereas tunneling keeps request and response headers and content
jpayne@69:         private.  IP address, target hostname, SNI, and port are always visible
jpayne@69:         to an HTTPS proxy even when this flag is disabled.
jpayne@69: 
jpayne@69:     :param proxy_assert_hostname:
jpayne@69:         The hostname of the certificate to verify against.
jpayne@69: 
jpayne@69:     :param proxy_assert_fingerprint:
jpayne@69:         The fingerprint of the certificate to verify against.
jpayne@69: 
jpayne@69:     Example:
jpayne@69: 
jpayne@69:     .. code-block:: python
jpayne@69: 
jpayne@69:         import urllib3
jpayne@69: 
jpayne@69:         proxy = urllib3.ProxyManager("https://localhost:3128/")
jpayne@69: 
jpayne@69:         resp1 = proxy.request("GET", "https://google.com/")
jpayne@69:         resp2 = proxy.request("GET", "https://httpbin.org/")
jpayne@69: 
jpayne@69:         print(len(proxy.pools))
jpayne@69:         # 1
jpayne@69: 
jpayne@69:         resp3 = proxy.request("GET", "https://httpbin.org/")
jpayne@69:         resp4 = proxy.request("GET", "https://twitter.com/")
jpayne@69: 
jpayne@69:         print(len(proxy.pools))
jpayne@69:         # 3
jpayne@69: 
jpayne@69:     """
jpayne@69: 
jpayne@69:     def __init__(
jpayne@69:         self,
jpayne@69:         proxy_url: str,
jpayne@69:         num_pools: int = 10,
jpayne@69:         headers: typing.Mapping[str, str] | None = None,
jpayne@69:         proxy_headers: typing.Mapping[str, str] | None = None,
jpayne@69:         proxy_ssl_context: ssl.SSLContext | None = None,
jpayne@69:         use_forwarding_for_https: bool = False,
jpayne@69:         proxy_assert_hostname: None | str | typing.Literal[False] = None,
jpayne@69:         proxy_assert_fingerprint: str | None = None,
jpayne@69:         **connection_pool_kw: typing.Any,
jpayne@69:     ) -> None:
jpayne@69:         if isinstance(proxy_url, HTTPConnectionPool):
jpayne@69:             str_proxy_url = f"{proxy_url.scheme}://{proxy_url.host}:{proxy_url.port}"
jpayne@69:         else:
jpayne@69:             str_proxy_url = proxy_url
jpayne@69:         proxy = parse_url(str_proxy_url)
jpayne@69: 
jpayne@69:         if proxy.scheme not in ("http", "https"):
jpayne@69:             raise ProxySchemeUnknown(proxy.scheme)
jpayne@69: 
jpayne@69:         if not proxy.port:
jpayne@69:             port = port_by_scheme.get(proxy.scheme, 80)
jpayne@69:             proxy = proxy._replace(port=port)
jpayne@69: 
jpayne@69:         self.proxy = proxy
jpayne@69:         self.proxy_headers = proxy_headers or {}
jpayne@69:         self.proxy_ssl_context = proxy_ssl_context
jpayne@69:         self.proxy_config = ProxyConfig(
jpayne@69:             proxy_ssl_context,
jpayne@69:             use_forwarding_for_https,
jpayne@69:             proxy_assert_hostname,
jpayne@69:             proxy_assert_fingerprint,
jpayne@69:         )
jpayne@69: 
jpayne@69:         connection_pool_kw["_proxy"] = self.proxy
jpayne@69:         connection_pool_kw["_proxy_headers"] = self.proxy_headers
jpayne@69:         connection_pool_kw["_proxy_config"] = self.proxy_config
jpayne@69: 
jpayne@69:         super().__init__(num_pools, headers, **connection_pool_kw)
jpayne@69: 
jpayne@69:     def connection_from_host(
jpayne@69:         self,
jpayne@69:         host: str | None,
jpayne@69:         port: int | None = None,
jpayne@69:         scheme: str | None = "http",
jpayne@69:         pool_kwargs: dict[str, typing.Any] | None = None,
jpayne@69:     ) -> HTTPConnectionPool:
jpayne@69:         if scheme == "https":
jpayne@69:             return super().connection_from_host(
jpayne@69:                 host, port, scheme, pool_kwargs=pool_kwargs
jpayne@69:             )
jpayne@69: 
jpayne@69:         return super().connection_from_host(
jpayne@69:             self.proxy.host, self.proxy.port, self.proxy.scheme, pool_kwargs=pool_kwargs  # type: ignore[union-attr]
jpayne@69:         )
jpayne@69: 
jpayne@69:     def _set_proxy_headers(
jpayne@69:         self, url: str, headers: typing.Mapping[str, str] | None = None
jpayne@69:     ) -> typing.Mapping[str, str]:
jpayne@69:         """
jpayne@69:         Sets headers needed by proxies: specifically, the Accept and Host
jpayne@69:         headers. Only sets headers not provided by the user.
jpayne@69:         """
jpayne@69:         headers_ = {"Accept": "*/*"}
jpayne@69: 
jpayne@69:         netloc = parse_url(url).netloc
jpayne@69:         if netloc:
jpayne@69:             headers_["Host"] = netloc
jpayne@69: 
jpayne@69:         if headers:
jpayne@69:             headers_.update(headers)
jpayne@69:         return headers_
jpayne@69: 
jpayne@69:     def urlopen(  # type: ignore[override]
jpayne@69:         self, method: str, url: str, redirect: bool = True, **kw: typing.Any
jpayne@69:     ) -> BaseHTTPResponse:
jpayne@69:         "Same as HTTP(S)ConnectionPool.urlopen, ``url`` must be absolute."
jpayne@69:         u = parse_url(url)
jpayne@69:         if not connection_requires_http_tunnel(self.proxy, self.proxy_config, u.scheme):
jpayne@69:             # For connections using HTTP CONNECT, httplib sets the necessary
jpayne@69:             # headers on the CONNECT to the proxy. If we're not using CONNECT,
jpayne@69:             # we'll definitely need to set 'Host' at the very least.
jpayne@69:             headers = kw.get("headers", self.headers)
jpayne@69:             kw["headers"] = self._set_proxy_headers(url, headers)
jpayne@69: 
jpayne@69:         return super().urlopen(method, url, redirect=redirect, **kw)
jpayne@69: 
jpayne@69: 
jpayne@69: def proxy_from_url(url: str, **kw: typing.Any) -> ProxyManager:
jpayne@69:     return ProxyManager(proxy_url=url, **kw)