Mercurial > repos > guerler > hhblits

comparison lib/python3.8/site-packages/pip/_internal/utils/filesystem.py @ 1:64071f2a4cf0 draft default tip

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
Deleted selected files
author guerler
date Mon, 27 Jul 2020 03:55:49 -0400
parents 9e54283cc701
children
comparison
equal deleted inserted replaced
0:9e54283cc701 1:64071f2a4cf0
1 import errno
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
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
16 from pip._internal.utils.compat import get_path_uid
17 from pip._internal.utils.typing import MYPY_CHECK_RUNNING, cast
18
19 if MYPY_CHECK_RUNNING:
20 from typing import BinaryIO, Iterator
21
22 class NamedTemporaryFileResult(BinaryIO):
23 @property
24 def file(self):
25 # type: () -> BinaryIO
26 pass
27
28
29 def check_path_owner(path):
30 # type: (str) -> bool
31 # If we don't have a way to check the effective uid of this process, then
32 # we'll just assume that we own the directory.
33 if sys.platform == "win32" or not hasattr(os, "geteuid"):
34 return True
35
36 assert os.path.isabs(path)
37
38 previous = None
39 while path != previous:
40 if os.path.lexists(path):
41 # Check if path is writable by current user.
42 if os.geteuid() == 0:
43 # Special handling for root user in order to handle properly
44 # cases where users use sudo without -H flag.
45 try:
46 path_uid = get_path_uid(path)
47 except OSError:
48 return False
49 return path_uid == 0
50 else:
51 return os.access(path, os.W_OK)
52 else:
53 previous, path = path, os.path.dirname(path)
54 return False # assume we don't own the path
55
56
57 def copy2_fixed(src, dest):
58 # type: (str, str) -> None
59 """Wrap shutil.copy2() but map errors copying socket files to
60 SpecialFileError as expected.
61
62 See also https://bugs.python.org/issue37700.
63 """
64 try:
65 shutil.copy2(src, dest)
66 except (OSError, IOError):
67 for f in [src, dest]:
68 try:
69 is_socket_file = is_socket(f)
70 except OSError:
71 # An error has already occurred. Another error here is not
72 # a problem and we can ignore it.
73 pass
74 else:
75 if is_socket_file:
76 raise shutil.SpecialFileError("`%s` is a socket" % f)
77
78 raise
79
80
81 def is_socket(path):
82 # type: (str) -> bool
83 return stat.S_ISSOCK(os.lstat(path).st_mode)
84
85
86 @contextmanager
87 def adjacent_tmp_file(path):
88 # type: (str) -> Iterator[NamedTemporaryFileResult]
89 """Given a path to a file, open a temp file next to it securely and ensure
90 it is written to disk after the context reaches its end.
91 """
92 with NamedTemporaryFile(
93 delete=False,
94 dir=os.path.dirname(path),
95 prefix=os.path.basename(path),
96 suffix='.tmp',
97 ) as f:
98 result = cast('NamedTemporaryFileResult', f)
99 try:
100 yield result
101 finally:
102 result.file.flush()
103 os.fsync(result.file.fileno())
104
105
106 _replace_retry = retry(stop_max_delay=1000, wait_fixed=250)
107
108 if PY2:
109 @_replace_retry
110 def replace(src, dest):
111 # type: (str, str) -> None
112 try:
113 os.rename(src, dest)
114 except OSError:
115 os.remove(dest)
116 os.rename(src, dest)
117
118 else:
119 replace = _replace_retry(os.replace)
120
121
122 # test_writable_dir and _test_writable_dir_win are copied from Flit,
123 # with the author's agreement to also place them under pip's license.
124 def test_writable_dir(path):
125 # type: (str) -> bool
126 """Check if a directory is writable.
127
128 Uses os.access() on POSIX, tries creating files on Windows.
129 """
130 # If the directory doesn't exist, find the closest parent that does.
131 while not os.path.isdir(path):
132 parent = os.path.dirname(path)
133 if parent == path:
134 break # Should never get here, but infinite loops are bad
135 path = parent
136
137 if os.name == 'posix':
138 return os.access(path, os.W_OK)
139
140 return _test_writable_dir_win(path)
141
142
143 def _test_writable_dir_win(path):
144 # type: (str) -> bool
145 # os.access doesn't work on Windows: http://bugs.python.org/issue2528
146 # and we can't use tempfile: http://bugs.python.org/issue22107
147 basename = 'accesstest_deleteme_fishfingers_custard_'
148 alphabet = 'abcdefghijklmnopqrstuvwxyz0123456789'
149 for i in range(10):
150 name = basename + ''.join(random.choice(alphabet) for _ in range(6))
151 file = os.path.join(path, name)
152 try:
153 fd = os.open(file, os.O_RDWR | os.O_CREAT | os.O_EXCL)
154 except OSError as e:
155 if e.errno == errno.EEXIST:
156 continue
157 if e.errno == errno.EPERM:
158 # This could be because there's a directory with the same name.
159 # But it's highly unlikely there's a directory called that,
160 # so we'll assume it's because the parent dir is not writable.
161 return False
162 raise
163 else:
164 os.close(fd)
165 os.unlink(file)
166 return True
167
168 # This should never be reached
169 raise EnvironmentError(
170 'Unexpected condition testing for writable directory'
171 )