Harden installer failure paths and add battle-test regressions

Engine/build-locked-engine.sh

@@ -95,7 +95,8 @@ done < <(
 )
 
 stage=$(mktemp -d /private/tmp/omarchy-locked-engine-build.XXXXXX)
-trap 'rm -rf -- "$stage"' EXIT
+temporary=
+trap 'rm -rf -- "$stage"; if [[ -n $temporary ]]; then rm -f -- "$temporary"; fi' EXIT
 cp -a "$checkout/." "$stage/"
 git -C "$stage" apply "$engine_root/$(jq -r '.downstream_overlay.patch.path' "$lock")"
 
@@ -143,11 +144,10 @@ source_date_epoch=$(jq -r '.build_toolchain.source_date_epoch' "$lock")
   --mtime="@$source_date_epoch" --owner=0 --group=0 --numeric-owner \
   -C "$stage/package" -cf - . | "$gzip" -n -9 >"$temporary"
 chmod 644 "$temporary"
-mv -f "$temporary" "$artifact"
-python3 "$engine_root/verify-archive-modes.py" "$artifact"
+python3 "$engine_root/verify-archive-modes.py" "$temporary"
 
-actual_size=$(stat -f '%z' "$artifact")
-actual_digest=$(/usr/bin/shasum -a 256 "$artifact")
+actual_size=$(stat -f '%z' "$temporary")
+actual_digest=$(/usr/bin/shasum -a 256 "$temporary")
 actual_digest=${actual_digest%% *}
 expected_filename=$(jq -er '.validation_artifact.filename' "$lock")
 expected_size=$(jq -er '.validation_artifact.size_bytes' "$lock")
@@ -172,4 +172,7 @@ if [[ $actual_digest != "$expected_digest" ]]; then
   echo "Built engine digest does not reproduce source lock" >&2
   exit 1
 fi
+# A failed rebuild must never replace an earlier authenticated artifact.
+mv -f "$temporary" "$artifact"
+temporary=
 printf '%s  %s  %s bytes\n' "$actual_digest" "$artifact" "$actual_size"

Engine/overlay/src/omarchy_asahi.py

@@ -75,6 +75,9 @@ def rewrite_existing_content(self, plan):
             names = payload.namelist()
             if len(names) != len(set(names)):
                 raise AsahiAdapterError("ambiguous repair payload")
+            replacements = []
+            # Validate the entire member table before opening any disk for write.
+            # A missing later image must not leave earlier partitions replaced.
             for role in ("stub", "efi", "boot", "root"):
                 expected = self.manifest["replacement_content"][role]
                 member = expected["payload_member"]
@@ -97,6 +100,15 @@ def rewrite_existing_content(self, plan):
                 info = payload.getinfo(member)
                 if info.file_size != expected["size_bytes"]:
                     raise AsahiAdapterError("repair payload size changed")
+                try:
+                    # Check local headers, encryption and decoder availability
+                    # without decompressing multi-gigabyte images twice.
+                    with payload.open(info):
+                        pass
+                except (RuntimeError, NotImplementedError, zipfile.BadZipFile) as error:
+                    raise AsahiAdapterError("unsupported repair payload member") from error
+                replacements.append((role, expected, info))
+            for role, expected, info in replacements:
                 identifier = self._partition_identifier(role)
                 digest = hashlib.sha256()
                 written = 0
@@ -453,6 +465,12 @@ def prepare_target(self, plan):
                 name=plan.source_identifier,
                 free=False,
             )
+            if (
+                source.offset + source.size != plan.offset_bytes + plan.length_bytes
+                or source.offset >= plan.offset_bytes
+                or source.type != "Apple_APFS"
+            ):
+                raise AsahiAdapterError("approved source partition changed")
             new_size = source.size - plan.length_bytes
             if new_size < plan.minimum_container_bytes:
                 raise AsahiAdapterError(
@@ -502,13 +520,8 @@ def install_stub_and_esp(self, plan):
             )
         return self._installed_evidence(plan)
 
-    def validate_installed_checkpoint(
-        self,
-        plan,
-        target_evidence,
-        installed_evidence,
-    ):
-        """Re-read the exact completed stage-one state without mutation."""
+    def validate_prepared_checkpoint(self, plan, target_evidence):
+        """Reconcile the recorded APFS identity before resuming any mutation."""
         self._require_preflight()
         target = self._parse_checkpoint_evidence(
             target_evidence,
@@ -520,6 +533,31 @@ def validate_installed_checkpoint(
                 "uuid",
             },
         )
+        if target["plan_digest"] != plan.plan_digest:
+            raise AsahiAdapterError("installed checkpoint plan changed")
+        self._refresh_parts()
+        prepared = self._find_prepared_target(plan)
+        # Bind the adapter only after every recorded identity field matches.
+        actual = {
+            "plan_digest": plan.plan_digest,
+            "partition_identifier": prepared.name,
+            "offset_bytes": prepared.offset,
+            "size_bytes": prepared.size,
+            "uuid": prepared.uuid,
+        }
+        if actual != target:
+            raise AsahiAdapterError("prepared target identity changed")
+        self.target_part = prepared
+        self.installer.part = prepared
+
+    def validate_installed_checkpoint(
+        self,
+        plan,
+        target_evidence,
+        installed_evidence,
+    ):
+        """Re-read the exact completed stage-one state without mutation."""
+        self.validate_prepared_checkpoint(plan, target_evidence)
         installed = self._parse_checkpoint_evidence(
             installed_evidence,
             {
@@ -531,18 +569,9 @@ def validate_installed_checkpoint(
                 "populated_partitions",
             },
         )
-        if (
-            target["plan_digest"] != plan.plan_digest
-            or installed["plan_digest"] != plan.plan_digest
-        ):
+        if installed["plan_digest"] != plan.plan_digest:
             raise AsahiAdapterError("installed checkpoint plan changed")
 
-        self._refresh_parts()
-        self.target_part = self._find_prepared_target(plan)
-        self.installer.part = self.target_part
-        if self._target_evidence(plan) != target_evidence:
-            raise AsahiAdapterError("prepared target identity changed")
-
         recorded = installed["populated_partitions"]
         if (
             not isinstance(recorded, list)

Engine/overlay/src/omarchy_repair.py

@@ -468,6 +468,10 @@ def _validate_manifest(manifest):
             }:
                 raise InPlaceRepairError("invalid preserved repair content")
             continue
+        capacity = next(item["size_bytes"] for item in partitions if item["role"] == role)
+        if any(identity["size_bytes"] > capacity or identity["size_bytes"] % 4096
+               for identity in (before, after)):
+            raise InPlaceRepairError("invalid repair content size")
         if (
             not isinstance(member, str)
             or not member

Engine/overlay/src/omarchy_stage1.py

@@ -71,6 +71,20 @@ def run_stage1(plan, journal, adapter):
     if completed is not None:
         return completed
 
+    # Historical inventory admits a resumed run, but cannot establish that its
+    # prepared target still belongs to this install. Re-read checkpoint identity
+    # before recording any new mutation intent. Fresh runs need no readback.
+    if "apfs-target-prepared" in journal.checkpoints:
+        arguments = [plan, journal.checkpoint_evidence("apfs-target-prepared")]
+        validator_name = "validate_prepared_checkpoint"
+        if "stub-and-esp-installed" in journal.checkpoints:
+            validator_name = "validate_installed_checkpoint"
+            arguments.append(journal.checkpoint_evidence("stub-and-esp-installed"))
+        validator = getattr(adapter, validator_name, None)
+        if not callable(validator):
+            raise Stage1Error(f"adapter does not implement {validator_name}")
+        validator(*arguments)
+
     # The privileged helper runs with umask 077 so nothing it creates is
     # readable by other users. The stub volume and its ESP are not the
     # helper's private state: macOS and the unprivileged inspection that

Engine/overlay/tests/test_omarchy_asahi.py

@@ -296,6 +296,41 @@ def test_changed_free_extent_is_rejected_before_partition_creation(self):
 
         self.assertEqual(dutil.add_calls, [])
 
+    def test_resize_geometry_drift_is_rejected_before_mutation(self):
+        source = FakePart("disk0s2", offset=0, size=500 * GIB, free=False)
+        plan = SimpleNamespace(**{
+            **self.plan.__dict__, "candidate_kind": "resize",
+            "source_identifier": source.name, "minimum_container_bytes": 320 * GIB,
+        })
+        dutil = FakeDiskUtil([[source]])
+        adapter = self._adapter(FakeInstaller(dutil))
+        adapter.preflight(plan)
+        with self.assertRaisesRegex(AsahiAdapterError, "approved source partition changed"):
+            adapter.prepare_target(plan)
+        self.assertEqual(dutil.resize_calls, [])
+        self.assertEqual(dutil.add_calls, [])
+
+    def test_prepared_checkpoint_rejects_each_changed_identity_field(self):
+        initial = self._adapter(FakeInstaller(FakeDiskUtil([[self.free]])))
+        initial.preflight(self.plan)
+        evidence = initial.prepare_target(self.plan)
+        for field, value in (
+            ("uuid", "DIFFERENT-UUID"), ("name", "disk0s9"),
+            ("size", 2 * GIB + 4096), ("offset", self.plan.offset_bytes + 4096),
+            ("type", "Linux Filesystem"),
+        ):
+            with self.subTest(field=field):
+                part = FakePart("disk0s4", offset=self.plan.offset_bytes,
+                                size=2 * GIB, free=False)
+                setattr(part, field, value)
+                installer = FakeInstaller(FakeDiskUtil([[part]]))
+                adapter = self._adapter(installer)
+                adapter.preflight(self.plan)
+                with self.assertRaises(AsahiAdapterError):
+                    adapter.validate_prepared_checkpoint(self.plan, evidence)
+                self.assertIsNone(adapter.target_part)
+                self.assertFalse(any(call[0] == "prepare_volume" for call in installer.ins.calls))
+
     def test_retry_reconciles_the_exact_prepared_apfs_partition(self):
         prepared = FakePart(
             "disk0s4",
@@ -688,6 +723,20 @@ def test_readback_is_exhaustive_for_declared_writes_and_skips_preserved_content(
         ):
             self.adapter.validate_repaired_content(self.plan)
 
+    def test_invalid_later_repair_member_does_not_write_earlier_partition(self):
+        for field, value in (("payload_member", "repair/missing.img"), ("size_bytes", 4096)):
+            with self.subTest(field=field):
+                expected = self.manifest["replacement_content"]["root"]
+                original = expected[field]
+                expected[field] = value
+                try:
+                    with self.assertRaises(AsahiAdapterError):
+                        self.adapter.rewrite_existing_content(self.plan)
+                    self.assertEqual(self.content["disk0s5"], b"boot-old")
+                    self.assertEqual(self.opened_partitions, [])
+                finally:
+                    expected[field] = original
+
     @staticmethod
     def _identity(content):
         return {

Engine/overlay/tests/test_omarchy_repair.py

@@ -96,6 +96,18 @@ def test_exact_existing_layout_produces_one_repair_candidate(self):
         self.assertEqual(candidate["length_bytes"], 137_438_953_472)
         self.assertRegex(candidate["identity_digest"], r"^sha256:[0-9a-f]{64}$")
 
+    def test_repair_image_must_fit_its_partition_and_raw_device_alignment(self):
+        for size in (1, 4097, self.parts[3].size + 4096):
+            with self.subTest(size=size):
+                self.manifest["replacement_content"]["boot"]["size_bytes"] = size
+                with self.assertRaisesRegex(InPlaceRepairError, "repair content size"):
+                    collect_repair_inventory(
+                        SimpleNamespace(sys_disk="disk0", parts=self.parts),
+                        self.manifest,
+                        disk_identity_reader=lambda _: "sha256:" + "5" * 64,
+                        filesystem_identity_reader=self._filesystem_identity,
+                    )
+
     def test_release_and_canary_share_the_concrete_repair_executor_adapter(self):
         plan = SimpleNamespace(plan_digest="a" * 64)
         release_adapter = FakeRepairAdapter()

Engine/overlay/tests/test_omarchy_stage1.py

@@ -134,12 +134,41 @@ def test_recovery_handoff_can_retry_after_authentication_failure(self):
         outcome = run_stage1(self.plan, resumed, retry)
 
         self.assertEqual(outcome, "awaiting_recovery")
-        self.assertEqual(retry.calls, ["prepare_recovery_handoff"])
+        self.assertEqual(retry.calls, [
+            "validate_installed_checkpoint", "prepare_recovery_handoff",
+        ])
         self.assertEqual(
             resumed.completion_outcome,
             "awaiting_recovery",
         )
 
+    def test_resume_validates_prepared_target_before_recording_mutation_intent(self):
+        self.journal.event("apfs_preparation_started")
+        self.journal.checkpoint("apfs-target-prepared", "apfs_preparation", b"target")
+        adapter = RecordingStage1Adapter(failure="validate_prepared_checkpoint")
+        with self.assertRaisesRegex(RuntimeError, "synthetic crash"):
+            run_stage1(self.plan, self.journal, adapter)
+        self.assertEqual(adapter.calls, ["validate_prepared_checkpoint"])
+        self.assertFalse(self.journal.has_event("stub_and_esp_started"))
+
+    def test_resume_validates_installed_content_before_recovery(self):
+        first = RecordingStage1Adapter(failure="prepare_recovery_handoff")
+        with self.assertRaises(RuntimeError):
+            run_stage1(self.plan, self.journal, first)
+        retry = RecordingStage1Adapter(failure="validate_installed_checkpoint")
+        with self.assertRaisesRegex(RuntimeError, "synthetic crash"):
+            run_stage1(self.plan, Journal(str(self.path)), retry)
+        self.assertEqual(retry.calls, ["validate_installed_checkpoint"])
+
+    def test_resume_without_a_checkpoint_validator_fails_closed(self):
+        self.journal.event("apfs_preparation_started")
+        self.journal.checkpoint("apfs-target-prepared", "apfs_preparation", b"target")
+        adapter = RecordingStage1Adapter()
+        adapter.validate_prepared_checkpoint = None
+        with self.assertRaisesRegex(Stage1Error, "validate_prepared_checkpoint"):
+            run_stage1(self.plan, self.journal, adapter)
+        self.assertEqual(adapter.calls, [])
+
     def test_explicit_recovery_retry_validates_checkpoint_then_runs_bless_only(self):
         target_evidence = b'{"partition_identifier":"disk0s4"}'
         installed_evidence = b'{"populated_partitions":["disk0s5"]}'
@@ -277,9 +306,13 @@ def validate_installed_checkpoint(
         target_evidence,
         installed_evidence,
     ):
-        self.calls.append("validate_installed_checkpoint")
+        self._record("validate_installed_checkpoint", plan)
         self.validated_evidence = (target_evidence, installed_evidence)
 
+    def validate_prepared_checkpoint(self, plan, target_evidence):
+        self._record("validate_prepared_checkpoint", plan)
+        self.validated_evidence = target_evidence
+
     def _record(self, name, plan):
         self.calls.append(name)
         if self.failure == name: