It caused rmtree to not ignore certain errors when ignore_errors was set.
Patch by Alessandro Moura and Serhiy Storchaka.
except os.error:
onerror(os.rmdir, path, sys.exc_info())
- except os.error:
+# Version using fd-based APIs to protect against races
+def _rmtree_safe_fd(topfd, path, onerror):
+ names = []
+ try:
+ names = os.listdir(topfd)
- except os.error:
++ 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 os.error:
++ except OSError:
+ mode = 0
+ if stat.S_ISDIR(mode):
+ try:
+ dirfd = os.open(name, os.O_RDONLY, dir_fd=topfd)
- except os.error:
++ 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 os.error:
++ 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)
- if (stat.S_ISDIR(orig_st.st_mode) and
- os.path.samestat(orig_st, os.fstat(fd))):
++ 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:
- raise NotADirectoryError(20,
- "Not a directory: '{}'".format(path))
++ if os.path.samestat(orig_st, os.fstat(fd)):
+ _rmtree_safe_fd(fd, path, onerror)
+ try:
+ os.rmdir(path)
+ except os.error:
+ 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)
+
+# Allow introspection of whether or not the hardening against symlink
+# attacks is supported on the current platform
+rmtree.avoids_symlink_attacks = _use_fd_functions
def _basename(path):
# A basename() variant which first strips the trailing slash, if present.
os.symlink(dir_, link)
self.assertRaises(OSError, shutil.rmtree, link)
self.assertTrue(os.path.exists(dir_))
+ self.assertTrue(os.path.lexists(link))
+ errors = []
+ def onerror(*args):
+ errors.append(args)
+ shutil.rmtree(link, onerror=onerror)
+ self.assertEqual(len(errors), 1)
+ self.assertIs(errors[0][0], os.path.islink)
+ self.assertEqual(errors[0][1], link)
+ self.assertIsInstance(errors[0][2][1], OSError)
+ @support.skip_unless_symlink
+ def test_rmtree_works_on_symlinks(self):
+ tmp = self.mkdtemp()
+ dir1 = os.path.join(tmp, 'dir1')
+ dir2 = os.path.join(dir1, 'dir2')
+ dir3 = os.path.join(tmp, 'dir3')
+ for d in dir1, dir2, dir3:
+ os.mkdir(d)
+ file1 = os.path.join(tmp, 'file1')
+ write_file(file1, 'foo')
+ link1 = os.path.join(dir1, 'link1')
+ os.symlink(dir2, link1)
+ link2 = os.path.join(dir1, 'link2')
+ os.symlink(dir3, link2)
+ link3 = os.path.join(dir1, 'link3')
+ os.symlink(file1, link3)
+ # make sure symlinks are removed but not followed
+ shutil.rmtree(dir1)
+ self.assertFalse(os.path.exists(dir1))
+ self.assertTrue(os.path.exists(dir3))
+ self.assertTrue(os.path.exists(file1))
+
def test_rmtree_errors(self):
# filename is guaranteed not to exist
filename = tempfile.mktemp()
-- self.assertRaises(OSError, shutil.rmtree, filename)
- # test that ignore_errors option is honoured
++ self.assertRaises(FileNotFoundError, shutil.rmtree, filename)
++ # test that ignore_errors option is honored
+ shutil.rmtree(filename, ignore_errors=True)
+
+ # existing file
+ tmpdir = self.mkdtemp()
- self.write_file((tmpdir, "tstfile"), "")
++ write_file((tmpdir, "tstfile"), "")
+ filename = os.path.join(tmpdir, "tstfile")
- with self.assertRaises(OSError) as cm:
++ with self.assertRaises(NotADirectoryError) as cm:
+ shutil.rmtree(filename)
+ self.assertEqual(cm.exception.filename, filename)
+ self.assertTrue(os.path.exists(filename))
- # test that ignore_errors option is honoured
++ # test that ignore_errors option is honored
+ shutil.rmtree(filename, ignore_errors=True)
+ self.assertTrue(os.path.exists(filename))
+ errors = []
+ def onerror(*args):
+ errors.append(args)
+ shutil.rmtree(filename, onerror=onerror)
+ self.assertEqual(len(errors), 2)
+ self.assertIs(errors[0][0], os.listdir)
+ self.assertEqual(errors[0][1], filename)
- self.assertIsInstance(errors[0][2][1], OSError)
++ self.assertIsInstance(errors[0][2][1], NotADirectoryError)
+ self.assertEqual(errors[0][2][1].filename, filename)
+ self.assertIs(errors[1][0], os.rmdir)
+ self.assertEqual(errors[1][1], filename)
- self.assertIsInstance(errors[1][2][1], OSError)
++ self.assertIsInstance(errors[1][2][1], NotADirectoryError)
+ self.assertEqual(errors[1][2][1].filename, filename)
# See bug #1071513 for why we don't run this on cygwin
# and bug #1076467 for why we don't run this as root.
Library
-------
++- Issue #15872: Fix 3.3 regression introduced by the new fd-based shutil.rmtree
++ that caused it to not ignore certain errors when ignore_errors was set.
++ Patch by Alessandro Moura and Serhiy Storchaka.
++
- Issue #16248: Disable code execution from the user's home directory by
tkinter when the -E flag is passed to Python. Patch by Zachary Ware.
projects / python / commitdiff