Загрузить файлы в «venv/Lib/site-packages/pip/_internal/utils»
This commit is contained in:
248
venv/Lib/site-packages/pip/_internal/utils/subprocess.py
Normal file
248
venv/Lib/site-packages/pip/_internal/utils/subprocess.py
Normal file
@@ -0,0 +1,248 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
import os
|
||||
import shlex
|
||||
import subprocess
|
||||
from collections.abc import Iterable, Mapping
|
||||
from typing import Any, Callable, Literal, Union
|
||||
|
||||
from pip._vendor.rich.markup import escape
|
||||
|
||||
from pip._internal.cli.spinners import SpinnerInterface, open_spinner
|
||||
from pip._internal.exceptions import InstallationSubprocessError
|
||||
from pip._internal.utils.logging import VERBOSE, subprocess_logger
|
||||
from pip._internal.utils.misc import HiddenText
|
||||
|
||||
CommandArgs = list[Union[str, HiddenText]]
|
||||
|
||||
|
||||
def make_command(*args: str | HiddenText | CommandArgs) -> CommandArgs:
|
||||
"""
|
||||
Create a CommandArgs object.
|
||||
"""
|
||||
command_args: CommandArgs = []
|
||||
for arg in args:
|
||||
# Check for list instead of CommandArgs since CommandArgs is
|
||||
# only known during type-checking.
|
||||
if isinstance(arg, list):
|
||||
command_args.extend(arg)
|
||||
else:
|
||||
# Otherwise, arg is str or HiddenText.
|
||||
command_args.append(arg)
|
||||
|
||||
return command_args
|
||||
|
||||
|
||||
def format_command_args(args: list[str] | CommandArgs) -> str:
|
||||
"""
|
||||
Format command arguments for display.
|
||||
"""
|
||||
# For HiddenText arguments, display the redacted form by calling str().
|
||||
# Also, we don't apply str() to arguments that aren't HiddenText since
|
||||
# this can trigger a UnicodeDecodeError in Python 2 if the argument
|
||||
# has type unicode and includes a non-ascii character. (The type
|
||||
# checker doesn't ensure the annotations are correct in all cases.)
|
||||
return " ".join(
|
||||
shlex.quote(str(arg)) if isinstance(arg, HiddenText) else shlex.quote(arg)
|
||||
for arg in args
|
||||
)
|
||||
|
||||
|
||||
def reveal_command_args(args: list[str] | CommandArgs) -> list[str]:
|
||||
"""
|
||||
Return the arguments in their raw, unredacted form.
|
||||
"""
|
||||
return [arg.secret if isinstance(arg, HiddenText) else arg for arg in args]
|
||||
|
||||
|
||||
def call_subprocess(
|
||||
cmd: list[str] | CommandArgs,
|
||||
show_stdout: bool = False,
|
||||
cwd: str | None = None,
|
||||
on_returncode: Literal["raise", "warn", "ignore"] = "raise",
|
||||
extra_ok_returncodes: Iterable[int] | None = None,
|
||||
extra_environ: Mapping[str, Any] | None = None,
|
||||
unset_environ: Iterable[str] | None = None,
|
||||
spinner: SpinnerInterface | None = None,
|
||||
log_failed_cmd: bool | None = True,
|
||||
stdout_only: bool | None = False,
|
||||
*,
|
||||
command_desc: str,
|
||||
) -> str:
|
||||
"""
|
||||
Args:
|
||||
show_stdout: if true, use INFO to log the subprocess's stderr and
|
||||
stdout streams. Otherwise, use DEBUG. Defaults to False.
|
||||
extra_ok_returncodes: an iterable of integer return codes that are
|
||||
acceptable, in addition to 0. Defaults to None, which means [].
|
||||
unset_environ: an iterable of environment variable names to unset
|
||||
prior to calling subprocess.Popen().
|
||||
log_failed_cmd: if false, failed commands are not logged, only raised.
|
||||
stdout_only: if true, return only stdout, else return both. When true,
|
||||
logging of both stdout and stderr occurs when the subprocess has
|
||||
terminated, else logging occurs as subprocess output is produced.
|
||||
"""
|
||||
if extra_ok_returncodes is None:
|
||||
extra_ok_returncodes = []
|
||||
if unset_environ is None:
|
||||
unset_environ = []
|
||||
# Most places in pip use show_stdout=False. What this means is--
|
||||
#
|
||||
# - We connect the child's output (combined stderr and stdout) to a
|
||||
# single pipe, which we read.
|
||||
# - We log this output to stderr at DEBUG level as it is received.
|
||||
# - If DEBUG logging isn't enabled (e.g. if --verbose logging wasn't
|
||||
# requested), then we show a spinner so the user can still see the
|
||||
# subprocess is in progress.
|
||||
# - If the subprocess exits with an error, we log the output to stderr
|
||||
# at ERROR level if it hasn't already been displayed to the console
|
||||
# (e.g. if --verbose logging wasn't enabled). This way we don't log
|
||||
# the output to the console twice.
|
||||
#
|
||||
# If show_stdout=True, then the above is still done, but with DEBUG
|
||||
# replaced by INFO.
|
||||
if show_stdout:
|
||||
# Then log the subprocess output at INFO level.
|
||||
log_subprocess: Callable[..., None] = subprocess_logger.info
|
||||
used_level = logging.INFO
|
||||
else:
|
||||
# Then log the subprocess output using VERBOSE. This also ensures
|
||||
# it will be logged to the log file (aka user_log), if enabled.
|
||||
log_subprocess = subprocess_logger.verbose
|
||||
used_level = VERBOSE
|
||||
|
||||
# Whether the subprocess will be visible in the console.
|
||||
showing_subprocess = subprocess_logger.getEffectiveLevel() <= used_level
|
||||
|
||||
# Only use the spinner if we're not showing the subprocess output
|
||||
# and we have a spinner.
|
||||
use_spinner = not showing_subprocess and spinner is not None
|
||||
|
||||
log_subprocess("Running command %s", command_desc)
|
||||
env = os.environ.copy()
|
||||
if extra_environ:
|
||||
env.update(extra_environ)
|
||||
for name in unset_environ:
|
||||
env.pop(name, None)
|
||||
try:
|
||||
proc = subprocess.Popen(
|
||||
# Convert HiddenText objects to the underlying str.
|
||||
reveal_command_args(cmd),
|
||||
stdin=subprocess.PIPE,
|
||||
stdout=subprocess.PIPE,
|
||||
stderr=subprocess.STDOUT if not stdout_only else subprocess.PIPE,
|
||||
cwd=cwd,
|
||||
env=env,
|
||||
errors="backslashreplace",
|
||||
)
|
||||
except Exception as exc:
|
||||
if log_failed_cmd:
|
||||
subprocess_logger.critical(
|
||||
"Error %s while executing command %s",
|
||||
exc,
|
||||
command_desc,
|
||||
)
|
||||
raise
|
||||
all_output = []
|
||||
if not stdout_only:
|
||||
assert proc.stdout
|
||||
assert proc.stdin
|
||||
proc.stdin.close()
|
||||
# In this mode, stdout and stderr are in the same pipe.
|
||||
while True:
|
||||
line: str = proc.stdout.readline()
|
||||
if not line:
|
||||
break
|
||||
line = line.rstrip()
|
||||
all_output.append(line + "\n")
|
||||
|
||||
# Show the line immediately.
|
||||
log_subprocess(line)
|
||||
# Update the spinner.
|
||||
if use_spinner:
|
||||
assert spinner
|
||||
spinner.spin()
|
||||
try:
|
||||
proc.wait()
|
||||
finally:
|
||||
if proc.stdout:
|
||||
proc.stdout.close()
|
||||
output = "".join(all_output)
|
||||
else:
|
||||
# In this mode, stdout and stderr are in different pipes.
|
||||
# We must use communicate() which is the only safe way to read both.
|
||||
out, err = proc.communicate()
|
||||
# log line by line to preserve pip log indenting
|
||||
for out_line in out.splitlines():
|
||||
log_subprocess(out_line)
|
||||
all_output.append(out)
|
||||
for err_line in err.splitlines():
|
||||
log_subprocess(err_line)
|
||||
all_output.append(err)
|
||||
output = out
|
||||
|
||||
proc_had_error = proc.returncode and proc.returncode not in extra_ok_returncodes
|
||||
if use_spinner:
|
||||
assert spinner
|
||||
if proc_had_error:
|
||||
spinner.finish("error")
|
||||
else:
|
||||
spinner.finish("done")
|
||||
if proc_had_error:
|
||||
if on_returncode == "raise":
|
||||
error = InstallationSubprocessError(
|
||||
command_description=command_desc,
|
||||
exit_code=proc.returncode,
|
||||
output_lines=all_output if not showing_subprocess else None,
|
||||
)
|
||||
if log_failed_cmd:
|
||||
subprocess_logger.error("%s", error, extra={"rich": True})
|
||||
subprocess_logger.verbose(
|
||||
"[bold magenta]full command[/]: [blue]%s[/]",
|
||||
escape(format_command_args(cmd)),
|
||||
extra={"markup": True},
|
||||
)
|
||||
subprocess_logger.verbose(
|
||||
"[bold magenta]cwd[/]: %s",
|
||||
escape(cwd or "[inherit]"),
|
||||
extra={"markup": True},
|
||||
)
|
||||
|
||||
raise error
|
||||
elif on_returncode == "warn":
|
||||
subprocess_logger.warning(
|
||||
'Command "%s" had error code %s in %s',
|
||||
command_desc,
|
||||
proc.returncode,
|
||||
cwd,
|
||||
)
|
||||
elif on_returncode == "ignore":
|
||||
pass
|
||||
else:
|
||||
raise ValueError(f"Invalid value: on_returncode={on_returncode!r}")
|
||||
return output
|
||||
|
||||
|
||||
def runner_with_spinner_message(message: str) -> Callable[..., None]:
|
||||
"""Provide a subprocess_runner that shows a spinner message.
|
||||
|
||||
Intended for use with for BuildBackendHookCaller. Thus, the runner has
|
||||
an API that matches what's expected by BuildBackendHookCaller.subprocess_runner.
|
||||
"""
|
||||
|
||||
def runner(
|
||||
cmd: list[str],
|
||||
cwd: str | None = None,
|
||||
extra_environ: Mapping[str, Any] | None = None,
|
||||
) -> None:
|
||||
with open_spinner(message) as spinner:
|
||||
call_subprocess(
|
||||
cmd,
|
||||
command_desc=message,
|
||||
cwd=cwd,
|
||||
extra_environ=extra_environ,
|
||||
spinner=spinner,
|
||||
)
|
||||
|
||||
return runner
|
||||
294
venv/Lib/site-packages/pip/_internal/utils/temp_dir.py
Normal file
294
venv/Lib/site-packages/pip/_internal/utils/temp_dir.py
Normal file
@@ -0,0 +1,294 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import errno
|
||||
import itertools
|
||||
import logging
|
||||
import os.path
|
||||
import tempfile
|
||||
import traceback
|
||||
from collections.abc import Generator
|
||||
from contextlib import ExitStack, contextmanager
|
||||
from pathlib import Path
|
||||
from typing import (
|
||||
Any,
|
||||
Callable,
|
||||
TypeVar,
|
||||
)
|
||||
|
||||
from pip._internal.utils.misc import enum, rmtree
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
_T = TypeVar("_T", bound="TempDirectory")
|
||||
|
||||
|
||||
# Kinds of temporary directories. Only needed for ones that are
|
||||
# globally-managed.
|
||||
tempdir_kinds = enum(
|
||||
BUILD_ENV="build-env",
|
||||
EPHEM_WHEEL_CACHE="ephem-wheel-cache",
|
||||
REQ_BUILD="req-build",
|
||||
)
|
||||
|
||||
|
||||
_tempdir_manager: ExitStack | None = None
|
||||
|
||||
|
||||
@contextmanager
|
||||
def global_tempdir_manager() -> Generator[None, None, None]:
|
||||
global _tempdir_manager
|
||||
with ExitStack() as stack:
|
||||
old_tempdir_manager, _tempdir_manager = _tempdir_manager, stack
|
||||
try:
|
||||
yield
|
||||
finally:
|
||||
_tempdir_manager = old_tempdir_manager
|
||||
|
||||
|
||||
class TempDirectoryTypeRegistry:
|
||||
"""Manages temp directory behavior"""
|
||||
|
||||
def __init__(self) -> None:
|
||||
self._should_delete: dict[str, bool] = {}
|
||||
|
||||
def set_delete(self, kind: str, value: bool) -> None:
|
||||
"""Indicate whether a TempDirectory of the given kind should be
|
||||
auto-deleted.
|
||||
"""
|
||||
self._should_delete[kind] = value
|
||||
|
||||
def get_delete(self, kind: str) -> bool:
|
||||
"""Get configured auto-delete flag for a given TempDirectory type,
|
||||
default True.
|
||||
"""
|
||||
return self._should_delete.get(kind, True)
|
||||
|
||||
|
||||
_tempdir_registry: TempDirectoryTypeRegistry | None = None
|
||||
|
||||
|
||||
@contextmanager
|
||||
def tempdir_registry() -> Generator[TempDirectoryTypeRegistry, None, None]:
|
||||
"""Provides a scoped global tempdir registry that can be used to dictate
|
||||
whether directories should be deleted.
|
||||
"""
|
||||
global _tempdir_registry
|
||||
old_tempdir_registry = _tempdir_registry
|
||||
_tempdir_registry = TempDirectoryTypeRegistry()
|
||||
try:
|
||||
yield _tempdir_registry
|
||||
finally:
|
||||
_tempdir_registry = old_tempdir_registry
|
||||
|
||||
|
||||
class _Default:
|
||||
pass
|
||||
|
||||
|
||||
_default = _Default()
|
||||
|
||||
|
||||
class TempDirectory:
|
||||
"""Helper class that owns and cleans up a temporary directory.
|
||||
|
||||
This class can be used as a context manager or as an OO representation of a
|
||||
temporary directory.
|
||||
|
||||
Attributes:
|
||||
path
|
||||
Location to the created temporary directory
|
||||
delete
|
||||
Whether the directory should be deleted when exiting
|
||||
(when used as a contextmanager)
|
||||
|
||||
Methods:
|
||||
cleanup()
|
||||
Deletes the temporary directory
|
||||
|
||||
When used as a context manager, if the delete attribute is True, on
|
||||
exiting the context the temporary directory is deleted.
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
path: str | None = None,
|
||||
delete: bool | None | _Default = _default,
|
||||
kind: str = "temp",
|
||||
globally_managed: bool = False,
|
||||
ignore_cleanup_errors: bool = True,
|
||||
):
|
||||
super().__init__()
|
||||
|
||||
if delete is _default:
|
||||
if path is not None:
|
||||
# If we were given an explicit directory, resolve delete option
|
||||
# now.
|
||||
delete = False
|
||||
else:
|
||||
# Otherwise, we wait until cleanup and see what
|
||||
# tempdir_registry says.
|
||||
delete = None
|
||||
|
||||
# The only time we specify path is in for editables where it
|
||||
# is the value of the --src option.
|
||||
if path is None:
|
||||
path = self._create(kind)
|
||||
|
||||
self._path = path
|
||||
self._deleted = False
|
||||
self.delete = delete
|
||||
self.kind = kind
|
||||
self.ignore_cleanup_errors = ignore_cleanup_errors
|
||||
|
||||
if globally_managed:
|
||||
assert _tempdir_manager is not None
|
||||
_tempdir_manager.enter_context(self)
|
||||
|
||||
@property
|
||||
def path(self) -> str:
|
||||
assert not self._deleted, f"Attempted to access deleted path: {self._path}"
|
||||
return self._path
|
||||
|
||||
def __repr__(self) -> str:
|
||||
return f"<{self.__class__.__name__} {self.path!r}>"
|
||||
|
||||
def __enter__(self: _T) -> _T:
|
||||
return self
|
||||
|
||||
def __exit__(self, exc: Any, value: Any, tb: Any) -> None:
|
||||
if self.delete is not None:
|
||||
delete = self.delete
|
||||
elif _tempdir_registry:
|
||||
delete = _tempdir_registry.get_delete(self.kind)
|
||||
else:
|
||||
delete = True
|
||||
|
||||
if delete:
|
||||
self.cleanup()
|
||||
|
||||
def _create(self, kind: str) -> str:
|
||||
"""Create a temporary directory and store its path in self.path"""
|
||||
# We realpath here because some systems have their default tmpdir
|
||||
# symlinked to another directory. This tends to confuse build
|
||||
# scripts, so we canonicalize the path by traversing potential
|
||||
# symlinks here.
|
||||
path = os.path.realpath(tempfile.mkdtemp(prefix=f"pip-{kind}-"))
|
||||
logger.debug("Created temporary directory: %s", path)
|
||||
return path
|
||||
|
||||
def cleanup(self) -> None:
|
||||
"""Remove the temporary directory created and reset state"""
|
||||
self._deleted = True
|
||||
if not os.path.exists(self._path):
|
||||
return
|
||||
|
||||
errors: list[BaseException] = []
|
||||
|
||||
def onerror(
|
||||
func: Callable[..., Any],
|
||||
path: Path,
|
||||
exc_val: BaseException,
|
||||
) -> None:
|
||||
"""Log a warning for a `rmtree` error and continue"""
|
||||
formatted_exc = "\n".join(
|
||||
traceback.format_exception_only(type(exc_val), exc_val)
|
||||
)
|
||||
formatted_exc = formatted_exc.rstrip() # remove trailing new line
|
||||
if func in (os.unlink, os.remove, os.rmdir):
|
||||
logger.debug(
|
||||
"Failed to remove a temporary file '%s' due to %s.\n",
|
||||
path,
|
||||
formatted_exc,
|
||||
)
|
||||
else:
|
||||
logger.debug("%s failed with %s.", func.__qualname__, formatted_exc)
|
||||
errors.append(exc_val)
|
||||
|
||||
if self.ignore_cleanup_errors:
|
||||
try:
|
||||
# first try with @retry; retrying to handle ephemeral errors
|
||||
rmtree(self._path, ignore_errors=False)
|
||||
except OSError:
|
||||
# last pass ignore/log all errors
|
||||
rmtree(self._path, onexc=onerror)
|
||||
if errors:
|
||||
logger.warning(
|
||||
"Failed to remove contents in a temporary directory '%s'.\n"
|
||||
"You can safely remove it manually.",
|
||||
self._path,
|
||||
)
|
||||
else:
|
||||
rmtree(self._path)
|
||||
|
||||
|
||||
class AdjacentTempDirectory(TempDirectory):
|
||||
"""Helper class that creates a temporary directory adjacent to a real one.
|
||||
|
||||
Attributes:
|
||||
original
|
||||
The original directory to create a temp directory for.
|
||||
path
|
||||
After calling create() or entering, contains the full
|
||||
path to the temporary directory.
|
||||
delete
|
||||
Whether the directory should be deleted when exiting
|
||||
(when used as a contextmanager)
|
||||
|
||||
"""
|
||||
|
||||
# The characters that may be used to name the temp directory
|
||||
# We always prepend a ~ and then rotate through these until
|
||||
# a usable name is found.
|
||||
# pkg_resources raises a different error for .dist-info folder
|
||||
# with leading '-' and invalid metadata
|
||||
LEADING_CHARS = "-~.=%0123456789"
|
||||
|
||||
def __init__(self, original: str, delete: bool | None = None) -> None:
|
||||
self.original = original.rstrip("/\\")
|
||||
super().__init__(delete=delete)
|
||||
|
||||
@classmethod
|
||||
def _generate_names(cls, name: str) -> Generator[str, None, None]:
|
||||
"""Generates a series of temporary names.
|
||||
|
||||
The algorithm replaces the leading characters in the name
|
||||
with ones that are valid filesystem characters, but are not
|
||||
valid package names (for both Python and pip definitions of
|
||||
package).
|
||||
"""
|
||||
for i in range(1, len(name)):
|
||||
for candidate in itertools.combinations_with_replacement(
|
||||
cls.LEADING_CHARS, i - 1
|
||||
):
|
||||
new_name = "~" + "".join(candidate) + name[i:]
|
||||
if new_name != name:
|
||||
yield new_name
|
||||
|
||||
# If we make it this far, we will have to make a longer name
|
||||
for i in range(len(cls.LEADING_CHARS)):
|
||||
for candidate in itertools.combinations_with_replacement(
|
||||
cls.LEADING_CHARS, i
|
||||
):
|
||||
new_name = "~" + "".join(candidate) + name
|
||||
if new_name != name:
|
||||
yield new_name
|
||||
|
||||
def _create(self, kind: str) -> str:
|
||||
root, name = os.path.split(self.original)
|
||||
for candidate in self._generate_names(name):
|
||||
path = os.path.join(root, candidate)
|
||||
try:
|
||||
os.mkdir(path)
|
||||
except OSError as ex:
|
||||
# Continue if the name exists already
|
||||
if ex.errno != errno.EEXIST:
|
||||
raise
|
||||
else:
|
||||
path = os.path.realpath(path)
|
||||
break
|
||||
else:
|
||||
# Final fallback on the default behavior.
|
||||
path = os.path.realpath(tempfile.mkdtemp(prefix=f"pip-{kind}-"))
|
||||
|
||||
logger.debug("Created temporary directory: %s", path)
|
||||
return path
|
||||
362
venv/Lib/site-packages/pip/_internal/utils/unpacking.py
Normal file
362
venv/Lib/site-packages/pip/_internal/utils/unpacking.py
Normal file
@@ -0,0 +1,362 @@
|
||||
"""Utilities related archives."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
import os
|
||||
import shutil
|
||||
import stat
|
||||
import sys
|
||||
import tarfile
|
||||
import zipfile
|
||||
from collections.abc import Iterable
|
||||
from zipfile import ZipInfo
|
||||
|
||||
from pip._internal.exceptions import InstallationError
|
||||
from pip._internal.utils.filetypes import (
|
||||
BZ2_EXTENSIONS,
|
||||
TAR_EXTENSIONS,
|
||||
XZ_EXTENSIONS,
|
||||
ZIP_EXTENSIONS,
|
||||
)
|
||||
from pip._internal.utils.misc import ensure_dir
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
SUPPORTED_EXTENSIONS = ZIP_EXTENSIONS + TAR_EXTENSIONS
|
||||
|
||||
try:
|
||||
import bz2 # noqa
|
||||
|
||||
SUPPORTED_EXTENSIONS += BZ2_EXTENSIONS
|
||||
except ImportError:
|
||||
logger.debug("bz2 module is not available")
|
||||
|
||||
try:
|
||||
# Only for Python 3.3+
|
||||
import lzma # noqa
|
||||
|
||||
SUPPORTED_EXTENSIONS += XZ_EXTENSIONS
|
||||
except ImportError:
|
||||
logger.debug("lzma module is not available")
|
||||
|
||||
|
||||
def current_umask() -> int:
|
||||
"""Get the current umask which involves having to set it temporarily."""
|
||||
mask = os.umask(0)
|
||||
os.umask(mask)
|
||||
return mask
|
||||
|
||||
|
||||
def split_leading_dir(path: str) -> list[str]:
|
||||
path = path.lstrip("/").lstrip("\\")
|
||||
if "/" in path and (
|
||||
("\\" in path and path.find("/") < path.find("\\")) or "\\" not in path
|
||||
):
|
||||
return path.split("/", 1)
|
||||
elif "\\" in path:
|
||||
return path.split("\\", 1)
|
||||
else:
|
||||
return [path, ""]
|
||||
|
||||
|
||||
def has_leading_dir(paths: Iterable[str]) -> bool:
|
||||
"""Returns true if all the paths have the same leading path name
|
||||
(i.e., everything is in one subdirectory in an archive)"""
|
||||
common_prefix = None
|
||||
for path in paths:
|
||||
prefix, rest = split_leading_dir(path)
|
||||
if not prefix:
|
||||
return False
|
||||
elif common_prefix is None:
|
||||
common_prefix = prefix
|
||||
elif prefix != common_prefix:
|
||||
return False
|
||||
return True
|
||||
|
||||
|
||||
def is_within_directory(directory: str, target: str) -> bool:
|
||||
"""
|
||||
Return true if the absolute path of target is within the directory
|
||||
"""
|
||||
abs_directory = os.path.abspath(directory)
|
||||
abs_target = os.path.abspath(target)
|
||||
|
||||
prefix = os.path.commonpath([abs_directory, abs_target])
|
||||
return prefix == abs_directory
|
||||
|
||||
|
||||
def _get_default_mode_plus_executable() -> int:
|
||||
return 0o777 & ~current_umask() | 0o111
|
||||
|
||||
|
||||
def set_extracted_file_to_default_mode_plus_executable(path: str) -> None:
|
||||
"""
|
||||
Make file present at path have execute for user/group/world
|
||||
(chmod +x) is no-op on windows per python docs
|
||||
"""
|
||||
os.chmod(path, _get_default_mode_plus_executable())
|
||||
|
||||
|
||||
def zip_item_is_executable(info: ZipInfo) -> bool:
|
||||
mode = info.external_attr >> 16
|
||||
# if mode and regular file and any execute permissions for
|
||||
# user/group/world?
|
||||
return bool(mode and stat.S_ISREG(mode) and mode & 0o111)
|
||||
|
||||
|
||||
def unzip_file(filename: str, location: str, flatten: bool = True) -> None:
|
||||
"""
|
||||
Unzip the file (with path `filename`) to the destination `location`. All
|
||||
files are written based on system defaults and umask (i.e. permissions are
|
||||
not preserved), except that regular file members with any execute
|
||||
permissions (user, group, or world) have "chmod +x" applied after being
|
||||
written. Note that for windows, any execute changes using os.chmod are
|
||||
no-ops per the python docs.
|
||||
"""
|
||||
ensure_dir(location)
|
||||
zipfp = open(filename, "rb")
|
||||
try:
|
||||
zip = zipfile.ZipFile(zipfp, allowZip64=True)
|
||||
leading = has_leading_dir(zip.namelist()) and flatten
|
||||
for info in zip.infolist():
|
||||
name = info.filename
|
||||
fn = name
|
||||
if leading:
|
||||
fn = split_leading_dir(name)[1]
|
||||
fn = os.path.join(location, fn)
|
||||
dir = os.path.dirname(fn)
|
||||
if not is_within_directory(location, fn):
|
||||
message = (
|
||||
"The zip file ({}) has a file ({}) trying to install "
|
||||
"outside target directory ({})"
|
||||
)
|
||||
raise InstallationError(message.format(filename, fn, location))
|
||||
if fn.endswith(("/", "\\")):
|
||||
# A directory
|
||||
ensure_dir(fn)
|
||||
else:
|
||||
ensure_dir(dir)
|
||||
# Don't use read() to avoid allocating an arbitrarily large
|
||||
# chunk of memory for the file's content
|
||||
fp = zip.open(name)
|
||||
try:
|
||||
with open(fn, "wb") as destfp:
|
||||
shutil.copyfileobj(fp, destfp)
|
||||
finally:
|
||||
fp.close()
|
||||
if zip_item_is_executable(info):
|
||||
set_extracted_file_to_default_mode_plus_executable(fn)
|
||||
finally:
|
||||
zipfp.close()
|
||||
|
||||
|
||||
def untar_file(filename: str, location: str) -> None:
|
||||
"""
|
||||
Untar the file (with path `filename`) to the destination `location`.
|
||||
All files are written based on system defaults and umask (i.e. permissions
|
||||
are not preserved), except that regular file members with any execute
|
||||
permissions (user, group, or world) have "chmod +x" applied on top of the
|
||||
default. Note that for windows, any execute changes using os.chmod are
|
||||
no-ops per the python docs.
|
||||
"""
|
||||
ensure_dir(location)
|
||||
if filename.lower().endswith(".gz") or filename.lower().endswith(".tgz"):
|
||||
mode = "r:gz"
|
||||
elif filename.lower().endswith(BZ2_EXTENSIONS):
|
||||
mode = "r:bz2"
|
||||
elif filename.lower().endswith(XZ_EXTENSIONS):
|
||||
mode = "r:xz"
|
||||
elif filename.lower().endswith(".tar"):
|
||||
mode = "r"
|
||||
else:
|
||||
logger.warning(
|
||||
"Cannot determine compression type for file %s",
|
||||
filename,
|
||||
)
|
||||
mode = "r:*"
|
||||
|
||||
tar = tarfile.open(filename, mode, encoding="utf-8") # type: ignore
|
||||
try:
|
||||
leading = has_leading_dir([member.name for member in tar.getmembers()])
|
||||
|
||||
# PEP 706 added `tarfile.data_filter`, and made some other changes to
|
||||
# Python's tarfile module (see below). The features were backported to
|
||||
# security releases.
|
||||
try:
|
||||
data_filter = tarfile.data_filter
|
||||
except AttributeError:
|
||||
_untar_without_filter(filename, location, tar, leading)
|
||||
else:
|
||||
default_mode_plus_executable = _get_default_mode_plus_executable()
|
||||
|
||||
if leading:
|
||||
# Strip the leading directory from all files in the archive,
|
||||
# including hardlink targets (which are relative to the
|
||||
# unpack location).
|
||||
for member in tar.getmembers():
|
||||
name_lead, name_rest = split_leading_dir(member.name)
|
||||
member.name = name_rest
|
||||
if member.islnk():
|
||||
lnk_lead, lnk_rest = split_leading_dir(member.linkname)
|
||||
if lnk_lead == name_lead:
|
||||
member.linkname = lnk_rest
|
||||
|
||||
def pip_filter(member: tarfile.TarInfo, path: str) -> tarfile.TarInfo:
|
||||
orig_mode = member.mode
|
||||
try:
|
||||
try:
|
||||
member = data_filter(member, location)
|
||||
except tarfile.LinkOutsideDestinationError:
|
||||
if sys.version_info[:3] in {
|
||||
(3, 9, 17),
|
||||
(3, 10, 12),
|
||||
(3, 11, 4),
|
||||
}:
|
||||
# The tarfile filter in specific Python versions
|
||||
# raises LinkOutsideDestinationError on valid input
|
||||
# (https://github.com/python/cpython/issues/107845)
|
||||
# Ignore the error there, but do use the
|
||||
# more lax `tar_filter`
|
||||
member = tarfile.tar_filter(member, location)
|
||||
else:
|
||||
raise
|
||||
except tarfile.TarError as exc:
|
||||
message = "Invalid member in the tar file {}: {}"
|
||||
# Filter error messages mention the member name.
|
||||
# No need to add it here.
|
||||
raise InstallationError(
|
||||
message.format(
|
||||
filename,
|
||||
exc,
|
||||
)
|
||||
)
|
||||
if member.isfile() and orig_mode & 0o111:
|
||||
member.mode = default_mode_plus_executable
|
||||
else:
|
||||
# See PEP 706 note above.
|
||||
# The PEP changed this from `int` to `Optional[int]`,
|
||||
# where None means "use the default". Mypy doesn't
|
||||
# know this yet.
|
||||
member.mode = None # type: ignore [assignment]
|
||||
return member
|
||||
|
||||
tar.extractall(location, filter=pip_filter)
|
||||
|
||||
finally:
|
||||
tar.close()
|
||||
|
||||
|
||||
def is_symlink_target_in_tar(tar: tarfile.TarFile, tarinfo: tarfile.TarInfo) -> bool:
|
||||
"""Check if the file pointed to by the symbolic link is in the tar archive"""
|
||||
linkname = os.path.join(os.path.dirname(tarinfo.name), tarinfo.linkname)
|
||||
|
||||
linkname = os.path.normpath(linkname)
|
||||
linkname = linkname.replace("\\", "/")
|
||||
|
||||
try:
|
||||
tar.getmember(linkname)
|
||||
return True
|
||||
except KeyError:
|
||||
return False
|
||||
|
||||
|
||||
def _untar_without_filter(
|
||||
filename: str,
|
||||
location: str,
|
||||
tar: tarfile.TarFile,
|
||||
leading: bool,
|
||||
) -> None:
|
||||
"""Fallback for Python without tarfile.data_filter"""
|
||||
# NOTE: This function can be removed once pip requires CPython ≥ 3.12.
|
||||
# PEP 706 added tarfile.data_filter, made tarfile extraction operations more secure.
|
||||
# This feature is fully supported from CPython 3.12 onward.
|
||||
for member in tar.getmembers():
|
||||
fn = member.name
|
||||
if leading:
|
||||
fn = split_leading_dir(fn)[1]
|
||||
path = os.path.join(location, fn)
|
||||
if not is_within_directory(location, path):
|
||||
message = (
|
||||
"The tar file ({}) has a file ({}) trying to install "
|
||||
"outside target directory ({})"
|
||||
)
|
||||
raise InstallationError(message.format(filename, path, location))
|
||||
if member.isdir():
|
||||
ensure_dir(path)
|
||||
elif member.issym():
|
||||
if not is_symlink_target_in_tar(tar, member):
|
||||
message = (
|
||||
"The tar file ({}) has a file ({}) trying to install "
|
||||
"outside target directory ({})"
|
||||
)
|
||||
raise InstallationError(
|
||||
message.format(filename, member.name, member.linkname)
|
||||
)
|
||||
try:
|
||||
tar._extract_member(member, path)
|
||||
except Exception as exc:
|
||||
# Some corrupt tar files seem to produce this
|
||||
# (specifically bad symlinks)
|
||||
logger.warning(
|
||||
"In the tar file %s the member %s is invalid: %s",
|
||||
filename,
|
||||
member.name,
|
||||
exc,
|
||||
)
|
||||
continue
|
||||
else:
|
||||
try:
|
||||
fp = tar.extractfile(member)
|
||||
except (KeyError, AttributeError) as exc:
|
||||
# Some corrupt tar files seem to produce this
|
||||
# (specifically bad symlinks)
|
||||
logger.warning(
|
||||
"In the tar file %s the member %s is invalid: %s",
|
||||
filename,
|
||||
member.name,
|
||||
exc,
|
||||
)
|
||||
continue
|
||||
ensure_dir(os.path.dirname(path))
|
||||
assert fp is not None
|
||||
with open(path, "wb") as destfp:
|
||||
shutil.copyfileobj(fp, destfp)
|
||||
fp.close()
|
||||
# Update the timestamp (useful for cython compiled files)
|
||||
tar.utime(member, path)
|
||||
# member have any execute permissions for user/group/world?
|
||||
if member.mode & 0o111:
|
||||
set_extracted_file_to_default_mode_plus_executable(path)
|
||||
|
||||
|
||||
def unpack_file(
|
||||
filename: str,
|
||||
location: str,
|
||||
content_type: str | None = None,
|
||||
) -> None:
|
||||
filename = os.path.realpath(filename)
|
||||
if (
|
||||
content_type == "application/zip"
|
||||
or filename.lower().endswith(ZIP_EXTENSIONS)
|
||||
or zipfile.is_zipfile(filename)
|
||||
):
|
||||
unzip_file(filename, location, flatten=not filename.endswith(".whl"))
|
||||
elif (
|
||||
content_type == "application/x-gzip"
|
||||
or tarfile.is_tarfile(filename)
|
||||
or filename.lower().endswith(TAR_EXTENSIONS + BZ2_EXTENSIONS + XZ_EXTENSIONS)
|
||||
):
|
||||
untar_file(filename, location)
|
||||
else:
|
||||
# FIXME: handle?
|
||||
# FIXME: magic signatures?
|
||||
logger.critical(
|
||||
"Cannot unpack file %s (downloaded from %s, content-type: %s); "
|
||||
"cannot detect archive format",
|
||||
filename,
|
||||
location,
|
||||
content_type,
|
||||
)
|
||||
raise InstallationError(f"Cannot determine archive format of {location}")
|
||||
55
venv/Lib/site-packages/pip/_internal/utils/urls.py
Normal file
55
venv/Lib/site-packages/pip/_internal/utils/urls.py
Normal file
@@ -0,0 +1,55 @@
|
||||
import os
|
||||
import string
|
||||
import urllib.parse
|
||||
import urllib.request
|
||||
|
||||
from .compat import WINDOWS
|
||||
|
||||
|
||||
def path_to_url(path: str) -> str:
|
||||
"""
|
||||
Convert a path to a file: URL. The path will be made absolute and have
|
||||
quoted path parts.
|
||||
"""
|
||||
path = os.path.normpath(os.path.abspath(path))
|
||||
url = urllib.parse.urljoin("file://", urllib.request.pathname2url(path))
|
||||
return url
|
||||
|
||||
|
||||
def url_to_path(url: str) -> str:
|
||||
"""
|
||||
Convert a file: URL to a path.
|
||||
"""
|
||||
assert url.startswith(
|
||||
"file:"
|
||||
), f"You can only turn file: urls into filenames (not {url!r})"
|
||||
|
||||
_, netloc, path, _, _ = urllib.parse.urlsplit(url)
|
||||
|
||||
if not netloc or netloc == "localhost":
|
||||
# According to RFC 8089, same as empty authority.
|
||||
netloc = ""
|
||||
elif WINDOWS:
|
||||
# If we have a UNC path, prepend UNC share notation.
|
||||
netloc = "\\\\" + netloc
|
||||
else:
|
||||
raise ValueError(
|
||||
f"non-local file URIs are not supported on this platform: {url!r}"
|
||||
)
|
||||
|
||||
path = urllib.request.url2pathname(netloc + path)
|
||||
|
||||
# On Windows, urlsplit parses the path as something like "/C:/Users/foo".
|
||||
# This creates issues for path-related functions like io.open(), so we try
|
||||
# to detect and strip the leading slash.
|
||||
if (
|
||||
WINDOWS
|
||||
and not netloc # Not UNC.
|
||||
and len(path) >= 3
|
||||
and path[0] == "/" # Leading slash to strip.
|
||||
and path[1] in string.ascii_letters # Drive letter.
|
||||
and path[2:4] in (":", ":/") # Colon + end of string, or colon + absolute path.
|
||||
):
|
||||
path = path[1:]
|
||||
|
||||
return path
|
||||
105
venv/Lib/site-packages/pip/_internal/utils/virtualenv.py
Normal file
105
venv/Lib/site-packages/pip/_internal/utils/virtualenv.py
Normal file
@@ -0,0 +1,105 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
import os
|
||||
import re
|
||||
import site
|
||||
import sys
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
_INCLUDE_SYSTEM_SITE_PACKAGES_REGEX = re.compile(
|
||||
r"include-system-site-packages\s*=\s*(?P<value>true|false)"
|
||||
)
|
||||
|
||||
|
||||
def _running_under_venv() -> bool:
|
||||
"""Checks if sys.base_prefix and sys.prefix match.
|
||||
|
||||
This handles PEP 405 compliant virtual environments.
|
||||
"""
|
||||
return sys.prefix != getattr(sys, "base_prefix", sys.prefix)
|
||||
|
||||
|
||||
def _running_under_legacy_virtualenv() -> bool:
|
||||
"""Checks if sys.real_prefix is set.
|
||||
|
||||
This handles virtual environments created with pypa's virtualenv.
|
||||
"""
|
||||
# pypa/virtualenv case
|
||||
return hasattr(sys, "real_prefix")
|
||||
|
||||
|
||||
def running_under_virtualenv() -> bool:
|
||||
"""True if we're running inside a virtual environment, False otherwise."""
|
||||
return _running_under_venv() or _running_under_legacy_virtualenv()
|
||||
|
||||
|
||||
def _get_pyvenv_cfg_lines() -> list[str] | None:
|
||||
"""Reads {sys.prefix}/pyvenv.cfg and returns its contents as list of lines
|
||||
|
||||
Returns None, if it could not read/access the file.
|
||||
"""
|
||||
pyvenv_cfg_file = os.path.join(sys.prefix, "pyvenv.cfg")
|
||||
try:
|
||||
# Although PEP 405 does not specify, the built-in venv module always
|
||||
# writes with UTF-8. (pypa/pip#8717)
|
||||
with open(pyvenv_cfg_file, encoding="utf-8") as f:
|
||||
return f.read().splitlines() # avoids trailing newlines
|
||||
except OSError:
|
||||
return None
|
||||
|
||||
|
||||
def _no_global_under_venv() -> bool:
|
||||
"""Check `{sys.prefix}/pyvenv.cfg` for system site-packages inclusion
|
||||
|
||||
PEP 405 specifies that when system site-packages are not supposed to be
|
||||
visible from a virtual environment, `pyvenv.cfg` must contain the following
|
||||
line:
|
||||
|
||||
include-system-site-packages = false
|
||||
|
||||
Additionally, log a warning if accessing the file fails.
|
||||
"""
|
||||
cfg_lines = _get_pyvenv_cfg_lines()
|
||||
if cfg_lines is None:
|
||||
# We're not in a "sane" venv, so assume there is no system
|
||||
# site-packages access (since that's PEP 405's default state).
|
||||
logger.warning(
|
||||
"Could not access 'pyvenv.cfg' despite a virtual environment "
|
||||
"being active. Assuming global site-packages is not accessible "
|
||||
"in this environment."
|
||||
)
|
||||
return True
|
||||
|
||||
for line in cfg_lines:
|
||||
match = _INCLUDE_SYSTEM_SITE_PACKAGES_REGEX.match(line)
|
||||
if match is not None and match.group("value") == "false":
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
def _no_global_under_legacy_virtualenv() -> bool:
|
||||
"""Check if "no-global-site-packages.txt" exists beside site.py
|
||||
|
||||
This mirrors logic in pypa/virtualenv for determining whether system
|
||||
site-packages are visible in the virtual environment.
|
||||
"""
|
||||
site_mod_dir = os.path.dirname(os.path.abspath(site.__file__))
|
||||
no_global_site_packages_file = os.path.join(
|
||||
site_mod_dir,
|
||||
"no-global-site-packages.txt",
|
||||
)
|
||||
return os.path.exists(no_global_site_packages_file)
|
||||
|
||||
|
||||
def virtualenv_no_global() -> bool:
|
||||
"""Returns a boolean, whether running in venv with no system site-packages."""
|
||||
# PEP 405 compliance needs to be checked first since virtualenv >=20 would
|
||||
# return True for both checks, but is only able to use the PEP 405 config.
|
||||
if _running_under_venv():
|
||||
return _no_global_under_venv()
|
||||
|
||||
if _running_under_legacy_virtualenv():
|
||||
return _no_global_under_legacy_virtualenv()
|
||||
|
||||
return False
|
||||
Reference in New Issue
Block a user