CLERC/Reskreen
6
0
Fork 0
Code Issues 5 Pull Requests Actions Packages Projects 1 Releases 1 Wiki Activity

Change venv

Browse Source
This commit is contained in:
Ambulance Clerc
2023-05-31 08:31:22 +02:00
parent fb6f579089
commit fdbb52c96f
466 changed files with 25899 additions and 64721 deletions
Download Patch File Download Diff File Expand all files Collapse all files

367
venv/Lib/site-packages/pip/_internal/network/auth.py
View File

@@ -3,9 +3,18 @@
Contains interface (MultiDomainBasicAuth) and associated glue code for
providing credentials in the context of network requests.
"""
import logging
import os
import shutil
import subprocess
import sysconfig
import typing
import urllib.parse
from typing import Any, Dict, List, Optional, Tuple
from abc import ABC, abstractmethod
from functools import lru_cache
from os.path import commonprefix
from pathlib import Path
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,59 +32,204 @@ from pip._internal.vcs.versioncontrol import AuthInfo
logger = getLogger(__name__)
Credentials = Tuple[str, str, str]
try:
import keyring
except ImportError:
keyring = None
except Exception as exc:
logger.warning(
"Keyring is skipped due to an exception: %s",
str(exc),
)
keyring = None
KEYRING_DISABLED = False
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:
class Credentials(NamedTuple):
url: str
username: str
password: str
class KeyRingBaseProvider(ABC):
"""Keyring base provider interface"""
has_keyring: 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 KeyRingNullProvider(KeyRingBaseProvider):
"""Keyring null provider"""
has_keyring = False
def get_auth_info(self, url: str, username: Optional[str]) -> Optional[AuthInfo]:
return None
try:
try:
get_credential = keyring.get_credential
except AttributeError:
pass
else:
def save_auth_info(self, url: str, username: str, password: str) -> None:
return None
class KeyRingPythonProvider(KeyRingBaseProvider):
"""Keyring interface which uses locally imported `keyring`"""
has_keyring = True
def __init__(self) -> None:
import keyring
self.keyring = keyring
def get_auth_info(self, url: str, username: Optional[str]) -> Optional[AuthInfo]:
# 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
except Exception as exc:
logger.warning(
"Keyring is skipped due to an exception: %s",
str(exc),
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.
"""
has_keyring = True
def __init__(self, cmd: str) -> None:
self.keyring = cmd
def get_auth_info(self, url: str, username: Optional[str]) -> Optional[AuthInfo]:
# 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:
return self._set_password(url, username, password)
def _get_password(self, service_name: str, username: str) -> Optional[str]:
"""Mirror the implementation 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,
stdout=subprocess.PIPE,
env=env,
)
keyring = None
return None
if res.returncode:
return None
return res.stdout.decode("utf-8").strip(os.linesep)
def _set_password(self, service_name: str, username: str, password: str) -> None:
"""Mirror the implementation of keyring.set_password using cli"""
if self.keyring is None:
return None
env = os.environ.copy()
env["PYTHONIOENCODING"] = "utf-8"
subprocess.run(
[self.keyring, "set", service_name, username],
input=f"{password}{os.linesep}".encode("utf-8"),
env=env,
check=True,
)
return None
@lru_cache(maxsize=None)
def get_keyring_provider(provider: str) -> KeyRingBaseProvider:
logger.verbose("Keyring provider requested: %s", provider)
# keyring has previously failed and been disabled
if KEYRING_DISABLED:
provider = "disabled"
if provider in ["import", "auto"]:
try:
impl = KeyRingPythonProvider()
logger.verbose("Keyring provider set: import")
return impl
except ImportError:
pass
except Exception as exc:
# In the event of an unexpected exception
# we should warn the user
msg = "Installed copy of keyring fails with exception %s"
if provider == "auto":
msg = msg + ", trying to find a keyring executable as a fallback"
logger.warning(msg, exc, exc_info=logger.isEnabledFor(logging.DEBUG))
if provider in ["subprocess", "auto"]:
cli = shutil.which("keyring")
if cli and cli.startswith(sysconfig.get_path("scripts")):
# all code within this function is stolen from shutil.which implementation
@typing.no_type_check
def PATH_as_shutil_which_determines_it() -> str:
path = os.environ.get("PATH", None)
if path is None:
try:
path = os.confstr("CS_PATH")
except (AttributeError, ValueError):
# os.confstr() or CS_PATH is not available
path = os.defpath
# bpo-35755: Don't use os.defpath if the PATH environment variable is
# set to an empty string
return path
scripts = Path(sysconfig.get_path("scripts"))
paths = []
for path in PATH_as_shutil_which_determines_it().split(os.pathsep):
p = Path(path)
try:
if not p.samefile(scripts):
paths.append(path)
except FileNotFoundError:
pass
path = os.pathsep.join(paths)
cli = shutil.which("keyring", path=path)
if cli:
logger.verbose("Keyring provider set: subprocess with executable %s", cli)
return KeyRingCliProvider(cli)
logger.verbose("Keyring provider set: disabled")
return KeyRingNullProvider()
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,
keyring_provider: str = "auto",
) -> None:
self.prompting = prompting
self.index_urls = index_urls
self.keyring_provider = keyring_provider # type: ignore[assignment]
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,
@@ -84,6 +238,47 @@ class MultiDomainBasicAuth(AuthBase):
# ``save_credentials`` to save these.
self._credentials_to_save: Optional[Credentials] = None
@property
def keyring_provider(self) -> KeyRingBaseProvider:
return get_keyring_provider(self._keyring_provider)
@keyring_provider.setter
def keyring_provider(self, provider: str) -> None:
# The free function get_keyring_provider has been decorated with
# functools.cache. If an exception occurs in get_keyring_auth that
# cache will be cleared and keyring disabled, take that into account
# if you want to remove this indirection.
self._keyring_provider = provider
@property
def use_keyring(self) -> bool:
# We won't use keyring when --no-input is passed unless
# a specific provider is requested because it might require
# user interaction
return self.prompting or self._keyring_provider not in ["auto", "disabled"]
def _get_keyring_auth(
self,
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
try:
return self.keyring_provider.get_auth_info(url, username)
except Exception as exc:
logger.warning(
"Keyring is skipped due to an exception: %s",
str(exc),
)
global KEYRING_DISABLED
KEYRING_DISABLED = True
get_keyring_provider.cache_clear()
return None
def _get_index_url(self, url: str) -> Optional[str]:
"""Return the original index URL matching the requested URL.
@@ -100,15 +295,42 @@ class MultiDomainBasicAuth(AuthBase):
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
url = remove_auth_from_url(url).rstrip("/") + "/"
parsed_url = urllib.parse.urlsplit(url)
candidates = []
for index in self.index_urls:
index = index.rstrip("/") + "/"
parsed_index = urllib.parse.urlsplit(remove_auth_from_url(index))
if parsed_url == parsed_index:
return index
if parsed_url.netloc != parsed_index.netloc:
continue
candidate = urllib.parse.urlsplit(index)
candidates.append(candidate)
if not candidates:
return None
candidates.sort(
reverse=True,
key=lambda candidate: commonprefix(
[
parsed_url.path,
candidate.path,
]
).rfind("/"),
)
return urllib.parse.urlunsplit(candidates[0])
def _get_new_credentials(
self,
original_url: str,
*,
allow_netrc: bool = True,
allow_keyring: bool = False,
) -> AuthInfo:
@@ -152,8 +374,8 @@ class MultiDomainBasicAuth(AuthBase):
# 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)
self._get_keyring_auth(index_url, username) or
self._get_keyring_auth(netloc, username)
)
# fmt: on
if kr_auth:
@@ -179,9 +401,16 @@ class MultiDomainBasicAuth(AuthBase):
# 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
if username is None and password is None:
username, password = self.passwords.get(netloc, (None, None))
# 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
@@ -223,18 +452,23 @@ class MultiDomainBasicAuth(AuthBase):
def _prompt_for_password(
self, netloc: str
) -> Tuple[Optional[str], Optional[str], bool]:
username = ask_input(f"User for {netloc}: ")
username = ask_input(f"User for {netloc}: ") if self.prompting else None
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
if self.use_keyring:
auth = self._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:
if (
not self.prompting
or not self.use_keyring
or not self.keyring_provider.has_keyring
):
return False
return ask("Save credentials to keyring [y/N]: ", ["y", "n"]) == "y"
@@ -244,19 +478,22 @@ class MultiDomainBasicAuth(AuthBase):
if resp.status_code != 401:
return resp
username, password = None, None
# Query the keyring for credentials:
if self.use_keyring:
username, password = self._get_new_credentials(
resp.url,
allow_netrc=False,
allow_keyring=True,
)
# We are not able to prompt the user so simply return the response
if not self.prompting:
if not self.prompting and not username and not password:
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:
@@ -269,7 +506,11 @@ class MultiDomainBasicAuth(AuthBase):
# 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.
@@ -302,15 +543,17 @@ class MultiDomainBasicAuth(AuthBase):
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
assert (
self.keyring_provider.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.set_password(*creds)
self.keyring_provider.save_auth_info(
creds.url, creds.username, creds.password
)
except Exception:
logger.exception("Failed to save credentials")