Add an interface to allow calling system keyring

news/11589.feature.rst

@@ -0,0 +1,2 @@
+Enable the use of ``keyring`` found on ``PATH``. This allows ``keyring``
+installed using ``pipx`` to be used by ``pip``.

src/pip/_internal/network/auth.py

@@ -4,8 +4,10 @@
 providing credentials in the context of network requests.
 """
 
+import shutil
+import subprocess
 import urllib.parse
-from typing import Any, Dict, List, Optional, Tuple
+from typing import Any, Dict, List, NamedTuple, Optional, Tuple
 
 from pip._vendor.requests.auth import AuthBase, HTTPBasicAuth
 from pip._vendor.requests.models import Request, Response
@@ -23,12 +25,52 @@
 
 logger = getLogger(__name__)
 
-Credentials = Tuple[str, str, str]
+
+class Credentials(NamedTuple):
+    service_name: str
+    username: str
+    password: str
+
+
+class KeyRingCli:
+    """Mirror the parts of keyring's API which pip uses
+
+    Instead of calling the keyring package installed alongside pip
+    we call keyring on the command line which will enable pip to
+    use which ever installation of keyring is available first in
+    PATH.
+    """
+
+    def __init__(self, keyring: str) -> None:
+        self.keyring = keyring
+
+    def get_password(self, service_name: str, username: str) -> Optional[str]:
+        cmd = [self.keyring, "get", service_name, username]
+        res = subprocess.run(
+            cmd,
+            stdin=subprocess.DEVNULL,
+            capture_output=True,
+            env=dict(PYTHONIOENCODING="utf-8"),
+        )
+        if res.returncode:
+            return None
+        return res.stdout.decode("utf-8").strip("\n")
+
+    def set_password(self, service_name: str, username: str, password: str) -> None:
+        cmd = [self.keyring, "set", service_name, username]
+        input_ = password.encode("utf-8") + b"\n"
+        res = subprocess.run(cmd, input=input_, env=dict(PYTHONIOENCODING="utf-8"))
+        res.check_returncode()
+        return None
+
 
 try:
     import keyring
 except ImportError:
     keyring = None  # type: ignore[assignment]
+    keyring_path = shutil.which("keyring")
+    if keyring_path is not None:
+        keyring = KeyRingCli(keyring_path)  # type: ignore[assignment]
 except Exception as exc:
     logger.warning(
         "Keyring is skipped due to an exception: %s",
@@ -276,7 +318,11 @@ def handle_401(self, resp: Response, **kwargs: Any) -> Response:
 
             # Prompt to save the password to keyring
             if save and self._should_save_password_to_keyring():
-                self._credentials_to_save = (parsed.netloc, username, password)
+                self._credentials_to_save = Credentials(
+                    service_name=parsed.netloc,
+                    username=username,
+                    password=password,
+                )
 
         # Consume content and release the original connection to allow our new
         #   request to reuse the same one.
@@ -318,6 +364,6 @@ def save_credentials(self, resp: Response, **kwargs: Any) -> None:
         if creds and resp.status_code < 400:
             try:
                 logger.info("Saving credentials to keyring")
-                keyring.set_password(*creds)
+                keyring.set_password(creds.service_name, creds.username, creds.password)
             except Exception:
                 logger.exception("Failed to save credentials")