venv/lib/python3.11/site-packages/pip/_internal/utils/filesystem.py

   1 import fnmatch
   2 import os
   3 import os.path
   4 import random
   5 import sys
   6 from contextlib import contextmanager
   7 from tempfile import NamedTemporaryFile
   8 from typing import Any, BinaryIO, Generator, List, Union, cast
   9
  10 from pip._vendor.tenacity import retry, stop_after_delay, wait_fixed
  11
  12 from pip._internal.utils.compat import get_path_uid
  13 from pip._internal.utils.misc import format_size
  14
  15
  16 def check_path_owner(path: str) -> bool:
  17     # If we don't have a way to check the effective uid of this process, then
  18     # we'll just assume that we own the directory.
  19     if sys.platform == "win32" or not hasattr(os, "geteuid"):
  20         return True
  21
  22     assert os.path.isabs(path)
  23
  24     previous = None
  25     while path != previous:
  26         if os.path.lexists(path):
  27             # Check if path is writable by current user.
  28             if os.geteuid() == 0:
  29                 # Special handling for root user in order to handle properly
  30                 # cases where users use sudo without -H flag.
  31                 try:
  32                     path_uid = get_path_uid(path)
  33                 except OSError:
  34                     return False
  35                 return path_uid == 0
  36             else:
  37                 return os.access(path, os.W_OK)
  38         else:
  39             previous, path = path, os.path.dirname(path)
  40     return False  # assume we don't own the path
  41
  42
  43 @contextmanager
  44 def adjacent_tmp_file(path: str, **kwargs: Any) -> Generator[BinaryIO, None, None]:
  45     """Return a file-like object pointing to a tmp file next to path.
  46
  47     The file is created securely and is ensured to be written to disk
  48     after the context reaches its end.
  49
  50     kwargs will be passed to tempfile.NamedTemporaryFile to control
  51     the way the temporary file will be opened.
  52     """
  53     with NamedTemporaryFile(
  54         delete=False,
  55         dir=os.path.dirname(path),
  56         prefix=os.path.basename(path),
  57         suffix=".tmp",
  58         **kwargs,
  59     ) as f:
  60         result = cast(BinaryIO, f)
  61         try:
  62             yield result
  63         finally:
  64             result.flush()
  65             os.fsync(result.fileno())
  66
  67
  68 # Tenacity raises RetryError by default, explicitly raise the original exception
  69 _replace_retry = retry(reraise=True, stop=stop_after_delay(1), wait=wait_fixed(0.25))
  70
  71 replace = _replace_retry(os.replace)
  72
  73
  74 # test_writable_dir and _test_writable_dir_win are copied from Flit,
  75 # with the author's agreement to also place them under pip's license.
  76 def test_writable_dir(path: str) -> bool:
  77     """Check if a directory is writable.
  78
  79     Uses os.access() on POSIX, tries creating files on Windows.
  80     """
  81     # If the directory doesn't exist, find the closest parent that does.
  82     while not os.path.isdir(path):
  83         parent = os.path.dirname(path)
  84         if parent == path:
  85             break  # Should never get here, but infinite loops are bad
  86         path = parent
  87
  88     if os.name == "posix":
  89         return os.access(path, os.W_OK)
  90
  91     return _test_writable_dir_win(path)
  92
  93
  94 def _test_writable_dir_win(path: str) -> bool:
  95     # os.access doesn't work on Windows: http://bugs.python.org/issue2528
  96     # and we can't use tempfile: http://bugs.python.org/issue22107
  97     basename = "accesstest_deleteme_fishfingers_custard_"
  98     alphabet = "abcdefghijklmnopqrstuvwxyz0123456789"
  99     for _ in range(10):
 100         name = basename + "".join(random.choice(alphabet) for _ in range(6))
 101         file = os.path.join(path, name)
 102         try:
 103             fd = os.open(file, os.O_RDWR | os.O_CREAT | os.O_EXCL)
 104         except FileExistsError:
 105             pass
 106         except PermissionError:
 107             # This could be because there's a directory with the same name.
 108             # But it's highly unlikely there's a directory called that,
 109             # so we'll assume it's because the parent dir is not writable.
 110             # This could as well be because the parent dir is not readable,
 111             # due to non-privileged user access.
 112             return False
 113         else:
 114             os.close(fd)
 115             os.unlink(file)
 116             return True
 117
 118     # This should never be reached
 119     raise OSError("Unexpected condition testing for writable directory")
 120
 121
 122 def find_files(path: str, pattern: str) -> List[str]:
 123     """Returns a list of absolute paths of files beneath path, recursively,
 124     with filenames which match the UNIX-style shell glob pattern."""
 125     result: List[str] = []
 126     for root, _, files in os.walk(path):
 127         matches = fnmatch.filter(files, pattern)
 128         result.extend(os.path.join(root, f) for f in matches)
 129     return result
 130
 131
 132 def file_size(path: str) -> Union[int, float]:
 133     # If it's a symlink, return 0.
 134     if os.path.islink(path):
 135         return 0
 136     return os.path.getsize(path)
 137
 138
 139 def format_file_size(path: str) -> str:
 140     return format_size(file_size(path))
 141
 142
 143 def directory_size(path: str) -> Union[int, float]:
 144     size = 0.0
 145     for root, _dirs, files in os.walk(path):
 146         for filename in files:
 147             file_path = os.path.join(root, filename)
 148             size += file_size(file_path)
 149     return size
 150
 151
 152 def format_directory_size(path: str) -> str:
 153     return format_size(directory_size(path))