filesystem.py 5.9 KB

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