Properly handle 7z archives

Co-authored-by: catileptic <alex.stefanescu@pm.me>

Author: tillprochaska

ingestors/packages/__init__.py

@@ -4,6 +4,9 @@
 import tarfile
 from pathlib import PurePath
 
+import py7zr
+from py7zr.exceptions import ArchiveError
+
 from ingestors.ingestor import Ingestor
 from ingestors.support.package import PackageSupport
 from ingestors.support.shell import ShellSupport
@@ -24,9 +27,11 @@ def unpack(self, file_path, entity, temp_dir):
                 *pure_file_path.parts[1:-1], reconstructed_filename
             )
 
-        self.exec_command(
-            "7z", "x", str(pure_file_path), "-y", "-r", "-bb0", "-bd", f"-oc:{temp_dir}"
-        )
+        try:
+            with py7zr.SevenZipFile(str(pure_file_path), mode="r") as z:
+                z.extractall(path=temp_dir)
+        except ArchiveError as e:
+            raise ProcessingException(f"Error: {e}")
 
 
 class SingleFilePackageIngestor(PackageSupport, Ingestor):

requirements.txt

@@ -14,6 +14,7 @@ numpy<2.0.0 # pinned because otherwise spacy requires an incompatible numpy
 fingerprints==1.1.1
 fasttext==0.9.2
 pika==1.3.2
+py7zr==1.0.0
 
 # Development
 pytest==8.2.0

tests/fixtures/bad7zip.7z

@@ -0,0 +1 @@
+This is a secret!

tests/fixtures/badrar.rar

@@ -1,5 +1,6 @@
 # -*- coding: utf-8 -*-
 from pprint import pprint  # noqa
+from pathlib import Path
 
 from .support import TestCase
 
@@ -11,14 +12,80 @@ def test_zip(self):
         self.assertEqual(entity.first("processingStatus"), self.manager.STATUS_SUCCESS)
         self.assertEqual(entity.schema.name, "Package")
 
+    def test_zip_symlink_escape(self):
+        fixture_path, entity = self.fixture("badzip.zip")
+
+        # Ensure that the symlink target exists
+        target = Path("/ingestors/tests/fixtures/secret.txt")
+        assert target.read_text() == "This is a secret!"
+
+        self.manager.ingest(fixture_path, entity)
+
+        # Python’s zipfile handles symlinks that point to files outside of the archive root
+        # treating them as normal files
+        assert len(self.manager.entities) == 2
+        assert self.manager.entities[0].first("fileName") == "secret.txt"
+        assert (
+            self.manager.entities[0].first("bodyText")
+            == "/ingestors/tests/fixtures/secret.txt"
+        )
+        assert self.manager.entities[1].first("fileName") == "badzip.zip"
+
     def test_rar(self):
         fixture_path, entity = self.fixture("test-documents.rar")
         self.manager.ingest(fixture_path, entity)
         self.assertEqual(entity.first("processingStatus"), self.manager.STATUS_SUCCESS)
         self.assertEqual(entity.schema.name, "Package")
 
+    def test_rar_symlink_escape(self):
+        fixture_path, entity = self.fixture("badrar.rar")
+
+        # Ensure that the symlink target exists
+        target = Path("/ingestors/tests/fixtures/secret.txt")
+        assert target.read_text() == "This is a secret!"
+
+        self.manager.ingest(fixture_path, entity)
+
+        # rarfile handles symlinks that point to files outside of the archive root
+        # treating them as normal files
+        assert len(self.manager.entities) == 2
+        assert self.manager.entities[0].first("fileName") == "secret.txt"
+        assert (
+            self.manager.entities[0].first("bodyText")
+            == "/ingestors/tests/fixtures/secret.txt"
+        )
+        assert self.manager.entities[1].first("fileName") == "badrar.rar"
+
     def test_tar(self):
         fixture_path, entity = self.fixture("test-documents.tar")
         self.manager.ingest(fixture_path, entity)
         self.assertEqual(entity.first("processingStatus"), self.manager.STATUS_SUCCESS)
         self.assertEqual(entity.schema.name, "Package")
+
+    def test_tar_symlink_escape(self):
+        fixture_path, entity = self.fixture("badtar.tar")
+
+        # Ensure that the symlink target exists
+        target = Path("/ingestors/tests/fixtures/secret.txt")
+        assert target.read_text() == "This is a secret!"
+
+        self.manager.ingest(fixture_path, entity)
+
+        # Python’s tarfile ignores symlinks that point to files outside of the archive root
+        assert len(self.manager.entities) == 1
+        assert self.manager.entities[0].first("fileName") == "badtar.tar"
+
+    def test_7zip_symlink_escape(self):
+        fixture_path, entity = self.fixture("bad7zip.7z")
+
+        # Ensure that the symlink target exists
+        target = Path("/ingestors/tests/fixtures/secret.txt")
+        assert target.read_text() == "This is a secret!"
+
+        self.manager.ingest(fixture_path, entity)
+
+        # py7zr raises an exception if it encounters a symlink that points to a file
+        # outside of the archive root
+        assert len(self.manager.entities) == 1
+        assert self.manager.entities[0].first("fileName") == "bad7zip.7z"
+        assert self.manager.entities[0].first("processingStatus") == "failure"