Add support for everything Firmware_extractor supported

dumpyara/dumpyara.py

@@ -3,6 +3,7 @@
 # SPDX-License-Identifier: GPL-3.0-or-later
 #
 
+from importlib.util import find_spec
 from pathlib import Path
 from sebaubuntu_libs.liblogging import LOGI
 from sebaubuntu_libs.libreorder import strcoll_files_key
@@ -14,13 +15,17 @@
 from dumpyara.steps.extract_images import extract_images
 from dumpyara.steps.prepare_images import prepare_images
 
+_HAS_FIRMWARE_PARSERS = find_spec("firmware_parsers") is not None
+
 # Package name to package commands
 REQUIRED_TOOLS = {
     "7-zip or p7zip": [SEVEN_ZIP_EXECUTABLE, P7ZIP_EXECUTABLE],
     "erofs-utils": ["fsck.erofs"],
-    "android-sdk-libsparse-utils or platform-utils": ["simg2img"],
 }
 
+if not _HAS_FIRMWARE_PARSERS:
+    REQUIRED_TOOLS["android-sdk-libsparse-utils or platform-utils"] = ["simg2img"]
+
 
 def dumpyara(file: Path, output_path: Path, debug: bool = False):
     """Dump an Android firmware."""
@@ -68,7 +73,7 @@ def dumpyara(file: Path, output_path: Path, debug: bool = False):
         # Create all_files.txt
         LOGI("Creating all_files.txt")
         (output_path / "all_files.txt").write_text(
-            "\n".join([str(file) for file in files_list]) + "\n"
+            "\n".join([str(file) for file in files_list]) + "\n", encoding="utf-8"
         )
 
         return output_path

dumpyara/steps/extract_archive.py

@@ -9,12 +9,61 @@
 """
 
 from pathlib import Path
+import re
 from re import Pattern, compile
 from shutil import unpack_archive
 from sebaubuntu_libs.liblogging import LOGD, LOGI
 from typing import Callable, Dict
+from zipfile import ZipFile, is_zipfile
 
 from dumpyara.utils.files import get_recursive_files_list
+from dumpyara.utils.partitions import get_partition_names_with_alias
+
+try:
+    import firmware_parsers
+except ImportError:
+    firmware_parsers = None
+
+
+def _strip_vendor_prefix(directory: Path):
+    """Strip a shared vendor prefix from extracted filenames."""
+    files = [file for file in directory.iterdir() if file.is_file()]
+    if len(files) < 3:
+        return
+
+    prefix_pattern = re.compile(r"^[A-Z]{2,4}(?:-[A-Za-z0-9]{1,6}){2,4}-")
+    prefixed = {}
+    for file in files:
+        match = prefix_pattern.match(file.name)
+        if match:
+            new_name = file.name[match.end() :]
+            if new_name and not (directory / new_name).exists():
+                prefixed[file] = directory / new_name
+
+    # Avoid false positives by requiring a clear majority.
+    if len(prefixed) >= len(files) * 0.6:
+        for old, new in prefixed.items():
+            LOGD(f"Stripping vendor prefix: {old.name} → {new.name}")
+            old.rename(new)
+
+
+def _has_nested_partition_markers(archive_path: Path) -> bool:
+    """Return True when a nested zip contains dumpable partition container markers."""
+    if not is_zipfile(archive_path):
+        LOGD(f"Skipping nested zip scan for non-zip archive: {archive_path.name}")
+        return False
+
+    try:
+        with ZipFile(archive_path, "r") as zip_file:
+            for file_name in zip_file.namelist():
+                for pattern in NESTED_ZIP_PARTITION_MARKERS:
+                    if pattern.search(file_name):
+                        return True
+    except Exception as e:
+        LOGD(f"Failed to inspect nested zip {archive_path.name}: {e}")
+        return False
+
+    return False
 
 
 def extract_archive(archive_path: Path, extracted_archive_path: Path, is_nested: bool = False):
@@ -23,8 +72,32 @@ def extract_archive(archive_path: Path, extracted_archive_path: Path, is_nested:
     """
     LOGD(f"Extracting archive: {archive_path.name}")
 
+    # Try firmware_parsers detection first
+    if firmware_parsers is not None:
+        try:
+            firmware_format = firmware_parsers.detect(str(archive_path))
+            if firmware_format != "unknown":
+                extractor = getattr(firmware_parsers, firmware_format, None)
+                if extractor is not None:
+                    LOGI(f"Detected firmware format: {firmware_format}")
+                    extractor(str(archive_path), str(extracted_archive_path))
+                    if is_nested:
+                        archive_path.unlink()
+                    return
+        except Exception as error:
+            LOGI(f"firmware_parsers failed ({error}), falling back to generic extraction")
+
     # Extract the archive
-    unpack_archive(archive_path, extracted_archive_path)
+    try:
+        unpack_archive(archive_path, extracted_archive_path)
+    except Exception:
+        # Handle zip archives with non-standard extensions such as .ozip and .ftf.
+        if not is_zipfile(archive_path):
+            raise
+        LOGD(f"Falling back to zipfile for {archive_path.name}")
+        with ZipFile(archive_path, "r") as archive:
+            archive.extractall(extracted_archive_path)
+
     if is_nested:
         LOGD("Archive is nested, unlinking")
         archive_path.unlink()
@@ -36,6 +109,22 @@ def extract_archive(archive_path: Path, extracted_archive_path: Path, is_nested:
 
         file.rename(extracted_archive_path / file.name)
 
+    # Re-detect firmware formats in extracted files
+    if firmware_parsers is not None:
+        for file in list(get_recursive_files_list(extracted_archive_path)):
+            try:
+                firmware_format = firmware_parsers.detect(str(file))
+                if firmware_format != "unknown":
+                    extractor = getattr(firmware_parsers, firmware_format, None)
+                    if extractor is not None:
+                        LOGI(f"Detected nested firmware format: {firmware_format} in {file.name}")
+                        extractor(str(file), str(extracted_archive_path))
+                        file.unlink()
+            except Exception as error:
+                LOGD(f"firmware_parsers failed on {file.name}: {error}")
+
+    _strip_vendor_prefix(extracted_archive_path)
+
     # Check for nested archives
     extracted_archive_tempdir_files_list = list(
         get_recursive_files_list(extracted_archive_path, True)
@@ -60,9 +149,75 @@ def extract_archive(archive_path: Path, extracted_archive_path: Path, is_nested:
 
             func(nested_archive, extracted_archive_path, True)
 
+    nested_archive_patterns = tuple(NESTED_ARCHIVES.keys())
+    for file in extracted_archive_tempdir_files_list:
+        if any(pattern.match(str(file)) for pattern in nested_archive_patterns):
+            continue
+
+        if not NESTED_ZIP_PATTERN.match(str(file)):
+            continue
+
+        nested_archive = extracted_archive_path / file
+        LOGI(f"Found nested zip candidate: {nested_archive.name}")
+
+        if not nested_archive.is_file():
+            LOGD(f"Nested zip {nested_archive.name} probably already handled, skipping")
+            continue
+
+        if not _has_nested_partition_markers(nested_archive):
+            LOGD(f"Skipping nested zip {nested_archive.name}: no partition markers")
+            continue
+
+        extract_archive(nested_archive, extracted_archive_path, True)
+
     LOGD(f"Extracted archive: {archive_path.name}")
 
 
+# Partition names, longest first so that e.g. "system_ext" wins over "system"
+# when the alternation is applied to a filename.
+_NESTED_ZIP_PARTITION_NAMES = "|".join(
+    re.escape(name) for name in sorted(get_partition_names_with_alias(), key=len, reverse=True)
+)
+
+NESTED_ZIP_PARTITION_MARKERS = (
+    compile(
+        r"(?:^|/)"
+        r"(?:boot|boot-debug|boot-verified|cust|dtbo|dtbo-verified|exaid|factory|india|"
+        r"init_boot|mi_ext|modem|my_bigball|my_carrier|my_company|my_country|my_custom|"
+        r"my_engineering|my_heytap|my_manifest|my_odm|my_operator|my_preload|my_product|"
+        r"my_region|my_stock|my_version|NON-HLOS|odm|odm_dlkm|odm_ext|oem|oppo_product|"
+        r"opproduct|preas|preavs|preload|preload_common|product|product_h|recovery|rescue|"
+        r"reserve|special_preload|super|system|system_dlkm|system_ext|system_other|"
+        r"systemex|tz|vendor|vendor_boot|vendor_boot-debug|vendor_dlkm|"
+        r"vendor_kernel_boot|xrom)(?:_[ab])?\.new\.dat\.br$"
+    ),
+    compile(
+        r"(?:^|/)"
+        r"(?:boot|boot-debug|boot-verified|cust|dtbo|dtbo-verified|exaid|factory|india|"
+        r"init_boot|mi_ext|modem|my_bigball|my_carrier|my_company|my_country|my_custom|"
+        r"my_engineering|my_heytap|my_manifest|my_odm|my_operator|my_preload|my_product|"
+        r"my_region|my_stock|my_version|NON-HLOS|odm|odm_dlkm|odm_ext|oem|oppo_product|"
+        r"opproduct|preas|preavs|preload|preload_common|product|product_h|recovery|rescue|"
+        r"reserve|special_preload|super|system|system_dlkm|system_ext|system_other|"
+        r"systemex|tz|vendor|vendor_boot|vendor_boot-debug|vendor_dlkm|"
+        r"vendor_kernel_boot|xrom)(?:_[ab])?\.transfer\.list$"
+    ),
+    compile(r"(?:^|/)payload\.bin$"),
+    compile(r"(?:^|/)super(?!.*(_empty)).*\.img$"),
+    # Loose raw partition images, e.g. Pixel factory images ship system.img /
+    # vendor.img / product.img directly inside the nested image-*.zip with no
+    # super.img or payload.bin. Anchor on known partition names + optional A/B
+    # slot + the image suffixes get_raw_image() understands so an unrelated
+    # stray *.img elsewhere doesn't flag a zip as dumpable. super_empty.img is
+    # excluded structurally: "super" only matches when followed by a slot suffix
+    # or extension, never "_empty".
+    compile(
+        rf"(?:^|/)(?:{_NESTED_ZIP_PARTITION_NAMES})(?:_[ab])?"
+        r"(?:\.(?:bin|ext4|image|mbn)|\.img(?:\.ext4|\.lz4)?|\.raw(?:\.img)?)$"
+    ),
+    compile(r"(?:^|/)[^/]+\.tar\.md5$"),
+)
+NESTED_ZIP_PATTERN = compile(r".*\.zip$")
 NESTED_ARCHIVES: Dict[Pattern[str], Callable[[Path, Path, bool], None]] = {
     compile(key): value
     for key, value in {

dumpyara/utils/multipartitions.py

@@ -3,28 +3,60 @@
 # SPDX-License-Identifier: GPL-3.0-or-later
 #
 
+from tempfile import mkdtemp
 from typing import Callable, Dict
 from liblp.partition_tools.lpunpack import lpunpack
 from pathlib import Path
 from re import Pattern, compile
 from sebaubuntu_libs.liblogging import LOGI
-from shutil import move
-from subprocess import STDOUT, check_output
+from shutil import move, rmtree, which
+from subprocess import STDOUT, check_output, run
 
 from dumpyara.lib.libpayload import extract_android_ota_payload
 
+SIMG2IMG_EXECUTABLE = which("simg2img") or "simg2img"
+OTADUMP_EXECUTABLE = which("otadump")
+
+try:
+    import firmware_parsers
+except ImportError:
+    firmware_parsers = None
+
+
+def _extract_payload_otadump(image: Path, output_dir: Path, otadump_bin: str):
+    # Stage into a sibling tempdir so a partial failure does not leave half-written
+    # images in output_dir alongside the real payload.bin. Move results across on success.
+    staging = Path(mkdtemp(prefix=".otadump-", dir=output_dir))
+    try:
+        run(  # nosec B603
+            [otadump_bin, "--output-dir", str(staging), str(image)],
+            check=True,
+        )
+        for img in staging.iterdir():
+            move(str(img), str(output_dir / img.name))
+    finally:
+        rmtree(staging, ignore_errors=True)
+
 
 def extract_payload(image: Path, output_dir: Path):
-    extract_android_ota_payload(image, output_dir)
+    if OTADUMP_EXECUTABLE:
+        LOGI(f"Extracting {image.name} with otadump ({OTADUMP_EXECUTABLE})")
+        _extract_payload_otadump(image, output_dir, OTADUMP_EXECUTABLE)
+    else:
+        LOGI(f"Extracting {image.name} with vendored Python payload parser")
+        extract_android_ota_payload(image, output_dir)
 
 
 def extract_super(image: Path, output_dir: Path):
     unsparsed_super = output_dir / "super.unsparsed.img"
 
     try:
-        check_output(
-            ["simg2img", image, unsparsed_super], stderr=STDOUT
-        )  # TODO: Rewrite libsparse...
+        if firmware_parsers is not None:
+            firmware_parsers.sparse_to_raw(str(image), str(unsparsed_super))
+        else:
+            check_output(  # nosec B603
+                [SIMG2IMG_EXECUTABLE, image, unsparsed_super], stderr=STDOUT
+            )
     except Exception:
         LOGI(f"Failed to unsparse {image.name}")
     else:

dumpyara/utils/partitions.py

@@ -55,6 +55,7 @@
     "factory": FILESYSTEM,
     "india": FILESYSTEM,
     "mi_ext": FILESYSTEM,
+    "mi_product": FILESYSTEM,
     "modem": FILESYSTEM,
     "my_bigball": FILESYSTEM,
     "my_carrier": FILESYSTEM,
@@ -98,12 +99,12 @@ def get_partition_name(partition_name: str):
     return ALTERNATIVE_PARTITION_NAMES.get(partition_name, partition_name)
 
 
-def get_partition_names():
+def get_partition_names() -> List[str]:
     """Get a list of partition names."""
     return list(PARTITIONS)
 
 
-def get_partition_names_with_alias():
+def get_partition_names_with_alias() -> List[str]:
     """Get a list of partition names with alias."""
     return get_partition_names() + list(ALTERNATIVE_PARTITION_NAMES)
 
@@ -133,18 +134,23 @@ def prepare_raw_images(files_path: Path, raw_images_path: Path):
 def fix_aliases(images_path: Path):
     """Move aliased partitions to their generic name."""
     for alt_name, name in ALTERNATIVE_PARTITION_NAMES.items():
-        alt_path = images_path / f"{alt_name}.img"
+        alt_paths = [
+            images_path / f"{alt_name}.img",
+            images_path / f"{alt_name}.bin",
+        ]
         partition_path = images_path / f"{name}.img"
 
-        if not alt_path.exists():
-            continue
+        for alt_path in alt_paths:
+            if not alt_path.exists():
+                continue
 
-        if partition_path.exists():
-            LOGI(f"Ignoring {alt_name} ({name} already extracted)")
-            alt_path.unlink()
+            if partition_path.exists():
+                LOGI(f"Ignoring {alt_name} ({name} already extracted)")
+                alt_path.unlink()
+                continue
 
-        LOGI(f"Fixing alias {alt_name} -> {name}")
-        move(alt_path, partition_path)
+            LOGI(f"Fixing alias {alt_name} -> {name}")
+            move(alt_path, partition_path)
 
 
 def get_filename_suffixes(file: Path):

dumpyara/utils/raw_image.py

@@ -9,9 +9,16 @@
 from lz4.frame import LZ4FrameFile
 from pathlib import Path
 from sebaubuntu_libs.liblogging import LOGD, LOGI
-from shutil import copyfile, move
+from shutil import copyfile, move, which
 from subprocess import STDOUT, check_output
 
+SIMG2IMG_EXECUTABLE = which("simg2img") or "simg2img"
+
+try:
+    import firmware_parsers
+except ImportError:
+    firmware_parsers = None
+
 
 def get_raw_image(partition: str, files_path: Path, output_image_path: Path):
     """
@@ -59,9 +66,12 @@ def get_raw_image(partition: str, files_path: Path, output_image_path: Path):
             continue
 
         try:
-            check_output(
-                ["simg2img", image_path, unsparsed_image], stderr=STDOUT
-            )  # TODO: Rewrite libsparse...
+            if firmware_parsers is not None:
+                firmware_parsers.sparse_to_raw(str(image_path), str(unsparsed_image))
+            else:
+                check_output(  # nosec B603
+                    [SIMG2IMG_EXECUTABLE, image_path, unsparsed_image], stderr=STDOUT
+                )
         except Exception:
             LOGD(f"Failed to unsparse {image_path.name}, should be a raw image")
             pass