fix: validate digest-pinned rollback images safely

docs/operations/CURRENT_STATE.md

@@ -84,6 +84,11 @@ UFW rule details, and protected file metadata remain unverified.
   `status` helper, guarded rollback helper, and installer with exact sudoers
   command rules. None is installed on the VPS. CI contains no production
   deployment job; starting/stopping services remains owner-run.
+- Rollback preflight validates digest-pinned image references against archived
+  image IDs and local availability, then refuses before stopping services or
+  creating a backup because the current flow cannot safely switch immutable
+  digest references. This behavior is covered by synthetic local tests; it has
+  not been exercised on the VPS.
 - A repository-scoped official `act_runner` `0.2.13` installation is
   configured in `infra/runner/`: rootless Docker under `akyldash-runner`, 15 GiB
   ext4 loop-backed image/workspace storage, capacity 1, per-job limits (2 CPU,

docs/operations/ROLLBACK.md

@@ -18,11 +18,18 @@ sudo -n /usr/local/sbin/akyldash-rollback \
 The helper verifies archive integrity, Compose image references, deployment
 file hashes, and the availability of both prior local images. It makes and
 verifies a second cold backup, restores only the prior backend/OpenSearch image
-tags, waits for Compose healthchecks, and checks HTTPS locally. It refuses to
-restore if the current Compose/Caddyfile differs from the saved release. This
-does not restore data or automatically undo a release with an unknown schema
-change. The application has no formal migration framework; compatibility must
-be checked for every change. If it is uncertain, stop and ask the owner.
+tags, waits for Compose healthchecks, and checks HTTPS locally. Digest-pinned
+references are validated against their recorded local image IDs, but rollback
+fails before stopping services or making a backup when either image reference
+uses `@sha256:`. The helper cannot safely switch immutable digest references
+with the current release-recording flow. Do not retag a digest manually; digest
+rollback needs a separately reviewed atomic release-reference mechanism.
+
+It refuses to restore if the current Compose/Caddyfile differs from the saved
+release. This does not restore data or automatically undo a release with an
+unknown schema change. The application has no formal migration framework;
+compatibility must be checked for every change. If it is uncertain, stop and ask
+the owner.
 
 The initial owner-run deployment must record its running backend image ID in
 `/opt/akyldash/deployed-image.txt` with mode `0600`; the rollback helper updates

infra/deploy/rollback.py

@@ -19,6 +19,12 @@ ENV_FILE = Path("/etc/akyldash/production.env")
 BACKUP_DIR = Path("/var/backups/akyldash")
 BACKUP_TOOL = Path("/usr/local/lib/akyldash/archive.py")
 APP_SERVICES = {"backend", "opensearch"}
+IMAGE_TAG_RE = re.compile(r"[A-Za-z0-9][A-Za-z0-9._:/-]*")
+IMAGE_DIGEST_RE = re.compile(r"[A-Za-z0-9][A-Za-z0-9._:/-]*@sha256:[0-9a-f]{64}")
+
+
+def is_digest_reference(reference: str) -> bool:
+    return IMAGE_DIGEST_RE.fullmatch(reference) is not None
 
 
 def image_restore_plan(manifest: dict, configured: dict[str, str]) -> dict[str, tuple[str, str]]:
@@ -34,14 +40,48 @@ def image_restore_plan(manifest: dict, configured: dict[str, str]) -> dict[str,
         image_id = image.get("image_id")
         if reference != configured.get(service):
             raise ValueError(f"Compose image reference changed for {service}; refusing rollback")
-        if not isinstance(reference, str) or not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9._:/-]*", reference):
-            raise ValueError(f"{service} image reference is not a local tag")
+        if not isinstance(reference, str) or not (
+            IMAGE_TAG_RE.fullmatch(reference) or is_digest_reference(reference)
+        ):
+            raise ValueError(f"invalid image reference for {service}")
         if not isinstance(image_id, str) or not re.fullmatch(r"sha256:[0-9a-f]{64}", image_id):
             raise ValueError(f"invalid previous image ID for {service}")
         plan[service] = (image_id, reference)
     return plan
 
 
+def verify_previous_images(plan: dict[str, tuple[str, str]]) -> None:
+    for image_id, reference in plan.values():
+        try:
+            actual = run(["docker", "image", "inspect", "--format", "{{.Id}}", image_id], capture=True)
+        except subprocess.CalledProcessError as error:
+            raise ValueError(f"previous image is unavailable: {image_id}") from error
+        if actual != image_id:
+            raise ValueError(f"previous image is unavailable: {image_id}")
+        if is_digest_reference(reference):
+            try:
+                referenced_id = run(
+                    ["docker", "image", "inspect", "--format", "{{.Id}}", reference], capture=True
+                )
+            except subprocess.CalledProcessError as error:
+                raise ValueError(f"previous digest reference is unavailable: {reference}") from error
+            if referenced_id != image_id:
+                raise ValueError(f"previous digest reference does not match its recorded image: {reference}")
+
+
+def require_tag_rollback(plan: dict[str, tuple[str, str]]) -> None:
+    if any(is_digest_reference(reference) for _image_id, reference in plan.values()):
+        raise ValueError(
+            "rollback of digest-pinned images is not supported safely; no services or image references were changed"
+        )
+
+
+def tag_previous_images(plan: dict[str, tuple[str, str]]) -> None:
+    require_tag_rollback(plan)
+    for image_id, reference in plan.values():
+        run(["docker", "image", "tag", image_id, reference])
+
+
 def run(command: list[str], *, capture: bool = False) -> str:
     result = subprocess.run(command, check=True, text=True, capture_output=capture)
     return result.stdout.strip() if capture else ""
@@ -96,10 +136,8 @@ def rollback(backup: Path, data_compatible: bool) -> None:
     rendered = json.loads(compose("config", "--format", "json", capture=True))
     configured = {service: rendered["services"][service]["image"] for service in APP_SERVICES}
     plan = image_restore_plan(manifest, configured)
-    for image_id, _reference in plan.values():
-        actual = run(["docker", "image", "inspect", "--format", "{{.Id}}", image_id], capture=True)
-        if actual != image_id:
-            raise ValueError(f"previous image is unavailable: {image_id}")
+    verify_previous_images(plan)
+    require_tag_rollback(plan)
 
     backup_name = f"akyldash-pre-rollback-{dt.datetime.now(dt.timezone.utc):%Y%m%dT%H%M%SZ}.tar.gz"
     backup_output = BACKUP_DIR / backup_name
@@ -123,8 +161,7 @@ def rollback(backup: Path, data_compatible: bool) -> None:
         compose("up", "--detach")
         raise
 
-    for image_id, reference in plan.values():
-        run(["docker", "image", "tag", image_id, reference])
+    tag_previous_images(plan)
     compose("up", "--detach", "--no-build", "--wait", "--wait-timeout", "300")
     run(
         [

infra/deploy/test_rollback.py

@@ -1,6 +1,8 @@
+import subprocess
 import unittest
+from unittest.mock import patch
 
-from rollback import image_restore_plan
+from rollback import image_restore_plan, require_tag_rollback, tag_previous_images, verify_previous_images
 
 
 class RollbackPlanTests(unittest.TestCase):
@@ -28,9 +30,56 @@ class RollbackPlanTests(unittest.TestCase):
     def test_refuses_digest_reference_that_cannot_be_retagged(self):
         self.manifest["images"]["backend"]["configured"] = "registry/backend@sha256:" + "c" * 64
         self.configured["backend"] = self.manifest["images"]["backend"]["configured"]
-        with self.assertRaisesRegex(ValueError, "local tag"):
+        plan = image_restore_plan(self.manifest, self.configured)
+        self.assertEqual(plan["backend"], ("sha256:" + "a" * 64, self.configured["backend"]))
+        with self.assertRaisesRegex(ValueError, "not supported safely"):
+            require_tag_rollback(plan)
+
+    def test_refuses_malformed_digest_reference(self):
+        self.manifest["images"]["backend"]["configured"] = "registry/backend@sha256:xyz"
+        self.configured["backend"] = self.manifest["images"]["backend"]["configured"]
+        with self.assertRaisesRegex(ValueError, "invalid image reference"):
             image_restore_plan(self.manifest, self.configured)
 
+    def test_refuses_invalid_recorded_image_id(self):
+        self.manifest["images"]["backend"]["image_id"] = "sha256:not-a-digest"
+        with self.assertRaisesRegex(ValueError, "invalid previous image ID"):
+            image_restore_plan(self.manifest, self.configured)
+
+    def test_digest_reference_must_resolve_to_recorded_image_id(self):
+        digest = "registry/backend@sha256:" + "c" * 64
+        self.manifest["images"]["backend"]["configured"] = digest
+        self.configured["backend"] = digest
+        plan = image_restore_plan(self.manifest, self.configured)
+        with patch("rollback.run", side_effect=["sha256:" + "a" * 64, "sha256:" + "a" * 64,
+                                                 "sha256:" + "b" * 64]):
+            verify_previous_images(plan)
+        with patch("rollback.run", side_effect=["sha256:" + "a" * 64, "sha256:" + "d" * 64]):
+            with self.assertRaisesRegex(ValueError, "does not match its recorded image"):
+                verify_previous_images(plan)
+
+    def test_missing_previous_image_fails_closed(self):
+        plan = image_restore_plan(self.manifest, self.configured)
+        with patch("rollback.run", side_effect=subprocess.CalledProcessError(1, "docker")):
+            with self.assertRaisesRegex(ValueError, "previous image is unavailable"):
+                verify_previous_images(plan)
+
+    def test_digest_reference_is_never_tagged(self):
+        digest = "registry/backend@sha256:" + "c" * 64
+        plan = {"backend": ("sha256:" + "a" * 64, digest)}
+        with patch("rollback.run") as run_mock:
+            with self.assertRaisesRegex(ValueError, "not supported safely"):
+                tag_previous_images(plan)
+        run_mock.assert_not_called()
+
+    def test_tag_rollback_keeps_existing_image_tag_operation(self):
+        plan = image_restore_plan(self.manifest, self.configured)
+        with patch("rollback.run") as run_mock:
+            tag_previous_images(plan)
+        self.assertEqual(run_mock.call_count, 2)
+        run_mock.assert_any_call(["docker", "image", "tag", "sha256:" + "a" * 64,
+                                  "akyldash/backend:local"])
+
 
 if __name__ == "__main__":
     unittest.main()