[3.12] gh-157579: Fix race condition in the cleanup of tempfile.TemporaryDirectory (GH-157580) (#158489)

* gh-157579: Fix race condition in the cleanup of `tempfile.TemporaryDirectory` (GH-157580)
(cherry picked from commit 5c20517)

* Add root user checks to test.support

This partially backports commit 86b8617
(GH-146195); making existing tests use the helper is omitted.

* gh-134993: Add os.lstat() to os.supports_dir_fd (#135188)

Co-authored-by: Stan Ulbrych <stan@python.org>
Co-authored-by: Victor Stinner <vstinner@python.org>
Co-authored-by: Petr Viktorin <encukou@gmail.com>

Author: 3 people

Doc/library/tempfile.rst

@@ -207,6 +207,15 @@ The module defines the following user-callable items:
    debugging or when you need your cleanup behavior to be conditional based on
    other logic.
 
+   .. warning::
+
+      Cleanup is not robust against the tree being modified while it is removed.
+      Files outside of the tree may have their permissions and file flags reset.
+
+      On systems where :data:`shutil.rmtree.avoids_symlink_attacks` is
+      false, manipulating symbolic links during cleanup
+      may cause files outside of the tree to be removed.
+
    .. audit-event:: tempfile.mkdtemp fullpath tempfile.TemporaryDirectory
 
    .. versionadded:: 3.2

Lib/os.py

@@ -110,6 +110,7 @@ def _add(str, fn):
     _add("HAVE_FCHMODAT",   "chmod")
     _add("HAVE_FCHOWNAT",   "chown")
     _add("HAVE_FSTATAT",    "stat")
+    _add("HAVE_LSTAT",      "lstat")
     _add("HAVE_FUTIMESAT",  "utime")
     _add("HAVE_LINKAT",     "link")
     _add("HAVE_MKDIRAT",    "mkdir")

Lib/shutil.py

@@ -654,6 +654,7 @@ def _rmtree_safe_fd(stack, onexc):
     #   save a call to os.lstat() when walking subdirectories.
     func, dirfd, path, orig_entry = stack.pop()
     name = path if orig_entry is None else orig_entry.name
+    parent_fd = None if func is os.close else dirfd
     try:
         if func is os.close:
             os.close(dirfd)
@@ -697,17 +698,18 @@ def _rmtree_safe_fd(stack, onexc):
             try:
                 os.unlink(entry.name, dir_fd=topfd)
             except OSError as err:
-                onexc(os.unlink, fullname, err)
+                onexc(os.unlink, fullname, err, direntry=entry, dir_fd=topfd)
     except OSError as err:
         err.filename = path
-        onexc(func, path, err)
+        onexc(func, path, err, direntry=orig_entry, dir_fd=parent_fd)
 
 _use_fd_functions = ({os.open, os.stat, os.unlink, os.rmdir} <=
                      os.supports_dir_fd and
                      os.scandir in os.supports_fd and
                      os.stat in os.supports_follow_symlinks)
 
-def rmtree(path, ignore_errors=False, onerror=None, *, onexc=None, dir_fd=None):
+def rmtree(path, ignore_errors=False, onerror=None, *, onexc=None, dir_fd=None,
+           _onexc_kwargs=False):
     """Recursively delete a directory tree.
 
     If dir_fd is not None, it should be a file descriptor open to a directory;
@@ -730,24 +732,29 @@ def rmtree(path, ignore_errors=False, onerror=None, *, onexc=None, dir_fd=None):
 
     sys.audit("shutil.rmtree", path, dir_fd)
     if ignore_errors:
-        def onexc(*args):
+        def onexc(*args, **kwargs):
             pass
     elif onerror is None and onexc is None:
-        def onexc(*args):
+        def onexc(*args, **kwargs):
             raise
     elif onexc is None:
         if onerror is None:
-            def onexc(*args):
+            def onexc(*args, **kwargs):
                 raise
         else:
             # delegate to onerror
-            def onexc(*args):
+            def onexc(*args, **kwargs):
                 func, path, exc = args
                 if exc is None:
                     exc_info = None, None, None
                 else:
                     exc_info = type(exc), exc, exc.__traceback__
                 return onerror(func, path, exc_info)
+    elif not _onexc_kwargs:
+        # Only the internal caller in tempfile asks for the extra arguments.
+        _onexc = onexc
+        def onexc(func, path, err, **kwargs):
+            return _onexc(func, path, err)
 
     if _use_fd_functions:
         # While the unsafe rmtree works fine on bytes, the fd based does not.

Lib/tempfile.py

@@ -43,6 +43,7 @@
 import shutil as _shutil
 import errno as _errno
 from random import Random as _Random
+import stat as _stat
 import sys as _sys
 import types as _types
 import weakref as _weakref
@@ -276,15 +277,68 @@ def _dont_follow_symlinks(func, path, *args):
     elif _os.name == 'nt' or not _os.path.islink(path):
         func(path, *args)
 
-def _resetperms(path):
+def _resetflags(path):
     try:
         chflags = _os.chflags
     except AttributeError:
         pass
     else:
         _dont_follow_symlinks(chflags, path, 0)
+
+def _resetperms(path):
+    _resetflags(path)
     _dont_follow_symlinks(_os.chmod, path, 0o700)
 
+# True if TemporaryDirectory._rmtree() can work relative to open directories
+# instead of resolving paths again.
+_rmtree_use_dir_fd = (
+    {_os.chmod, _os.unlink, _os.lstat} <= _os.supports_dir_fd
+    and _os.chmod in _os.supports_fd
+)
+
+def _resetperms_fd(dir_fd, path):
+    # Same as _resetperms(), but for the directory referred to by dir_fd.
+    if dir_fd is None:
+        _resetperms(path)
+        return
+    _resetflags(path)
+    _os.chmod(dir_fd, 0o700)
+
+try:
+    _nofollow_mode = _os.O_RDONLY | _os.O_NONBLOCK | _os.O_NOFOLLOW
+except AttributeError:
+    _nofollow_mode = None
+
+def _resetperms_at(name, dir_fd, path):
+    # Same as _resetperms(), but name is resolved relative to the directory
+    # file descriptor dir_fd. path is only used for os.chflags(), which
+    # doesn't support dir_fd or file descriptors.
+    if dir_fd is None:
+        _resetperms(path)
+        return
+    _resetflags(path)
+    if _os.chmod in _os.supports_follow_symlinks:
+        _os.chmod(name, 0o700, dir_fd=dir_fd, follow_symlinks=False)
+    else:
+        # dir_fd & follow_symlinks is not supported on this platform.
+        # Try chmod opening the file with O_NOFOLLOW.
+        if _nofollow_mode is not None:
+            try:
+                fd = _os.open(name, _nofollow_mode, dir_fd=dir_fd)
+            except OSError:
+                pass
+            else:
+                try:
+                    _os.chmod(fd, 0o700)
+                finally:
+                    _os.close(fd)
+                return
+        # If that did not work, we change by name, which is subject to a race
+        # condition.
+        stat = _os.lstat(name, dir_fd=dir_fd)
+        if not _stat.S_ISLNK(stat.st_mode):
+            _os.chmod(name, 0o700, dir_fd=dir_fd)
+
 
 # User visible interfaces.
 
@@ -892,23 +946,42 @@ def __init__(self, suffix=None, prefix=None, dir=None,
             ignore_errors=self._ignore_cleanup_errors, delete=self._delete)
 
     @classmethod
-    def _rmtree(cls, name, ignore_errors=False, repeated=False):
-        def onexc(func, path, exc):
+    def _rmtree(cls, name, ignore_errors=False, repeated=False, dir_fd=None,
+                fullname=None):
+        if fullname is None:
+            fullname = name
+
+        def onexc(func, path, exc, direntry=None, dir_fd=None):
             if isinstance(exc, PermissionError):
                 if repeated and path == name:
                     if ignore_errors:
                         return
                     raise
 
+                # fullpath is path as seen from the working directory
+                fullpath = fullname + path[len(name):]
+                # base is path relative to dir_fd, the directory rmtree()
+                # reached it through, or the whole path when there is none
+                if dir_fd is None or not _rmtree_use_dir_fd:
+                    base, dir_fd = path, None
+                elif direntry is None:
+                    base = path
+                else:
+                    base = direntry.name
+
                 try:
                     if path != name:
-                        _resetperms(_os.path.dirname(path))
-                    _resetperms(path)
+                        # The parent directory of path is the one referred to
+                        # by dir_fd.
+                        _resetperms_fd(dir_fd, _os.path.dirname(fullpath))
+                    _resetperms_at(base, dir_fd, fullpath)
 
                     try:
-                        _os.unlink(path)
+                        _os.unlink(base, dir_fd=dir_fd)
                     except IsADirectoryError:
-                        cls._rmtree(path, ignore_errors=ignore_errors)
+                        cls._rmtree(base, ignore_errors=ignore_errors,
+                                    repeated=(path == name),
+                                    dir_fd=dir_fd, fullname=fullpath)
                     except PermissionError:
                         # The PermissionError handler was originally added for
                         # FreeBSD in directories, but it seems that it is raised
@@ -917,21 +990,27 @@ def onexc(func, path, exc):
                         # raise NotADirectoryError and mask the PermissionError.
                         # So we must re-raise the current PermissionError if
                         # path is not a directory.
-                        if not _os.path.isdir(path) or _os.path.isjunction(path):
+                        if (not _os.path.isdir(fullpath)
+                                or _os.path.isjunction(fullpath)):
                             if ignore_errors:
                                 return
                             raise
-                        cls._rmtree(path, ignore_errors=ignore_errors,
-                                    repeated=(path == name))
+                        cls._rmtree(base, ignore_errors=ignore_errors,
+                                    repeated=(path == name),
+                                    dir_fd=dir_fd, fullname=fullpath)
                 except FileNotFoundError:
                     pass
+                except OSError:
+                    if ignore_errors:
+                        return
+                    raise
             elif isinstance(exc, FileNotFoundError):
                 pass
             else:
                 if not ignore_errors:
                     raise
 
-        _shutil.rmtree(name, onexc=onexc)
+        _shutil.rmtree(name, onexc=onexc, dir_fd=dir_fd, _onexc_kwargs=True)
 
     @classmethod
     def _cleanup(cls, name, warn_message, ignore_errors=False, delete=True):

Lib/test/support/__init__.py

@@ -2598,3 +2598,8 @@ def control_characters_c0() -> list[str]:
     C0 control characters defined as the byte range 0x00-0x1F, and 0x7F.
     """
     return [chr(c) for c in range(0x00, 0x20)] + ["\x7F"]
+
+
+_ROOT_IN_POSIX = hasattr(os, 'geteuid') and os.geteuid() == 0
+requires_root_user = unittest.skipUnless(_ROOT_IN_POSIX, "test needs root privilege")
+requires_non_root_user = unittest.skipIf(_ROOT_IN_POSIX, "test needs non-root account")

Lib/test/test_shutil.py

@@ -497,6 +497,37 @@ def check_args_to_onexc(self, func, arg, exc):
             self.assertTrue(isinstance(exc, OSError))
             self.errorState = 3
 
+    @os_helper.skip_if_dac_override
+    @os_helper.skip_unless_working_chmod
+    @unittest.skipUnless(shutil.rmtree.avoids_symlink_attacks,
+                         'requires the fd based implementation of rmtree()')
+    def test_on_exc_kwargs(self):
+        os.mkdir(TESTFN)
+        self.addCleanup(shutil.rmtree, TESTFN)
+
+        child_dir_path = os.path.join(TESTFN, 'b')
+        child_file_path = os.path.join(child_dir_path, 'a')
+        os.mkdir(child_dir_path)
+        os_helper.create_empty_file(child_file_path)
+        old_child_dir_mode = os.stat(child_dir_path).st_mode
+        # Make unwritable.
+        new_mode = stat.S_IREAD|stat.S_IEXEC
+        os.chmod(child_dir_path, new_mode)
+
+        self.addCleanup(os.chmod, child_dir_path, old_child_dir_mode)
+
+        calls = []
+        def onexc(func, path, err, direntry=None, dir_fd=None):
+            calls.append((func, path, err))
+            if func is os.unlink:
+                self.assertEqual(direntry.name, os.path.basename(path))
+                self.assertTrue(os.path.samestat(
+                    os.stat(path), os.stat(direntry.name, dir_fd=dir_fd)))
+
+        shutil.rmtree(TESTFN, onexc=onexc, _onexc_kwargs=True)
+        self.assertIn((os.unlink, child_file_path),
+                      [(func, path) for func, path, err in calls])
+
     @unittest.skipIf(sys.platform[:6] == 'cygwin',
                      "This test can't be run on Cygwin (issue #1071513).")
     @os_helper.skip_if_dac_override