diff options
| -rw-r--r-- | gemato/exceptions.py | 17 | ||||
| -rw-r--r-- | gemato/recursiveloader.py | 37 | ||||
| -rw-r--r-- | tests/test_recursiveloader.py | 13 |
3 files changed, 60 insertions, 7 deletions
diff --git a/gemato/exceptions.py b/gemato/exceptions.py index d544f62..d2c818e 100644 --- a/gemato/exceptions.py +++ b/gemato/exceptions.py @@ -89,6 +89,23 @@ class ManifestCrossDevice(GematoException): .format(self.path)) +class ManifestSymlinkLoop(GematoException): + """ + An exception caused by hitting a symlink loop (symlink to itself + or a parent directory). + """ + + __slots__ = ['path'] + + def __init__(self, path): + super(ManifestSymlinkLoop, self).__init__(path) + self.path = path + + def __str__(self): + return ("Path {} is a symlink to one of its parent directories, it must be IGNORE-d explicitly" + .format(self.path)) + + class ManifestUnsignedData(GematoException): """ An exception caused by a Manifest file containing non-whitespace diff --git a/gemato/recursiveloader.py b/gemato/recursiveloader.py index 53a328e..b89df87 100644 --- a/gemato/recursiveloader.py +++ b/gemato/recursiveloader.py @@ -593,11 +593,21 @@ class ManifestRecursiveLoader(object): Pre-process os.walk() result for verification. Yield objects suitable to passing to subprocesses. """ + directory_ids = {} + for dirpath, dirnames, filenames in it: dir_st = os.stat(dirpath) if dir_st.st_dev != self.manifest_device: raise gemato.exceptions.ManifestCrossDevice(dirpath) + dir_id = (dir_st.st_dev, dir_st.st_ino) + # if this directory was already processed for one of its + # parents, we're in a loop + parent_dir = os.path.dirname(dirpath) + parent_dir_ids = directory_ids.get(parent_dir, []) + if dir_id in parent_dir_ids: + raise gemato.exceptions.ManifestSymlinkLoop(dirpath) + relpath = os.path.relpath(dirpath, self.root_directory) # strip dot to avoid matching problems if relpath == '.': @@ -625,6 +635,9 @@ class ManifestRecursiveLoader(object): # skip scanning ignored directories for d in skip_dirs: dirnames.remove(d) + # if we are planning to recur, record this dir + if dirnames: + directory_ids[dirpath] = parent_dir_ids + [dir_id] yield (dirpath, relpath, dirnames, filenames, dirdict) @@ -962,6 +975,7 @@ class ManifestRecursiveLoader(object): entry_dict = self.get_file_entry_dict(path, only_types=['IGNORE'], verify_manifests=verify_manifests) new_manifests = [] + directory_ids = {} it = os.walk(os.path.join(self.root_directory, path), onerror=gemato.util.throw_exception, followlinks=True) @@ -971,6 +985,14 @@ class ManifestRecursiveLoader(object): if dir_st.st_dev != self.manifest_device: raise gemato.exceptions.ManifestCrossDevice(dirpath) + dir_id = (dir_st.st_dev, dir_st.st_ino) + # if this directory was already processed for one of its + # parents, we're in a loop + parent_dir = os.path.dirname(dirpath) + parent_dir_ids = directory_ids.get(parent_dir, []) + if dir_id in parent_dir_ids: + raise gemato.exceptions.ManifestSymlinkLoop(dirpath) + relpath = os.path.relpath(dirpath, self.root_directory) # strip dot to avoid matching problems if relpath == '.': @@ -994,6 +1016,9 @@ class ManifestRecursiveLoader(object): # skip scanning ignored directories for d in skip_dirs: dirnames.remove(d) + # if we are planning to recur, record this dir + if dirnames: + directory_ids[dirpath] = parent_dir_ids + [dir_id] # check for unregistered Manifest for mname in manifest_filenames: @@ -1061,6 +1086,7 @@ class ManifestRecursiveLoader(object): ._iter_manifests_for_path(path)): manifest_stack.append((mpath, mrpath, m)) break + directory_ids = {} it = os.walk(os.path.join(self.root_directory, path), onerror=gemato.util.throw_exception, @@ -1071,6 +1097,14 @@ class ManifestRecursiveLoader(object): if dir_st.st_dev != self.manifest_device: raise gemato.exceptions.ManifestCrossDevice(dirpath) + dir_id = (dir_st.st_dev, dir_st.st_ino) + # if this directory was already processed for one of its + # parents, we're in a loop + parent_dir = os.path.dirname(dirpath) + parent_dir_ids = directory_ids.get(parent_dir, []) + if dir_id in parent_dir_ids: + raise gemato.exceptions.ManifestSymlinkLoop(dirpath) + relpath = os.path.relpath(dirpath, self.root_directory) # strip dot to avoid matching problems if relpath == '.': @@ -1110,6 +1144,9 @@ class ManifestRecursiveLoader(object): # skip scanning ignored directories for d in skip_dirs: dirnames.remove(d) + # if we are planning to recur, record this dir + if dirnames: + directory_ids[dirpath] = parent_dir_ids + [dir_id] new_entries = [] for f in filenames: diff --git a/tests/test_recursiveloader.py b/tests/test_recursiveloader.py index 94002d2..19646df 100644 --- a/tests/test_recursiveloader.py +++ b/tests/test_recursiveloader.py @@ -2737,21 +2737,20 @@ class SymlinkLoopTest(TempDirTestCase): def test_assert_directory_verifies(self): m = gemato.recursiveloader.ManifestRecursiveLoader( os.path.join(self.dir, 'Manifest')) - self.assertRaises(OSError, + self.assertRaises(gemato.exceptions.ManifestSymlinkLoop, m.assert_directory_verifies, '') def test_cli_verifies(self): - self.assertRaises(OSError, - gemato.cli.main, ['gemato', 'verify', self.dir]) + self.assertEqual(gemato.cli.main(['gemato', 'verify', self.dir]), + 1) def test_update_entries_for_directory(self): m = gemato.recursiveloader.ManifestRecursiveLoader( os.path.join(self.dir, 'Manifest'), hashes=['SHA256', 'SHA512']) - self.assertRaises(OSError, + self.assertRaises(gemato.exceptions.ManifestSymlinkLoop, m.update_entries_for_directory, '') def test_cli_update(self): - self.assertRaises(OSError, - gemato.cli.main, ['gemato', 'update', - '--hashes=SHA256 SHA512', self.dir]) + self.assertEqual(gemato.cli.main(['gemato', 'update', + '--hashes=SHA256 SHA512', self.dir]), 1) |