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,22 +9,94 @@
 """
 
 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
 
+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):
     """
     Extract the archive into a folder.
     """
     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 +108,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 +148,58 @@ 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}")
 
 
+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$"),
+    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

@@ -8,11 +8,18 @@
 from pathlib import Path
 from re import Pattern, compile
 from sebaubuntu_libs.liblogging import LOGI
-from shutil import move
+from shutil import move, which
 from subprocess import STDOUT, check_output
 
 from dumpyara.lib.libpayload import extract_android_ota_payload
 
+SIMG2IMG_EXECUTABLE = which("simg2img") or "simg2img"
+
+try:
+    import firmware_parsers
+except ImportError:
+    firmware_parsers = None
+
 
 def extract_payload(image: Path, output_dir: Path):
     extract_android_ota_payload(image, output_dir)
@@ -22,9 +29,12 @@ 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

@@ -133,18 +133,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

dumpyara/utils/sparsed_images.py

@@ -9,6 +9,11 @@
 
 from dumpyara.utils.partitions import get_partition_names_with_alias
 
+try:
+    import firmware_parsers
+except ImportError:
+    firmware_parsers = None
+
 
 def prepare_sparsed_images(files_path: Path):
     """
@@ -31,6 +36,12 @@ def prepare_sparsed_images(files_path: Path):
         if sparsechunk_image_files:
             LOGI(f"Preparing sparsechunk images for {partition}")
             LOGI(f"Converting {sparsechunk_image_files[0]} to {output_image.name}")
-            check_output(["simg2img", *sparsechunk_image_files, output_image], stderr=STDOUT)
+            if firmware_parsers is not None:
+                firmware_parsers.sparse_chunks_to_raw(
+                    [str(file) for file in sparsechunk_image_files],
+                    str(output_image),
+                )
+            else:
+                check_output(["simg2img", *sparsechunk_image_files, output_image], stderr=STDOUT)
             for sparsechunk_image_file in sparsechunk_image_files:
                 sparsechunk_image_file.unlink()