Skip to content
Open
Show file tree
Hide file tree
Changes from all 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
20 changes: 15 additions & 5 deletions Lib/shutil.py
Original file line number Diff line number Diff line change
Expand Up @@ -543,12 +543,22 @@ def _ignore_patterns(path, names):
return _ignore_patterns

def _copytree(entries, src, dst, symlinks, ignore, copy_function,
ignore_dangling_symlinks, dirs_exist_ok=False):
ignore_dangling_symlinks, dirs_exist_ok=False, _seen=None):
if ignore is not None:
ignored_names = ignore(os.fspath(src), [x.name for x in entries])
else:
ignored_names = ()

# Track visited directories to detect cycles (e.g., Windows junctions)
if _seen is None:
_seen = frozenset()
Comment on lines +553 to +554
Copy link
Member

Choose a reason for hiding this comment

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

I do not understand why we do this. Why not just using a default argument? (it is no more mutable so it is fine)

src_st = os.stat(src)
Copy link
Member

Choose a reason for hiding this comment

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

This call can fail and (can be) expensive. But I assume that we need it right?

src_id = (src_st.st_dev, src_st.st_ino)
if src_id in _seen:
raise Error([(src, dst, "Infinite recursion detected")])

_seen = _seen | {src_id}

os.makedirs(dst, exist_ok=dirs_exist_ok)
errors = []
use_srcentry = copy_function is copy2 or copy_function is copy
Expand Down Expand Up @@ -583,12 +593,12 @@ def _copytree(entries, src, dst, symlinks, ignore, copy_function,
if srcentry.is_dir():
copytree(srcobj, dstname, symlinks, ignore,
copy_function, ignore_dangling_symlinks,
dirs_exist_ok)
dirs_exist_ok, _seen=_seen)
else:
copy_function(srcobj, dstname)
elif srcentry.is_dir():
copytree(srcobj, dstname, symlinks, ignore, copy_function,
ignore_dangling_symlinks, dirs_exist_ok)
ignore_dangling_symlinks, dirs_exist_ok, _seen=_seen)
else:
# Will raise a SpecialFileError for unsupported file types
copy_function(srcobj, dstname)
Expand All @@ -609,7 +619,7 @@ def _copytree(entries, src, dst, symlinks, ignore, copy_function,
return dst

def copytree(src, dst, symlinks=False, ignore=None, copy_function=copy2,
ignore_dangling_symlinks=False, dirs_exist_ok=False):
ignore_dangling_symlinks=False, dirs_exist_ok=False, _seen=None):
Copy link
Contributor

Choose a reason for hiding this comment

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

I'm not sure, but I think adding a private parameter to a public function is a little odd. I see we added the _seen here because _copytree calls the public copytree, so we need to add it here too.

I wonder if we could just change _copytree to call itself recursively instead of adding this private parameter. This would also avoid triggering the audit recursively (should we do that?).

If we should trigger the audit recursively, maybe we could create another private function for the recursive logic and let _copytree call it, with a public wrapper that keeps the original function signature (without adding the _seen).

Copy link
Member

Choose a reason for hiding this comment

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

I agree. It is better to have a private helper for that.

"""Recursively copy a directory tree and return the destination directory.

If exception(s) occur, an Error is raised with a list of reasons.
Expand Down Expand Up @@ -654,7 +664,7 @@ def copytree(src, dst, symlinks=False, ignore=None, copy_function=copy2,
return _copytree(entries=entries, src=src, dst=dst, symlinks=symlinks,
ignore=ignore, copy_function=copy_function,
ignore_dangling_symlinks=ignore_dangling_symlinks,
dirs_exist_ok=dirs_exist_ok)
dirs_exist_ok=dirs_exist_ok, _seen=_seen)

if hasattr(os.stat_result, 'st_file_attributes'):
def _rmtree_islink(st):
Expand Down
36 changes: 36 additions & 0 deletions Lib/test/test_shutil.py
Original file line number Diff line number Diff line change
Expand Up @@ -1099,6 +1099,42 @@ def test_copytree_subdirectory(self):
rv = shutil.copytree(src_dir, dst_dir)
self.assertEqual(['pol'], os.listdir(rv))

@unittest.skipUnless(sys.platform == "win32", "Windows-specific test")
def test_copytree_recursive_junction(self):
# Test that copytree raises Error for recursive junctions (Windows)
base_dir = self.mkdtemp()
self.addCleanup(shutil.rmtree, base_dir, ignore_errors=True)

# Create source directory structure
src_dir = os.path.join(base_dir, "source")
junction_dir = os.path.join(src_dir, "junction")
os.makedirs(junction_dir)

# Create a junction pointing to its parent, creating a cycle
junction_target = os.path.dirname(junction_dir) # Points to Source
Copy link
Member

Choose a reason for hiding this comment

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

Suggested change
junction_target = os.path.dirname(junction_dir) # Points to Source
junction_target = os.path.dirname(junction_dir)

try:
result = subprocess.run(
["mklink", "/J", junction_dir, junction_target],
shell=True, check=False, capture_output=True, text=True
)
if result.returncode != 0:
# Skip if we don't have permission to create junctions
self.skipTest(f"Failed to create junction: {result.stderr.strip()}")
except Exception as e:
# Skip if mklink is not available or fails for any reason
self.skipTest(f"Failed to create junction: {e}")
Comment on lines +1123 to +1125
Copy link
Member

Choose a reason for hiding this comment

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

I don't think it's a good idea. Can't we create a junction from Python directly? (cc @barneygale).

Copy link
Member

Choose a reason for hiding this comment

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

This is still not addressed.


# Create destination directory
dst_dir = os.path.join(base_dir, "Dest")
Copy link
Member

Choose a reason for hiding this comment

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

Dest -> dest


# Test that copytree raises Error with infinite recursion message
with self.assertRaises(shutil.Error) as cm:
shutil.copytree(src_dir, dst_dir)

# Verify the error message contains "Infinite recursion detected"
self.assertIn("Infinite recursion detected", str(cm.exception))


class TestCopy(BaseTest, unittest.TestCase):

### shutil.copymode
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1 @@
Fix infinite recursion crash in :func:`shutil.copytree` when a cycle involving a directory junction is present on Windows.
Loading