Optional truststore support

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)

tests/functional/test_truststore.py

@@ -0,0 +1,61 @@
+import sys
+from typing import Any, Callable
+
+import pytest
+
+from tests.lib import PipTestEnvironment, TestPipResult
+
+PipRunner = Callable[..., TestPipResult]
+
+
+@pytest.fixture()
+def pip(script: PipTestEnvironment) -> PipRunner:
+    def pip(*args: str, **kwargs: Any) -> TestPipResult:
+        return script.pip(*args, "--use-feature=truststore", **kwargs)
+
+    return pip
+
+
+@pytest.mark.skipif(sys.version_info >= (3, 10), reason="3.10 can run truststore")
+def test_truststore_error_on_old_python(pip: PipRunner) -> None:
+    result = pip(
+        "install",
+        "--no-index",
+        "does-not-matter",
+        expect_error=True,
+    )
+    assert "The truststore feature is only available for Python 3.10+" in result.stderr
+
+
+@pytest.mark.skipif(sys.version_info < (3, 10), reason="3.10+ required for truststore")
+def test_truststore_error_without_preinstalled(pip: PipRunner) -> None:
+    result = pip(
+        "install",
+        "--no-index",
+        "does-not-matter",
+        expect_error=True,
+    )
+    assert (
+        "To use the truststore feature, 'truststore' must be installed into "
+        "pip's current environment."
+    ) in result.stderr
+
+
+@pytest.mark.skipif(sys.version_info < (3, 10), reason="3.10+ required for truststore")
+@pytest.mark.network
+@pytest.mark.parametrize(
+    "package",
+    [
+        "INITools",
+        "https://github.com/pypa/pip-test-package/archive/refs/heads/master.zip",
+    ],
+    ids=["PyPI", "GitHub"],
+)
+def test_trustore_can_install(
+    script: PipTestEnvironment,
+    pip: PipRunner,
+    package: str,
+) -> None:
+    script.pip("install", "truststore")
+    result = pip("install", package)
+    assert "Successfully installed" in result.stdout