feat(hander): add par2 directory handler

Author: rxpha3l

python/unblob/handlers/__init__.py

@@ -6,6 +6,7 @@
     cab,
     cpio,
     dmg,
+    par2,
     rar,
     sevenzip,
     stuffit,
@@ -54,6 +55,7 @@
 BUILTIN_HANDLERS: Handlers = (
     cramfs.CramFSHandler,
     extfs.EXTHandler,
+    erofs.EROFSHandler,
     fat.FATHandler,
     jffs2.JFFS2NewHandler,
     jffs2.JFFS2OldHandler,
@@ -124,4 +126,5 @@
 BUILTIN_DIR_HANDLERS: DirectoryHandlers = (
     sevenzip.MultiVolumeSevenZipHandler,
     gzip.MultiVolumeGzipHandler,
+    par2.MultiVolumePAR2Handler,
 )

python/unblob/handlers/archive/par2.py

@@ -0,0 +1,63 @@
+import hashlib
+import io
+from pathlib import Path
+from typing import Optional
+
+from unblob.file_utils import Endian, StructParser
+from unblob.models import (
+    DirectoryHandler,
+    Glob,
+    MultiFile,
+)
+
+C_DEFINITIONS = r"""
+    typedef struct par2_header{
+        char magic[8];
+        uint64 packet_length;
+        char md5_hash[16];
+        char recovery_set_id[16];
+        char type[16];
+    } par2_header_t;
+"""
+
+PAR2_MAGIC = b"PAR2\x00PKT"
+HEADER_STRUCT = "par2_header_t"
+HEADER_PARSER = StructParser(C_DEFINITIONS)
+
+
+class MultiVolumePAR2Handler(DirectoryHandler):
+    NAME = "multi-par2"
+    PATTERN = Glob("*.par2")
+    EXTRACTOR = None
+
+    def is_valid_header(self, file_paths: list) -> bool:
+        for path in file_paths:
+            with path.open("rb") as f:
+                header = HEADER_PARSER.parse(HEADER_STRUCT, f, Endian.LITTLE)
+                if header.magic != PAR2_MAGIC:
+                    return False
+
+                offset_to_recovery_id = 32
+                # seek to beginning of recovery set ID
+                f.seek(offset_to_recovery_id, io.SEEK_SET)
+                packet_content = f.read(
+                    header.packet_length - len(header) + offset_to_recovery_id
+                )
+                packet_checksum = hashlib.md5(packet_content).digest()  # noqa: S324
+
+                if packet_checksum != header.md5_hash:
+                    return False
+        return True
+
+    def calculate_multifile(self, file: Path) -> Optional[MultiFile]:
+        paths = sorted(
+            [p for p in file.parent.glob(f"{file.stem}.*") if p.resolve().exists()]
+        )
+
+        if len(paths) <= 1 or not self.is_valid_header(paths):
+            return None
+
+        return MultiFile(
+            name=file.stem,
+            paths=paths,
+        )

tests/integration/archive/par2/__input__/foo.erofs.img.par2

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

tests/integration/archive/par2/__input__/foo.erofs.img.vol00+01.par2

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

tests/integration/archive/par2/__input__/foo.erofs.img.vol01+02.par2

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

tests/integration/archive/par2/__input__/foo.erofs.img.vol03+04.par2

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

tests/integration/archive/par2/__input__/foo.erofs.img.vol07+08.par2

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

tests/integration/archive/par2/__input__/foo.erofs.img.vol15+16.par2

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

tests/integration/archive/par2/__input__/foo.erofs.img.vol31+20.par2

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