Description: Debian: Degrade tempfile gracefully without shutil python3.X-minimal includes tempfile but not shutil. Use a fallback racy rmtree, if shutil can't be imported. This needs constant maintainance, as the code is a copy. Forwarded: not-needed --- a/Lib/tempfile.py +++ b/Lib/tempfile.py @@ -40,7 +40,212 @@ import warnings as _warnings import io as _io import os as _os -import shutil as _shutil +try: + import shutil as _shutil + _rmtree = _shutil.rmtree +except ImportError: + import sys as _sys + import stat as _stat + + 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_unsafe(path, onexc): + 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) + + # Version using fd-based APIs to protect against races + def _rmtree_safe_fd(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) + + _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): + """Recursively delete a directory tree. + + If dir_fd is not None, it should be a file descriptor open to a directory; + path will then be relative to that directory. + dir_fd may not be implemented on your platform. + If it is unavailable, using it will raise a NotImplementedError. + + If ignore_errors is set, errors are ignored; otherwise, if onexc or + onerror is set, it is called to handle the error with arguments (func, + path, exc_info) where func is platform and implementation dependent; + path is the argument to that function that caused it to fail; and + the value of exc_info describes the exception. For onexc it is the + exception instance, and for onerror it is a tuple as returned by + sys.exc_info(). If ignore_errors is false and both onexc and + onerror are None, the exception is reraised. + + onerror is deprecated and only remains for backwards compatibility. + If both onerror and onexc are set, onerror is ignored and onexc is used. + """ + + _sys.audit("shutil.rmtree", path, dir_fd) + if ignore_errors: + def onexc(*args): + pass + elif onerror is None and onexc is None: + def onexc(*args): + raise + elif onexc is None: + if onerror is None: + def onexc(*args): + raise + else: + # delegate to onerror + 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) + + if _use_fd_functions: + # 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(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) + else: + 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 + return _rmtree_unsafe(path, onexc) + + # Allow introspection of whether or not the hardening against symlink + # attacks is supported on the current platform + _rmtree.avoids_symlink_attacks = _use_fd_functions + + import errno as _errno from random import Random as _Random import sys as _sys @@ -932,7 +1137,7 @@ if not ignore_errors: raise - _shutil.rmtree(name, onexc=onexc) + _rmtree(name, onexc=onexc) @classmethod def _cleanup(cls, name, warn_message, ignore_errors=False, delete=True):