[3.12] GH-157579: Revert GH-158489 (prematurely merged) (#158528)

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

This reverts commit 458e713.

Author: Yhg1s

Doc/library/tempfile.rst

@@ -207,15 +207,6 @@ 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,7 +110,6 @@ 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,7 +654,6 @@ 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)
@@ -698,18 +697,17 @@ def _rmtree_safe_fd(stack, onexc):
             try:
                 os.unlink(entry.name, dir_fd=topfd)
             except OSError as err:
-                onexc(os.unlink, fullname, err, direntry=entry, dir_fd=topfd)
+                onexc(os.unlink, fullname, err)
     except OSError as err:
         err.filename = path
-        onexc(func, path, err, direntry=orig_entry, dir_fd=parent_fd)
+        onexc(func, path, err)
 
 _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,
-           _onexc_kwargs=False):
+def rmtree(path, ignore_errors=False, onerror=None, *, onexc=None, dir_fd=None):
     """Recursively delete a directory tree.
 
     If dir_fd is not None, it should be a file descriptor open to a directory;
@@ -732,29 +730,24 @@ 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, **kwargs):
+        def onexc(*args):
             pass
     elif onerror is None and onexc is None:
-        def onexc(*args, **kwargs):
+        def onexc(*args):
             raise
     elif onexc is None:
         if onerror is None:
-            def onexc(*args, **kwargs):
+            def onexc(*args):
                 raise
         else:
             # delegate to onerror
-            def onexc(*args, **kwargs):
+            def onexc(*args):
                 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,7 +43,6 @@
 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
@@ -277,68 +276,15 @@ def _dont_follow_symlinks(func, path, *args):
     elif _os.name == 'nt' or not _os.path.islink(path):
         func(path, *args)
 
-def _resetflags(path):
+def _resetperms(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.
 
@@ -946,42 +892,23 @@ 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, dir_fd=None,
-                fullname=None):
-        if fullname is None:
-            fullname = name
-
-        def onexc(func, path, exc, direntry=None, dir_fd=None):
+    def _rmtree(cls, name, ignore_errors=False, repeated=False):
+        def onexc(func, path, exc):
             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:
-                        # 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)
+                        _resetperms(_os.path.dirname(path))
+                    _resetperms(path)
 
                     try:
-                        _os.unlink(base, dir_fd=dir_fd)
+                        _os.unlink(path)
                     except IsADirectoryError:
-                        cls._rmtree(base, ignore_errors=ignore_errors,
-                                    repeated=(path == name),
-                                    dir_fd=dir_fd, fullname=fullpath)
+                        cls._rmtree(path, ignore_errors=ignore_errors)
                     except PermissionError:
                         # The PermissionError handler was originally added for
                         # FreeBSD in directories, but it seems that it is raised
@@ -990,27 +917,21 @@ def onexc(func, path, exc, direntry=None, dir_fd=None):
                         # 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(fullpath)
-                                or _os.path.isjunction(fullpath)):
+                        if not _os.path.isdir(path) or _os.path.isjunction(path):
                             if ignore_errors:
                                 return
                             raise
-                        cls._rmtree(base, ignore_errors=ignore_errors,
-                                    repeated=(path == name),
-                                    dir_fd=dir_fd, fullname=fullpath)
+                        cls._rmtree(path, ignore_errors=ignore_errors,
+                                    repeated=(path == name))
                 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, dir_fd=dir_fd, _onexc_kwargs=True)
+        _shutil.rmtree(name, onexc=onexc)
 
     @classmethod
     def _cleanup(cls, name, warn_message, ignore_errors=False, delete=True):

Lib/test/support/__init__.py

@@ -2598,8 +2598,3 @@ 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,37 +497,6 @@ 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