filesystem.py 6.7 KB

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