(stack, onexc)
| 731 | onexc(os.close, path, err) |
| 732 | |
| 733 | def _rmtree_safe_fd_step(stack, onexc): |
| 734 | # Each stack item has four elements: |
| 735 | # * func: The first operation to perform: os.lstat, os.close or os.rmdir. |
| 736 | # Walking a directory starts with an os.lstat() to detect symlinks; in |
| 737 | # this case, func is updated before subsequent operations and passed to |
| 738 | # onexc() if an error occurs. |
| 739 | # * dirfd: Open file descriptor, or None if we're processing the top-level |
| 740 | # directory given to rmtree() and the user didn't supply dir_fd. |
| 741 | # * path: Path of file to operate upon. This is passed to onexc() if an |
| 742 | # error occurs. |
| 743 | # * orig_entry: os.DirEntry, or None if we're processing the top-level |
| 744 | # directory given to rmtree(). We used the cached stat() of the entry to |
| 745 | # save a call to os.lstat() when walking subdirectories. |
| 746 | func, dirfd, path, orig_entry = stack.pop() |
| 747 | name = path if orig_entry is None else orig_entry.name |
| 748 | try: |
| 749 | if func is os.close: |
| 750 | os.close(dirfd) |
| 751 | return |
| 752 | if func is os.rmdir: |
| 753 | os.rmdir(name, dir_fd=dirfd) |
| 754 | return |
| 755 | |
| 756 | # Note: To guard against symlink races, we use the standard |
| 757 | # lstat()/open()/fstat() trick. |
| 758 | assert func is os.lstat |
| 759 | if orig_entry is None: |
| 760 | orig_st = os.lstat(name, dir_fd=dirfd) |
| 761 | else: |
| 762 | orig_st = orig_entry.stat(follow_symlinks=False) |
| 763 | |
| 764 | func = os.open # For error reporting. |
| 765 | topfd = os.open(name, os.O_RDONLY | os.O_NONBLOCK, dir_fd=dirfd) |
| 766 | |
| 767 | func = os.path.islink # For error reporting. |
| 768 | try: |
| 769 | if not os.path.samestat(orig_st, os.fstat(topfd)): |
| 770 | # Symlinks to directories are forbidden, see GH-46010. |
| 771 | raise OSError("Cannot call rmtree on a symbolic link") |
| 772 | stack.append((os.rmdir, dirfd, path, orig_entry)) |
| 773 | finally: |
| 774 | stack.append((os.close, topfd, path, orig_entry)) |
| 775 | |
| 776 | func = os.scandir # For error reporting. |
| 777 | with os.scandir(topfd) as scandir_it: |
| 778 | entries = list(scandir_it) |
| 779 | for entry in entries: |
| 780 | fullname = os.path.join(path, entry.name) |
| 781 | try: |
| 782 | if entry.is_dir(follow_symlinks=False): |
| 783 | # Traverse into sub-directory. |
| 784 | stack.append((os.lstat, topfd, fullname, entry)) |
| 785 | continue |
| 786 | except FileNotFoundError: |
| 787 | continue |
| 788 | except OSError: |
| 789 | pass |
| 790 | try: |
no test coverage detected
searching dependent graphs…