Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

gh-91133: tempfile.TemporaryDirectory: fix symlink bug in cleanup #99930

Merged
merged 15 commits into from
Dec 7, 2023
Merged
Show file tree
Hide file tree
Changes from 3 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
21 changes: 14 additions & 7 deletions Lib/tempfile.py
Original file line number Diff line number Diff line change
Expand Up @@ -864,15 +864,22 @@ def __init__(self, suffix=None, prefix=None, dir=None,

@classmethod
def _rmtree(cls, name, ignore_errors=False):
def without_following_symlinks(func, path, *args):
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I suggest to rename the function to "no_follow_symlinks" or "dont_follow_symlinks".

I dislike this trend to declare nested functions. Can't you move these functions at the module level, just make them private? This function doesn't use name nor ignore_errors.

# Pass follow_symlinks=False, unless not supported on this platform.
if func in _os.supports_follow_symlinks:
func(path, *args, follow_symlinks=False)
elif not _os.path.islink(path):
func(path, *args)

def resetperms(path):
try:
without_following_symlinks(_os.chflags, path, 0)
except AttributeError:
pass
without_following_symlinks(_os.chmod, path, 0o700)

def onerror(func, path, exc_info):
if issubclass(exc_info[0], PermissionError):
def resetperms(path):
try:
_os.chflags(path, 0)
except AttributeError:
pass
_os.chmod(path, 0o700)

try:
if path != name:
resetperms(_os.path.dirname(path))
Expand Down
64 changes: 57 additions & 7 deletions Lib/test/test_tempfile.py
Original file line number Diff line number Diff line change
Expand Up @@ -1652,6 +1652,45 @@ def test_cleanup_with_symlink_to_a_directory(self):
"were deleted")
d2.cleanup()

@os_helper.skip_unless_symlink
@os_helper.skip_unless_working_chmod
@unittest.skipUnless(os.chmod in os.supports_follow_symlinks, 'needs chmod follow_symlinks support')
kwi-dk marked this conversation as resolved.
Show resolved Hide resolved
def test_cleanup_with_error_deleting_symlink(self):
# cleanup() should not follow symlinks when fixing mode bits etc. (#91133)
d1 = self.do_create()
d2 = self.do_create(recurse=0)

# Symlink d1/my_symlink -> d2, then give d2 a custom mode to see if changes.
kwi-dk marked this conversation as resolved.
Show resolved Hide resolved
os.symlink(d2.name, os.path.join(d1.name, "my_symlink"))
os.chmod(d2.name, 0o567)
expected_mode = os.stat(d2.name).st_mode # can be impacted by umask etc.

# There are a variety of reasons why the OS may raise a PermissionError,
# but provoking those reliably and cross-platform is not straightforward,
# so raise the error synthetically instead.
real_unlink = os.unlink
error_was_raised = False
def patched_unlink(path, **kwargs):
nonlocal error_was_raised
# unlink may be called with full path or path relative to 'fd' kwarg.
if path.endswith("my_symlink") and not error_was_raised:
error_was_raised = True
raise PermissionError()
real_unlink(path, **kwargs)

with mock.patch("tempfile._os.unlink", patched_unlink):
# This call to cleanup() should not follow my_symlink when fixing permissions
d1.cleanup()

self.assertTrue(error_was_raised, "did not see expected 'unlink' call")
self.assertFalse(os.path.exists(d1.name),
"TemporaryDirectory %s exists after cleanup" % d1.name)
self.assertTrue(os.path.exists(d2.name),
"Directory pointed to by a symlink was deleted")
self.assertEqual(os.stat(d2.name).st_mode, expected_mode,
"Mode of the directory pointed to by a symlink changed")
d2.cleanup()

@support.cpython_only
def test_del_on_collection(self):
# A TemporaryDirectory is deleted when garbage collected
Expand Down Expand Up @@ -1826,17 +1865,28 @@ def test_modes(self):

@unittest.skipUnless(hasattr(os, 'chflags'), 'requires os.lchflags')
def test_flags(self):
flags = stat.UF_IMMUTABLE | stat.UF_NOUNLINK
d = self.do_create(recurse=3, dirs=2, files=2)
with d:
# Change files and directories flags recursively.
for root, dirs, files in os.walk(d.name, topdown=False):
def chflags_recursively(name, flags):
for root, dirs, files in os.walk(name, topdown=False):
for name in files:
os.chflags(os.path.join(root, name), flags)
os.chflags(root, flags)
d.cleanup()
self.assertFalse(os.path.exists(d.name))

d = self.do_create(recurse=3, dirs=2, files=2)
try:
with d:
chflags_recursively(d.name, stat.UF_IMMUTABLE | stat.UF_NOUNLINK)
d.cleanup()

self.assertFalse(os.path.exists(d.name))

finally:
# Even if test fails, make a best-effort attempt at not leaving
kwi-dk marked this conversation as resolved.
Show resolved Hide resolved
# behind NOUNLINK files, because they require manual removal,
# which becomes tiresome during development.
try:
chflags_recursively(d.name, 0)
except Exception:
pass # we tried

if __name__ == "__main__":
unittest.main()
Original file line number Diff line number Diff line change
@@ -0,0 +1,2 @@
Fix a bug :class:`tempfile.TemporaryDirectory` cleanup, which now no longer
kwi-dk marked this conversation as resolved.
Show resolved Hide resolved
dereferences symlinks when working around file system permission errors.