Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1"""Network Authentication Helpers
3Contains interface (MultiDomainBasicAuth) and associated glue code for
4providing credentials in the context of network requests.
5"""
7from __future__ import annotations
9import json
10import logging
11import os
12import shutil
13import subprocess
14import sysconfig
15import typing
16import urllib.parse
17from abc import ABC, abstractmethod
18from functools import cache
19from os.path import commonpath
20from pathlib import Path
21from typing import Any, NamedTuple
23from pip._vendor.requests.auth import AuthBase, HTTPBasicAuth
24from pip._vendor.requests.utils import get_netrc_auth
26from pip._internal.utils.logging import getLogger
27from pip._internal.utils.misc import (
28 ask,
29 ask_input,
30 ask_password,
31 remove_auth_from_url,
32 split_auth_netloc_from_url,
33)
34from pip._internal.vcs.versioncontrol import AuthInfo
36if typing.TYPE_CHECKING:
37 from pip._vendor.requests import PreparedRequest
38 from pip._vendor.requests.models import Response
40logger = getLogger(__name__)
42KEYRING_DISABLED = False
45class Credentials(NamedTuple):
46 url: str
47 username: str
48 password: str
51class KeyRingBaseProvider(ABC):
52 """Keyring base provider interface"""
54 has_keyring: bool
56 @abstractmethod
57 def get_auth_info(self, url: str, username: str | None) -> AuthInfo | None: ...
59 @abstractmethod
60 def save_auth_info(self, url: str, username: str, password: str) -> None: ...
63class KeyRingNullProvider(KeyRingBaseProvider):
64 """Keyring null provider"""
66 has_keyring = False
68 def get_auth_info(self, url: str, username: str | None) -> AuthInfo | None:
69 return None
71 def save_auth_info(self, url: str, username: str, password: str) -> None:
72 return None
75class KeyRingPythonProvider(KeyRingBaseProvider):
76 """Keyring interface which uses locally imported `keyring`"""
78 has_keyring = True
80 def __init__(self) -> None:
81 import keyring
83 self.keyring = keyring
85 def get_auth_info(self, url: str, username: str | None) -> AuthInfo | None:
86 # Support keyring's get_credential interface which supports getting
87 # credentials without a username. This is only available for
88 # keyring>=15.2.0.
89 if hasattr(self.keyring, "get_credential"):
90 logger.debug("Getting credentials from keyring for %s", url)
91 cred = self.keyring.get_credential(url, username)
92 if cred is not None:
93 return cred.username, cred.password
94 return None
96 if username is not None:
97 logger.debug("Getting password from keyring for %s", url)
98 password = self.keyring.get_password(url, username)
99 if password:
100 return username, password
101 return None
103 def save_auth_info(self, url: str, username: str, password: str) -> None:
104 self.keyring.set_password(url, username, password)
107class KeyRingCliProvider(KeyRingBaseProvider):
108 """Provider which uses `keyring` cli
110 Instead of calling the keyring package installed alongside pip
111 we call keyring on the command line which will enable pip to
112 use which ever installation of keyring is available first in
113 PATH.
114 """
116 has_keyring = True
118 def __init__(self, cmd: str) -> None:
119 self.keyring = cmd
121 def get_auth_info(self, url: str, username: str | None) -> AuthInfo | None:
122 return self._get_creds(url, username)
124 def save_auth_info(self, url: str, username: str, password: str) -> None:
125 return self._set_password(url, username, password)
127 def _get_creds(self, service_name: str, username: str | None) -> AuthInfo | None:
128 """Mirror the implementation of keyring.get_credential using cli"""
129 if self.keyring is None:
130 return None
132 cmd = [self.keyring, "--mode=creds", "--output=json", "get", service_name]
133 if username is not None:
134 cmd.append(username)
136 env = os.environ.copy()
137 env["PYTHONIOENCODING"] = "utf-8"
138 res = subprocess.run( # noqa: UP022
139 cmd,
140 stdin=subprocess.DEVNULL,
141 stdout=subprocess.PIPE,
142 stderr=subprocess.PIPE,
143 env=env,
144 )
146 # Detect if the user is running an outdated version of keyring without support
147 # for querying credentials without username
148 errs = res.stderr.decode("utf-8")
149 if (
150 res.returncode == 2
151 and "unrecognized arguments" in errs
152 and "--mode=creds" in errs
153 ):
154 raise RuntimeError(
155 "Keyring util is outdated; must be at least version 25.2.1, "
156 "please upgrade it"
157 )
159 if res.returncode:
160 return None
162 data = json.loads(res.stdout.decode("utf-8"))
163 return (data["username"], data["password"])
165 def _set_password(self, service_name: str, username: str, password: str) -> None:
166 """Mirror the implementation of keyring.set_password using cli"""
167 if self.keyring is None:
168 return None
169 env = os.environ.copy()
170 env["PYTHONIOENCODING"] = "utf-8"
171 subprocess.run(
172 [self.keyring, "set", service_name, username],
173 input=f"{password}{os.linesep}".encode(),
174 env=env,
175 check=True,
176 )
177 return None
180@cache
181def get_keyring_provider(provider: str) -> KeyRingBaseProvider:
182 logger.verbose("Keyring provider requested: %s", provider)
184 # keyring has previously failed and been disabled
185 if KEYRING_DISABLED:
186 provider = "disabled"
187 if provider in ["import", "auto"]:
188 try:
189 impl = KeyRingPythonProvider()
190 logger.verbose("Keyring provider set: import")
191 return impl
192 except ImportError:
193 pass
194 except Exception as exc:
195 # In the event of an unexpected exception
196 # we should warn the user
197 msg = "Installed copy of keyring fails with exception %s"
198 if provider == "auto":
199 msg = msg + ", trying to find a keyring executable as a fallback"
200 logger.warning(msg, exc, exc_info=logger.isEnabledFor(logging.DEBUG))
201 if provider in ["subprocess", "auto"]:
202 cli = shutil.which("keyring")
203 if cli and cli.startswith(sysconfig.get_path("scripts")):
204 # all code within this function is stolen from shutil.which implementation
205 @typing.no_type_check
206 def PATH_as_shutil_which_determines_it() -> str:
207 path = os.environ.get("PATH", None)
208 if path is None:
209 try:
210 path = os.confstr("CS_PATH")
211 except (AttributeError, ValueError):
212 # os.confstr() or CS_PATH is not available
213 path = os.defpath
214 # bpo-35755: Don't use os.defpath if the PATH environment variable is
215 # set to an empty string
217 return path
219 scripts = Path(sysconfig.get_path("scripts"))
221 paths = []
222 for path in PATH_as_shutil_which_determines_it().split(os.pathsep):
223 p = Path(path)
224 try:
225 if not p.samefile(scripts):
226 paths.append(path)
227 except FileNotFoundError:
228 pass
230 path = os.pathsep.join(paths)
232 cli = shutil.which("keyring", path=path)
234 if cli:
235 logger.verbose("Keyring provider set: subprocess with executable %s", cli)
236 return KeyRingCliProvider(cli)
238 logger.verbose("Keyring provider set: disabled")
239 return KeyRingNullProvider()
242class MultiDomainBasicAuth(AuthBase):
243 def __init__(
244 self,
245 prompting: bool = True,
246 index_urls: list[str] | None = None,
247 keyring_provider: str = "auto",
248 ) -> None:
249 self.prompting = prompting
250 self.index_urls = index_urls
251 self.keyring_provider = keyring_provider
252 self.passwords: dict[str, AuthInfo] = {}
253 # When the user is prompted to enter credentials and keyring is
254 # available, we will offer to save them. If the user accepts,
255 # this value is set to the credentials they entered. After the
256 # request authenticates, the caller should call
257 # ``save_credentials`` to save these.
258 self._credentials_to_save: Credentials | None = None
260 @property
261 def keyring_provider(self) -> KeyRingBaseProvider:
262 return get_keyring_provider(self._keyring_provider)
264 @keyring_provider.setter
265 def keyring_provider(self, provider: str) -> None:
266 # The free function get_keyring_provider has been decorated with
267 # functools.cache. If an exception occurs in get_keyring_auth that
268 # cache will be cleared and keyring disabled, take that into account
269 # if you want to remove this indirection.
270 self._keyring_provider = provider
272 @property
273 def use_keyring(self) -> bool:
274 # We won't use keyring when --no-input is passed unless
275 # a specific provider is requested because it might require
276 # user interaction
277 return self.prompting or self._keyring_provider not in ["auto", "disabled"]
279 def _get_keyring_auth(
280 self,
281 url: str | None,
282 username: str | None,
283 ) -> AuthInfo | None:
284 """Return the tuple auth for a given url from keyring."""
285 # Do nothing if no url was provided
286 if not url:
287 return None
289 try:
290 return self.keyring_provider.get_auth_info(url, username)
291 except Exception as exc:
292 # Log the full exception (with stacktrace) at debug, so it'll only
293 # show up when running in verbose mode.
294 logger.debug("Keyring is skipped due to an exception", exc_info=True)
295 # Always log a shortened version of the exception.
296 logger.warning(
297 "Keyring is skipped due to an exception: %s",
298 str(exc),
299 )
300 global KEYRING_DISABLED
301 KEYRING_DISABLED = True
302 get_keyring_provider.cache_clear()
303 return None
305 def _get_index_url(self, url: str) -> str | None:
306 """Return the original index URL matching the requested URL.
308 Cached or dynamically generated credentials may work against
309 the original index URL rather than just the netloc.
311 The provided url should have had its username and password
312 removed already. If the original index url had credentials then
313 they will be included in the return value.
315 Returns None if no matching index was found, or if --no-index
316 was specified by the user.
317 """
318 if not url or not self.index_urls:
319 return None
321 url = remove_auth_from_url(url).rstrip("/") + "/"
322 parsed_url = urllib.parse.urlsplit(url)
324 candidates = []
326 for index in self.index_urls:
327 index = index.rstrip("/") + "/"
328 parsed_index = urllib.parse.urlsplit(remove_auth_from_url(index))
329 if parsed_url == parsed_index:
330 return index
332 if parsed_url.netloc != parsed_index.netloc:
333 continue
335 candidate = urllib.parse.urlsplit(index)
336 candidates.append(candidate)
338 if not candidates:
339 return None
341 candidates.sort(
342 reverse=True,
343 key=lambda candidate: len(
344 commonpath(
345 [
346 parsed_url.path,
347 candidate.path,
348 ]
349 )
350 ),
351 )
353 return urllib.parse.urlunsplit(candidates[0])
355 def _get_new_credentials(
356 self,
357 original_url: str,
358 *,
359 allow_netrc: bool = True,
360 allow_keyring: bool = False,
361 ) -> AuthInfo:
362 """Find and return credentials for the specified URL."""
363 # Split the credentials and netloc from the url.
364 url, netloc, url_user_password = split_auth_netloc_from_url(
365 original_url,
366 )
368 # Start with the credentials embedded in the url
369 username, password = url_user_password
370 if username is not None and password is not None:
371 logger.debug("Found credentials in url for %s", netloc)
372 return url_user_password
374 # Find a matching index url for this request
375 index_url = self._get_index_url(url)
376 if index_url:
377 # Split the credentials from the url.
378 index_info = split_auth_netloc_from_url(index_url)
379 if index_info:
380 index_url, _, index_url_user_password = index_info
381 logger.debug("Found index url %s", index_url)
383 # If an index URL was found, try its embedded credentials
384 if index_url and index_url_user_password[0] is not None:
385 username, password = index_url_user_password
386 if username is not None and password is not None:
387 logger.debug("Found credentials in index url for %s", netloc)
388 return index_url_user_password
390 # Get creds from netrc if we still don't have them
391 if allow_netrc:
392 netrc_auth = get_netrc_auth(original_url)
393 if netrc_auth:
394 logger.debug("Found credentials in netrc for %s", netloc)
395 return netrc_auth
397 # If we don't have a password and keyring is available, use it.
398 if allow_keyring:
399 # The index url is more specific than the netloc, so try it first
400 # fmt: off
401 kr_auth = (
402 self._get_keyring_auth(index_url, username) or
403 self._get_keyring_auth(netloc, username)
404 )
405 # fmt: on
406 if kr_auth:
407 logger.debug("Found credentials in keyring for %s", netloc)
408 return kr_auth
410 return username, password
412 def _get_url_and_credentials(
413 self, original_url: str
414 ) -> tuple[str, str | None, str | None]:
415 """Return the credentials to use for the provided URL.
417 If allowed, netrc and keyring may be used to obtain the
418 correct credentials.
420 Returns (url_without_credentials, username, password). Note
421 that even if the original URL contains credentials, this
422 function may return a different username and password.
423 """
424 url, netloc, _ = split_auth_netloc_from_url(original_url)
426 # Try to get credentials from original url
427 username, password = self._get_new_credentials(original_url)
429 # If credentials not found, use any stored credentials for this netloc.
430 # Do this if either the username or the password is missing.
431 # This accounts for the situation in which the user has specified
432 # the username in the index url, but the password comes from keyring.
433 if (username is None or password is None) and netloc in self.passwords:
434 un, pw = self.passwords[netloc]
435 # It is possible that the cached credentials are for a different username,
436 # in which case the cache should be ignored.
437 if username is None or username == un:
438 username, password = un, pw
440 if username is not None or password is not None:
441 # Convert the username and password if they're None, so that
442 # this netloc will show up as "cached" in the conditional above.
443 # Further, HTTPBasicAuth doesn't accept None, so it makes sense to
444 # cache the value that is going to be used.
445 username = username or ""
446 password = password or ""
448 # Store any acquired credentials.
449 self.passwords[netloc] = (username, password)
451 assert (
452 # Credentials were found
453 (username is not None and password is not None)
454 # Credentials were not found
455 or (username is None and password is None)
456 ), f"Could not load credentials from url: {original_url}"
458 return url, username, password
460 def __call__(self, req: PreparedRequest) -> PreparedRequest:
461 # Get credentials for this request
462 assert req.url is not None
463 url, username, password = self._get_url_and_credentials(req.url)
465 # Set the url of the request to the url without any credentials
466 req.url = url
468 if username is not None and password is not None:
469 # Send the basic auth with this request
470 req = HTTPBasicAuth(username, password)(req)
472 # Attach a hook to handle 401 responses
473 req.register_hook("response", self.handle_401)
475 return req
477 # Factored out to allow for easy patching in tests
478 def _prompt_for_password(self, netloc: str) -> tuple[str | None, str | None, bool]:
479 username = ask_input(f"User for {netloc}: ") if self.prompting else None
480 if not username:
481 return None, None, False
482 if self.use_keyring:
483 auth = self._get_keyring_auth(netloc, username)
484 if auth and auth[0] is not None and auth[1] is not None:
485 return auth[0], auth[1], False
486 password = ask_password("Password: ")
487 return username, password, True
489 # Factored out to allow for easy patching in tests
490 def _should_save_password_to_keyring(self) -> bool:
491 if (
492 not self.prompting
493 or not self.use_keyring
494 or not self.keyring_provider.has_keyring
495 ):
496 return False
497 return ask("Save credentials to keyring [y/N]: ", ["y", "n"]) == "y"
499 def handle_401(self, resp: Response, **kwargs: Any) -> Response:
500 # We only care about 401 responses, anything else we want to just
501 # pass through the actual response
502 if resp.status_code != 401:
503 return resp
505 # Look for credentials for the (possibly redirected) URL. Credentials
506 # embedded in the URL -- e.g. carried in the ``Location`` of a
507 # cross-origin redirect, whose ``Authorization`` header requests strips
508 # -- are always honoured, since recovering them needs no user
509 # interaction. Keyring is only consulted when it is enabled, because it
510 # may require interaction and is therefore disabled under --no-input.
511 username, password = self._get_new_credentials(
512 resp.url,
513 allow_netrc=False,
514 allow_keyring=self.use_keyring,
515 )
517 # We are not able to prompt the user so simply return the response
518 if not self.prompting and not username and not password:
519 return resp
521 parsed = urllib.parse.urlparse(resp.url)
523 # Prompt the user for a new username and password
524 save = False
525 if not username and not password:
526 username, password, save = self._prompt_for_password(parsed.netloc)
528 # Store the new username and password to use for future requests
529 self._credentials_to_save = None
530 if username is not None and password is not None:
531 self.passwords[parsed.netloc] = (username, password)
533 # Prompt to save the password to keyring
534 if save and self._should_save_password_to_keyring():
535 self._credentials_to_save = Credentials(
536 url=parsed.netloc,
537 username=username,
538 password=password,
539 )
541 # Consume content and release the original connection to allow our new
542 # request to reuse the same one.
543 # The result of the assignment isn't used, it's just needed to consume
544 # the content.
545 _ = resp.content
546 resp.raw.release_conn()
548 # Add our new username and password to the request
549 req = HTTPBasicAuth(username or "", password or "")(resp.request)
550 req.register_hook("response", self.warn_on_401)
552 # On successful request, save the credentials that were used to
553 # keyring. (Note that if the user responded "no" above, this member
554 # is not set and nothing will be saved.)
555 if self._credentials_to_save:
556 req.register_hook("response", self.save_credentials)
558 # Send our new request
559 new_resp = resp.connection.send(req, **kwargs)
560 new_resp.history.append(resp)
562 return new_resp
564 def warn_on_401(self, resp: Response, **kwargs: Any) -> None:
565 """Response callback to warn about incorrect credentials."""
566 if resp.status_code == 401:
567 logger.warning(
568 "401 Error, Credentials not correct for %s",
569 resp.request.url,
570 )
572 def save_credentials(self, resp: Response, **kwargs: Any) -> None:
573 """Response callback to save credentials on success."""
574 assert (
575 self.keyring_provider.has_keyring
576 ), "should never reach here without keyring"
578 creds = self._credentials_to_save
579 self._credentials_to_save = None
580 if creds and resp.status_code < 400:
581 try:
582 logger.info("Saving credentials to keyring")
583 self.keyring_provider.save_auth_info(
584 creds.url, creds.username, creds.password
585 )
586 except Exception:
587 logger.exception("Failed to save credentials")