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,13 @@
 providing credentials in the context of network requests.
 """
 
+import functools
+import os
+import shutil
+import subprocess
 import urllib.parse
-from typing import Any, Dict, List, Optional, Tuple
+from abc import ABC, abstractmethod
+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,51 +28,168 @@
 
 logger = getLogger(__name__)
 
-Credentials = Tuple[str, str, str]
+KEYRING_DISABLED = False
 
-try:
-    import keyring
-except ImportError:
-    keyring = None  # type: ignore[assignment]
-except Exception as exc:
-    logger.warning(
-        "Keyring is skipped due to an exception: %s",
-        str(exc),
-    )
-    keyring = None  # type: ignore[assignment]
 
+class Credentials(NamedTuple):
+    url: str
+    username: str
+    password: str
 
-def get_keyring_auth(url: Optional[str], username: Optional[str]) -> Optional[AuthInfo]:
-    """Return the tuple auth for a given url from keyring."""
-    global keyring
-    if not url or not keyring:
-        return None
 
-    try:
+class KeyRingBaseProvider(ABC):
+    """Keyring base provider interface"""
+
+    @abstractmethod
+    def is_available(self) -> bool:
+        ...
+
+    @abstractmethod
+    def get_auth_info(self, url: str, username: Optional[str]) -> Optional[AuthInfo]:
+        ...
+
+    @abstractmethod
+    def save_auth_info(self, url: str, username: str, password: str) -> None:
+        ...
+
+
+class KeyRingPythonProvider(KeyRingBaseProvider):
+    """Keyring interface which uses locally imported `keyring`"""
+
+    def __init__(self) -> None:
         try:
-            get_credential = keyring.get_credential
-        except AttributeError:
-            pass
-        else:
+            import keyring
+        except ImportError:
+            keyring = None  # type: ignore[assignment]
+
+        self.keyring = keyring
+
+    def is_available(self) -> bool:
+        return self.keyring is not None
+
+    def get_auth_info(self, url: str, username: Optional[str]) -> Optional[AuthInfo]:
+        if self.is_available is False:
+            return None
+
+        # Support keyring's get_credential interface which supports getting
+        # credentials without a username. This is only available for
+        # keyring>=15.2.0.
+        if hasattr(self.keyring, "get_credential"):
             logger.debug("Getting credentials from keyring for %s", url)
-            cred = get_credential(url, username)
+            cred = self.keyring.get_credential(url, username)
             if cred is not None:
                 return cred.username, cred.password
             return None
 
-        if username:
+        if username is not None:
             logger.debug("Getting password from keyring for %s", url)
-            password = keyring.get_password(url, username)
+            password = self.keyring.get_password(url, username)
             if password:
                 return username, password
+        return None
+
+    def save_auth_info(self, url: str, username: str, password: str) -> None:
+        self.keyring.set_password(url, username, password)
+
+
+class KeyRingCliProvider(KeyRingBaseProvider):
+    """Provider which uses `keyring` cli
+
+    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) -> None:
+        self.keyring = shutil.which("keyring")
+
+    def is_available(self) -> bool:
+        return self.keyring is not None
+
+    def get_auth_info(self, url: str, username: Optional[str]) -> Optional[AuthInfo]:
+        if self.is_available is False:
+            return None
+
+        # This is the default implementation of keyring.get_credential
+        # https://github.com/jaraco/keyring/blob/97689324abcf01bd1793d49063e7ca01e03d7d07/keyring/backend.py#L134-L139
+        if username is not None:
+            password = self._get_password(url, username)
+            if password is not None:
+                return username, password
+        return None
 
+    def save_auth_info(self, url: str, username: str, password: str) -> None:
+        if not self.is_available:
+            raise RuntimeError("keyring is not available")
+        return self._set_password(url, username, password)
+
+    def _get_password(self, service_name: str, username: str) -> Optional[str]:
+        """Mirror the implemenation of keyring.get_password using cli"""
+        if self.keyring is None:
+            return None
+
+        cmd = [self.keyring, "get", service_name, username]
+        env = os.environ.copy()
+        env["PYTHONIOENCODING"] = "utf-8"
+        res = subprocess.run(
+            cmd,
+            stdin=subprocess.DEVNULL,
+            capture_output=True,
+            env=env,
+        )
+        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:
+        """Mirror the implemenation of keyring.set_password using cli"""
+        if self.keyring is None:
+            return None
+
+        cmd = [self.keyring, "set", service_name, username]
+        input_ = password.encode("utf-8") + b"\n"
+        env = os.environ.copy()
+        env["PYTHONIOENCODING"] = "utf-8"
+        res = subprocess.run(cmd, input=input_, env=env)
+        res.check_returncode()
+        return None
+
+
+@functools.lru_cache(maxsize=1)
+def get_keyring_provider() -> Optional[KeyRingBaseProvider]:
+    python_keyring = KeyRingPythonProvider()
+    if python_keyring.is_available():
+        return python_keyring
+
+    cli_keyring = KeyRingCliProvider()
+    if cli_keyring.is_available():
+        return cli_keyring
+
+    return None
+
+
+def get_keyring_auth(url: Optional[str], username: Optional[str]) -> Optional[AuthInfo]:
+    """Return the tuple auth for a given url from keyring."""
+    # Do nothing if no url was provided
+    if not url:
+        return None
+
+    keyring = get_keyring_provider()
+    # Do nothing if keyring is not available
+    global KEYRING_DISABLED
+    if keyring is None or KEYRING_DISABLED:
+        return None
+
+    try:
+        return keyring.get_auth_info(url, username)
     except Exception as exc:
         logger.warning(
             "Keyring is skipped due to an exception: %s",
             str(exc),
         )
-        keyring = None  # type: ignore[assignment]
-    return None
+        KEYRING_DISABLED = True
+        return None
 
 
 class MultiDomainBasicAuth(AuthBase):
@@ -241,7 +363,7 @@ def _prompt_for_password(
 
     # Factored out to allow for easy patching in tests
     def _should_save_password_to_keyring(self) -> bool:
-        if not keyring:
+        if get_keyring_provider() is None:
             return False
         return ask("Save credentials to keyring [y/N]: ", ["y", "n"]) == "y"
 
@@ -276,7 +398,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(
+                    url=parsed.netloc,
+                    username=username,
+                    password=password,
+                )
 
         # Consume content and release the original connection to allow our new
         #   request to reuse the same one.
@@ -309,15 +435,16 @@ def warn_on_401(self, resp: Response, **kwargs: Any) -> None:
 
     def save_credentials(self, resp: Response, **kwargs: Any) -> None:
         """Response callback to save credentials on success."""
+        keyring = get_keyring_provider()
         assert keyring is not None, "should never reach here without keyring"
         if not keyring:
-            return
+            return None
 
         creds = self._credentials_to_save
         self._credentials_to_save = None
         if creds and resp.status_code < 400:
             try:
                 logger.info("Saving credentials to keyring")
-                keyring.set_password(*creds)
+                keyring.save_auth_info(creds.url, creds.username, creds.password)
             except Exception:
                 logger.exception("Failed to save credentials")