feat(extraction): allow returning reports from Extractor.extract calls

Author: vlaci

docs/development.md

@@ -515,7 +515,7 @@ class Extractor(abc.ABC):
         return []
 
     @abc.abstractmethod
-    def extract(self, inpath: Path, outdir: Path):
+    def extract(self, inpath: Path, outdir: Path) -> Optional[ExtractResult]:
         """Extract the carved out chunk. Raises ExtractError on failure."""
 ```
 
@@ -538,7 +538,7 @@ class DirectoryExtractor(abc.ABC):
         return []
 
     @abc.abstractmethod
-    def extract(self, paths: List[Path], outdir: Path):
+    def extract(self, paths: List[Path], outdir: Path) -> Optional[ExtractResult]:
         """Extract from a multi file path list.
 
         Raises ExtractError on failure.

unblob/handlers/archive/_safe_tarfile.py

@@ -2,28 +2,21 @@
 import tarfile
 from pathlib import Path
 
-import attrs
 from structlog import get_logger
 
 from unblob.extractor import is_safe_path
+from unblob.report import ExtractionProblem
 
 logger = get_logger()
 
 RUNNING_AS_ROOT = os.getuid() == 0
 MAX_PATH_LEN = 255
 
 
-@attrs.define
-class ProblematicTarMember:
-    tarinfo: tarfile.TarInfo
-    problem: str
-    resolution: str
-
-
 class SafeTarFile:
     def __init__(self, inpath: Path):
         self.inpath = inpath
-        self.problems = []
+        self.reports = []
         self.tarfile = tarfile.open(inpath)
         self.directories = {}
 
@@ -143,4 +136,10 @@ def fix_directories(self, extract_root):
 
     def record_problem(self, tarinfo, problem, resolution):
         logger.warning(f"{problem} {resolution}", path=tarinfo.name)  # noqa: G004
-        self.problems.append(ProblematicTarMember(tarinfo, problem, resolution))
+        self.reports.append(
+            ExtractionProblem(
+                path=tarinfo.name,
+                problem=problem,
+                resolution=resolution,
+            )
+        )

unblob/handlers/archive/tar.py

@@ -7,7 +7,14 @@
 from structlog import get_logger
 
 from ...file_utils import OffsetFile, SeekError, decode_int, round_up, snull
-from ...models import Extractor, File, HexString, StructHandler, ValidChunk
+from ...models import (
+    Extractor,
+    ExtractResult,
+    File,
+    HexString,
+    StructHandler,
+    ValidChunk,
+)
 from ._safe_tarfile import SafeTarFile
 
 logger = get_logger()
@@ -88,6 +95,7 @@ class TarExtractor(Extractor):
     def extract(self, inpath: Path, outdir: Path):
         with contextlib.closing(SafeTarFile(inpath)) as tarfile:
             tarfile.extractall(outdir)
+        return ExtractResult(reports=tarfile.reports)
 
 
 class TarHandler(StructHandler):

unblob/models.py

@@ -103,7 +103,7 @@ class ValidChunk(Chunk):
     handler: "Handler" = attr.ib(init=False, eq=False)
     is_encrypted: bool = attr.ib(default=False)
 
-    def extract(self, inpath: Path, outdir: Path):
+    def extract(self, inpath: Path, outdir: Path) -> Optional["ExtractResult"]:
         if self.is_encrypted:
             logger.warning(
                 "Encrypted file is not extracted",
@@ -112,7 +112,7 @@ def extract(self, inpath: Path, outdir: Path):
             )
             raise ExtractError
 
-        self.handler.extract(inpath, outdir)
+        return self.handler.extract(inpath, outdir)
 
     def as_report(self, extraction_reports: List[Report]) -> ChunkReport:
         return ChunkReport(
@@ -154,8 +154,8 @@ class MultiFile(Blob):
 
     handler: "DirectoryHandler" = attr.ib(init=False, eq=False)
 
-    def extract(self, outdir: Path):
-        self.handler.extract(self.paths, outdir)
+    def extract(self, outdir: Path) -> Optional["ExtractResult"]:
+        return self.handler.extract(self.paths, outdir)
 
     def as_report(self, extraction_reports: List[Report]) -> MultiFileReport:
         return MultiFileReport(
@@ -253,13 +253,18 @@ def __init__(self, *reports: Report):
         self.reports: Tuple[Report, ...] = reports
 
 
+@attr.define(kw_only=True)
+class ExtractResult:
+    reports: List[Report]
+
+
 class Extractor(abc.ABC):
     def get_dependencies(self) -> List[str]:
         """Return the external command dependencies."""
         return []
 
     @abc.abstractmethod
-    def extract(self, inpath: Path, outdir: Path):
+    def extract(self, inpath: Path, outdir: Path) -> Optional[ExtractResult]:
         """Extract the carved out chunk.
 
         Raises ExtractError on failure.
@@ -272,7 +277,7 @@ def get_dependencies(self) -> List[str]:
         return []
 
     @abc.abstractmethod
-    def extract(self, paths: List[Path], outdir: Path):
+    def extract(self, paths: List[Path], outdir: Path) -> Optional[ExtractResult]:
         """Extract from a multi file path list.
 
         Raises ExtractError on failure.
@@ -381,15 +386,15 @@ def get_dependencies(cls):
     def calculate_multifile(self, file: Path) -> Optional[MultiFile]:
         """Calculate the MultiFile in a directory, using a file matched by the pattern as a starting point."""
 
-    def extract(self, paths: List[Path], outdir: Path):
+    def extract(self, paths: List[Path], outdir: Path) -> Optional[ExtractResult]:
         if self.EXTRACTOR is None:
             logger.debug("Skipping file: no extractor.", paths=paths)
             raise ExtractError
 
         # We only extract every blob once, it's a mistake to extract the same blob again
         outdir.mkdir(parents=True, exist_ok=False)
 
-        self.EXTRACTOR.extract(paths, outdir)
+        return self.EXTRACTOR.extract(paths, outdir)
 
 
 class Handler(abc.ABC):
@@ -414,15 +419,15 @@ def get_dependencies(cls):
     def calculate_chunk(self, file: File, start_offset: int) -> Optional[ValidChunk]:
         """Calculate the Chunk offsets from the File and the file type headers."""
 
-    def extract(self, inpath: Path, outdir: Path):
+    def extract(self, inpath: Path, outdir: Path) -> Optional[ExtractResult]:
         if self.EXTRACTOR is None:
             logger.debug("Skipping file: no extractor.", path=inpath)
             raise ExtractError
 
         # We only extract every blob once, it's a mistake to extract the same blob again
         outdir.mkdir(parents=True, exist_ok=False)
 
-        self.EXTRACTOR.extract(inpath, outdir)
+        return self.EXTRACTOR.extract(inpath, outdir)
 
 
 class StructHandler(Handler):

unblob/processing.py

@@ -403,7 +403,8 @@ def _extract_multi_file(self, multi_file: MultiFile) -> Path:
 
         extraction_reports = []
         try:
-            multi_file.extract(extract_dir)
+            if result := multi_file.extract(extract_dir):
+                extraction_reports.extend(result.reports)
         except ExtractError as e:
             extraction_reports.extend(e.reports)
         except Exception as exc:
@@ -522,7 +523,7 @@ def _calculate_entropy(self, path: Path) -> Optional[EntropyReport]:
             return report
         return None
 
-    def _extract_chunk(self, file, chunk: ValidChunk):
+    def _extract_chunk(self, file, chunk: ValidChunk):  # noqa: C901
         skip_carving = chunk.is_whole_file
         if skip_carving:
             inpath = self.task.path
@@ -554,7 +555,8 @@ def _extract_chunk(self, file, chunk: ValidChunk):
 
         extraction_reports = []
         try:
-            chunk.extract(inpath, extract_dir)
+            if result := chunk.extract(inpath, extract_dir):
+                extraction_reports.extend(result.reports)
 
             if carved_path and not self.config.keep_extracted_chunks:
                 logger.debug("Removing extracted chunk", path=carved_path)

unblob/report.py

@@ -244,3 +244,23 @@ class MultiFileReport(Report):
     name: str
     paths: List[Path]
     extraction_reports: List[Report]
+
+
+@attr.define(kw_only=True, frozen=True)
+class ExtractionProblem(Report):
+    """A non-fatal problem discovered during extraction.
+
+    A report like this still means, that the extraction was successful,
+    but there were problems that got resolved.
+    The output is expected to be complete, with the exception of
+    the reported path.
+
+    Examples
+    --------
+    - duplicate entries for certain archive formats (tar, zip)
+    - unsafe symlinks pointing outside of extraction directory
+    """
+
+    problem: str
+    resolution: str
+    path: Optional[str] = None