diff --git a/Doc/library/tempfile.rst b/Doc/library/tempfile.rst index 316d82775cf626..8bc1cdb81d65c6 100644 --- a/Doc/library/tempfile.rst +++ b/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 diff --git a/Lib/shutil.py b/Lib/shutil.py index ab75ba9da8894b..d6d7806802bd17 100644 --- a/Lib/shutil.py +++ b/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,14 +809,14 @@ 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 @@ -823,7 +824,8 @@ def _rmtree_safe_fd_step(stack, onexc): 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) diff --git a/Lib/tempfile.py b/Lib/tempfile.py index b5f799f8d68554..ad81c5d3417b4c 100644 --- a/Lib/tempfile.py +++ b/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,8 +981,12 @@ 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: @@ -936,15 +994,30 @@ def onexc(func, path, exc): 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): diff --git a/Lib/test/test_shutil.py b/Lib/test/test_shutil.py index a7825adf923c19..87b6453fba8024 100644 --- a/Lib/test/test_shutil.py +++ b/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 diff --git a/Lib/test/test_tempfile.py b/Lib/test/test_tempfile.py index e33cc65e090e3b..cd960ed99117b6 100644 --- a/Lib/test/test_tempfile.py +++ b/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() diff --git a/Misc/NEWS.d/next/Security/2026-08-14-11-55-00.gh-issue-157579.Kq3Vt2.rst b/Misc/NEWS.d/next/Security/2026-08-14-11-55-00.gh-issue-157579.Kq3Vt2.rst new file mode 100644 index 00000000000000..46e87575dba983 --- /dev/null +++ b/Misc/NEWS.d/next/Security/2026-08-14-11-55-00.gh-issue-157579.Kq3Vt2.rst @@ -0,0 +1,6 @@ +Fix a race condition in the cleanup of :class:`tempfile.TemporaryDirectory`. +When working around file system permission errors, files are now removed +relative to open directory file descriptors instead of resolving their path +again, so that replacing a directory of the tree with a symbolic link can no +longer make the cleanup delete files outside of the temporary directory. +This addresses :cve:`2026-12345`.