fix: make custom boot animation survive OTA repacking

src/boot_animation.py

@@ -8,13 +8,13 @@
 
 from __future__ import annotations
 
-import hashlib
 import argparse
+import hashlib
+import io
 import os
 import re
 import stat
 import sys
-import tempfile
 import zipfile
 from collections.abc import Iterable
 from pathlib import Path, PurePosixPath
@@ -28,6 +28,10 @@
 MAX_COMPRESSION_RATIO = 200
 MAX_DESCRIPTION_BYTES = 4096
 PAYLOAD_ENVIRONMENT = "PIXENEOS_BOOT_ANIMATION_PATH"
+BOOT_ANIMATION_TARGETS = (
+    ("product", "/product/media/bootanimation.zip"),
+    ("product", "/product/media/bootanimation-dark.zip"),
+)
 
 
 class BootAnimationError(ValueError):
@@ -205,6 +209,44 @@ def validate_payload(path: str | os.PathLike[str]) -> str:
     return digest
 
 
+def build_runtime_payload(path: str | os.PathLike[str]) -> bytes:
+    """Return a deterministic Android-compatible, uncompressed animation ZIP."""
+
+    validate_payload(path)
+    output = io.BytesIO()
+    with zipfile.ZipFile(Path(path), "r") as source:
+        with zipfile.ZipFile(
+            output,
+            "w",
+            compression=zipfile.ZIP_STORED,
+            allowZip64=False,
+        ) as runtime:
+            for info in source.infolist():
+                if info.is_dir():
+                    continue
+                runtime.writestr(
+                    info.filename,
+                    _read_member(source, info),
+                    compress_type=zipfile.ZIP_STORED,
+                )
+    return output.getvalue()
+
+
+def install_runtime_payload(ext_fs: dict[str, Any], payload: bytes) -> None:
+    """Install through ExtFs so AFSR metadata and SELinux labels stay in sync."""
+
+    for partition, raw_target in BOOT_ANIMATION_TARGETS:
+        fs = ext_fs.get(partition)
+        if fs is None:
+            raise RuntimeError(
+                f"boot animation target partition is missing: {partition}"
+            )
+        target = PurePosixPath(raw_target)
+        fs.mkdir(str(target.parent), mode=0o755, parents=True, exist_ok=True)
+        with fs.open(str(target), "wb", mode=0o644) as stream:
+            stream.write(payload)
+
+
 def _module_class() -> type[Any]:
     # Import only when the pinned helper imports this file.  This keeps the
     # standalone validator usable before the helper is fetched.
@@ -245,7 +287,7 @@ def from_args(cls, args: argparse.Namespace) -> "BootAnimationMod":
         def requirements(self) -> ModuleRequirements:
             return ModuleRequirements(
                 boot_images=set(),
-                ext_images={"system"},
+                ext_images={"product"},
                 selinux_patching=False,
             )
 
@@ -260,23 +302,8 @@ def inject(
             payload_path = os.environ.get(PAYLOAD_ENVIRONMENT)
             if not payload_path:
                 raise RuntimeError(f"{PAYLOAD_ENVIRONMENT} is not set")
-            validate_payload(payload_path)
-            payload = Path(payload_path)
-            target = ext_fs["system"].tree / "system" / "media" / "bootanimation.zip"
-            target.parent.mkdir(parents=True, exist_ok=True)
-            descriptor, temporary = tempfile.mkstemp(
-                prefix=".bootanimation.", dir=target.parent
-            )
-            try:
-                with os.fdopen(descriptor, "wb") as stream:
-                    stream.write(payload.read_bytes())
-                    stream.flush()
-                    os.fsync(stream.fileno())
-                os.chmod(temporary, 0o644)
-                os.replace(temporary, target)
-            finally:
-                if os.path.exists(temporary):
-                    os.unlink(temporary)
+            payload = build_runtime_payload(payload_path)
+            install_runtime_payload(ext_fs, payload)
 
     return BootAnimationMod
 

tests/boot_animation_test.py

@@ -4,6 +4,7 @@
 from __future__ import annotations
 
 import hashlib
+import io
 import stat
 import sys
 import tempfile
@@ -15,6 +16,8 @@
 from boot_animation import (  # noqa: E402
     MAX_MEMBER_COUNT,
     BootAnimationError,
+    build_runtime_payload,
+    install_runtime_payload,
     validate_payload,
 )
 
@@ -33,7 +36,72 @@ def assert_rejected(path: Path, context: str) -> None:
     raise AssertionError(f"{context}: invalid archive was accepted")
 
 
+class FakeExtFs:
+    def __init__(self, root: Path) -> None:
+        self.root = root
+        self.mkdir_calls: list[tuple[str, int, bool, bool]] = []
+        self.open_calls: list[tuple[str, str, int]] = []
+
+    @property
+    def tree(self) -> Path:
+        raise AssertionError("boot animation injection bypassed ExtFs metadata APIs")
+
+    def mkdir(
+        self,
+        path: str,
+        mode: int = 0o755,
+        parents: bool = False,
+        exist_ok: bool = False,
+    ) -> None:
+        self.mkdir_calls.append((path, mode, parents, exist_ok))
+        (self.root / path.lstrip("/")).mkdir(
+            mode=mode,
+            parents=parents,
+            exist_ok=exist_ok,
+        )
+
+    def open(self, path: str, open_mode: str, mode: int = 0o644):
+        self.open_calls.append((path, open_mode, mode))
+        target = self.root / path.lstrip("/")
+        target.parent.mkdir(parents=True, exist_ok=True)
+        return target.open(open_mode)
+
+
+def test_runtime_payload_installation() -> None:
+    with tempfile.TemporaryDirectory() as temporary:
+        root = Path(temporary)
+        source = root / "source.zip"
+        write_valid(source)
+
+        runtime = build_runtime_payload(source)
+        with zipfile.ZipFile(io.BytesIO(runtime), "r") as archive:
+            files = [info for info in archive.infolist() if not info.is_dir()]
+            assert files
+            assert all(info.compress_type == zipfile.ZIP_STORED for info in files)
+
+        product = FakeExtFs(root / "product-fs")
+        install_runtime_payload({"product": product}, runtime)
+
+        expected_targets = [
+            "/product/media/bootanimation.zip",
+            "/product/media/bootanimation-dark.zip",
+        ]
+        assert [call[0] for call in product.open_calls] == expected_targets
+        assert all(call[1:] == ("wb", 0o644) for call in product.open_calls)
+        assert all(call[0] == "/product/media" for call in product.mkdir_calls)
+        for target in expected_targets:
+            assert (product.root / target.lstrip("/")).read_bytes() == runtime
+
+        try:
+            install_runtime_payload({}, runtime)
+        except RuntimeError:
+            pass
+        else:
+            raise AssertionError("missing product partition was accepted")
+
+
 def main() -> None:
+    test_runtime_payload_installation()
     checked_in = Path("custom/boot-animation/bootanimation.zip")
     expected = hashlib.sha256(checked_in.read_bytes()).hexdigest()
     assert validate_payload(checked_in) == expected