teleo/teleo-infrastructure
8
0
Fork 0
Code Issues Pull requests 1 Projects Releases Packages Wiki Activity Actions 1

Pin lifecycle receipts to the reviewed parity manifest

Browse source
This commit is contained in:
twentyOne2x 2026-07-15 10:59:43 +02:00
parent 52fc0a88fe
commit 1c6b7b182e
2 changed files with 41 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

7
ops/verify_gcp_canonical_lifecycle.py
View file

@ -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"),

34
tests/test_verify_gcp_canonical_lifecycle.py
View file

@ -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())