diff --git a/ops/verify_gcp_canonical_lifecycle.py b/ops/verify_gcp_canonical_lifecycle.py index b70aa3c..5d182b5 100644 --- a/ops/verify_gcp_canonical_lifecycle.py +++ b/ops/verify_gcp_canonical_lifecycle.py @@ -16,10 +16,12 @@ from typing import Any try: from .private_receipt_io import print_private_receipt_summary from .restore_gcp_generated_postgres_snapshot import DEFAULT_RESTORE_SERVICE_ACCOUNT + from .verify_postgres_parity_manifest import REVIEWED_MANIFEST_SQL_SHA256 from .verify_vps_canonical_snapshot_delta import DELTA_SCHEMA, parse_timestamp, write_private_json except ImportError: # pragma: no cover - direct script execution from private_receipt_io import print_private_receipt_summary from restore_gcp_generated_postgres_snapshot import DEFAULT_RESTORE_SERVICE_ACCOUNT + from verify_postgres_parity_manifest import REVIEWED_MANIFEST_SQL_SHA256 from verify_vps_canonical_snapshot_delta import DELTA_SCHEMA, parse_timestamp, write_private_json SOURCE_SCHEMA = "livingip.sourceSnapshotReceipt.v2" @@ -405,6 +407,11 @@ def verify_lifecycle( and delta_current.get("manifest_sql_sha256") == (current_source.get("manifest") or {}).get("manifest_sql_sha256") and delta_baseline.get("manifest_sql_sha256") == delta_current.get("manifest_sql_sha256"), + "source_delta_manifest_sql_reviewed": delta_baseline.get("manifest_sql_sha256") == REVIEWED_MANIFEST_SQL_SHA256 + and delta_current.get("manifest_sql_sha256") == REVIEWED_MANIFEST_SQL_SHA256 + and (source.get("manifest") or {}).get("manifest_sql_sha256") == REVIEWED_MANIFEST_SQL_SHA256 + and (current_source.get("manifest") or {}).get("manifest_sql_sha256") == REVIEWED_MANIFEST_SQL_SHA256 + and capture_binding.get("manifest_sql_sha256") == REVIEWED_MANIFEST_SQL_SHA256, "source_delta_authorizations_bound": delta_baseline.get("capture_authorization_ref") == source.get("capture_authorization_ref") and delta_current.get("capture_authorization_ref") == current_source.get("capture_authorization_ref"), diff --git a/tests/test_verify_gcp_canonical_lifecycle.py b/tests/test_verify_gcp_canonical_lifecycle.py index b2a26da..c0da1f0 100644 --- a/tests/test_verify_gcp_canonical_lifecycle.py +++ b/tests/test_verify_gcp_canonical_lifecycle.py @@ -120,6 +120,7 @@ def lifecycle_receipts(tmp_path: Path) -> dict[str, Path]: "provenance_binding_sha256": "c" * 64, "dump_sha256": "a" * 64, "manifest_sha256": "b" * 64, + "manifest_sql_sha256": REVIEWED_MANIFEST_SQL_SHA256, "checks": { name: True for name in ( @@ -504,6 +505,39 @@ def test_caller_authored_bundle_is_preparation_only_even_when_all_checks_pass(tm assert "current private GCP staging" not in result["strongest_claim_allowed"] +def test_rebound_unreviewed_manifest_sql_digest_fails_closed(tmp_path: Path) -> None: + paths = lifecycle_receipts(tmp_path) + unreviewed_digest = "f" * 64 + + source = json.loads(paths["source_path"].read_text()) + source["manifest"]["manifest_sql_sha256"] = unreviewed_digest + write_json(paths["source_path"], source) + current_source = json.loads(paths["current_source_path"].read_text()) + current_source["manifest"]["manifest_sql_sha256"] = unreviewed_digest + write_json(paths["current_source_path"], current_source) + + restore = json.loads(paths["restore_path"].read_text()) + restore["source"]["capture_receipt"]["sha256"] = sha256(paths["source_path"]) + restore["source"]["capture_receipt"]["manifest_sql_sha256"] = unreviewed_digest + write_json(paths["restore_path"], restore) + + delta = json.loads(paths["source_delta_path"].read_text()) + delta["baseline"]["capture_receipt_sha256"] = sha256(paths["source_path"]) + delta["baseline"]["manifest_sql_sha256"] = unreviewed_digest + delta["current"]["capture_receipt_sha256"] = sha256(paths["current_source_path"]) + delta["current"]["manifest_sql_sha256"] = unreviewed_digest + write_json(paths["source_delta_path"], delta) + + cleanup = json.loads(paths["cleanup_path"].read_text()) + cleanup["restore_receipt"]["sha256"] = sha256(paths["restore_path"]) + write_json(paths["cleanup_path"], cleanup) + + result = verify_lifecycle(**paths) + + assert result["status"] == "fail" + assert "source_delta_manifest_sql_reviewed" in result["failed_checks"] + + def test_local_parity_or_missing_cleanup_fails_closed(tmp_path: Path) -> None: paths = lifecycle_receipts(tmp_path) parity = json.loads(paths["parity_path"].read_text())