diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 1db7ccd..fd752f9 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -45,6 +45,7 @@ jobs: ops/check_gcp_infra_readiness.py \ ops/plan_gcp_iam_split.py \ ops/sqlite_to_postgres_dump.py \ + ops/verify_gcp_cloudsql_restore_readback.py \ telegram/approvals.py \ scripts/check_crabbox_ci_contract.py \ scripts/check_llm_refinement_contract.py \ @@ -54,6 +55,7 @@ jobs: tests/test_decision_engine_replay.py \ tests/test_evaluate_agent_routing.py \ tests/test_gcp_cloudsql_restore_drill.py \ + tests/test_gcp_cloudsql_restore_readback.py \ tests/test_gcp_iam_split_apply.py \ tests/test_gcp_iam_split_plan.py \ tests/test_gcp_readiness_workflow.py \ diff --git a/docs/gcp-infra-hardening-20260707.md b/docs/gcp-infra-hardening-20260707.md index e7bce59..f16dc79 100644 --- a/docs/gcp-infra-hardening-20260707.md +++ b/docs/gcp-infra-hardening-20260707.md @@ -137,6 +137,17 @@ be converted and restored into PostgreSQL with table/row-count parity, but the same import still needs to run against the GCP Cloud SQL standby through an approved GCP auth and network path. +After Cloud SQL import, use the generated `target-counts.sql` and verify it with: + +```bash +python3 ops/verify_gcp_cloudsql_restore_readback.py \ + --drill-proof outputs/gcp-infra-hardening-20260707/proofs/gcp-cloudsql-restore-drill-.json \ + --target-counts-csv outputs/gcp-infra-hardening-20260707/proofs/gcp-cloudsql-target-counts-.csv \ + --output outputs/gcp-infra-hardening-20260707/proofs/gcp-cloudsql-restore-readback-verification-.json +``` + +The verifier must return `status = pass` before claiming row-count parity in GCP. + ## IAM Split Plan Do not make `sa-artifact-builder@teleo-501523.iam.gserviceaccount.com` the broad diff --git a/docs/gcp-kb-cloudsql-restore-runbook.md b/docs/gcp-kb-cloudsql-restore-runbook.md index f669d0e..e3b87c4 100644 --- a/docs/gcp-kb-cloudsql-restore-runbook.md +++ b/docs/gcp-kb-cloudsql-restore-runbook.md @@ -96,6 +96,28 @@ The import operation alone is still not the final proof. The final proof needs `target-counts.sql` run against `teleo-pgvector-standby` and compared to the source counts in the drill proof. +After the import operation is `DONE`, run the generated count query from a +trusted VPC runtime or Cloud SQL connector path and retain CSV output: + +```bash +psql "$TELEO_CLOUDSQL_DATABASE_URL" \ + --csv \ + -f outputs/gcp-infra-hardening-20260707/private-cloudsql-restore-drills/gcp-cloudsql-restore-drill-/target-counts.sql \ + > outputs/gcp-infra-hardening-20260707/proofs/gcp-cloudsql-target-counts-.csv +``` + +Then compare the Cloud SQL readback to the source proof: + +```bash +python3 ops/verify_gcp_cloudsql_restore_readback.py \ + --drill-proof outputs/gcp-infra-hardening-20260707/proofs/gcp-cloudsql-restore-drill-.json \ + --target-counts-csv outputs/gcp-infra-hardening-20260707/proofs/gcp-cloudsql-target-counts-.csv \ + --output outputs/gcp-infra-hardening-20260707/proofs/gcp-cloudsql-restore-readback-verification-.json +``` + +Only a `status = pass` verifier output is enough for row-count parity. It still +does not prove application cutover or continuous replication. + ## Required Proof A successful restore or replication canary must retain: @@ -139,6 +161,7 @@ select schemaname, tablename from pg_tables where schemaname not in ('pg_catalog ``` 8. Retain the SQLite backup hash, GCS object generation, import/conversion operation, query output, and row-count sample. +9. Run `ops/verify_gcp_cloudsql_restore_readback.py` and retain a passing parity proof. ## Logical Replication Path diff --git a/ops/check_gcp_infra_readiness.py b/ops/check_gcp_infra_readiness.py index 986af6c..49b30ee 100755 --- a/ops/check_gcp_infra_readiness.py +++ b/ops/check_gcp_infra_readiness.py @@ -600,8 +600,9 @@ def check_local_sqlite_postgres_restore_canary() -> Check: def check_gcp_cloudsql_restore_drill_contract() -> Check: script = Path("ops/run_gcp_cloudsql_restore_drill.sh") + verifier = Path("ops/verify_gcp_cloudsql_restore_readback.py") runbook = Path("docs/gcp-kb-cloudsql-restore-runbook.md") - missing = [str(path) for path in (script, runbook) if not path.exists()] + missing = [str(path) for path in (script, verifier, runbook) if not path.exists()] if missing: return Check( "gcp_cloudsql_restore_drill_contract", @@ -618,10 +619,17 @@ def check_gcp_cloudsql_restore_drill_contract() -> Check: "gcloud sql operations wait", "EXECUTE", "target-counts.sql", + "source_table_counts", ] absent = [item for item in required if item not in script_text] + verifier_text = verifier.read_text() if verifier.exists() else "" + for item in ("target_counts_csv", "mismatched_tables", "source_total_rows", "target_total_rows"): + if item not in verifier_text: + absent.append(f"verifier:{item}") if "run_gcp_cloudsql_restore_drill.sh" not in runbook_text: absent.append("runbook_reference") + if "verify_gcp_cloudsql_restore_readback.py" not in runbook_text: + absent.append("readback_verifier_runbook_reference") if absent: return Check( "gcp_cloudsql_restore_drill_contract", diff --git a/ops/run_gcp_cloudsql_restore_drill.sh b/ops/run_gcp_cloudsql_restore_drill.sh index 2d6f367..db888d4 100755 --- a/ops/run_gcp_cloudsql_restore_drill.sh +++ b/ops/run_gcp_cloudsql_restore_drill.sh @@ -91,6 +91,7 @@ proof = { "source_integrity_check": summary["source_integrity_check"], "source_table_count": summary["table_count"], "source_total_rows": summary["source_total_rows"], + "source_table_counts": {table["name"]: table["row_count"] for table in summary["tables"]}, "gcs_import_uri": gcs_import_uri, "local_private_artifacts": { "postgres_import_sql": str(import_sql), diff --git a/ops/verify_gcp_cloudsql_restore_readback.py b/ops/verify_gcp_cloudsql_restore_readback.py new file mode 100644 index 0000000..1994170 --- /dev/null +++ b/ops/verify_gcp_cloudsql_restore_readback.py @@ -0,0 +1,125 @@ +#!/usr/bin/env python3 +"""Verify Cloud SQL restore readback against a Teleo restore drill proof. + +This script is intentionally post-import and non-mutating. Run the +`target-counts.sql` generated by `ops/run_gcp_cloudsql_restore_drill.sh` against +Cloud SQL with CSV output, then pass that CSV here. The verifier compares table +counts and total rows with the source counts retained in the drill proof. +""" + +from __future__ import annotations + +import argparse +import csv +import json +from datetime import UTC, datetime +from pathlib import Path + + +def load_json(path: Path) -> dict[str, object]: + return json.loads(path.read_text()) + + +def read_target_counts(path: Path) -> dict[str, int]: + rows: dict[str, int] = {} + with path.open(newline="") as handle: + reader = csv.DictReader(handle) + required = {"table_name", "row_count"} + missing = required - set(reader.fieldnames or []) + if missing: + raise ValueError(f"target counts CSV missing columns: {sorted(missing)}") + for row in reader: + table = (row.get("table_name") or "").strip() + if not table: + continue + rows[table] = int(row.get("row_count") or "0") + return rows + + +def source_counts_from_drill(proof: dict[str, object]) -> dict[str, int] | None: + source_counts = proof.get("source_table_counts") + if isinstance(source_counts, dict): + return {str(table): int(count) for table, count in source_counts.items()} + return None + + +def compare_counts(source_counts: dict[str, int] | None, target_counts: dict[str, int]) -> list[dict[str, object]]: + if source_counts is None: + return [] + all_tables = sorted(set(source_counts) | set(target_counts)) + return [ + { + "table": table, + "source_rows": source_counts.get(table), + "target_rows": target_counts.get(table), + } + for table in all_tables + if source_counts.get(table) != target_counts.get(table) + ] + + +def build_payload(drill_proof_path: Path, target_counts_csv: Path) -> dict[str, object]: + proof = load_json(drill_proof_path) + target_counts = read_target_counts(target_counts_csv) + source_counts = source_counts_from_drill(proof) + target_table_count = len(target_counts) + target_total_rows = sum(target_counts.values()) + source_table_count = int(proof.get("source_table_count") or 0) + source_total_rows = int(proof.get("source_total_rows") or 0) + mismatches = compare_counts(source_counts, target_counts) + + problems = [] + if proof.get("status") not in {"import_operation_done", "readback_verified"}: + problems.append(f"drill_status={proof.get('status')}") + if source_table_count != target_table_count: + problems.append(f"table_count source={source_table_count} target={target_table_count}") + if source_total_rows != target_total_rows: + problems.append(f"total_rows source={source_total_rows} target={target_total_rows}") + if mismatches: + problems.append(f"mismatched_tables={len(mismatches)}") + + return { + "artifact": "gcp_cloudsql_restore_readback_verification", + "generated_at_utc": datetime.now(UTC).isoformat(), + "drill_proof": str(drill_proof_path), + "target_counts_csv": str(target_counts_csv), + "project_id": proof.get("project_id"), + "cloudsql_instance": proof.get("cloudsql_instance"), + "database": proof.get("database"), + "postgres_schema": proof.get("postgres_schema"), + "source_sqlite_backup_sha256": proof.get("source_sqlite_backup_sha256"), + "source_table_count": source_table_count, + "target_table_count": target_table_count, + "source_total_rows": source_total_rows, + "target_total_rows": target_total_rows, + "mismatched_tables": mismatches, + "status": "pass" if not problems else "fail", + "problems": problems, + "not_proven_by_this_artifact": [ + "application cutover to Cloud SQL", + "ongoing replication after the readback timestamp", + "semantic query correctness beyond table row-count parity", + ], + } + + +def main() -> int: + parser = argparse.ArgumentParser() + parser.add_argument("--drill-proof", required=True, type=Path) + parser.add_argument("--target-counts-csv", required=True, type=Path) + parser.add_argument( + "--output", + default="outputs/gcp-infra-hardening-20260707/proofs/gcp-cloudsql-restore-readback-verification.json", + type=Path, + ) + args = parser.parse_args() + + payload = build_payload(args.drill_proof, args.target_counts_csv) + args.output.parent.mkdir(parents=True, exist_ok=True) + args.output.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n") + print(json.dumps(payload, indent=2, sort_keys=True)) + return 0 if payload["status"] == "pass" else 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/test_gcp_cloudsql_restore_drill.py b/tests/test_gcp_cloudsql_restore_drill.py index 9540879..0fe2566 100644 --- a/tests/test_gcp_cloudsql_restore_drill.py +++ b/tests/test_gcp_cloudsql_restore_drill.py @@ -40,5 +40,6 @@ def test_gcp_cloudsql_restore_drill_dry_run(tmp_path): proof = json.loads((proof_root / "gcp-cloudsql-restore-drill-20260707Tdryrun.json").read_text()) assert proof["status"] == "prepared" assert proof["gcs_import_uri"].startswith("gs://teleo-501523-prod-backups/") + assert proof["source_table_counts"] == {"response_audit": 1} assert "gcloud sql import sql" in "\n".join(proof["planned_commands"]) assert (output_root / "gcp-cloudsql-restore-drill-20260707Tdryrun" / "target-counts.sql").exists() diff --git a/tests/test_gcp_cloudsql_restore_readback.py b/tests/test_gcp_cloudsql_restore_readback.py new file mode 100644 index 0000000..54a8d4f --- /dev/null +++ b/tests/test_gcp_cloudsql_restore_readback.py @@ -0,0 +1,101 @@ +import json +import subprocess +from pathlib import Path + +REPO_ROOT = Path(__file__).resolve().parents[1] + + +def write_drill_proof(path: Path, *, status: str = "import_operation_done") -> None: + path.write_text( + json.dumps( + { + "status": status, + "project_id": "teleo-501523", + "cloudsql_instance": "teleo-pgvector-standby", + "database": "teleo_kb", + "postgres_schema": "teleo_restore", + "source_sqlite_backup_sha256": "abc123", + "source_table_count": 2, + "source_total_rows": 4, + "source_table_counts": { + "claims": 3, + "evidence": 1, + }, + } + ) + ) + + +def run_verifier(drill_proof: Path, counts_csv: Path, output: Path) -> subprocess.CompletedProcess[str]: + return subprocess.run( + [ + "python3", + "ops/verify_gcp_cloudsql_restore_readback.py", + "--drill-proof", + str(drill_proof), + "--target-counts-csv", + str(counts_csv), + "--output", + str(output), + ], + cwd=REPO_ROOT, + text=True, + capture_output=True, + check=False, + ) + + +def test_cloudsql_restore_readback_passes_matching_counts(tmp_path: Path) -> None: + proof = tmp_path / "drill.json" + counts = tmp_path / "target-counts.csv" + output = tmp_path / "verification.json" + write_drill_proof(proof) + counts.write_text("table_name,row_count\nclaims,3\nevidence,1\n") + + completed = run_verifier(proof, counts, output) + + assert completed.returncode == 0, completed.stderr + payload = json.loads(output.read_text()) + assert payload["status"] == "pass" + assert payload["source_table_count"] == 2 + assert payload["target_table_count"] == 2 + assert payload["source_total_rows"] == 4 + assert payload["target_total_rows"] == 4 + assert payload["mismatched_tables"] == [] + + +def test_cloudsql_restore_readback_fails_mismatched_counts(tmp_path: Path) -> None: + proof = tmp_path / "drill.json" + counts = tmp_path / "target-counts.csv" + output = tmp_path / "verification.json" + write_drill_proof(proof) + counts.write_text("table_name,row_count\nclaims,2\nevidence,1\n") + + completed = run_verifier(proof, counts, output) + + assert completed.returncode == 1 + payload = json.loads(output.read_text()) + assert payload["status"] == "fail" + assert "total_rows source=4 target=3" in payload["problems"] + assert payload["mismatched_tables"] == [ + { + "source_rows": 3, + "table": "claims", + "target_rows": 2, + } + ] + + +def test_cloudsql_restore_readback_requires_completed_import(tmp_path: Path) -> None: + proof = tmp_path / "drill.json" + counts = tmp_path / "target-counts.csv" + output = tmp_path / "verification.json" + write_drill_proof(proof, status="prepared") + counts.write_text("table_name,row_count\nclaims,3\nevidence,1\n") + + completed = run_verifier(proof, counts, output) + + assert completed.returncode == 1 + payload = json.loads(output.read_text()) + assert payload["status"] == "fail" + assert "drill_status=prepared" in payload["problems"]