keyring2.0

Author: Aleksandr Mangin

src/pip/_internal/cli/cmdoptions.py

@@ -883,6 +883,13 @@ def _handle_config_settings(
     help="Don't periodically check PyPI to determine whether a new version "
     "of pip is available for download. Implied with --no-index.",
 )
+default_key_ring_user: Callable[..., Option] = partial(
+    Option,
+    "--default-key-ring-user",
+    dest="default_key_ring_user",
+    default=None,
+    help="Default key ring user that pip can use to get credentials from keyring",
+)
 
 root_user_action: Callable[..., Option] = partial(
     Option,

src/pip/_internal/cli/req_command.py

@@ -128,6 +128,7 @@ def _build_session(
             trusted_hosts=options.trusted_hosts,
             index_urls=self._get_index_urls(options),
             ssl_context=ssl_context,
+            default_key_ring_user=getattr(options, "default_key_ring_user", None),
         )
 
         # Handle custom ca-bundles from the user

src/pip/_internal/commands/install.py

@@ -261,6 +261,7 @@ def add_options(self) -> None:
         self.cmd_opts.add_option(cmdoptions.require_hashes())
         self.cmd_opts.add_option(cmdoptions.progress_bar())
         self.cmd_opts.add_option(cmdoptions.root_user_action())
+        self.cmd_opts.add_option(cmdoptions.default_key_ring_user())
 
         index_opts = cmdoptions.make_option_group(
             cmdoptions.index_group,

src/pip/_internal/network/auth.py

@@ -7,22 +7,15 @@
 import os
 import shutil
 import subprocess
-import urllib.parse
 from abc import ABC, abstractmethod
-from typing import Any, Dict, List, NamedTuple, Optional, Tuple
+from typing import Dict, List, NamedTuple, Optional, Tuple
 
 from pip._vendor.requests.auth import AuthBase, HTTPBasicAuth
-from pip._vendor.requests.models import Request, Response
+from pip._vendor.requests.models import Request
 from pip._vendor.requests.utils import get_netrc_auth
 
 from pip._internal.utils.logging import getLogger
-from pip._internal.utils.misc import (
-    ask,
-    ask_input,
-    ask_password,
-    remove_auth_from_url,
-    split_auth_netloc_from_url,
-)
+from pip._internal.utils.misc import remove_auth_from_url, split_auth_netloc_from_url
 from pip._internal.vcs.versioncontrol import AuthInfo
 
 logger = getLogger(__name__)
@@ -198,11 +191,15 @@ def get_keyring_auth(url: Optional[str], username: Optional[str]) -> Optional[Au
 
 class MultiDomainBasicAuth(AuthBase):
     def __init__(
-        self, prompting: bool = True, index_urls: Optional[List[str]] = None
+        self,
+        prompting: bool = True,
+        index_urls: Optional[List[str]] = None,
+        default_key_ring_user: Optional[str] = None,
     ) -> None:
         self.prompting = prompting
         self.index_urls = index_urls
         self.passwords: Dict[str, AuthInfo] = {}
+        self.default_key_ring_user = default_key_ring_user
         # When the user is prompted to enter credentials and keyring is
         # available, we will offer to save them. If the user accepts,
         # this value is set to the credentials they entered. After the
@@ -235,29 +232,31 @@ def _get_index_url(self, url: str) -> Optional[str]:
     def _get_new_credentials(
         self,
         original_url: str,
-        allow_netrc: bool = True,
-        allow_keyring: bool = False,
     ) -> AuthInfo:
         """Find and return credentials for the specified URL."""
         # Split the credentials and netloc from the url.
         url, netloc, url_user_password = split_auth_netloc_from_url(
             original_url,
         )
 
-        # Start with the credentials embedded in the url
         username, password = url_user_password
         if username is not None and password is not None:
             logger.debug("Found credentials in url for %s", netloc)
             return url_user_password
 
-        # Find a matching index url for this request
-        index_url = self._get_index_url(url)
-        if index_url:
-            # Split the credentials from the url.
-            index_info = split_auth_netloc_from_url(index_url)
-            if index_info:
-                index_url, _, index_url_user_password = index_info
-                logger.debug("Found index url %s", index_url)
+        def split_index_url_on_url_and_credentials(url):
+            if not url:
+                return url, None
+            index_info = split_auth_netloc_from_url(url)
+            if not index_info:
+                return url, None
+            index_url, _, index_url_user_password = index_info
+            logger.debug("Found index url %s", index_url)
+            return index_url, index_url_user_password
+
+        index_url, index_url_user_password = split_index_url_on_url_and_credentials(
+            self._get_index_url(url)
+        )
 
         # If an index URL was found, try its embedded credentials
         if index_url and index_url_user_password[0] is not None:
@@ -266,27 +265,57 @@ def _get_new_credentials(
                 logger.debug("Found credentials in index url for %s", netloc)
                 return index_url_user_password
 
-        # Get creds from netrc if we still don't have them
-        if allow_netrc:
-            netrc_auth = get_netrc_auth(original_url)
-            if netrc_auth:
-                logger.debug("Found credentials in netrc for %s", netloc)
-                return netrc_auth
-
-        # If we don't have a password and keyring is available, use it.
-        if allow_keyring:
-            # The index url is more specific than the netloc, so try it first
-            # fmt: off
-            kr_auth = (
-                get_keyring_auth(index_url, username) or
-                get_keyring_auth(netloc, username)
-            )
-            # fmt: on
+        netrc_auth = get_netrc_auth(original_url)
+        if netrc_auth is not None:
+            logger.debug("Found credentials in netrc for %s", netloc)
+            return netrc_auth
+
+        kr_auth = self._find_key_ring_credentials(
+            index_url, index_url_user_password, netloc, username
+        )
+        if kr_auth is not None:
+            return kr_auth
+
+        return username, password
+
+    def _find_key_ring_credentials(
+        self,
+        index_url: Optional[str],
+        index_url_user_password: Optional[str],
+        netloc: str,
+        artifact_username: str,
+    ) -> Optional[AuthInfo]:
+        def get_key_ring_user() -> Optional[str]:
+            if artifact_username is not None:
+                return artifact_username
+            if index_url_user_password:
+                if (
+                    index_url_user_password[0] is not None
+                    and index_url_user_password[1] is None
+                ):
+                    logger.debug("Found key ring username in index_url")
+                    return index_url_user_password[0]
+            if artifact_username is None and self.default_key_ring_user is not None:
+                logger.debug("Using default_key_ring_user")
+                return self.default_key_ring_user
+            return None
+
+        key_ring_user = get_key_ring_user()
+        if key_ring_user is None:
+            return None
+
+        if index_url is not None:
+            kr_auth = get_keyring_auth(index_url, key_ring_user)
             if kr_auth:
-                logger.debug("Found credentials in keyring for %s", netloc)
+                logger.debug("Found credentials in keyring for %s", index_url)
                 return kr_auth
 
-        return username, password
+        kr_auth = get_keyring_auth(netloc, key_ring_user)
+        if kr_auth:
+            logger.debug("Found credentials in keyring for %s", netloc)
+            return kr_auth
+
+        return None
 
     def _get_url_and_credentials(
         self, original_url: str
@@ -346,107 +375,4 @@ def __call__(self, req: Request) -> Request:
         if username is not None and password is not None:
             # Send the basic auth with this request
             req = HTTPBasicAuth(username, password)(req)
-
-        # Attach a hook to handle 401 responses
-        req.register_hook("response", self.handle_401)
-
         return req
-
-    # Factored out to allow for easy patching in tests
-    def _prompt_for_password(
-        self, netloc: str
-    ) -> Tuple[Optional[str], Optional[str], bool]:
-        username = ask_input(f"User for {netloc}: ")
-        if not username:
-            return None, None, False
-        auth = get_keyring_auth(netloc, username)
-        if auth and auth[0] is not None and auth[1] is not None:
-            return auth[0], auth[1], False
-        password = ask_password("Password: ")
-        return username, password, True
-
-    # Factored out to allow for easy patching in tests
-    def _should_save_password_to_keyring(self) -> bool:
-        if not get_keyring_provider().has_keyring:
-            return False
-        return ask("Save credentials to keyring [y/N]: ", ["y", "n"]) == "y"
-
-    def handle_401(self, resp: Response, **kwargs: Any) -> Response:
-        # We only care about 401 responses, anything else we want to just
-        #   pass through the actual response
-        if resp.status_code != 401:
-            return resp
-
-        # We are not able to prompt the user so simply return the response
-        if not self.prompting:
-            return resp
-
-        parsed = urllib.parse.urlparse(resp.url)
-
-        # Query the keyring for credentials:
-        username, password = self._get_new_credentials(
-            resp.url,
-            allow_netrc=False,
-            allow_keyring=True,
-        )
-
-        # Prompt the user for a new username and password
-        save = False
-        if not username and not password:
-            username, password, save = self._prompt_for_password(parsed.netloc)
-
-        # Store the new username and password to use for future requests
-        self._credentials_to_save = None
-        if username is not None and password is not None:
-            self.passwords[parsed.netloc] = (username, password)
-
-            # Prompt to save the password to keyring
-            if save and self._should_save_password_to_keyring():
-                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.
-        resp.content
-        resp.raw.release_conn()
-
-        # Add our new username and password to the request
-        req = HTTPBasicAuth(username or "", password or "")(resp.request)
-        req.register_hook("response", self.warn_on_401)
-
-        # On successful request, save the credentials that were used to
-        # keyring. (Note that if the user responded "no" above, this member
-        # is not set and nothing will be saved.)
-        if self._credentials_to_save:
-            req.register_hook("response", self.save_credentials)
-
-        # Send our new request
-        new_resp = resp.connection.send(req, **kwargs)
-        new_resp.history.append(resp)
-
-        return new_resp
-
-    def warn_on_401(self, resp: Response, **kwargs: Any) -> None:
-        """Response callback to warn about incorrect credentials."""
-        if resp.status_code == 401:
-            logger.warning(
-                "401 Error, Credentials not correct for %s",
-                resp.request.url,
-            )
-
-    def save_credentials(self, resp: Response, **kwargs: Any) -> None:
-        """Response callback to save credentials on success."""
-        keyring = get_keyring_provider()
-        assert keyring.has_keyring, "should never reach here without keyring"
-
-        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.save_auth_info(creds.url, creds.username, creds.password)
-            except Exception:
-                logger.exception("Failed to save credentials")

src/pip/_internal/network/session.py

@@ -327,6 +327,7 @@ def __init__(
         trusted_hosts: Sequence[str] = (),
         index_urls: Optional[List[str]] = None,
         ssl_context: Optional["SSLContext"] = None,
+        default_key_ring_user: Optional[str] = None,
         **kwargs: Any,
     ) -> None:
         """
@@ -343,7 +344,10 @@ def __init__(
         self.headers["User-Agent"] = user_agent()
 
         # Attach our Authentication handler to the session
-        self.auth = MultiDomainBasicAuth(index_urls=index_urls)
+        self.auth = MultiDomainBasicAuth(
+            index_urls=index_urls,
+            default_key_ring_user=default_key_ring_user,
+        )
 
         # Create our urllib3.Retry instance which will allow us to customize
         # how we handle retries.

src/pip/_internal/utils/misc.py

@@ -3,7 +3,6 @@
 
 import contextlib
 import errno
-import getpass
 import hashlib
 import io
 import logging
@@ -199,18 +198,6 @@ def ask(message: str, options: Iterable[str]) -> str:
             return response
 
 
-def ask_input(message: str) -> str:
-    """Ask for input interactively."""
-    _check_no_input(message)
-    return input(message)
-
-
-def ask_password(message: str) -> str:
-    """Ask for a password interactively."""
-    _check_no_input(message)
-    return getpass.getpass(message)
-
-
 def strtobool(val: str) -> int:
     """Convert a string representation of truth to true (1) or false (0).