Optional truststore support

This adds a --use-feature=truststore flag that, when specified on Python
3.10+ with truststore installed, switches pip to use truststore to
provide HTTPS certificate validation, instead of certifi. This allows
pip to verify certificates against custom certificates in the system
store.

truststore is deliberately NOT vendored because it is expected the
library to be under active development in the short term, and this
prevents users having to wait for a pip release to get potentially vital
bug fixes needed to be made in truststore.

Supplying the use-feature flag without installing truststore beforehand,
or on Python versions prior to 3.10, results in a command error.

news/11082.feature.rst

@@ -0,0 +1,3 @@
+Add support to use `truststore <https://pypi.org/project/truststore/>`_ as an alternative SSL certificate verification backend. The backend can be enabled on Python 3.10 and later by installing ``truststore`` into the environment, and adding the ``--use-feature=truststore`` flag to various pip commands.
+
+``truststore`` differs from the current default verification backend (provided by ``certifi``) in it uses the operating system’s trust store, which can be better controlled and augmented to better support non-standard certificates. Depending on feedback, pip may switch to this as the default certificate verification backend in the future.

src/pip/_internal/cli/cmdoptions.py

@@ -991,7 +991,7 @@ def check_list_path_option(options: Values) -> None:
     metavar="feature",
     action="append",
     default=[],
-    choices=["2020-resolver", "fast-deps"],
+    choices=["2020-resolver", "fast-deps", "truststore"],
     help="Enable new functionality, that may be backward incompatible.",
 )
 

src/pip/_internal/cli/req_command.py

@@ -10,7 +10,7 @@
 import sys
 from functools import partial
 from optparse import Values
-from typing import Any, List, Optional, Tuple
+from typing import TYPE_CHECKING, Any, List, Optional, Tuple
 
 from pip._internal.cache import WheelCache
 from pip._internal.cli import cmdoptions
@@ -42,9 +42,33 @@
 )
 from pip._internal.utils.virtualenv import running_under_virtualenv
 
+if TYPE_CHECKING:
+    from ssl import SSLContext
+
 logger = logging.getLogger(__name__)
 
 
+def _create_truststore_ssl_context() -> Optional["SSLContext"]:
+    if sys.version_info < (3, 10):
+        raise CommandError("The truststore feature is only available for Python 3.10+")
+
+    try:
+        import ssl
+    except ImportError:
+        logger.warning("Disabling truststore since ssl support is missing")
+        return None
+
+    try:
+        import truststore
+    except ImportError:
+        raise CommandError(
+            "To use the truststore feature, 'truststore' must be installed into "
+            "pip's current environment."
+        )
+
+    return truststore.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
+
+
 class SessionCommandMixin(CommandContextMixIn):
 
     """
@@ -84,15 +108,27 @@ def _build_session(
         options: Values,
         retries: Optional[int] = None,
         timeout: Optional[int] = None,
+        fallback_to_certifi: bool = False,
     ) -> PipSession:
-        assert not options.cache_dir or os.path.isabs(options.cache_dir)
+        cache_dir = options.cache_dir
+        assert not cache_dir or os.path.isabs(cache_dir)
+
+        if "truststore" in options.features_enabled:
+            try:
+                ssl_context = _create_truststore_ssl_context()
+            except Exception:
+                if not fallback_to_certifi:
+                    raise
+                ssl_context = None
+        else:
+            ssl_context = None
+
         session = PipSession(
-            cache=(
-                os.path.join(options.cache_dir, "http") if options.cache_dir else None
-            ),
+            cache=os.path.join(cache_dir, "http") if cache_dir else None,
             retries=retries if retries is not None else options.retries,
             trusted_hosts=options.trusted_hosts,
             index_urls=self._get_index_urls(options),
+            ssl_context=ssl_context,
         )
 
         # Handle custom ca-bundles from the user
@@ -142,7 +178,14 @@ def handle_pip_version_check(self, options: Values) -> None:
 
         # Otherwise, check if we're using the latest version of pip available.
         session = self._build_session(
-            options, retries=0, timeout=min(5, options.timeout)
+            options,
+            retries=0,
+            timeout=min(5, options.timeout),
+            # This is set to ensure the function does not fail when truststore is
+            # specified in use-feature but cannot be loaded. This usually raises a
+            # CommandError and shows a nice user-facing error, but this function is not
+            # called in that try-except block.
+            fallback_to_certifi=True,
         )
         with session:
             pip_self_version_check(session, options)

src/pip/_internal/network/session.py

@@ -15,11 +15,23 @@
 import sys
 import urllib.parse
 import warnings
-from typing import Any, Dict, Generator, List, Mapping, Optional, Sequence, Tuple, Union
+from typing import (
+    TYPE_CHECKING,
+    Any,
+    Dict,
+    Generator,
+    List,
+    Mapping,
+    Optional,
+    Sequence,
+    Tuple,
+    Union,
+)
 
 from pip._vendor import requests, urllib3
-from pip._vendor.cachecontrol import CacheControlAdapter
-from pip._vendor.requests.adapters import BaseAdapter, HTTPAdapter
+from pip._vendor.cachecontrol import CacheControlAdapter as _BaseCacheControlAdapter
+from pip._vendor.requests.adapters import DEFAULT_POOLBLOCK, BaseAdapter
+from pip._vendor.requests.adapters import HTTPAdapter as _BaseHTTPAdapter
 from pip._vendor.requests.models import PreparedRequest, Response
 from pip._vendor.requests.structures import CaseInsensitiveDict
 from pip._vendor.urllib3.connectionpool import ConnectionPool
@@ -37,6 +49,12 @@
 from pip._internal.utils.misc import build_url_from_netloc, parse_netloc
 from pip._internal.utils.urls import url_to_path
 
+if TYPE_CHECKING:
+    from ssl import SSLContext
+
+    from pip._vendor.urllib3.poolmanager import PoolManager
+
+
 logger = logging.getLogger(__name__)
 
 SecureOrigin = Tuple[str, str, Optional[Union[int, str]]]
@@ -233,6 +251,48 @@ def close(self) -> None:
         pass
 
 
+class _SSLContextAdapterMixin:
+    """Mixin to add the ``ssl_context`` contructor argument to HTTP adapters.
+
+    The additional argument is forwarded directly to the pool manager. This allows us
+    to dynamically decide what SSL store to use at runtime, which is used to implement
+    the optional ``truststore`` backend.
+    """
+
+    def __init__(
+        self,
+        *,
+        ssl_context: Optional["SSLContext"] = None,
+        **kwargs: Any,
+    ) -> None:
+        self._ssl_context = ssl_context
+        super().__init__(**kwargs)
+
+    def init_poolmanager(
+        self,
+        connections: int,
+        maxsize: int,
+        block: bool = DEFAULT_POOLBLOCK,
+        **pool_kwargs: Any,
+    ) -> "PoolManager":
+        if self._ssl_context is not None:
+            pool_kwargs.setdefault("ssl_context", self._ssl_context)
+        return super().init_poolmanager(  # type: ignore[misc]
+            connections=connections,
+            maxsize=maxsize,
+            block=block,
+            **pool_kwargs,
+        )
+
+
+class HTTPAdapter(_SSLContextAdapterMixin, _BaseHTTPAdapter):
+    pass
+
+
+class CacheControlAdapter(_SSLContextAdapterMixin, _BaseCacheControlAdapter):
+    pass
+
+
 class InsecureHTTPAdapter(HTTPAdapter):
     def cert_verify(
         self,
@@ -266,6 +326,7 @@ def __init__(
         cache: Optional[str] = None,
         trusted_hosts: Sequence[str] = (),
         index_urls: Optional[List[str]] = None,
+        ssl_context: Optional["SSLContext"] = None,
         **kwargs: Any,
     ) -> None:
         """
@@ -318,13 +379,14 @@ def __init__(
             secure_adapter = CacheControlAdapter(
                 cache=SafeFileCache(cache),
                 max_retries=retries,
+                ssl_context=ssl_context,
             )
             self._trusted_host_adapter = InsecureCacheControlAdapter(
                 cache=SafeFileCache(cache),
                 max_retries=retries,
             )
         else:
-            secure_adapter = HTTPAdapter(max_retries=retries)
+            secure_adapter = HTTPAdapter(max_retries=retries, ssl_context=ssl_context)
             self._trusted_host_adapter = insecure_adapter
 
         self.mount("https://", secure_adapter)