gh-157579: Fix race condition in the cleanup of tempfile.TemporaryDirectory (GH-157580)

Co-authored-by: Petr Viktorin <encukou@gmail.com>

Author: StanFromIreland

Doc/library/tempfile.rst

@@ -234,6 +234,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/shutil.py

@@ -761,6 +761,7 @@ def _rmtree_safe_fd_step(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)
@@ -808,22 +809,23 @@ def _rmtree_safe_fd_step(stack, onexc):
             except FileNotFoundError:
                 continue
             except OSError as err:
-                onexc(os.unlink, fullname, err)
+                onexc(os.unlink, fullname, err, direntry=entry, dir_fd=topfd)
     except FileNotFoundError as err:
         if orig_entry is None or func is os.close:
             err.filename = path
-            onexc(func, path, err)
+            onexc(func, path, err, direntry=orig_entry, dir_fd=parent_fd)
     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)
 _rmtree_impl = _rmtree_safe_fd if _use_fd_functions else _rmtree_unsafe
 
-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;
@@ -846,24 +848,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)
 
     _rmtree_impl(path, dir_fd, onexc)
 

Lib/tempfile.py

@@ -44,6 +44,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
@@ -274,15 +275,68 @@ def _dont_follow_symlinks(func, path, *args):
     elif 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.
 
@@ -927,24 +981,43 @@ 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):
             # On DragonFly BSD, UF_NOUNLINK removal fails with EISDIR, not EPERM.
             if isinstance(exc, (PermissionError, IsADirectoryError)):
                 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
@@ -953,21 +1026,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/test_shutil.py

@@ -495,6 +495,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

Lib/test/test_tempfile.py

@@ -14,6 +14,7 @@
 import gc
 import shutil
 import subprocess
+import sysconfig
 from unittest import mock
 
 import unittest
@@ -1861,6 +1862,54 @@ def test(target, target_is_directory):
                 new_flags = os.stat(dir1).st_flags
                 self.assertEqual(new_flags, old_flags)
 
+    @os_helper.skip_unless_symlink
+    @os_helper.skip_unless_working_chmod
+    @support.requires_non_root_user
+    @unittest.skipIf(support.is_emscripten, 'Fails due to Emscripten bug:'
+                                            'emscripten-core/emscripten#27761')
+    @unittest.skipUnless(shutil.rmtree.avoids_symlink_attacks,
+                         'requires the fd based implementation of rmtree()')
+    def test_cleanup_with_symlink_race(self):
+        # cleanup() should not operate on files outside of the temporary
+        # directory when a directory is replaced with a symlink while it
+        # recovers from a PermissionError (CVE-2026-12345).
+        with self.do_create(recurse=0) as target:
+            target_file = os.path.join(target, 'file1')
+            open(target_file, 'wb').close()
+            target_mode = os.stat(target_file).st_mode
+
+            d1 = self.do_create(recurse=0)
+            dir1 = os.path.join(d1.name, 'dir1')
+            os.mkdir(dir1)
+            open(os.path.join(dir1, 'file1'), 'wb').close()
+            # Removing contents of dir1 fails with a PermissionError, and
+            # dir1 is replaced with a symlink to target at the very moment
+            # cleanup() starts to recover from that error.
+            os.chmod(dir1, 0o500)
+            unlink = os.unlink
+            def hook(path, *, dir_fd=None):
+                try:
+                    return unlink(path, dir_fd=dir_fd)
+                except PermissionError:
+                    if not os.path.islink(dir1):
+                        os.chmod(dir1, 0o700)
+                        os.rename(dir1, dir1 + '_moved')
+                        os.symlink(target, dir1)
+                    raise
+            try:
+                with mock.patch('os.unlink', hook):
+                    with contextlib.suppress(OSError):
+                        d1.cleanup()
+            finally:
+                if os.path.islink(dir1):
+                    os.unlink(dir1)
+                    os.rename(dir1 + '_moved', dir1)
+                os.chmod(dir1, 0o700)
+                d1.cleanup()
+
+            self.assertTrue(os.path.exists(target_file))
+            self.assertEqual(os.stat(target_file).st_mode, target_mode)
+
     @support.cpython_only
     def test_del_on_collection(self):
         # A TemporaryDirectory is deleted when garbage collected
@@ -2033,6 +2082,29 @@ def test_modes(self):
                     d.cleanup()
                 self.assertFalse(os.path.exists(d.name))
 
+    @support.subTests('ignore_errors', (True, False))
+    def test_parent_mode_preserved(self, ignore_errors):
+        # Test that cleanup does not touch the parent directory,
+        # even if that prevents removal.
+        for mode in range(8):
+            mode <<= 6
+            with self.subTest(mode=format(mode, '03o')):
+                outer = self.do_create()
+                with outer:
+                    d = self.do_create(dir=outer.name, dirs=2, files=2,
+                                       ignore_cleanup_errors=ignore_errors)
+                    with d:
+                        os.chmod(outer.name, mode)
+                        orig_mode = os.stat(outer.name).st_mode
+                        try:
+                            d.cleanup()
+                        except PermissionError:
+                            if ignore_errors:
+                                raise
+                        self.assertEqual(os.stat(outer.name).st_mode, orig_mode)
+                        outer.cleanup()
+                self.assertFalse(os.path.exists(outer.name))
+
     def check_flags(self, flags):
         # skip the test if these flags are not supported (ex: FreeBSD 13)
         filename = os_helper.TESTFN
@@ -2070,5 +2142,17 @@ def test_delete_false(self):
         self.assertTrue(os.path.exists(working_dir))
         shutil.rmtree(working_dir)
 
+    @unittest.skipUnless(
+        sysconfig.get_config_var('PY_SUPPORT_TIER')
+        and sysconfig.get_config_var('PY_SUPPORT_TIER') <= 3,
+        'regression test for supported platforms')
+    @unittest.skipIf(support.MS_WINDOWS, 'dirfd not used on Windows')
+    @unittest.skipIf(support.is_wasi, 'WASI has no chmod')
+    def test_cleanup_safe(self):
+        """Verify that cleanup uses the safer code path"""
+        # This is a regression test. Feel free to add exceptions for new
+        # platforms, but don't forget to update the docs.
+        self.assertTrue(tempfile._rmtree_use_dir_fd)
+
 if __name__ == "__main__":
     unittest.main()