| """Network Authentication Helpers |
| |
| Contains interface (MultiDomainBasicAuth) and associated glue code for |
| providing credentials in the context of network requests. |
| """ |
| |
| import urllib.parse |
| from typing import Any, Dict, List, Optional, Tuple |
| |
| from pip._vendor.requests.auth import AuthBase, HTTPBasicAuth |
| from pip._vendor.requests.models import Request, Response |
| 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.vcs.versioncontrol import AuthInfo |
| |
| logger = getLogger(__name__) |
| |
| Credentials = Tuple[str, str, str] |
| |
| 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] |
| |
| |
| 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: |
| try: |
| get_credential = keyring.get_credential |
| except AttributeError: |
| pass |
| else: |
| logger.debug("Getting credentials from keyring for %s", url) |
| cred = get_credential(url, username) |
| if cred is not None: |
| return cred.username, cred.password |
| return None |
| |
| if username: |
| logger.debug("Getting password from keyring for %s", url) |
| password = keyring.get_password(url, username) |
| if password: |
| return username, password |
| |
| except Exception as exc: |
| logger.warning( |
| "Keyring is skipped due to an exception: %s", |
| str(exc), |
| ) |
| keyring = None # type: ignore[assignment] |
| return None |
| |
| |
| class MultiDomainBasicAuth(AuthBase): |
| def __init__( |
| self, prompting: bool = True, index_urls: Optional[List[str]] = None |
| ) -> None: |
| self.prompting = prompting |
| self.index_urls = index_urls |
| self.passwords: Dict[str, AuthInfo] = {} |
| # 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 |
| # request authenticates, the caller should call |
| # ``save_credentials`` to save these. |
| self._credentials_to_save: Optional[Credentials] = None |
| |
| def _get_index_url(self, url: str) -> Optional[str]: |
| """Return the original index URL matching the requested URL. |
| |
| Cached or dynamically generated credentials may work against |
| the original index URL rather than just the netloc. |
| |
| The provided url should have had its username and password |
| removed already. If the original index url had credentials then |
| they will be included in the return value. |
| |
| Returns None if no matching index was found, or if --no-index |
| was specified by the user. |
| """ |
| if not url or not self.index_urls: |
| return None |
| |
| for u in self.index_urls: |
| prefix = remove_auth_from_url(u).rstrip("/") + "/" |
| if url.startswith(prefix): |
| return u |
| return None |
| |
| 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) |
| |
| # If an index URL was found, try its embedded credentials |
| if index_url and index_url_user_password[0] is not None: |
| username, password = index_url_user_password |
| if username is not None and password is not None: |
| 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 |
| if kr_auth: |
| logger.debug("Found credentials in keyring for %s", netloc) |
| return kr_auth |
| |
| return username, password |
| |
| def _get_url_and_credentials( |
| self, original_url: str |
| ) -> Tuple[str, Optional[str], Optional[str]]: |
| """Return the credentials to use for the provided URL. |
| |
| If allowed, netrc and keyring may be used to obtain the |
| correct credentials. |
| |
| Returns (url_without_credentials, username, password). Note |
| that even if the original URL contains credentials, this |
| function may return a different username and password. |
| """ |
| url, netloc, _ = split_auth_netloc_from_url(original_url) |
| |
| # Try to get credentials from original url |
| username, password = self._get_new_credentials(original_url) |
| |
| # If credentials not found, use any stored credentials for this netloc. |
| # Do this if either the username or the password is missing. |
| # This accounts for the situation in which the user has specified |
| # the username in the index url, but the password comes from keyring. |
| if (username is None or password is None) and netloc in self.passwords: |
| un, pw = self.passwords[netloc] |
| # It is possible that the cached credentials are for a different username, |
| # in which case the cache should be ignored. |
| if username is None or username == un: |
| username, password = un, pw |
| |
| if username is not None or password is not None: |
| # Convert the username and password if they're None, so that |
| # this netloc will show up as "cached" in the conditional above. |
| # Further, HTTPBasicAuth doesn't accept None, so it makes sense to |
| # cache the value that is going to be used. |
| username = username or "" |
| password = password or "" |
| |
| # Store any acquired credentials. |
| self.passwords[netloc] = (username, password) |
| |
| assert ( |
| # Credentials were found |
| (username is not None and password is not None) |
| # Credentials were not found |
| or (username is None and password is None) |
| ), f"Could not load credentials from url: {original_url}" |
| |
| return url, username, password |
| |
| def __call__(self, req: Request) -> Request: |
| # Get credentials for this request |
| url, username, password = self._get_url_and_credentials(req.url) |
| |
| # Set the url of the request to the url without any credentials |
| req.url = url |
| |
| 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 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 = (parsed.netloc, username, 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.""" |
| assert keyring is not None, "should never reach here without keyring" |
| if not keyring: |
| return |
| |
| 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) |
| except Exception: |
| logger.exception("Failed to save credentials") |
