Merge pull request #3784 from Flamefire/avoid_recursive_copies

Check for recursive symlinks by default before copying a folder

Author: akesandgren

easybuild/tools/filetools.py

@@ -44,6 +44,7 @@
 import hashlib
 import imp
 import inspect
+import itertools
 import os
 import re
 import shutil
@@ -2340,14 +2341,36 @@ def copy_files(paths, target_path, force_in_dry_run=False, target_single_file=Fa
         raise EasyBuildError("One or more files to copy should be specified!")
 
 
-def copy_dir(path, target_path, force_in_dry_run=False, dirs_exist_ok=False, **kwargs):
+def has_recursive_symlinks(path):
+    """
+    Check the given directory for recursive symlinks.
+
+    That means symlinks to folders inside the path which would cause infinite loops when traversed regularily.
+
+    :param path: Path to directory to check
+    """
+    for dirpath, dirnames, filenames in os.walk(path, followlinks=True):
+        for name in itertools.chain(dirnames, filenames):
+            fullpath = os.path.join(dirpath, name)
+            if os.path.islink(fullpath):
+                linkpath = os.path.realpath(fullpath)
+                fullpath += os.sep  # To catch the case where both are equal
+                if fullpath.startswith(linkpath + os.sep):
+                    _log.info("Recursive symlink detected at %s", fullpath)
+                    return True
+    return False
+
+
+def copy_dir(path, target_path, force_in_dry_run=False, dirs_exist_ok=False, check_for_recursive_symlinks=True,
+             **kwargs):
     """
     Copy a directory from specified location to specified location
 
     :param path: the original directory path
     :param target_path: path to copy the directory to
     :param force_in_dry_run: force running the command during dry run
     :param dirs_exist_ok: boolean indicating whether it's OK if the target directory already exists
+    :param check_for_recursive_symlinks: If symlink arg is not given or False check for recursive symlinks first
 
     shutil.copytree is used if the target path does not exist yet;
     if the target path already exists, the 'copy' function will be used to copy the contents of
@@ -2359,6 +2382,13 @@ def copy_dir(path, target_path, force_in_dry_run=False, dirs_exist_ok=False, **k
         dry_run_msg("copied directory %s to %s" % (path, target_path))
     else:
         try:
+            if check_for_recursive_symlinks and not kwargs.get('symlinks'):
+                if has_recursive_symlinks(path):
+                    raise EasyBuildError("Recursive symlinks detected in %s. "
+                                         "Will not try copying this unless `symlinks=True` is passed",
+                                         path)
+                else:
+                    _log.debug("No recursive symlinks in %s", path)
             if not dirs_exist_ok and os.path.exists(target_path):
                 raise EasyBuildError("Target location %s to copy %s to already exists", target_path, path)
 
@@ -2386,7 +2416,9 @@ def copy_dir(path, target_path, force_in_dry_run=False, dirs_exist_ok=False, **k
                 paths_to_copy = [os.path.join(path, x) for x in entries]
 
                 copy(paths_to_copy, target_path,
-                     force_in_dry_run=force_in_dry_run, dirs_exist_ok=dirs_exist_ok, **kwargs)
+                     force_in_dry_run=force_in_dry_run, dirs_exist_ok=dirs_exist_ok,
+                     check_for_recursive_symlinks=False,  # Don't check again
+                     **kwargs)
 
             else:
                 # if dirs_exist_ok is not enabled or target directory doesn't exist, just use shutil.copytree

test/framework/filetools.py

@@ -1790,6 +1790,46 @@ def test_copy_files(self):
         regex = re.compile("^copied 2 files to .*/target")
         self.assertTrue(regex.match(stdout), "Pattern '%s' should be found in: %s" % (regex.pattern, stdout))
 
+    def test_has_recursive_symlinks(self):
+        """Test has_recursive_symlinks function"""
+        test_folder = tempfile.mkdtemp()
+        self.assertFalse(ft.has_recursive_symlinks(test_folder))
+        # Clasic Loop: Symlink to .
+        os.symlink('.', os.path.join(test_folder, 'self_link_dot'))
+        self.assertTrue(ft.has_recursive_symlinks(test_folder))
+        # Symlink to self
+        test_folder = tempfile.mkdtemp()
+        os.symlink('self_link', os.path.join(test_folder, 'self_link'))
+        self.assertTrue(ft.has_recursive_symlinks(test_folder))
+        # Symlink from 2 folders up
+        test_folder = tempfile.mkdtemp()
+        sub_folder = os.path.join(test_folder, 'sub1', 'sub2')
+        os.makedirs(sub_folder)
+        os.symlink(os.path.join('..', '..'), os.path.join(sub_folder, 'uplink'))
+        self.assertTrue(ft.has_recursive_symlinks(test_folder))
+        # Non-issue: Symlink to sibling folders
+        test_folder = tempfile.mkdtemp()
+        sub_folder = os.path.join(test_folder, 'sub1', 'sub2')
+        os.makedirs(sub_folder)
+        sibling_folder = os.path.join(test_folder, 'sub1', 'sibling')
+        os.mkdir(sibling_folder)
+        os.symlink('sibling', os.path.join(test_folder, 'sub1', 'sibling_link'))
+        os.symlink(os.path.join('..', 'sibling'), os.path.join(test_folder, sub_folder, 'sibling_link'))
+        self.assertFalse(ft.has_recursive_symlinks(test_folder))
+        # Tricky case: Sibling symlink to folder starting with the same name
+        os.mkdir(os.path.join(test_folder, 'sub11'))
+        os.symlink(os.path.join('..', 'sub11'), os.path.join(test_folder, 'sub1', 'trick_link'))
+        self.assertFalse(ft.has_recursive_symlinks(test_folder))
+        # Symlink cycle: sub1/cycle_2 -> sub2, sub2/cycle_1 -> sub1, ...
+        test_folder = tempfile.mkdtemp()
+        sub_folder1 = os.path.join(test_folder, 'sub1')
+        sub_folder2 = sub_folder = os.path.join(test_folder, 'sub2')
+        os.mkdir(sub_folder1)
+        os.mkdir(sub_folder2)
+        os.symlink(os.path.join('..', 'sub2'), os.path.join(sub_folder1, 'cycle_1'))
+        os.symlink(os.path.join('..', 'sub1'), os.path.join(sub_folder2, 'cycle_2'))
+        self.assertTrue(ft.has_recursive_symlinks(test_folder))
+
     def test_copy_dir(self):
         """Test copy_dir function."""
         testdir = os.path.dirname(os.path.abspath(__file__))
@@ -1861,6 +1901,15 @@ def ignore_func(_, names):
         ft.mkdir(subdir)
         ft.copy_dir(srcdir, target_dir, symlinks=True, dirs_exist_ok=True)
 
+        # Detect recursive symlinks by default instead of infinite loop during copy
+        ft.remove_dir(target_dir)
+        os.symlink('.', os.path.join(subdir, 'recursive_link'))
+        self.assertErrorRegex(EasyBuildError, 'Recursive symlinks detected', ft.copy_dir, srcdir, target_dir)
+        self.assertFalse(os.path.exists(target_dir))
+        # Ok for symlinks=True
+        ft.copy_dir(srcdir, target_dir, symlinks=True)
+        self.assertTrue(os.path.exists(target_dir))
+
         # also test behaviour of copy_file under --dry-run
         build_options = {
             'extended_dry_run': True,