feat(handler): add multi-volume sevenzip directory handler

Author: martonilles

tests/integration/archive/sevenzip/__input__/multi-volume.tgz

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:c17f34d380545b1f139b52486f48b1852a9e74c2079a8e0338b0b7600a720fd6
+size 10240

tests/integration/archive/sevenzip/__output__/multi-volume.tgz_extract/mv.7z.001

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:a89ae76a6a3624af5eef2bbebe3ac0ac9916d66f3a65b055a4931f28065fd55e
+size 100

tests/integration/archive/sevenzip/__output__/multi-volume.tgz_extract/mv.7z.002

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:f3af8f79806dc5059ce0c746906f6d4c1f4e0206abd5e1742f8a8215bf6ebae0
+size 81

tests/integration/archive/sevenzip/__output__/multi-volume.tgz_extract/mv.7z_extract/file1

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:47d741b6059c6d7e99be25ce46fb9ba099cfd6515de1ef7681f93479d25996a4
+size 9

tests/integration/archive/sevenzip/__output__/multi-volume.tgz_extract/mv.7z_extract/file2

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:e0763097d2327a89fb7fc6a1fad40f87d2261dcdd6c09e65ee00b200a0128e1c
+size 9

unblob/handlers/__init__.py

@@ -100,4 +100,4 @@
     engenius.EngeniusHandler,
 )
 
-BUILTIN_DIR_HANDLERS: DirectoryHandlers = ()
+BUILTIN_DIR_HANDLERS: DirectoryHandlers = (sevenzip.MultiVolumeSevenZipHandler,)

unblob/handlers/archive/sevenzip.py

@@ -18,20 +18,62 @@
 https://py7zr.readthedocs.io/en/latest/archive_format.html
 """
 import binascii
+from pathlib import Path
 from typing import Optional
 
 from structlog import get_logger
 
 from unblob.extractors import Command
 
-from ...models import File, HexString, StructHandler, ValidChunk
+from ...extractors.command import MultiFileCommand
+from ...file_utils import Endian, InvalidInputFormat, StructParser
+from ...models import (
+    DirectoryHandler,
+    File,
+    Glob,
+    HexString,
+    MultiFile,
+    StructHandler,
+    ValidChunk,
+)
 
 logger = get_logger()
 
+C_DEFINITIONS = r"""
+    typedef struct sevenzip_header {
+        char magic[6];
+        uint8 version_maj;
+        uint8 version_min;
+        uint32 crc;
+        uint64 next_header_offset;
+        uint64 next_header_size;
+        uint32 next_header_crc;
+    } sevenzip_header_t;
+"""
+HEADER_STRUCT = "sevenzip_header_t"
+HEADER_SIZE = 6 + 1 + 1 + 4 + 8 + 8 + 4
+
+HEADER_PARSER = StructParser(C_DEFINITIONS)
+
 # StartHeader (next_header_offset, next_header_size, next_header_crc)
 START_HEADER_SIZE = 8 + 8 + 4
 
 
+SEVENZIP_MAGIC = b"7z\xbc\xaf\x27\x1c"
+
+
+def check_header_crc(header):
+    # CRC includes the StartHeader (next_header_offset, next_header_size, next_header_crc)
+    # CPP/7zip/Archive/7z/7zOut.cpp COutArchive::WriteStartHeader
+    calculated_crc = binascii.crc32(header.dumps()[-START_HEADER_SIZE:])
+    if header.crc != calculated_crc:
+        raise InvalidInputFormat("Invalid sevenzip header CRC")
+
+
+def calculate_sevenzip_size(header) -> int:
+    return len(header) + header.next_header_offset + header.next_header_size
+
+
 class SevenZipHandler(StructHandler):
     NAME = "sevenzip"
 
@@ -43,31 +85,48 @@ class SevenZipHandler(StructHandler):
         """
         )
     ]
-    C_DEFINITIONS = r"""
-        typedef struct sevenzip_header {
-            char magic[6];
-            uint8 version_maj;
-            uint8 version_min;
-            uint32 crc;
-            uint64 next_header_offset;
-            uint64 next_header_size;
-            uint32 next_header_crc;
-        } sevenzip_header_t;
-    """
-    HEADER_STRUCT = "sevenzip_header_t"
+    C_DEFINITIONS = C_DEFINITIONS
+    HEADER_STRUCT = HEADER_STRUCT
     EXTRACTOR = Command("7z", "x", "-p", "-y", "{inpath}", "-o{outdir}")
 
     def calculate_chunk(self, file: File, start_offset: int) -> Optional[ValidChunk]:
         header = self.parse_header(file)
 
-        # CRC includes the StartHeader (next_header_offset, next_header_size, next_header_crc)
-        # CPP/7zip/Archive/7z/7zOut.cpp COutArchive::WriteStartHeader
-        calculated_crc = binascii.crc32(header.dumps()[-START_HEADER_SIZE:])
-        if header.crc != calculated_crc:
-            logger.debug("Invalid header CRC", _verbosity=2)
+        check_header_crc(header)
+
+        size = calculate_sevenzip_size(header)
+
+        return ValidChunk(start_offset=start_offset, end_offset=start_offset + size)
+
+
+class MultiVolumeSevenZipHandler(DirectoryHandler):
+    NAME = "multi-sevenzip"
+    EXTRACTOR = MultiFileCommand("7z", "x", "-p", "-y", "{inpath}", "-o{outdir}")
+
+    PATTERN = Glob("*.7z.001")
+
+    def calculate_multifile(self, file: Path) -> Optional[MultiFile]:
+        with file.open("rb") as f:
+            header_data = f.read(HEADER_SIZE)
+
+        header = HEADER_PARSER.parse(HEADER_STRUCT, header_data, Endian.LITTLE)
+        if header.magic != SEVENZIP_MAGIC:
             return None
 
-        # We read the signature header here to get the offset to the header database
-        first_db_header = start_offset + len(header) + header.next_header_offset
-        end_offset = first_db_header + header.next_header_size
-        return ValidChunk(start_offset=start_offset, end_offset=end_offset)
+        check_header_crc(header)
+        size = calculate_sevenzip_size(header)
+        logger.debug("Sevenzip header", header=header, size=size, _verbosity=3)
+
+        paths = sorted(file.parent.glob(f"{file.stem}.*"))
+
+        files_size = sum(path.stat().st_size for path in paths)
+        logger.debug(
+            "Multi-volume files", paths=paths, files_size=files_size, _verbosity=2
+        )
+        if files_size != size:
+            return None
+
+        return MultiFile(
+            name=file.stem,
+            paths=paths,
+        )