diff --git a/.github/scripts/release/bundle-tools.py b/.github/scripts/release/bundle-tools.py index 8ae88223..23494035 100644 --- a/.github/scripts/release/bundle-tools.py +++ b/.github/scripts/release/bundle-tools.py @@ -56,25 +56,22 @@ def bundle_platform_tools(url: str) -> None: def bundle_avb_tools() -> None: - """Copy AVB tools from vendor/avb submodule into bin/tools/ for packaging.""" - avb_dir = REPO_ROOT / "vendor" / "avb" + """Copy AVB test keys from vendor/avbtool-rs into bin/tools/ for packaging.""" + keys_dir = REPO_ROOT / "vendor" / "avbtool-rs" / "src" / "keys" copy_map = { - avb_dir / "avbtool.py": TOOLS_DIR / "avbtool.py", - avb_dir / "test" / "data" / "testkey_rsa4096.pem": TOOLS_DIR - / "testkey_rsa4096.pem", - avb_dir / "test" / "data" / "testkey_rsa2048.pem": TOOLS_DIR - / "testkey_rsa2048.pem", + keys_dir / "testkey_rsa4096.pem": TOOLS_DIR / "testkey_rsa4096.pem", + keys_dir / "testkey_rsa2048.pem": TOOLS_DIR / "testkey_rsa2048.pem", } if all(dst.exists() for dst in copy_map.values()): - print("[bundle-tools] AVB tools already present, skipping.") + print("[bundle-tools] AVB test keys already present, skipping.") return for src, dst in copy_map.items(): if not src.exists(): raise RuntimeError( - f"vendor/avb submodule missing {src.relative_to(REPO_ROOT)}. " - f"Run: git submodule update --init vendor/avb" + f"vendor/avbtool-rs submodule missing {src.relative_to(REPO_ROOT)}. " + f"Run: git submodule update --init vendor/avbtool-rs" ) shutil.copy2(src, dst) print(f"[bundle-tools] Copied {src.name} -> {dst.relative_to(REPO_ROOT)}") diff --git a/.github/workflows/build-archive.yml b/.github/workflows/build-archive.yml index bf3c28b5..76e1e79e 100644 --- a/.github/workflows/build-archive.yml +++ b/.github/workflows/build-archive.yml @@ -45,6 +45,12 @@ jobs: name: qdl-rs-windows path: bin/tools/ + - name: Download avbtool-rs artifact + uses: actions/download-artifact@v8 + with: + name: avbtool-rs-windows + path: bin/tools/ + - name: Bundle base tools run: python .github/scripts/release/bundle-tools.py @@ -59,7 +65,7 @@ jobs: 'bin/tools/magiskboot_xz_helper.exe', 'bin/tools/adb.exe', 'bin/tools/fastboot.exe', - 'bin/tools/avbtool.py', + 'bin/tools/avbtool-rs.exe', 'bin/tools/kptools.exe', 'bin/tools/testkey_rsa4096.pem', 'bin/tools/testkey_rsa2048.pem', diff --git a/.github/workflows/build-avbtool-rs.yml b/.github/workflows/build-avbtool-rs.yml new file mode 100644 index 00000000..33898ff6 --- /dev/null +++ b/.github/workflows/build-avbtool-rs.yml @@ -0,0 +1,47 @@ +name: Build avbtool-rs + +on: + push: + paths: + - 'vendor/avbtool-rs/**' + - '.github/workflows/build-avbtool-rs.yml' + pull_request: + paths: + - 'vendor/avbtool-rs/**' + - '.github/workflows/build-avbtool-rs.yml' + workflow_call: + +jobs: + build-windows: + runs-on: windows-latest + timeout-minutes: 20 + defaults: + run: + shell: pwsh + steps: + - uses: actions/checkout@v6 + with: + submodules: true + + - uses: dtolnay/rust-toolchain@stable + with: + targets: x86_64-pc-windows-msvc + + - uses: Swatinem/rust-cache@v2 + with: + workspaces: vendor/avbtool-rs + cache-on-failure: true + + - name: Build avbtool-rs + working-directory: vendor/avbtool-rs + run: cargo build --release + + - name: Smoke test + run: vendor/avbtool-rs/target/release/avbtool-rs.exe version + + - name: Upload artifact + uses: actions/upload-artifact@v7 + with: + name: avbtool-rs-windows + path: vendor/avbtool-rs/target/release/avbtool-rs.exe + if-no-files-found: error diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml index 2e407364..1532b91d 100644 --- a/.github/workflows/release.yml +++ b/.github/workflows/release.yml @@ -20,6 +20,9 @@ jobs: build-qdlrs: uses: ./.github/workflows/build-qdlrs.yml + build-avbtool-rs: + uses: ./.github/workflows/build-avbtool-rs.yml + preflight: needs: test runs-on: windows-latest @@ -47,7 +50,7 @@ jobs: "- Config version matches tag ?" >> $env:GITHUB_STEP_SUMMARY build: - needs: [preflight, build-qdlrs] + needs: [preflight, build-qdlrs, build-avbtool-rs] uses: ./.github/workflows/build-archive.yml with: tag-name: ${{ needs.preflight.outputs.tag_name }} diff --git a/.github/workflows/windows_test.yml b/.github/workflows/windows_test.yml index e339d363..dc27c90e 100644 --- a/.github/workflows/windows_test.yml +++ b/.github/workflows/windows_test.yml @@ -18,6 +18,9 @@ jobs: with: task: architecture + build-avbtool-rs: + uses: ./.github/workflows/build-avbtool-rs.yml + build-tools: runs-on: windows-latest timeout-minutes: 15 @@ -59,11 +62,10 @@ jobs: path: | bin/tools/magiskboot.exe bin/tools/magiskboot_xz_helper.exe - bin/tools/openssl.exe bin/tools/*.dll integration-test: - needs: [unit-test, build-tools] + needs: [unit-test, build-tools, build-avbtool-rs] runs-on: windows-latest timeout-minutes: 30 env: @@ -96,6 +98,12 @@ jobs: name: tools-exe path: bin/tools + - name: Download avbtool-rs artifact + uses: actions/download-artifact@v8 + with: + name: avbtool-rs-windows + path: bin/tools + - name: Run Integration Tests run: | pytest -s tests/ -m integration --run-integration --cov=bin/ltbox --cov-report=term-missing diff --git a/.gitmodules b/.gitmodules index 7602edbd..c825f7f1 100644 --- a/.gitmodules +++ b/.gitmodules @@ -1,6 +1,6 @@ [submodule "vendor/qdlrs"] path = vendor/qdlrs - url = https://github.com/qualcomm/qdlrs.git + url = https://github.com/miner7222/qdlrs.git [submodule "vendor/avb"] path = vendor/avb url = https://android.googlesource.com/platform/external/avb @@ -8,3 +8,6 @@ [submodule "vendor/MagiskbootAlone"] path = vendor/MagiskbootAlone url = https://github.com/miner7222/MagiskbootAlone.git +[submodule "vendor/avbtool-rs"] + path = vendor/avbtool-rs + url = https://github.com/miner7222/avbtool-rs diff --git a/bin/ltbox/actions/root/workflow.py b/bin/ltbox/actions/root/workflow.py index 42e2f593..ee538096 100644 --- a/bin/ltbox/actions/root/workflow.py +++ b/bin/ltbox/actions/root/workflow.py @@ -625,8 +625,7 @@ def _sign_recovery_image( subprocess.run( [ - str(const.PYTHON_EXE), - str(const.AVBTOOL_PY), + str(const.AVBTOOL_RS), "erase_footer", "--image", str(final_twrp), diff --git a/bin/ltbox/constants.py b/bin/ltbox/constants.py index 56bf8b3d..69fafa21 100644 --- a/bin/ltbox/constants.py +++ b/bin/ltbox/constants.py @@ -61,8 +61,8 @@ def __init__(self): # --- Vendor Paths --- self.vendor_dir = self.base_dir / "vendor" - self.avb_dir = self.vendor_dir / "avb" - self.avb_testkeys_dir = self.avb_dir / "test" / "data" + self.avbtool_rs_dir = self.vendor_dir / "avbtool-rs" + self.avb_testkeys_dir = self.avbtool_rs_dir / "src" / "keys" # --- Executables --- self.python_exe = self.python_dir / "python.exe" @@ -70,9 +70,11 @@ def __init__(self): self.python_exe = Path(sys.executable) self.adb_exe = self.tools_dir / "adb.exe" self.fastboot_exe = self.tools_dir / "fastboot.exe" - self.avbtool_py = self.avb_dir / "avbtool.py" - if not self.avbtool_py.exists(): - self.avbtool_py = self.tools_dir / "avbtool.py" + self.avbtool_rs = self.tools_dir / "avbtool-rs.exe" + if not self.avbtool_rs.exists(): + self.avbtool_rs = ( + self.avbtool_rs_dir / "target" / "release" / "avbtool-rs.exe" + ) self.qdlrs_exe = self.tools_dir / "qdl-rs.exe" self.magiskboot_exe = self.tools_dir / "magiskboot.exe" @@ -248,13 +250,13 @@ def load_settings_raw() -> Dict[str, Any]: FN_VENDOR_BOOT_PRC = CONF.fn_vendor_boot_prc VENDOR_DIR = CONF.vendor_dir -AVB_DIR = CONF.avb_dir +AVBTOOL_RS_DIR = CONF.avbtool_rs_dir AVB_TESTKEYS_DIR = CONF.avb_testkeys_dir PYTHON_EXE = CONF.python_exe ADB_EXE = CONF.adb_exe FASTBOOT_EXE = CONF.fastboot_exe -AVBTOOL_PY = CONF.avbtool_py +AVBTOOL_RS = CONF.avbtool_rs QDLRS_EXE = CONF.qdlrs_exe MAGISKBOOT_EXE = CONF.magiskboot_exe diff --git a/bin/ltbox/info_scan.py b/bin/ltbox/info_scan.py index a28892c4..04ca4593 100644 --- a/bin/ltbox/info_scan.py +++ b/bin/ltbox/info_scan.py @@ -24,8 +24,7 @@ def collect_info_scan_files(paths: List[str]) -> List[Path]: def build_info_scan_command(image_path: Path, constants: Any) -> List[str]: return [ - str(constants.PYTHON_EXE), - str(constants.AVBTOOL_PY), + str(constants.AVBTOOL_RS), "info_image", "--image", str(image_path), diff --git a/bin/ltbox/patch/avb.py b/bin/ltbox/patch/avb.py index 612ceaa3..3e7fd2e5 100644 --- a/bin/ltbox/patch/avb.py +++ b/bin/ltbox/patch/avb.py @@ -1,14 +1,7 @@ -import hashlib -import importlib.util -import io +import json import shutil -import sys -import tempfile -import warnings -from dataclasses import dataclass -from functools import lru_cache +import subprocess from pathlib import Path -from types import ModuleType from typing import Any, Dict, List, Optional from .. import constants as const @@ -16,85 +9,11 @@ from ..i18n import get_string -@dataclass -class _ParsedAvbImage: - path: Path - partition_name: str - footer: Any - header: Any - descriptors: List[Any] - image_size: int - public_key: bytes - public_key_metadata: bytes - - -def _resolve_avbtool_source_path() -> Path: - candidates = [ - const.AVBTOOL_PY, - const.TOOLS_DIR / "avbtool.py", - ] - for candidate in candidates: - if candidate.exists(): - return candidate - raise FileNotFoundError( - "Unable to locate avbtool.py source for preserved vbmeta rebuild." - ) - - -def _resolve_avbtool_openssl_binary(source_path: Path) -> Optional[str]: - tool_dir = source_path.resolve().parent - candidates = ( - "avb_openssl", - "avb_openssl.exe", - "openssl", - "openssl.exe", - ) - for candidate in candidates: - candidate_path = tool_dir / candidate - if candidate_path.exists(): - return str(candidate_path) - return None - - -@lru_cache(maxsize=4) -def _load_avbtool_module(source_path: str) -> ModuleType: - module_name = f"_ltbox_avbtool_{abs(hash(source_path))}" - spec = importlib.util.spec_from_file_location(module_name, source_path) - if spec is None or spec.loader is None: - raise ImportError(f"Unable to load avbtool module from {source_path}") - module = importlib.util.module_from_spec(spec) - with warnings.catch_warnings(): - warnings.filterwarnings("ignore", category=DeprecationWarning) - warnings.filterwarnings("ignore", category=PendingDeprecationWarning) - spec.loader.exec_module(module) - return module - - -def _get_avbtool_module() -> ModuleType: - source_path = _resolve_avbtool_source_path() - module = _load_avbtool_module(str(source_path)) - openssl_binary = _resolve_avbtool_openssl_binary(source_path) - if openssl_binary: - setattr(module, "AVB_OPENSSL", openssl_binary) - mldsa_cls = getattr(module, "MLDSAPublicKey", None) - if mldsa_cls is not None: - mldsa_cls._IS_SUPPORTED = None - return module - - def _run_avbtool(*args: Any) -> str: - """Run an avbtool subcommand in-process and return captured stdout.""" - avb_module = _get_avbtool_module() - tool = avb_module.AvbTool() - str_args = [str(a) for a in args] - capture = io.StringIO() - original_stdout = sys.stdout - try: - sys.stdout = capture - tool.run(["avbtool"] + str_args) - finally: - sys.stdout = original_stdout - return capture.getvalue() + """Run an avbtool-rs subcommand and return captured stdout.""" + cmd = [str(const.AVBTOOL_RS)] + [str(a) for a in args] + result = subprocess.run(cmd, capture_output=True, text=True, check=True) + return result.stdout def _resolve_signing_key(pubkey_sha1: Optional[str], image_name: str) -> Optional[Path]: @@ -108,316 +27,38 @@ def _resolve_signing_key(pubkey_sha1: Optional[str], image_name: str) -> Optiona return key_file -def _close_image_handler(image_handler: Any) -> None: - image_file = getattr(image_handler, "_image", None) - if image_file is not None: - image_file.close() - - -def _parse_avb_image( - image_path: Path, - partition_name: Optional[str] = None, -) -> _ParsedAvbImage: - avb_module = _get_avbtool_module() - avb = avb_module.Avb() - image_handler = avb_module.ImageHandler(str(image_path), read_only=True) - - try: - footer, header, descriptors, image_size = avb._parse_image(image_handler) - vbmeta_offset = footer.vbmeta_offset if footer else 0 - aux_block_offset = ( - vbmeta_offset - + avb_module.AvbVBMetaHeader.SIZE - + header.authentication_data_block_size - ) - - public_key = b"" - if header.public_key_size: - image_handler.seek(aux_block_offset + header.public_key_offset) - public_key = image_handler.read(header.public_key_size) - - public_key_metadata = b"" - if header.public_key_metadata_size: - image_handler.seek(aux_block_offset + header.public_key_metadata_offset) - public_key_metadata = image_handler.read(header.public_key_metadata_size) - - resolved_partition_name = partition_name or image_path.stem - partition_descriptors = [ - descriptor - for descriptor in descriptors - if isinstance( - descriptor, - (avb_module.AvbHashDescriptor, avb_module.AvbHashtreeDescriptor), - ) - ] - if len(partition_descriptors) == 1: - resolved_partition_name = partition_descriptors[0].partition_name - - return _ParsedAvbImage( - path=image_path, - partition_name=resolved_partition_name, - footer=footer, - header=header, - descriptors=descriptors, - image_size=image_size, - public_key=public_key, - public_key_metadata=public_key_metadata, - ) - finally: - _close_image_handler(image_handler) - - -def _build_chain_partition_descriptor( - avb_module: ModuleType, - original_descriptor: Any, - public_key: bytes, -) -> Any: - descriptor = avb_module.AvbChainPartitionDescriptor() - descriptor.rollback_index_location = original_descriptor.rollback_index_location - descriptor.partition_name = original_descriptor.partition_name - descriptor.public_key = public_key - descriptor.flags = original_descriptor.flags - return descriptor - - -def _select_partition_descriptor( - avb_module: ModuleType, - parsed_image: _ParsedAvbImage, -) -> Any: - partition_descriptors = [ - descriptor - for descriptor in parsed_image.descriptors - if isinstance( - descriptor, - (avb_module.AvbHashDescriptor, avb_module.AvbHashtreeDescriptor), - ) - ] - if not partition_descriptors: - raise ValueError( - f"{parsed_image.path.name} does not contain a hash or hashtree descriptor." - ) - if len(partition_descriptors) == 1: - return partition_descriptors[0] - - for descriptor in partition_descriptors: - if descriptor.partition_name == parsed_image.partition_name: - return descriptor - - raise ValueError( - f"Unable to determine replacement descriptor for {parsed_image.path.name}." - ) - - -def _replace_vbmeta_descriptors( - avb_module: ModuleType, - original_descriptors: List[Any], - parsed_images: List[_ParsedAvbImage], -) -> int: - required_minor = 0 - hash_descriptor_types = ( - avb_module.AvbHashDescriptor, - avb_module.AvbHashtreeDescriptor, - ) - - for parsed_image in parsed_images: - chain_indexes = [ - index - for index, descriptor in enumerate(original_descriptors) - if isinstance(descriptor, avb_module.AvbChainPartitionDescriptor) - and descriptor.partition_name == parsed_image.partition_name - ] - - if len(chain_indexes) > 1: - raise ValueError( - f"Multiple chain descriptors found for {parsed_image.partition_name}." - ) - if chain_indexes: - if not parsed_image.public_key: - raise ValueError( - f"{parsed_image.path.name} does not expose a public key for chain replacement." - ) - descriptor_index = chain_indexes[0] - original_descriptors[descriptor_index] = _build_chain_partition_descriptor( - avb_module, - original_descriptors[descriptor_index], - parsed_image.public_key, - ) - required_minor = max( - required_minor, - int(parsed_image.header.required_libavb_version_minor), - ) - continue - - replacement_descriptor = _select_partition_descriptor(avb_module, parsed_image) - matching_indexes = [ - index - for index, descriptor in enumerate(original_descriptors) - if isinstance(descriptor, hash_descriptor_types) - and getattr(descriptor, "partition_name", None) - == replacement_descriptor.partition_name - ] - - if len(matching_indexes) > 1: - raise ValueError( - f"Multiple hash descriptors found for {replacement_descriptor.partition_name}." - ) - if not matching_indexes: - continue - - original_descriptors[matching_indexes[0]] = replacement_descriptor - required_minor = max( - required_minor, - int(parsed_image.header.required_libavb_version_minor), - ) - - return required_minor - - -def _generate_preserved_vbmeta_blob( - avb_module: ModuleType, - original_image: _ParsedAvbImage, - descriptors: List[Any], - key_file: Path, - algorithm: str, - required_minor: int, -) -> bytes: - avb = avb_module.Avb() - metadata_temp_path: Optional[str] = None - - try: - if original_image.public_key_metadata: - metadata_file = tempfile.NamedTemporaryFile(delete=False) - try: - metadata_file.write(original_image.public_key_metadata) - metadata_temp_path = metadata_file.name - finally: - metadata_file.close() - - return avb._generate_vbmeta_blob( - algorithm, - str(key_file), - metadata_temp_path, - descriptors, - None, - None, - original_image.header.rollback_index, - original_image.header.flags, - original_image.header.rollback_index_location, - None, - None, - None, - None, - None, - None, - None, - None, - original_image.header.release_string, - None, - max( - int(original_image.header.required_libavb_version_minor), - required_minor, - ), - ) - finally: - if metadata_temp_path is not None: - Path(metadata_temp_path).unlink(missing_ok=True) - - -def _write_preserved_vbmeta_blob( - avb_module: ModuleType, - original_image: _ParsedAvbImage, - original_vbmeta_path: Path, - output_path: Path, - vbmeta_blob: bytes, - padding_size: str, -) -> None: - if original_image.footer is not None: - shutil.copy2(original_vbmeta_path, output_path) - image_handler = avb_module.ImageHandler(str(output_path)) - try: - avb_module.Avb()._write_resigned_image( - image_handler, - original_image.footer, - vbmeta_blob, - True, - ) - finally: - _close_image_handler(image_handler) - return - - pad_to = int(padding_size) - padded_size = len(vbmeta_blob) - if pad_to > 0: - padded_size = avb_module.round_to_multiple(padded_size, pad_to) - padded_size = max(padded_size, int(original_image.image_size)) - padding_needed = padded_size - len(vbmeta_blob) - output_path.write_bytes(vbmeta_blob + (b"\0" * padding_needed)) - - -def _analyze_rollback_target( - image_name: str, - current_rb_index: int, - new_image_path: Path, - patched_image_path: Path, -) -> Optional[Dict[str, Any]]: - utils.ui.info(get_string("img_analyze_new").format(name=image_name)) - info = extract_image_avb_info(new_image_path) - new_rb_index = int(info.get("rollback", "0")) - utils.ui.info(get_string("img_new_index").format(index=new_rb_index)) - - if new_rb_index == current_rb_index: - utils.ui.info(get_string("img_index_ok").format(name=image_name)) - shutil.copy(new_image_path, patched_image_path) - return None - - utils.ui.info( - get_string("img_patch_bypass").format( - name=image_name, old=new_rb_index, new=current_rb_index - ) - ) - - return info - - -def require_info_keys( - info: Dict[str, Any], - required_keys: List[str], - image_path: Path, - defaults: Optional[Dict[str, str]] = None, -) -> None: - for key in required_keys: - if key not in info: - if key == "partition_size" and "data_size" in info: - info["partition_size"] = info["data_size"] - elif defaults and key in defaults: - info[key] = defaults[key] - else: - raise KeyError( - get_string("img_err_missing_key").format( - key=key, name=image_path.name - ) - ) +def _get_avb_info(image_path: Path) -> Dict[str, Any]: + """Run info_image --format json and return parsed Avb entry.""" + raw = _run_avbtool("info_image", "--image", image_path, "--format", "json") + data = json.loads(raw) + return data[0]["result"]["Avb"] def extract_image_avb_info(image_path: Path) -> Dict[str, Any]: - avb_module = _get_avbtool_module() - parsed = _parse_avb_image(image_path) - header = parsed.header + avb = _get_avb_info(image_path) + header = avb["header"] + footer = avb.get("footer") + + # For footer images, partition_size = actual file size (matches avbtool.py + # _parse_image behavior which returns ImageHandler.image_size = file size). + # For vbmeta-only images, partition_size = vbmeta blob size. + if footer is not None: + file_size = image_path.stat().st_size + partition_size_str = str(file_size) + else: + partition_size_str = str(avb["vbmeta_size"]) - alg_name, _ = avb_module.lookup_algorithm_by_type(header.algorithm_type) info: Dict[str, Any] = { - "partition_size": str(parsed.image_size), - "algorithm": alg_name, - "rollback": str(header.rollback_index), - "flags": str(header.flags), + "partition_size": partition_size_str, + "algorithm": avb["algorithm_name"], + "rollback": str(header["rollback_index"]), + "flags": str(header["flags"]), } - if parsed.footer is not None: - info["data_size"] = str(parsed.footer.original_image_size) + if footer is not None: + info["data_size"] = str(footer["original_image_size"]) - pubkey_sha1 = ( - hashlib.sha1(parsed.public_key).hexdigest() if parsed.public_key else None - ) + pubkey_sha1 = avb.get("public_key_sha1") if pubkey_sha1: info["pubkey_sha1"] = pubkey_sha1 @@ -425,24 +66,25 @@ def extract_image_avb_info(image_path: Path) -> Dict[str, Any]: utils.ui.info(get_string("img_info_flags").format(flags=info["flags"])) props_args: List[str] = [] - for descriptor in parsed.descriptors: - if isinstance(descriptor, avb_module.AvbPropertyDescriptor): - value = ( - descriptor.value.decode("utf-8", errors="replace") - if isinstance(descriptor.value, bytes) - else descriptor.value - ) - info[descriptor.key] = value - props_args.extend(["--prop", f"{descriptor.key}:{value}"]) - elif isinstance( - descriptor, - (avb_module.AvbHashDescriptor, avb_module.AvbHashtreeDescriptor), - ): + for descriptor in avb["descriptors"]: + if "Property" in descriptor: + prop = descriptor["Property"] + info[prop["key"]] = prop["value"] + props_args.extend(["--prop", f"{prop['key']}:{prop['value']}"]) + elif "Hash" in descriptor: if "name" not in info: - info["name"] = descriptor.partition_name - info["salt"] = descriptor.salt.hex() + h = descriptor["Hash"] + info["name"] = h["partition_name"] + info["salt"] = h["salt"] if "data_size" not in info: - info["data_size"] = str(descriptor.image_size) + info["data_size"] = str(h["image_size"]) + elif "Hashtree" in descriptor: + if "name" not in info: + ht = descriptor["Hashtree"] + info["name"] = ht["partition_name"] + info["salt"] = ht["salt"] + if "data_size" not in info: + info["data_size"] = str(ht["image_size"]) info["props_args"] = props_args if props_args: @@ -452,14 +94,12 @@ def extract_image_avb_info(image_path: Path) -> Dict[str, Any]: def vbmeta_has_chain_partition(vbmeta_path: Path, partition_name: str) -> bool: - avb_module = _get_avbtool_module() - parsed = _parse_avb_image(vbmeta_path) - chain_names = { - descriptor.partition_name - for descriptor in parsed.descriptors - if isinstance(descriptor, avb_module.AvbChainPartitionDescriptor) - } - return partition_name in chain_names + avb = _get_avb_info(vbmeta_path) + return any( + "ChainPartition" in d + and d["ChainPartition"]["partition_name"] == partition_name + for d in avb["descriptors"] + ) def apply_avb_integrity_footer( @@ -562,6 +202,51 @@ def _update_vbmeta_partition_descriptor( ) +def require_info_keys( + info: Dict[str, Any], + required_keys: List[str], + image_path: Path, + defaults: Optional[Dict[str, str]] = None, +) -> None: + for key in required_keys: + if key not in info: + if key == "partition_size" and "data_size" in info: + info["partition_size"] = info["data_size"] + elif defaults and key in defaults: + info[key] = defaults[key] + else: + raise KeyError( + get_string("img_err_missing_key").format( + key=key, name=image_path.name + ) + ) + + +def _analyze_rollback_target( + image_name: str, + current_rb_index: int, + new_image_path: Path, + patched_image_path: Path, +) -> Optional[Dict[str, Any]]: + utils.ui.info(get_string("img_analyze_new").format(name=image_name)) + info = extract_image_avb_info(new_image_path) + new_rb_index = int(info.get("rollback", "0")) + utils.ui.info(get_string("img_new_index").format(index=new_rb_index)) + + if new_rb_index == current_rb_index: + utils.ui.info(get_string("img_index_ok").format(name=image_name)) + shutil.copy(new_image_path, patched_image_path) + return None + + utils.ui.info( + get_string("img_patch_bypass").format( + name=image_name, old=new_rb_index, new=current_rb_index + ) + ) + + return info + + def patch_chained_image_rollback( image_name: str, current_rb_index: int, @@ -696,18 +381,13 @@ def process_boot_image_avb( def _resolve_vbmeta_key_and_algorithm( - avb_module: ModuleType, - parsed_image: _ParsedAvbImage, + avb_info: Dict[str, Any], key_file: Optional[Path], algorithm: Optional[str], ) -> tuple[Path, str]: resolved_key_file = key_file if resolved_key_file is None: - vbmeta_pubkey = ( - hashlib.sha1(parsed_image.public_key).hexdigest() - if parsed_image.public_key - else None - ) + vbmeta_pubkey = avb_info.get("public_key_sha1") resolved_key_file = const.KEY_MAP.get(str(vbmeta_pubkey)) utils.ui.info(get_string("act_verify_vbmeta_key")) @@ -718,66 +398,10 @@ def _resolve_vbmeta_key_and_algorithm( raise KeyError(get_string("act_err_unknown_key").format(key=vbmeta_pubkey)) utils.ui.info(get_string("img_key_matched").format(name=resolved_key_file.name)) - if algorithm: - resolved_algorithm = algorithm - else: - alg_name, _ = avb_module.lookup_algorithm_by_type( - parsed_image.header.algorithm_type - ) - resolved_algorithm = alg_name - + resolved_algorithm = algorithm or avb_info["algorithm_name"] return resolved_key_file, resolved_algorithm -def rebuild_vbmeta_preserving_descriptors( - output_path: Path, - original_vbmeta_path: Path, - chained_images: List[Path], - padding_size: str = "8192", - key_file: Optional[Path] = None, - algorithm: Optional[str] = None, -) -> None: - utils.ui.info(get_string("act_remake_vbmeta")) - if not chained_images: - shutil.copy2(original_vbmeta_path, output_path) - return - - avb_module = _get_avbtool_module() - original_image = _parse_avb_image( - original_vbmeta_path, - partition_name=original_vbmeta_path.stem, - ) - - resolved_key_file, resolved_algorithm = _resolve_vbmeta_key_and_algorithm( - avb_module, original_image, key_file, algorithm - ) - utils.ui.info(get_string("act_remaking_vbmeta")) - - parsed_images = [_parse_avb_image(image_path) for image_path in chained_images] - descriptors = list(original_image.descriptors) - required_minor = _replace_vbmeta_descriptors( - avb_module, - descriptors, - parsed_images, - ) - vbmeta_blob = _generate_preserved_vbmeta_blob( - avb_module, - original_image, - descriptors, - resolved_key_file, - resolved_algorithm, - required_minor, - ) - _write_preserved_vbmeta_blob( - avb_module, - original_image, - original_vbmeta_path, - output_path, - vbmeta_blob, - padding_size, - ) - - def rebuild_vbmeta_with_chained_images( output_path: Path, original_vbmeta_path: Path, @@ -787,18 +411,14 @@ def rebuild_vbmeta_with_chained_images( algorithm: Optional[str] = None, ) -> None: utils.ui.info(get_string("act_remake_vbmeta")) - avb_module = _get_avbtool_module() - parsed_vbmeta = _parse_avb_image( - original_vbmeta_path, - partition_name=original_vbmeta_path.stem, - ) + avb_info = _get_avb_info(original_vbmeta_path) resolved_key_file, resolved_algorithm = _resolve_vbmeta_key_and_algorithm( - avb_module, parsed_vbmeta, key_file, algorithm + avb_info, key_file, algorithm ) - rollback_str = str(parsed_vbmeta.header.rollback_index) - flags_str = str(parsed_vbmeta.header.flags) + rollback_str = str(avb_info["header"]["rollback_index"]) + flags_str = str(avb_info["header"]["flags"]) utils.ui.info(get_string("act_remaking_vbmeta")) diff --git a/bin/ltbox/utils.py b/bin/ltbox/utils.py index 1fdcf105..a1ff5f60 100644 --- a/bin/ltbox/utils.py +++ b/bin/ltbox/utils.py @@ -210,7 +210,7 @@ def check_dependencies() -> None: "Python Environment": const.PYTHON_EXE, "ADB": const.ADB_EXE, "Fastboot": const.FASTBOOT_EXE, - "avbtool": const.AVBTOOL_PY, + "avbtool-rs": const.AVBTOOL_RS, } if not is_git_checkout: diff --git a/tests/actions/test_actions_unit.py b/tests/actions/test_actions_unit.py index 0f726a9d..8e8c3b8e 100644 --- a/tests/actions/test_actions_unit.py +++ b/tests/actions/test_actions_unit.py @@ -1,5 +1,4 @@ import xml.etree.ElementTree as ET -from pathlib import Path from unittest.mock import MagicMock, patch import pytest @@ -10,7 +9,6 @@ from ltbox.actions import xml as xml_action from ltbox.actions.root import workflow as root_workflow from ltbox.menus import data as menu_data -from ltbox.patch import avb as avb_patch from ltbox.actions.root.strategies import GkiRootStrategy from ltbox.patch.avb import ( patch_chained_image_rollback, @@ -69,99 +67,6 @@ def test_xml_select_prefers_ota_keep_data_xml(mock_env): assert "rawprogram_save_persist_unsparse0.xml" not in r_names -def test_replace_vbmeta_descriptors_keeps_root_chain_descriptors_intact(): - class FakeChainDescriptor: - def __init__( - self, - partition_name="", - rollback_index_location=0, - public_key=b"", - flags=0, - ): - self.partition_name = partition_name - self.rollback_index_location = rollback_index_location - self.public_key = public_key - self.flags = flags - - class FakeHashDescriptor: - def __init__(self, partition_name=""): - self.partition_name = partition_name - - class FakeHashtreeDescriptor: - def __init__(self, partition_name=""): - self.partition_name = partition_name - - class FakeModule: - AvbChainPartitionDescriptor = FakeChainDescriptor - AvbHashDescriptor = FakeHashDescriptor - AvbHashtreeDescriptor = FakeHashtreeDescriptor - - original_descriptors = [ - FakeChainDescriptor("boot", 3, b"boot-old", 0), - FakeChainDescriptor("recovery", 1, b"recovery-old", 0), - FakeChainDescriptor("vbmeta_system", 2, b"vbmeta-system-old", 0), - FakeHashDescriptor("dtbo"), - FakeHashtreeDescriptor("vendor"), - ] - - replacement_images = [ - avb_patch._ParsedAvbImage( - path=Path("boot.img"), - partition_name="boot", - footer=None, - header=MagicMock(required_libavb_version_minor=0), - descriptors=[FakeHashDescriptor("boot")], - image_size=0, - public_key=b"boot-new", - public_key_metadata=b"", - ), - avb_patch._ParsedAvbImage( - path=Path("vbmeta_system.img"), - partition_name="vbmeta_system", - footer=None, - header=MagicMock(required_libavb_version_minor=0), - descriptors=[ - FakeHashDescriptor("pvmfw"), - FakeHashtreeDescriptor("product"), - FakeHashtreeDescriptor("system"), - FakeHashtreeDescriptor("system_ext"), - ], - image_size=0, - public_key=b"vbmeta-system-new", - public_key_metadata=b"", - ), - ] - - required_minor = avb_patch._replace_vbmeta_descriptors( - FakeModule, - original_descriptors, - replacement_images, - ) - - assert required_minor == 0 - assert len(original_descriptors) == 5 - assert isinstance(original_descriptors[0], FakeChainDescriptor) - assert original_descriptors[0].public_key == b"boot-new" - assert isinstance(original_descriptors[2], FakeChainDescriptor) - assert original_descriptors[2].public_key == b"vbmeta-system-new" - assert not any( - getattr(descriptor, "partition_name", None) - in {"pvmfw", "product", "system", "system_ext"} - for descriptor in original_descriptors - ) - - -def test_resolve_avbtool_openssl_binary_prefers_local_tool(tmp_path): - tool_dir = tmp_path / "tools" - tool_dir.mkdir() - source_path = tool_dir / "avbtool.py" - source_path.write_text("# stub", encoding="utf-8") - openssl_path = tool_dir / "openssl.exe" - openssl_path.write_text("stub", encoding="utf-8") - - assert avb_patch._resolve_avbtool_openssl_binary(source_path) == str(openssl_path) - - def test_flash_args(mock_env): img_dir = mock_env["IMAGE_DIR"] files = ["rawprogram1.xml", "rawprogram_unsparse0.xml", "patch0.xml"] @@ -331,21 +236,24 @@ def test_vbmeta_has_chain_partition_parses_descriptor(tmp_path): vbmeta_img = tmp_path / "vbmeta.img" vbmeta_img.write_bytes(b"dummy") - mock_chain_boot = MagicMock() - mock_chain_boot.partition_name = "boot" - mock_chain_recovery = MagicMock() - mock_chain_recovery.partition_name = "recovery" - - mock_parsed = MagicMock() - mock_parsed.descriptors = [mock_chain_boot, mock_chain_recovery] - - mock_avb_module = MagicMock() - mock_avb_module.AvbChainPartitionDescriptor = type(mock_chain_boot) + mock_avb_info = { + "descriptors": [ + { + "ChainPartition": { + "partition_name": "boot", + "rollback_index_location": 3, + } + }, + { + "ChainPartition": { + "partition_name": "recovery", + "rollback_index_location": 1, + } + }, + ], + } - with ( - patch("ltbox.patch.avb._parse_avb_image", return_value=mock_parsed), - patch("ltbox.patch.avb._get_avbtool_module", return_value=mock_avb_module), - ): + with patch("ltbox.patch.avb._get_avb_info", return_value=mock_avb_info): assert vbmeta_has_chain_partition(vbmeta_img, "boot") is True assert vbmeta_has_chain_partition(vbmeta_img, "init_boot") is False @@ -466,27 +374,16 @@ def test_rebuild_vbmeta_with_single_image_uses_descriptor_update(tmp_path): key_file = tmp_path / "vbmeta.pem" key_file.write_text("key", encoding="utf-8") - # sha1 of b"fake-public-key" - import hashlib - - fake_pubkey = b"fake-public-key" - pubkey_sha1 = hashlib.sha1(fake_pubkey).hexdigest() - - mock_header = MagicMock() - mock_header.algorithm_type = 5 - mock_header.rollback_index = 0 - mock_header.flags = 0 - mock_parsed = MagicMock() - mock_parsed.public_key = fake_pubkey - mock_parsed.header = mock_header - - mock_avb_module = MagicMock() - mock_avb_module.lookup_algorithm_by_type.return_value = ("SHA256_RSA4096", None) + mock_avb_info = { + "header": {"rollback_index": 0, "flags": 0}, + "algorithm_name": "SHA256_RSA4096", + "public_key_sha1": "fake-pubkey-sha1", + "descriptors": [], + } with ( - patch("ltbox.patch.avb._parse_avb_image", return_value=mock_parsed), - patch("ltbox.patch.avb._get_avbtool_module", return_value=mock_avb_module), - patch("ltbox.patch.avb.const.KEY_MAP", {pubkey_sha1: key_file}), + patch("ltbox.patch.avb._get_avb_info", return_value=mock_avb_info), + patch("ltbox.patch.avb.const.KEY_MAP", {"fake-pubkey-sha1": key_file}), patch("ltbox.patch.avb._run_avbtool") as mock_run, ): from ltbox.patch.avb import rebuild_vbmeta_with_chained_images @@ -525,26 +422,16 @@ def test_rebuild_vbmeta_with_multiple_images_falls_back_to_make_vbmeta(tmp_path) key_file = tmp_path / "vbmeta.pem" key_file.write_text("key", encoding="utf-8") - import hashlib - - fake_pubkey = b"fake-public-key" - pubkey_sha1 = hashlib.sha1(fake_pubkey).hexdigest() - - mock_header = MagicMock() - mock_header.algorithm_type = 5 - mock_header.rollback_index = 0 - mock_header.flags = 0 - mock_parsed = MagicMock() - mock_parsed.public_key = fake_pubkey - mock_parsed.header = mock_header - - mock_avb_module = MagicMock() - mock_avb_module.lookup_algorithm_by_type.return_value = ("SHA256_RSA4096", None) + mock_avb_info = { + "header": {"rollback_index": 0, "flags": 0}, + "algorithm_name": "SHA256_RSA4096", + "public_key_sha1": "fake-pubkey-sha1", + "descriptors": [], + } with ( - patch("ltbox.patch.avb._parse_avb_image", return_value=mock_parsed), - patch("ltbox.patch.avb._get_avbtool_module", return_value=mock_avb_module), - patch("ltbox.patch.avb.const.KEY_MAP", {pubkey_sha1: key_file}), + patch("ltbox.patch.avb._get_avb_info", return_value=mock_avb_info), + patch("ltbox.patch.avb.const.KEY_MAP", {"fake-pubkey-sha1": key_file}), patch("ltbox.patch.avb._run_avbtool") as mock_run, ): from ltbox.patch.avb import rebuild_vbmeta_with_chained_images @@ -583,20 +470,15 @@ def test_rebuild_vbmeta_with_override_key_skips_pubkey_validation(tmp_path): key_file = tmp_path / "testkey_rsa4096.pem" key_file.write_text("key", encoding="utf-8") - mock_header = MagicMock() - mock_header.algorithm_type = 3 - mock_header.rollback_index = 7 - mock_header.flags = 0 - mock_parsed = MagicMock() - mock_parsed.public_key = b"unknown-key" - mock_parsed.header = mock_header - - mock_avb_module = MagicMock() - mock_avb_module.lookup_algorithm_by_type.return_value = ("SHA256_RSA2048", None) + mock_avb_info = { + "header": {"rollback_index": 7, "flags": 0}, + "algorithm_name": "SHA256_RSA2048", + "public_key_sha1": "unknown-key-sha1", + "descriptors": [], + } with ( - patch("ltbox.patch.avb._parse_avb_image", return_value=mock_parsed), - patch("ltbox.patch.avb._get_avbtool_module", return_value=mock_avb_module), + patch("ltbox.patch.avb._get_avb_info", return_value=mock_avb_info), patch("ltbox.patch.avb._run_avbtool") as mock_run, ): from ltbox.patch.avb import rebuild_vbmeta_with_chained_images diff --git a/tests/conftest.py b/tests/conftest.py index 84ecf0ff..f78ef709 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -12,7 +12,6 @@ INTEGRATION_TOOL_FILES = ( "magiskboot.exe", "magiskboot_xz_helper.exe", - "openssl.exe", ) if str(BIN_PATH) not in sys.path: diff --git a/tests/core/test_conftest.py b/tests/core/test_conftest.py index 358ef5ef..6007b40b 100644 --- a/tests/core/test_conftest.py +++ b/tests/core/test_conftest.py @@ -21,7 +21,6 @@ def test_integration_tools_ready_false_when_any_required_tool_missing( tools_dir = tmp_path / "bin" / "tools" tools_dir.mkdir(parents=True) (tools_dir / "magiskboot.exe").write_text("stub", encoding="utf-8") - (tools_dir / "openssl.exe").write_text("stub", encoding="utf-8") with patch.object(conftest, "ROOT", tmp_path): assert conftest._integration_tools_ready() is False diff --git a/tests/core/test_main_scan.py b/tests/core/test_main_scan.py index be048102..8bada21e 100644 --- a/tests/core/test_main_scan.py +++ b/tests/core/test_main_scan.py @@ -20,13 +20,11 @@ def test_collect_info_scan_files_filters_img_only(tmp_path): def test_build_info_scan_command_uses_constants_paths(): - constants = SimpleNamespace( - PYTHON_EXE=Path("python"), AVBTOOL_PY=Path("avbtool.py") - ) + constants = SimpleNamespace(AVBTOOL_RS=Path("avbtool-rs.exe")) command = scan_api.build_info_scan_command(Path("boot.img"), constants) - assert command == ["python", "avbtool.py", "info_image", "--image", "boot.img"] + assert command == ["avbtool-rs.exe", "info_image", "--image", "boot.img"] def test_run_info_scan_creates_log(tmp_path): @@ -48,8 +46,7 @@ def run(self, cmd, options): constants = SimpleNamespace( BASE_DIR=tmp_path / "bin", - PYTHON_EXE=Path("python"), - AVBTOOL_PY=Path("avbtool.py"), + AVBTOOL_RS=Path("avbtool-rs.exe"), ) scan_api.run_info_scan( [str(image_dir), str(extra_img)], diff --git a/tests/core/test_utils.py b/tests/core/test_utils.py index ecad1b1d..4df421f4 100644 --- a/tests/core/test_utils.py +++ b/tests/core/test_utils.py @@ -217,10 +217,10 @@ def test_check_dependencies_blocks_source_download_without_edl_tools( python_exe = tmp_path / "python.exe" adb_exe = tmp_path / "adb.exe" fastboot_exe = tmp_path / "fastboot.exe" - avbtool_py = tmp_path / "avbtool.py" + avbtool_rs = tmp_path / "avbtool-rs.exe" qdlrs_exe = tmp_path / "qdl-rs.exe" - for path in (python_exe, adb_exe, fastboot_exe, avbtool_py): + for path in (python_exe, adb_exe, fastboot_exe, avbtool_rs): path.write_text("ok", encoding="utf-8") with ( @@ -228,7 +228,7 @@ def test_check_dependencies_blocks_source_download_without_edl_tools( patch("ltbox.utils.const.PYTHON_EXE", python_exe), patch("ltbox.utils.const.ADB_EXE", adb_exe), patch("ltbox.utils.const.FASTBOOT_EXE", fastboot_exe), - patch("ltbox.utils.const.AVBTOOL_PY", avbtool_py), + patch("ltbox.utils.const.AVBTOOL_RS", avbtool_rs), patch("ltbox.utils.const.QDLRS_EXE", qdlrs_exe), patch("ltbox.utils.const.KEY_MAP", {}), patch("ltbox.utils._check_required_windows_drivers"), @@ -297,10 +297,10 @@ def test_check_dependencies_allows_release_package_edl_tools( python_exe = tmp_path / "python.exe" adb_exe = tmp_path / "adb.exe" fastboot_exe = tmp_path / "fastboot.exe" - avbtool_py = tmp_path / "avbtool.py" + avbtool_rs = tmp_path / "avbtool-rs.exe" qdlrs_exe = tmp_path / "qdl-rs.exe" - for path in (python_exe, adb_exe, fastboot_exe, avbtool_py, qdlrs_exe): + for path in (python_exe, adb_exe, fastboot_exe, avbtool_rs, qdlrs_exe): path.write_text("ok", encoding="utf-8") with ( @@ -308,7 +308,7 @@ def test_check_dependencies_allows_release_package_edl_tools( patch("ltbox.utils.const.PYTHON_EXE", python_exe), patch("ltbox.utils.const.ADB_EXE", adb_exe), patch("ltbox.utils.const.FASTBOOT_EXE", fastboot_exe), - patch("ltbox.utils.const.AVBTOOL_PY", avbtool_py), + patch("ltbox.utils.const.AVBTOOL_RS", avbtool_rs), patch("ltbox.utils.const.QDLRS_EXE", qdlrs_exe), patch("ltbox.utils.const.KEY_MAP", {}), patch("ltbox.utils._check_required_windows_drivers"), diff --git a/tests/scripts/build_tools.py b/tests/scripts/build_tools.py index 5c092d26..9e292834 100644 --- a/tests/scripts/build_tools.py +++ b/tests/scripts/build_tools.py @@ -8,7 +8,6 @@ SUBMODULE_DIR = REPO_ROOT / "vendor" / "MagiskbootAlone" MAGISKBOOT_EXE = TOOLS_DIR / "magiskboot.exe" MAGISKBOOT_XZ_HELPER_EXE = TOOLS_DIR / "magiskboot_xz_helper.exe" -OPENSSL_EXE = TOOLS_DIR / "openssl.exe" VERSION_FILE = TOOLS_DIR / "magiskboot.version" @@ -46,14 +45,8 @@ def build(): current_sha = _get_submodule_sha() - openssl_ready = OPENSSL_EXE.exists() if os.name == "nt" else True helper_ready = MAGISKBOOT_XZ_HELPER_EXE.exists() if os.name == "nt" else True - if ( - MAGISKBOOT_EXE.exists() - and VERSION_FILE.exists() - and openssl_ready - and helper_ready - ): + if MAGISKBOOT_EXE.exists() and VERSION_FILE.exists() and helper_ready: cached_sha = VERSION_FILE.read_text(encoding="utf-8").strip() if cached_sha == current_sha: print("[INFO] Tools are up-to-date. Skipping build.") @@ -91,14 +84,12 @@ def build(): ) return - print( - "[INFO] Installing MSYS2 dependencies (gcc, cmake, make, zlib-devel, openssl)..." - ) + print("[INFO] Installing MSYS2 dependencies (gcc, cmake, make, zlib-devel)...") subprocess.run( [ str(bash_exe), "-lc", - "pacman -S --noconfirm --overwrite '*' --needed gcc cmake make zlib-devel openssl", + "pacman -S --noconfirm --overwrite '*' --needed gcc cmake make zlib-devel", ], check=True, ) @@ -147,19 +138,9 @@ def build(): raise RuntimeError("magiskboot_xz_helper.exe was not produced") shutil.copy(compiled_helper, MAGISKBOOT_XZ_HELPER_EXE) - msys_openssl = msys_root / "usr/bin/openssl.exe" - if msys_openssl.exists(): - shutil.copy(msys_openssl, OPENSSL_EXE) - print("[INFO] Copied openssl.exe from MSYS2.") - dlls_to_copy = ["msys-2.0.dll", "msys-z.dll"] usr_bin = msys_root / "usr/bin" - for dll_file in usr_bin.glob("msys-crypto-*.dll"): - dlls_to_copy.append(dll_file.name) - for dll_file in usr_bin.glob("msys-ssl-*.dll"): - dlls_to_copy.append(dll_file.name) - for dll in list(set(dlls_to_copy)): src_dll = usr_bin / dll if src_dll.exists(): diff --git a/vendor/avb b/vendor/avb deleted file mode 160000 index 4a4e2c8a..00000000 --- a/vendor/avb +++ /dev/null @@ -1 +0,0 @@ -Subproject commit 4a4e2c8a6592b88cf18b10fe5406f53a2a5d26cf diff --git a/vendor/avbtool-rs b/vendor/avbtool-rs new file mode 160000 index 00000000..172ee4dd --- /dev/null +++ b/vendor/avbtool-rs @@ -0,0 +1 @@ +Subproject commit 172ee4dd9fd0176f2f171be6501cc8bbf9257a23