Use implementation from shutil

python · barneygale · Jul 20, 2024 · May 15, 2024 · May 15, 2024 · May 27, 2024
commit ab7b0863a82908cf16eadebe2f50ee07b09a44bd
@@ -106,6 +106,8 @@ pathlib
 * Add :meth:`pathlib.Path.copy`, which copies the content of one file to
   another, like :func:`shutil.copyfile`.
   (Contributed by Barney Gale in :gh:`73991`.)
+* Add :meth:`pathlib.Path.rmtree`, which recursively removes a directory.
+  (Contributed by Barney Gale in :gh:`73991`.)
 
 symtable
 --------
@@ -118,12 +120,6 @@ symtable
 
   (Contributed by Bénédikt Tran in :gh:`120029`.)
 
-pathlib
--------
-
-* Add :meth:`pathlib.Path.rmtree`, which recursively removes a directory.
-  (Contributed by Barney Gale in :gh:`73991`.)
-
 
 Optimizations
 =============

@@ -883,21 +883,20 @@ def on_error(error):
                 top_down=False,
                 on_error=on_error,
                 follow_symlinks=False)
-            for dirpath, _, filenames in results:
-                for filename in filenames:
-                    filepath = dirpath / filename
+            for dirpath, dirnames, filenames in results:
+                for name in filenames:
+                    child = dirpath / name
                     try:
-                        filepath.unlink()
-                    except FileNotFoundError:
-                        pass
+                        child.unlink()
                     except OSError as error:
                         on_error(error)
-                try:
-                    dirpath.rmdir()
-                except FileNotFoundError:
-                    pass
-                except OSError as error:
-                    on_error(error)
+                for name in dirnames:
+                    child = dirpath / name
+                    try:
+                        child.rmdir()
+                    except OSError as error:
+                        on_error(error)
+            self.rmdir()
         except OSError as error:
             error.filename = str(self)
             on_error(error)

@@ -18,7 +18,7 @@
     grp = None
 
 from ._abc import UnsupportedOperation, PurePathBase, PathBase
-from ._os import copyfile
+from ._os import copyfile, rmtree as rmtree_impl
 
 
 __all__ = [
@@ -819,59 +819,27 @@ def rmdir(self):
         """
         os.rmdir(self)
 
-    _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)
-
-    if _use_fd_functions:
-        def rmtree(self, ignore_errors=False, on_error=None):
-            """
-            Recursively delete this directory tree.
+    def rmtree(self, ignore_errors=False, on_error=None):
+        """
+        Recursively delete this directory tree.
 
-            If *ignore_errors* is true, exceptions raised from scanning the tree
-            and removing files and directories are ignored. Otherwise, if
-            *on_error* is set, it will be called to handle the error. If neither
-            *ignore_errors* nor *on_error* are set, exceptions are propagated to
-            the caller.
-            """
-            path = os.fspath(self)
-            if ignore_errors:
-                def on_error(error):
-                    pass
-            elif on_error is None:
-                def on_error(error):
-                    raise
-            try:
-                if os.path.islink(path):
-                    raise OSError("Cannot call rmtree on a symbolic link")
-                results = os.fwalk(
-                    path,
-                    topdown=False,
-                    onerror=on_error,
-                    follow_symlinks=os._walk_symlinks_as_files)
-                for dirpath, dirnames, filenames, fd in results:
-                    for filename in filenames:
-                        try:
-                            os.unlink(filename, dir_fd=fd)
-                        except FileNotFoundError:
-                            pass
-                        except OSError as error:
-                            error.filename = os.path.join(dirpath, filename)
-                            on_error(error)
-                    for dirname in dirnames:
-                        try:
-                            os.rmdir(dirname, dir_fd=fd)
-                        except FileNotFoundError:
-                            pass
-                        except OSError as error:
-                            error.filename = os.path.join(dirpath, dirname)
-                            on_error(error)
-                os.rmdir(path)
-            except OSError as error:
-                error.filename = path
-                on_error(error)
-        rmtree.avoids_symlink_attacks = True
+        If *ignore_errors* is true, exceptions raised from scanning the tree
+        and removing files and directories are ignored. Otherwise, if
+        *on_error* is set, it will be called to handle the error. If neither
+        *ignore_errors* nor *on_error* are set, exceptions are propagated to
+        the caller.
+        """
+        if ignore_errors:
+            def onexc(func, filename, err):
+                pass
+        elif on_error:
+            def onexc(func, filename, err):
+                on_error(err)
+        else:
+            def onexc(func, filename, err):
+                raise err
+        rmtree_impl(str(self), None, onexc)
+    rmtree.avoids_symlink_attacks = rmtree_impl.avoids_symlink_attacks
 
     def rename(self, target):
         """

diff --git a/Lib/pathlib/_os.py b/Lib/pathlib/_os.py
@@ -4,6 +4,7 @@
 
 from errno import EBADF, EOPNOTSUPP, ETXTBSY, EXDEV
 import os
+import stat
 import sys
 try:
     import fcntl
@@ -136,3 +137,159 @@ def copyfileobj(source_f, target_f):
     write_target = target_f.write
     while buf := read_source(1024 * 1024):
         write_target(buf)
+
+if ({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_safe_fd_step(stack, onexc):
+        # Each stack item has four elements:
+        # * func: The first operation to perform: os.lstat, os.close or os.rmdir.
+        #   Walking a directory starts with an os.lstat() to detect symlinks; in
+        #   this case, func is updated before subsequent operations and passed to
+        #   onexc() if an error occurs.
+        # * dirfd: Open file descriptor, or None if we're processing the top-level
+        #   directory given to rmtree() and the user didn't supply dir_fd.
+        # * path: Path of file to operate upon. This is passed to onexc() if an
+        #   error occurs.
+        # * orig_entry: os.DirEntry, or None if we're processing the top-level
+        #   directory given to rmtree(). We used the cached stat() of the entry to
+        #   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
+        try:
+            if func is os.close:
+                os.close(dirfd)
+                return
+            if func is os.rmdir:
+                os.rmdir(name, dir_fd=dirfd)
+                return
+
+            # Note: To guard against symlink races, we use the standard
+            # lstat()/open()/fstat() trick.
+            assert func is os.lstat
+            if orig_entry is None:
+                orig_st = os.lstat(name, dir_fd=dirfd)
+            else:
+                orig_st = orig_entry.stat(follow_symlinks=False)
+
+            func = os.open  # For error reporting.
+            topfd = os.open(name, os.O_RDONLY | os.O_NONBLOCK, dir_fd=dirfd)
+
+            func = os.path.islink  # For error reporting.
+            try:
+                if not os.path.samestat(orig_st, os.fstat(topfd)):
+                    # Symlinks to directories are forbidden, see GH-46010.
+                    raise OSError("Cannot call rmtree on a symbolic link")
+                stack.append((os.rmdir, dirfd, path, orig_entry))
+            finally:
+                stack.append((os.close, topfd, path, orig_entry))
+
+            func = os.scandir  # For error reporting.
+            with os.scandir(topfd) as scandir_it:
+                entries = list(scandir_it)
+            for entry in entries:
+                fullname = os.path.join(path, entry.name)
+                try:
+                    if entry.is_dir(follow_symlinks=False):
+                        # Traverse into sub-directory.
+                        stack.append((os.lstat, topfd, fullname, entry))
+                        continue
+                except FileNotFoundError:
+                    continue
+                except OSError:
+                    pass
+                try:
+                    os.unlink(entry.name, dir_fd=topfd)
+                except FileNotFoundError:
+                    continue
+                except OSError as err:
+                    onexc(os.unlink, fullname, err)
+        except FileNotFoundError as err:
+            if orig_entry is None or func is os.close:
+                err.filename = path
+                onexc(func, path, err)
+        except OSError as err:
+            err.filename = path
+            onexc(func, path, err)
+
+    # Version using fd-based APIs to protect against races
+    def rmtree(path, dir_fd, onexc):
+        # While the unsafe rmtree works fine on bytes, the fd based does not.
+        if isinstance(path, bytes):
+            path = os.fsdecode(path)
+        stack = [(os.lstat, dir_fd, path, None)]
+        try:
+            while stack:
+                _rmtree_safe_fd_step(stack, onexc)
+        finally:
+            # Close any file descriptors still on the stack.
+            while stack:
+                func, fd, path, entry = stack.pop()
+                if func is not os.close:
+                    continue
+                try:
+                    os.close(fd)
+                except OSError as err:
+                    onexc(os.close, path, err)
+
+    rmtree.avoids_symlink_attacks = True
+
+else:
+    if hasattr(os.stat_result, 'st_file_attributes'):
+        def _rmtree_islink(st):
+            return (stat.S_ISLNK(st.st_mode) or
+                    (st.st_file_attributes & stat.FILE_ATTRIBUTE_REPARSE_POINT
+                     and st.st_reparse_tag == stat.IO_REPARSE_TAG_MOUNT_POINT))
+    else:
+        def _rmtree_islink(st):
+            return stat.S_ISLNK(st.st_mode)
+
+    # version vulnerable to race conditions
+    def rmtree(path, dir_fd, onexc):
+        if dir_fd is not None:
+            raise NotImplementedError("dir_fd unavailable on this platform")
+        try:
+            st = os.lstat(path)
+        except OSError as err:
+            onexc(os.lstat, path, err)
+            return
+        try:
+            if _rmtree_islink(st):
+                # symlinks to directories are forbidden, see bug #1669
+                raise OSError("Cannot call rmtree on a symbolic link")
+        except OSError as err:
+            onexc(os.path.islink, path, err)
+            # can't continue even if onexc hook returns
+            return
+
+        def onerror(err):
+            if not isinstance(err, FileNotFoundError):
+                onexc(os.scandir, err.filename, err)
+
+        results = os.walk(path, topdown=False, onerror=onerror,
+                          followlinks=os._walk_symlinks_as_files)
+        for dirpath, dirnames, filenames in results:
+            for name in dirnames:
+                fullname = os.path.join(dirpath, name)
+                try:
+                    os.rmdir(fullname)
+                except FileNotFoundError:
+                    continue
+                except OSError as err:
+                    onexc(os.rmdir, fullname, err)
+            for name in filenames:
+                fullname = os.path.join(dirpath, name)
+                try:
+                    os.unlink(fullname)
+                except FileNotFoundError:
+                    continue
+                except OSError as err:
+                    onexc(os.unlink, fullname, err)
+        try:
+            os.rmdir(path)
+        except FileNotFoundError:
+            pass
+        except OSError as err:
+            onexc(os.rmdir, path, err)
+
+    rmtree.avoids_symlink_attacks = False