1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
|
# DP: Avoid shutil import when it is not available.
Index: b/Lib/tempfile.py
===================================================================
--- a/Lib/tempfile.py
+++ b/Lib/tempfile.py
@@ -40,7 +40,146 @@ import functools as _functools
import warnings as _warnings
import io as _io
import os as _os
-import shutil as _shutil
+try:
+ import shutil as _shutil
+ _rmtree = _shutil.rmtree
+except ImportError:
+ import sys as _sys
+ import stat as _stat
+ # version vulnerable to race conditions
+ def _rmtree_unsafe(path, onerror):
+ try:
+ if _os.path.islink(path):
+ # symlinks to directories are forbidden, see bug #1669
+ raise OSError("Cannot call rmtree on a symbolic link")
+ except OSError:
+ onerror(_os.path.islink, path, _sys.exc_info())
+ # can't continue even if onerror hook returns
+ return
+ names = []
+ try:
+ names = _os.listdir(path)
+ except OSError:
+ onerror(_os.listdir, path, _sys.exc_info())
+ for name in names:
+ fullname = _os.path.join(path, name)
+ try:
+ mode = _os.lstat(fullname).st_mode
+ except OSError:
+ mode = 0
+ if _stat.S_ISDIR(mode):
+ _rmtree_unsafe(fullname, onerror)
+ else:
+ try:
+ _os.unlink(fullname)
+ except OSError:
+ onerror(_os.unlink, fullname, _sys.exc_info())
+ try:
+ _os.rmdir(path)
+ except OSError:
+ onerror(_os.rmdir, path, _sys.exc_info())
+
+ # Version using fd-based APIs to protect against races
+ def _rmtree_safe_fd(topfd, path, onerror):
+ names = []
+ try:
+ names = _os.listdir(topfd)
+ except OSError as err:
+ err.filename = path
+ onerror(_os.listdir, path, _sys.exc_info())
+ for name in names:
+ fullname = _os.path.join(path, name)
+ try:
+ orig_st = _os.stat(name, dir_fd=topfd, follow_symlinks=False)
+ mode = orig_st.st_mode
+ except OSError:
+ mode = 0
+ if _stat.S_ISDIR(mode):
+ try:
+ dirfd = _os.open(name, _os.O_RDONLY, dir_fd=topfd)
+ except OSError:
+ onerror(_os.open, fullname, _sys.exc_info())
+ else:
+ try:
+ if _os.path.samestat(orig_st, _os.fstat(dirfd)):
+ _rmtree_safe_fd(dirfd, fullname, onerror)
+ try:
+ _os.rmdir(name, dir_fd=topfd)
+ except OSError:
+ onerror(_os.rmdir, fullname, _sys.exc_info())
+ else:
+ try:
+ # This can only happen if someone replaces
+ # a directory with a symlink after the call to
+ # stat.S_ISDIR above.
+ raise OSError("Cannot call rmtree on a symbolic "
+ "link")
+ except OSError:
+ onerror(_os.path.islink, fullname, _sys.exc_info())
+ finally:
+ _os.close(dirfd)
+ else:
+ try:
+ _os.unlink(name, dir_fd=topfd)
+ except OSError:
+ onerror(_os.unlink, fullname, _sys.exc_info())
+
+ _use_fd_functions = ({_os.open, _os.stat, _os.unlink, _os.rmdir} <=
+ _os.supports_dir_fd and
+ _os.listdir in _os.supports_fd and
+ _os.stat in _os.supports_follow_symlinks)
+
+ def _rmtree(path, ignore_errors=False, onerror=None):
+ """Recursively delete a directory tree.
+
+ If ignore_errors is set, errors are ignored; otherwise, if onerror
+ is set, it is called to handle the error with arguments (func,
+ path, exc_info) where func is platform and implementation dependent;
+ path is the argument to that function that caused it to fail; and
+ exc_info is a tuple returned by sys.exc_info(). If ignore_errors
+ is false and onerror is None, an exception is raised.
+
+ """
+ if ignore_errors:
+ def onerror(*args):
+ pass
+ elif onerror is None:
+ def onerror(*args):
+ raise
+ if _use_fd_functions:
+ # While the unsafe rmtree works fine on bytes, the fd based does not.
+ if isinstance(path, bytes):
+ path = _os.fsdecode(path)
+ # Note: To guard against symlink races, we use the standard
+ # lstat()/open()/fstat() trick.
+ try:
+ orig_st = _os.lstat(path)
+ except Exception:
+ onerror(_os.lstat, path, _sys.exc_info())
+ return
+ try:
+ fd = _os.open(path, _os.O_RDONLY)
+ except Exception:
+ onerror(_os.lstat, path, _sys.exc_info())
+ return
+ try:
+ if _os.path.samestat(orig_st, _os.fstat(fd)):
+ _rmtree_safe_fd(fd, path, onerror)
+ try:
+ _os.rmdir(path)
+ except OSError:
+ onerror(_os.rmdir, path, _sys.exc_info())
+ else:
+ try:
+ # symlinks to directories are forbidden, see bug #1669
+ raise OSError("Cannot call rmtree on a symbolic link")
+ except OSError:
+ onerror(_os.path.islink, path, _sys.exc_info())
+ finally:
+ _os.close(fd)
+ else:
+ return _rmtree_unsafe(path, onerror)
+
import errno as _errno
from random import Random as _Random
import weakref as _weakref
@@ -794,7 +933,7 @@ class TemporaryDirectory(object):
@classmethod
def _cleanup(cls, name, warn_message):
- _shutil.rmtree(name)
+ _rmtree(name)
_warnings.warn(warn_message, ResourceWarning)
def __repr__(self):
@@ -808,4 +947,4 @@ class TemporaryDirectory(object):
def cleanup(self):
if self._finalizer.detach():
- _shutil.rmtree(self.name)
+ _rmtree(self.name)
|
