#!/usr/bin/env python3 """Bind a no-send reasoning receipt to live GCE metadata on the replay host.""" from __future__ import annotations import argparse import json import stat from datetime import UTC, datetime from pathlib import Path from typing import Any try: from .private_receipt_io import print_private_receipt_summary, write_private_json from .restore_gcp_generated_postgres_snapshot import ( DEFAULT_COMPUTE_INSTANCE, DEFAULT_COMPUTE_ZONE, DEFAULT_PRIVATE_NETWORK, DEFAULT_PROJECT, DEFAULT_RESTORE_SERVICE_ACCOUNT, RUN_ID_RE, TARGET_DATABASE_RE, gce_compute_identity, sha256_file, ) except ImportError: # pragma: no cover - direct script execution on the GCP VM from private_receipt_io import print_private_receipt_summary, write_private_json from restore_gcp_generated_postgres_snapshot import ( DEFAULT_COMPUTE_INSTANCE, DEFAULT_COMPUTE_ZONE, DEFAULT_PRIVATE_NETWORK, DEFAULT_PROJECT, DEFAULT_RESTORE_SERVICE_ACCOUNT, RUN_ID_RE, TARGET_DATABASE_RE, gce_compute_identity, sha256_file, ) REASONING_SCHEMA = "livingip.gcpGeneratedDbBlindClaimCanary.v1" ATTESTATION_SCHEMA = "livingip.gcpReasoningComputeAttestation.v1" def utc_now() -> str: return datetime.now(UTC).isoformat() def parse_timestamp(value: Any, label: str) -> datetime: text = str(value or "") if text.endswith("Z"): text = text[:-1] + "+00:00" try: parsed = datetime.fromisoformat(text) except ValueError as exc: raise ValueError(f"{label} is not an ISO-8601 timestamp") from exc if parsed.tzinfo is None: raise ValueError(f"{label} must include a timezone") return parsed.astimezone(UTC) def load_reasoning_receipt(path: Path) -> dict[str, Any]: try: mode = path.lstat().st_mode except OSError as exc: raise ValueError(f"reasoning receipt is unavailable: {exc}") from exc if not stat.S_ISREG(mode) or path.is_symlink(): raise ValueError("reasoning receipt must be a regular non-symlink file") try: payload = json.loads(path.read_text(encoding="utf-8")) except json.JSONDecodeError as exc: raise ValueError("reasoning receipt is invalid JSON") from exc if not isinstance(payload, dict): raise ValueError("reasoning receipt must be a JSON object") return payload def attest_reasoning_compute( *, reasoning_receipt_path: Path, restore_run_id: str, target_database: str, project: str = DEFAULT_PROJECT, expected_compute_instance: str = DEFAULT_COMPUTE_INSTANCE, expected_compute_zone: str = DEFAULT_COMPUTE_ZONE, expected_private_network: str = DEFAULT_PRIVATE_NETWORK, expected_restore_service_account: str = DEFAULT_RESTORE_SERVICE_ACCOUNT, metadata_timeout: float = 10.0, generated_at_utc: str | None = None, ) -> dict[str, Any]: if not RUN_ID_RE.fullmatch(restore_run_id): raise ValueError("restore run id is invalid") if not TARGET_DATABASE_RE.fullmatch(target_database): raise ValueError("target database must be a disposable teleo_clone_* database") reasoning_receipt_path = reasoning_receipt_path.resolve() reasoning = load_reasoning_receipt(reasoning_receipt_path) started = parse_timestamp(reasoning.get("generated_at_utc"), "reasoning start") completed = parse_timestamp(reasoning.get("completed_at_utc"), "reasoning completion") checks = { "schema": reasoning.get("schema") == REASONING_SCHEMA, "status": reasoning.get("status") == "pass" and not reasoning.get("errors"), "target_database": reasoning.get("target_database") == target_database, "source_compute": reasoning.get("source_compute") == expected_compute_instance, "chronology": started <= completed, "no_telegram_send": reasoning.get("posted_to_telegram") is False, "no_database_write": reasoning.get("database_write_attempted") is False, } failed = sorted(name for name, passed in checks.items() if not passed) if failed: raise ValueError("reasoning receipt failed attestation preflight: " + ", ".join(failed)) compute = gce_compute_identity( project, expected_compute_instance, expected_compute_zone, expected_private_network, expected_restore_service_account, timeout=metadata_timeout, ) return { "artifact": "gcp_reasoning_compute_attestation", "schema": ATTESTATION_SCHEMA, "status": "pass", "generated_at_utc": generated_at_utc or utc_now(), "restore_run_id": restore_run_id, "target_database": target_database, "reasoning_receipt": { "sha256": sha256_file(reasoning_receipt_path), "schema": reasoning["schema"], "generated_at_utc": reasoning["generated_at_utc"], "completed_at_utc": reasoning["completed_at_utc"], "claimed_source_compute": reasoning["source_compute"], }, "compute": compute, "checks": {**checks, "gce_metadata_identity": all(compute.get("checks", {}).values())}, "method": "gce_metadata_server_v1", "mutation": { "cloudsql_changed": False, "compute_changed": False, "database_changed": False, "telegram_message_sent": False, }, } def main() -> int: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--reasoning-receipt", required=True, type=Path) parser.add_argument("--restore-run-id", required=True) parser.add_argument("--target-db", required=True) parser.add_argument("--project", default=DEFAULT_PROJECT) parser.add_argument("--expected-compute-instance", default=DEFAULT_COMPUTE_INSTANCE) parser.add_argument("--expected-compute-zone", default=DEFAULT_COMPUTE_ZONE) parser.add_argument("--expected-private-network", default=DEFAULT_PRIVATE_NETWORK) parser.add_argument("--expected-restore-service-account", default=DEFAULT_RESTORE_SERVICE_ACCOUNT) parser.add_argument("--metadata-timeout", default=10.0, type=float) parser.add_argument("--output", required=True, type=Path) args = parser.parse_args() try: receipt = attest_reasoning_compute( reasoning_receipt_path=args.reasoning_receipt, restore_run_id=args.restore_run_id, target_database=args.target_db, project=args.project, expected_compute_instance=args.expected_compute_instance, expected_compute_zone=args.expected_compute_zone, expected_private_network=args.expected_private_network, expected_restore_service_account=args.expected_restore_service_account, metadata_timeout=args.metadata_timeout, ) write_private_json(args.output, receipt) except (OSError, ValueError, RuntimeError) as exc: parser.error(str(exc)) print_private_receipt_summary(args.output, receipt) return 0 if __name__ == "__main__": raise SystemExit(main())