Fix link target generation for collection test

The link_target was generated only using the extract_dir and the linkname, but without the directory path that contains the link. Therefore the link_target could be outside of extract_dir. No-Issue Signed-off-by: Thomas Woerner <[email protected]>
ansible · Dec 6, 2023 · 320bc28 · 320bc28
1 parent e8212c6
commit 320bc28
Show file tree

Hide file tree

Showing 2 changed files with 39 additions and 1 deletion.
diff --git a/galaxy_importer/collection.py b/galaxy_importer/collection.py
@@ -165,7 +165,9 @@ def _extract_archive(fileobj, extract_dir):
                 raise exc.ImporterError("Invalid file paths detected.")
             if item.linkname:
                 # Ensure the link target is within the extraction root
-                link_target = os.path.normpath(os.path.join(extract_dir, item.linkname))
+                link_target = os.path.normpath(
+                    os.path.join(extract_dir, os.path.dirname(item.name), item.linkname)
+                )
                 if not link_target.startswith(os.path.abspath(extract_dir)):
                     raise exc.ImporterError("Invalid link target detected.")
         tf.extractall(extract_dir)
diff --git a/tests/unit/test_collection_archive_extract.py b/tests/unit/test_collection_archive_extract.py
@@ -119,3 +119,39 @@ def test_invalid_link_destination(self):
 
         # Clean up the temporary extraction directory
         os.rmdir(extract_dir)
+
+    def test_valid_relative_symlink_in_subdir(self):
+        # Create a valid archive with a relative symlink in a subdir
+        # and the target in the top dir
+        archive_data = b"testfile content"
+        archive_file = BytesIO()
+        with tarfile.open(fileobj=archive_file, mode="w") as tf:
+            tarinfo = tarfile.TarInfo(".")
+            tarinfo.type = tarfile.DIRTYPE
+            tarinfo.mode = 493
+            tf.addfile(tarinfo, BytesIO(archive_data))
+            tarinfo.name = "testdir1"
+            tf.addfile(tarinfo, BytesIO(archive_data))
+            tarinfo.name = "testdir2"
+            tf.addfile(tarinfo, BytesIO(archive_data))
+            tarinfo.name = "testdir2/link"
+            tarinfo.type = tarfile.SYMTYPE
+            tarinfo.mode = 511
+            tarinfo.linkname = "../testdir1"
+            tf.addfile(tarinfo, BytesIO(archive_data))
+        archive_file.seek(0)
+
+        # Create a temporary extraction directory
+        extract_dir = tempfile.mkdtemp(prefix="collection-archive-extract-test-")
+        os.makedirs(extract_dir, exist_ok=True)
+
+        try:
+            _extract_archive(archive_file, extract_dir)
+        finally:
+            pass
+
+        extracted_file_path = os.path.join(extract_dir, "testdir2/link")
+        self.assertTrue(os.path.islink(extracted_file_path))
+
+        # Clean up the temporary extraction directory
+        shutil.rmtree(extract_dir)