Merge remote-tracking branch 'origin/main' into fix/rollback-image-digest-reference

# Conflicts:
#	CHANGELOG.md
#	docs/operations/CURRENT_STATE.md

.gitea/workflows/ci.yaml

@@ -51,7 +51,8 @@ jobs:
             infra/runner/test-install-runner-idempotency.sh infra/runner/test-install-runner-guard.sh \
             infra/runner/test-ci-container-builds.sh
           python3 -c "import ast,pathlib; ast.parse(pathlib.Path('infra/runner/secure-runner-image.py').read_text())"
-      - name: Run rollback helper tests
+          python3 -c "import ast,pathlib; ast.parse(pathlib.Path('infra/deploy/validate_release.py').read_text())"
+      - name: Run deployment contract and rollback tests
         run: python3 -m unittest discover -s infra/deploy -p 'test_*.py' -v
       - name: Validate production Compose and Caddyfile
         run: |

CHANGELOG.md

@@ -2,6 +2,13 @@
 
 ## Не выпущено
 
+- Rollback preflight проверяет digest-pinned image references по сохранённому
+  image ID и останавливается до изменений deployment; переключение production
+  на digest и полноценный digest rollback пока не реализованы.
+- Infrastructure: добавлен офлайн-контракт release manifest с точной схемой,
+  digest-ссылками и repository allowlist; проверен Compose override с
+  `--no-build` как требованием будущего deploy. Публикация, provenance и
+  production-деплой не включены.
 - Добавлен read-only возобновляемый inventory корпуса с полным хешированием,
   потоковой сверкой, индексируемыми диапазонами путей и provenance SHA-256
   проверяющего; значения `stat` хранятся без ограничения SQLite INTEGER.

docs/operations/CURRENT_STATE.md

@@ -89,6 +89,15 @@ 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.
+- This branch adds an offline release-manifest validator and tests for the
+  repository allowlist intended for a future root-owned host integration. The
+  manifest format does not prove source provenance. No registry, host policy,
+  publish credential, or deployment integration is configured or installed.
 - 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/DEPLOYMENT.md

@@ -75,6 +75,45 @@ repository defines sudo helpers for fixed verification, read-only Compose
 `config`/`status`, and guarded rollback, but none is installed on the VPS. They
 do not start or stop services.
 
+## Release image contract (preparation only)
+
+`infra/deploy/validate_release.py` checks an offline JSON release manifest
+against an explicitly supplied repository allowlist. Schema version 1 requires
+exactly `schema_version`, `source_ref`, `commit`, and `images`; the source ref
+must be `refs/heads/main`, the commit must be a 40-character lowercase Git
+SHA-1, and the images must contain exactly `backend` and `opensearch` as
+`repository@sha256:<64 lowercase hex>` references. The policy has schema
+version 1 and one non-empty repository allowlist for each service. Unknown or
+duplicate JSON keys are rejected.
+
+The validator takes the manifest path and an explicit policy path:
+
+```bash
+python3 infra/deploy/validate_release.py release.json --policy /etc/akyldash/release-policy.json
+```
+
+The policy shape is `{"schema_version":1,"repositories":{"backend":["registry.invalid/akyldash/backend"],"opensearch":["registry.invalid/akyldash/opensearch"]}}`.
+These `registry.invalid` names are format examples only; this change does not
+create a policy file or approve a registry.
+
+The repository name in the policy must be approved for the registry. No real
+registry namespace or root-owned host policy is configured by this change; do
+not use the synthetic `registry.invalid` names in tests as production values.
+The host integration must load policy only from its fixed root-owned path.
+This validator checks format and allowlist only: a manifest can claim a main
+ref and commit, but is not proof of source, build provenance, signature, or
+trusted publication. CI still has no publish credentials, image publication,
+host helper, or deploy trigger.
+
+Compose merges an image override with the existing service `build` entries.
+The regression test renders fake digest references and confirms both entries
+remain; any future deploy command must use `docker compose up --no-build` so
+the host cannot silently build a different image. This does not make digest
+rollback available: the current rollback helper expects local tags. A separate
+host-side deploy/rollback design, owner-approved registry permissions and
+retention, and manual backup/health/rollback end-to-end validation are still
+required before publishing or enabling CD.
+
 ## CI
 
 `.gitea/workflows/ci.yaml` runs the Python unit suites and builds both

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_release_manifest.py

@@ -0,0 +1,131 @@
+import json
+import os
+import shutil
+import subprocess
+import tempfile
+import unittest
+from pathlib import Path
+
+from validate_release import validate_manifest
+
+ROOT = Path(__file__).resolve().parents[2]
+REPOSITORIES = {
+    "backend": ["registry.invalid/akyldash/backend"],
+    "opensearch": ["registry.invalid/akyldash/opensearch"],
+}
+
+
+class ReleaseManifestTests(unittest.TestCase):
+    def setUp(self):
+        self.policy = {"schema_version": 1, "repositories": REPOSITORIES}
+        self.manifest = {
+            "schema_version": 1,
+            "source_ref": "refs/heads/main",
+            "commit": "a" * 40,
+            "images": {
+                "backend": "registry.invalid/akyldash/backend@sha256:" + "b" * 64,
+                "opensearch": "registry.invalid/akyldash/opensearch@sha256:" + "c" * 64,
+            },
+        }
+
+    def test_accepts_both_policy_pinned_images(self):
+        self.assertEqual(validate_manifest(self.manifest, self.policy), self.manifest)
+
+    def test_rejects_untrusted_ref(self):
+        self.manifest["source_ref"] = "refs/pull/12/head"
+        with self.assertRaisesRegex(ValueError, "source_ref"):
+            validate_manifest(self.manifest, self.policy)
+
+    def test_rejects_tag_instead_of_digest(self):
+        self.manifest["images"]["backend"] = "registry.invalid/akyldash/backend:latest"
+        with self.assertRaisesRegex(ValueError, "sha256"):
+            validate_manifest(self.manifest, self.policy)
+
+    def test_rejects_empty_image_reference(self):
+        self.manifest["images"]["backend"] = ""
+        with self.assertRaisesRegex(ValueError, "sha256"):
+            validate_manifest(self.manifest, self.policy)
+
+    def test_rejects_invalid_commit(self):
+        self.manifest["commit"] = "not-a-commit"
+        with self.assertRaisesRegex(ValueError, "commit"):
+            validate_manifest(self.manifest, self.policy)
+
+    def test_rejects_repository_outside_policy(self):
+        self.manifest["images"]["backend"] = "registry.invalid/other/backend@sha256:" + "b" * 64
+        with self.assertRaisesRegex(ValueError, "not allowed"):
+            validate_manifest(self.manifest, self.policy)
+
+    def test_rejects_missing_or_extra_images(self):
+        self.manifest["images"].pop("opensearch")
+        with self.assertRaisesRegex(ValueError, "exactly backend and opensearch"):
+            validate_manifest(self.manifest, self.policy)
+        self.manifest["images"]["opensearch"] = "registry.invalid/akyldash/opensearch@sha256:" + "c" * 64
+        self.manifest["images"]["extra"] = "registry.invalid/akyldash/extra@sha256:" + "d" * 64
+        with self.assertRaisesRegex(ValueError, "exactly backend and opensearch"):
+            validate_manifest(self.manifest, self.policy)
+
+    def test_rejects_duplicate_json_keys(self):
+        with tempfile.TemporaryDirectory() as directory:
+            manifest_path = Path(directory) / "manifest.json"
+            policy_path = Path(directory) / "policy.json"
+            manifest_path.write_text(
+                '{"schema_version":1,"source_ref":"refs/heads/main","commit":"' + "a" * 40
+                + '","images":{"backend":"first","backend":"second","opensearch":"third"}}',
+                encoding="utf-8",
+            )
+            policy_path.write_text(json.dumps(self.policy), encoding="utf-8")
+            command = [
+                "python3", str(ROOT / "infra/deploy/validate_release.py"), str(manifest_path),
+                "--policy", str(policy_path),
+            ]
+            result = subprocess.run(command, text=True, capture_output=True)
+        self.assertEqual(result.returncode, 1)
+        self.assertIn("duplicate JSON key: backend", result.stderr)
+
+    def test_rejects_malformed_json(self):
+        with tempfile.TemporaryDirectory() as directory:
+            manifest_path = Path(directory) / "manifest.json"
+            policy_path = Path(directory) / "policy.json"
+            manifest_path.write_text('{"schema_version":', encoding="utf-8")
+            policy_path.write_text(json.dumps(self.policy), encoding="utf-8")
+            result = subprocess.run(
+                ["python3", str(ROOT / "infra/deploy/validate_release.py"), str(manifest_path), "--policy", str(policy_path)],
+                text=True,
+                capture_output=True,
+            )
+        self.assertEqual(result.returncode, 1)
+        self.assertIn("cannot read valid JSON", result.stderr)
+
+    @unittest.skipUnless(shutil.which("docker"), "Docker CLI is unavailable")
+    def test_compose_override_renders_digest_images_and_preserves_build(self):
+        backend = "registry.invalid/akyldash/backend@sha256:" + "b" * 64
+        opensearch = "registry.invalid/akyldash/opensearch@sha256:" + "c" * 64
+        with tempfile.TemporaryDirectory() as directory:
+            temp = Path(directory)
+            secret = temp / "review_secret"
+            secret.touch()
+            override = temp / "release-override.json"
+            override.write_text(json.dumps({"services": {
+                "backend": {"image": backend},
+                "opensearch": {"image": opensearch},
+            }}), encoding="utf-8")
+            command = [
+                "docker", "compose", "--env-file", str(ROOT / "deploy/production/.env.example"),
+                "-f", str(ROOT / "deploy/production/compose.yaml"), "-f", str(override),
+                "config", "--format", "json",
+            ]
+            result = subprocess.run(
+                command, text=True, capture_output=True,
+                env={**os.environ, "REVIEW_SECRET_FILE": str(secret)},
+            )
+        self.assertEqual(result.returncode, 0, result.stderr)
+        services = json.loads(result.stdout)["services"]
+        self.assertEqual(services["backend"]["image"], backend)
+        self.assertEqual(services["opensearch"]["image"], opensearch)
+        self.assertIn("build", services["backend"])
+        self.assertIn("build", services["opensearch"])
+
+
+if __name__ == "__main__":
+    unittest.main()

infra/deploy/test_rollback.py

@@ -1,6 +1,12 @@
+import json
+import subprocess
+import sys
 import unittest
+from types import SimpleNamespace
+from unittest.mock import patch
 
-from rollback import image_restore_plan
+import rollback
+from rollback import image_restore_plan, require_tag_rollback, tag_previous_images, verify_previous_images
 
 
 class RollbackPlanTests(unittest.TestCase):
@@ -28,9 +34,96 @@ 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_digest_rollback_fails_before_stopping_or_changing_references(self):
+        backend_id = "sha256:" + "a" * 64
+        opensearch_id = "sha256:" + "b" * 64
+        digest = "registry/backend@sha256:" + "c" * 64
+        manifest = {
+            "images": {
+                "backend": {"configured": digest, "image_id": backend_id},
+                "opensearch": self.manifest["images"]["opensearch"],
+            },
+            "deployment_files": {
+                "deploy/production/compose.yaml": "compose-hash",
+                "deploy/production/Caddyfile": "caddy-hash",
+            },
+        }
+        rendered = {
+            "services": {
+                "backend": {"image": digest},
+                "opensearch": {"image": self.configured["opensearch"]},
+            }
+        }
+        fake_archive = SimpleNamespace(
+            digest=lambda path: "compose-hash" if path.name == "compose.yaml" else "caddy-hash",
+            verify_archive=lambda _backup: manifest,
+        )
+        backup = rollback.BACKUP_DIR / "synthetic.tar.gz"
+        with (
+            patch("rollback.Path.is_file", return_value=True),
+            patch("rollback.Path.is_symlink", return_value=False),
+            patch("rollback.Path.stat", return_value=SimpleNamespace(st_uid=0, st_mode=0o100600)),
+            patch("rollback.compose", return_value=json.dumps(rendered)) as compose_mock,
+            patch("rollback.run", side_effect=[backend_id, backend_id, opensearch_id]) as run_mock,
+            patch.dict(sys.modules, {"archive": fake_archive}),
+        ):
+            with self.assertRaisesRegex(ValueError, "not supported safely"):
+                rollback.rollback(backup, data_compatible=True)
+        compose_mock.assert_called_once_with("config", "--format", "json", capture=True)
+        self.assertEqual(run_mock.call_count, 3)
+        self.assertTrue(all(call.args[0][:3] == ["docker", "image", "inspect"]
+                            for call in run_mock.call_args_list))
+
+    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()

infra/deploy/validate_release.py

@@ -0,0 +1,106 @@
+#!/usr/bin/env python3
+"""Validate the shape and repository allowlist of a prepared release manifest."""
+
+from __future__ import annotations
+
+import argparse
+import json
+import re
+import sys
+from pathlib import Path
+
+SERVICES = {"backend", "opensearch"}
+COMMIT_RE = re.compile(r"[0-9a-f]{40}\Z")
+REPOSITORY_RE = re.compile(
+    r"[a-z0-9]+(?:[._-][a-z0-9]+)*(?::[0-9]+)?"
+    r"(?:/[a-z0-9]+(?:[._-][a-z0-9]+)*)*\Z"
+)
+IMAGE_RE = re.compile(r"(.+)@sha256:([0-9a-f]{64})\Z")
+
+
+def unique_object(pairs: list[tuple[str, object]]) -> dict[str, object]:
+    result: dict[str, object] = {}
+    for key, value in pairs:
+        if key in result:
+            raise ValueError(f"duplicate JSON key: {key}")
+        result[key] = value
+    return result
+
+
+def read_json(path: Path) -> object:
+    try:
+        return json.loads(path.read_text(encoding="utf-8"), object_pairs_hook=unique_object)
+    except (OSError, UnicodeError, json.JSONDecodeError) as error:
+        raise ValueError(f"cannot read valid JSON from {path}: {error}") from error
+
+
+def validate_policy(policy: object) -> dict[str, list[str]]:
+    if not isinstance(policy, dict) or set(policy) != {"schema_version", "repositories"}:
+        raise ValueError("policy must contain exactly schema_version and repositories")
+    if type(policy["schema_version"]) is not int or policy["schema_version"] != 1:
+        raise ValueError("unsupported policy schema_version")
+    repositories = policy["repositories"]
+    if not isinstance(repositories, dict) or set(repositories) != SERVICES:
+        raise ValueError("policy repositories must contain exactly backend and opensearch")
+    result: dict[str, list[str]] = {}
+    for service in sorted(SERVICES):
+        allowed = repositories[service]
+        if not isinstance(allowed, list) or not allowed:
+            raise ValueError(f"policy repository allowlist for {service} must be a non-empty list")
+        if any(not isinstance(repository, str) or not REPOSITORY_RE.fullmatch(repository) for repository in allowed):
+            raise ValueError(f"policy contains an invalid repository for {service}")
+        if len(set(allowed)) != len(allowed):
+            raise ValueError(f"policy repository allowlist for {service} contains duplicates")
+        result[service] = allowed
+    return result
+
+
+def validate_manifest(manifest: object, policy: object) -> dict[str, object]:
+    allowed_repositories = validate_policy(policy)
+    if not isinstance(manifest, dict) or set(manifest) != {
+        "schema_version", "source_ref", "commit", "images"
+    }:
+        raise ValueError("manifest must contain exactly schema_version, source_ref, commit, and images")
+    if type(manifest["schema_version"]) is not int or manifest["schema_version"] != 1:
+        raise ValueError("unsupported manifest schema_version")
+    if manifest["source_ref"] != "refs/heads/main":
+        raise ValueError("manifest source_ref must be refs/heads/main")
+    commit = manifest["commit"]
+    if not isinstance(commit, str) or not COMMIT_RE.fullmatch(commit):
+        raise ValueError("manifest commit must be a 40-character lowercase Git SHA-1")
+    images = manifest["images"]
+    if not isinstance(images, dict) or set(images) != SERVICES:
+        raise ValueError("manifest images must contain exactly backend and opensearch")
+    for service in sorted(SERVICES):
+        image = images[service]
+        match = IMAGE_RE.fullmatch(image) if isinstance(image, str) else None
+        if not match:
+            raise ValueError(f"{service} image must use repository@sha256:<64 lowercase hex>")
+        repository = match.group(1)
+        if not REPOSITORY_RE.fullmatch(repository):
+            raise ValueError(f"{service} image has an invalid OCI repository")
+        if repository not in allowed_repositories[service]:
+            raise ValueError(f"{service} image repository is not allowed by policy")
+    return manifest
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("manifest", type=Path)
+    parser.add_argument("--policy", required=True, type=Path)
+    args = parser.parse_args()
+    try:
+        manifest = validate_manifest(read_json(args.manifest), read_json(args.policy))
+    except ValueError as error:
+        print(f"ERROR: {error}", file=sys.stderr)
+        return 1
+    images = manifest["images"]
+    print(
+        "PASS release manifest "
+        f"commit={manifest['commit']} backend={images['backend']} opensearch={images['opensearch']}"
+    )
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())