diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index b47dec4..8cc3be3 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -41,6 +41,8 @@ jobs: lib/evaluate.py \ lib/llm.py \ lib/post_extract.py \ + ops/check_gcp_infra_readiness.py \ + ops/sqlite_to_postgres_dump.py \ telegram/approvals.py \ scripts/check_crabbox_ci_contract.py \ scripts/check_llm_refinement_contract.py \ @@ -49,10 +51,18 @@ jobs: tests/test_agent_routing.py \ tests/test_decision_engine_replay.py \ tests/test_evaluate_agent_routing.py \ + tests/test_gcp_cloudsql_restore_drill.py \ tests/test_phase1b_end_to_end.py \ + tests/test_sqlite_to_postgres_dump.py \ tests/test_eval_parse.py \ tests/test_contributor.py \ tests/test_search.py + - name: Shell syntax + run: | + bash -n \ + ops/backup_vps_sqlite_kb.sh \ + ops/run_gcp_cloudsql_restore_drill.sh \ + ops/run_sqlite_postgres_restore_canary.sh test: name: Unit tests diff --git a/docs/gcp-kb-cloudsql-restore-runbook.md b/docs/gcp-kb-cloudsql-restore-runbook.md index a5d3a49..f669d0e 100644 --- a/docs/gcp-kb-cloudsql-restore-runbook.md +++ b/docs/gcp-kb-cloudsql-restore-runbook.md @@ -66,6 +66,36 @@ This is a local restore/parity proof, not GCP redundancy by itself. It is the preflight that should pass before the same generated import is applied through the approved Cloud SQL connector/VPC path. +## Cloud SQL Restore Drill Runner + +Prepare the exact GCS import and Cloud SQL import operation without mutating GCP: + +```bash +SQLITE_BACKUP=./outputs/gcp-infra-hardening-20260707/private-backups/teleo-pipeline-sqlite-.db.gz \ + ops/run_gcp_cloudsql_restore_drill.sh +``` + +Execute it only from an authenticated operator environment that can write the +versioned backup bucket and administer the standby Cloud SQL instance: + +```bash +EXECUTE=1 \ +SQLITE_BACKUP=./outputs/gcp-infra-hardening-20260707/private-backups/teleo-pipeline-sqlite-.db.gz \ + ops/run_gcp_cloudsql_restore_drill.sh +``` + +The runner: + +- regenerates the explicit PostgreSQL import script; +- targets the shadow schema `teleo_restore` inside `teleo_kb`; +- uploads the import script to `gs://teleo-501523-prod-backups/kb-dumps/cloudsql-restore-drills/...` when `EXECUTE=1`; +- starts and waits for `gcloud sql import sql`; +- writes `target-counts.sql` for the required trusted VPC/Cloud SQL connector query readback. + +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. + ## Required Proof A successful restore or replication canary must retain: @@ -92,14 +122,15 @@ Use this while the canonical DB remains SQLite on the VPS and we need a GCP rest 1. Run `ops/backup_vps_sqlite_kb.sh`. 2. Upload the resulting SQLite backup and Leo/KB tarball to a versioned GCS bucket such as `gs://teleo-501523-prod-backups/kb-dumps/`. 3. Run the local SQLite-to-Postgres restore canary above and retain its proof. -4. Convert or replay the SQLite data into Cloud SQL with the explicit generated migration/import script. Do not run blind string rewrites against the SQLite dump. -5. Install required extensions on Cloud SQL: +4. Run `ops/run_gcp_cloudsql_restore_drill.sh` in dry-run mode to generate the GCS import plan. +5. Run `EXECUTE=1 ops/run_gcp_cloudsql_restore_drill.sh` from an authenticated operator environment to upload and import the generated SQL. Do not run blind string rewrites against the SQLite dump. +6. Install required extensions on Cloud SQL: ```sql create extension if not exists vector; ``` -6. From a trusted VPC runtime or Cloud SQL connector path, run readbacks: +7. From a trusted VPC runtime or Cloud SQL connector path, run readbacks: ```sql select current_database(); @@ -107,7 +138,7 @@ select extname, extversion from pg_extension where extname = 'vector'; select schemaname, tablename from pg_tables where schemaname not in ('pg_catalog', 'information_schema') order by 1, 2 limit 50; ``` -7. Retain the SQLite backup hash, GCS object generation, import/conversion operation, query output, and row-count sample. +8. Retain the SQLite backup hash, GCS object generation, import/conversion operation, query output, and row-count sample. ## Logical Replication Path diff --git a/ops/check_gcp_infra_readiness.py b/ops/check_gcp_infra_readiness.py index b02167a..3ff1517 100755 --- a/ops/check_gcp_infra_readiness.py +++ b/ops/check_gcp_infra_readiness.py @@ -558,6 +558,47 @@ 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") + runbook = Path("docs/gcp-kb-cloudsql-restore-runbook.md") + missing = [str(path) for path in (script, runbook) if not path.exists()] + if missing: + return Check( + "gcp_cloudsql_restore_drill_contract", + "blocked", + f"missing={missing}", + required_tier="T2_restore_execution_contract", + current_tier="T1_foundation", + ) + script_text = script.read_text() + runbook_text = runbook.read_text() + required = [ + "gcloud storage cp", + "gcloud sql import sql", + "gcloud sql operations wait", + "EXECUTE", + "target-counts.sql", + ] + absent = [item for item in required if item not in script_text] + if "run_gcp_cloudsql_restore_drill.sh" not in runbook_text: + absent.append("runbook_reference") + if absent: + return Check( + "gcp_cloudsql_restore_drill_contract", + "fail", + f"missing_contract={absent}", + required_tier="T2_restore_execution_contract", + current_tier="T1_foundation", + ) + return Check( + "gcp_cloudsql_restore_drill_contract", + "pass", + "script can prepare GCS import SQL, optionally upload/import via Cloud SQL Admin, and emits target-count readback SQL", + required_tier="T2_restore_execution_contract", + current_tier="T2_restore_execution_contract", + ) + + def check_kb_restore_or_replication() -> Check: return Check( "kb_restore_or_replication", @@ -581,6 +622,7 @@ def main() -> int: safe_check("cloud_sql_standby_target", check_cloud_sql_standby), safe_check("kb_source_restore_access", check_kb_source_restore_access), safe_check("local_sqlite_postgres_restore_canary", check_local_sqlite_postgres_restore_canary), + safe_check("gcp_cloudsql_restore_drill_contract", check_gcp_cloudsql_restore_drill_contract), safe_check("kb_restore_or_replication", check_kb_restore_or_replication), ] payload = { diff --git a/ops/run_gcp_cloudsql_restore_drill.sh b/ops/run_gcp_cloudsql_restore_drill.sh new file mode 100755 index 0000000..2d6f367 --- /dev/null +++ b/ops/run_gcp_cloudsql_restore_drill.sh @@ -0,0 +1,235 @@ +#!/usr/bin/env bash +set -euo pipefail + +# Prepare, and optionally execute, a Cloud SQL shadow-restore drill from a Teleo +# SQLite backup. Default mode is dry-run: it generates the Postgres import SQL +# and proof manifest, but does not upload to GCS or mutate Cloud SQL. + +PROJECT_ID="${PROJECT_ID:-teleo-501523}" +INSTANCE="${INSTANCE:-teleo-pgvector-standby}" +DATABASE="${DATABASE:-teleo_kb}" +REGION="${REGION:-europe-west6}" +BUCKET="${BUCKET:-teleo-501523-prod-backups}" +GCS_PREFIX="${GCS_PREFIX:-kb-dumps/cloudsql-restore-drills}" +SCHEMA="${SCHEMA:-teleo_restore}" +SQLITE_BACKUP="${SQLITE_BACKUP:-}" +EXECUTE="${EXECUTE:-0}" +OUTPUT_ROOT="${OUTPUT_ROOT:-./outputs/gcp-infra-hardening-20260707/private-cloudsql-restore-drills}" +PROOF_ROOT="${PROOF_ROOT:-./outputs/gcp-infra-hardening-20260707/proofs}" +TS="${TS:-$(date -u +%Y%m%dT%H%M%SZ)}" + +if [[ -z "$SQLITE_BACKUP" ]]; then + echo "SQLITE_BACKUP is required" >&2 + exit 2 +fi + +if [[ ! -f "$SQLITE_BACKUP" ]]; then + echo "SQLite backup not found: $SQLITE_BACKUP" >&2 + exit 2 +fi + +mkdir -p "$OUTPUT_ROOT" "$PROOF_ROOT" +chmod 700 "$OUTPUT_ROOT" + +WORK_DIR="$OUTPUT_ROOT/gcp-cloudsql-restore-drill-${TS}" +mkdir -p "$WORK_DIR" +chmod 700 "$WORK_DIR" + +IMPORT_SQL="$WORK_DIR/teleo-pipeline-postgres-import.sql" +SOURCE_SUMMARY="$WORK_DIR/source-summary.json" +TARGET_COUNTS_SQL="$WORK_DIR/target-counts.sql" +CONVERT_LOG="$WORK_DIR/sqlite-to-postgres-dump.stdout" +OBJECT_NAME="${GCS_PREFIX%/}/${TS}/teleo-pipeline-postgres-import.sql" +GCS_IMPORT_URI="gs://${BUCKET}/${OBJECT_NAME}" +PROOF_JSON="$PROOF_ROOT/gcp-cloudsql-restore-drill-${TS}.json" +IMPORT_START_JSON="$WORK_DIR/cloudsql-import-start.json" +IMPORT_WAIT_JSON="$WORK_DIR/cloudsql-import-wait.json" +GCS_OBJECT_JSON="$WORK_DIR/gcs-import-object.json" + +python3 ops/sqlite_to_postgres_dump.py \ + --sqlite-backup "$SQLITE_BACKUP" \ + --output-sql "$IMPORT_SQL" \ + --summary-json "$SOURCE_SUMMARY" \ + --schema "$SCHEMA" \ + --target-counts-sql "$TARGET_COUNTS_SQL" \ + > "$CONVERT_LOG" + +python3 - "$SOURCE_SUMMARY" "$SQLITE_BACKUP" "$IMPORT_SQL" "$TARGET_COUNTS_SQL" "$PROOF_JSON" "$PROJECT_ID" "$INSTANCE" "$DATABASE" "$REGION" "$GCS_IMPORT_URI" "$EXECUTE" <<'PY' +import hashlib +import json +import sys +from datetime import UTC, datetime +from pathlib import Path + +summary_path = Path(sys.argv[1]) +sqlite_backup = Path(sys.argv[2]) +import_sql = Path(sys.argv[3]) +target_counts_sql = Path(sys.argv[4]) +proof_path = Path(sys.argv[5]) +project_id, instance, database, region, gcs_import_uri, execute = sys.argv[6:12] +summary = json.loads(summary_path.read_text()) + +def sha256(path: Path) -> str: + digest = hashlib.sha256() + with path.open("rb") as handle: + for chunk in iter(lambda: handle.read(1024 * 1024), b""): + digest.update(chunk) + return digest.hexdigest() + +proof = { + "artifact": "gcp_cloudsql_restore_drill", + "generated_at_utc": datetime.now(UTC).isoformat(), + "mode": "execute" if execute == "1" else "dry_run", + "status": "prepared" if execute != "1" else "prepared_for_execute", + "project_id": project_id, + "region": region, + "cloudsql_instance": instance, + "database": database, + "postgres_schema": summary["schema"], + "source_sqlite_backup": str(sqlite_backup), + "source_sqlite_backup_sha256": sha256(sqlite_backup), + "source_integrity_check": summary["source_integrity_check"], + "source_table_count": summary["table_count"], + "source_total_rows": summary["source_total_rows"], + "gcs_import_uri": gcs_import_uri, + "local_private_artifacts": { + "postgres_import_sql": str(import_sql), + "source_summary_json": str(summary_path), + "target_counts_sql": str(target_counts_sql), + }, + "planned_commands": [ + f"gcloud storage cp {import_sql} {gcs_import_uri}", + f"gcloud sql import sql {instance} {gcs_import_uri} --project={project_id} --database={database} --quiet --format=json", + "run target-counts.sql from a trusted VPC/Cloud SQL connector path and compare source_total_rows/source_table_count", + ], + "not_proven_by_this_artifact": [ + "GCS object generation until EXECUTE=1 succeeds", + "Cloud SQL import operation until EXECUTE=1 succeeds", + "Cloud SQL query readback until target-counts.sql is run from a trusted VPC/connector path", + ], +} +proof_path.write_text(json.dumps(proof, indent=2, sort_keys=True) + "\n") +print(json.dumps({ + "artifact": proof["artifact"], + "mode": proof["mode"], + "status": proof["status"], + "source_integrity_check": proof["source_integrity_check"], + "source_table_count": proof["source_table_count"], + "source_total_rows": proof["source_total_rows"], + "gcs_import_uri": proof["gcs_import_uri"], + "proof": str(proof_path), +}, indent=2, sort_keys=True)) +PY + +if [[ "$EXECUTE" != "1" ]]; then + exit 0 +fi + +mark_execute_failure() { + local rc="$1" + local line="$2" + python3 - "$PROOF_JSON" "$rc" "$line" <<'PY' +import json +import sys +from datetime import UTC, datetime +from pathlib import Path + +proof_path = Path(sys.argv[1]) +rc = int(sys.argv[2]) +line = sys.argv[3] +proof = json.loads(proof_path.read_text()) +proof["status"] = "execute_failed" +proof["execute_failure"] = { + "exitcode": rc, + "line": line, + "recorded_at_utc": datetime.now(UTC).isoformat(), + "stderr_files": { + "gcs_upload": str(Path(proof["local_private_artifacts"]["postgres_import_sql"]).with_name("gcs-upload.stderr")), + }, +} +proof_path.write_text(json.dumps(proof, indent=2, sort_keys=True) + "\n") +PY +} + +trap 'rc=$?; mark_execute_failure "$rc" "$LINENO"; exit "$rc"' ERR + +gcloud storage cp "$IMPORT_SQL" "$GCS_IMPORT_URI" \ + --project="$PROJECT_ID" \ + > "$WORK_DIR/gcs-upload.stdout" \ + 2> "$WORK_DIR/gcs-upload.stderr" + +gcloud storage objects describe "$GCS_IMPORT_URI" \ + --project="$PROJECT_ID" \ + --format=json \ + > "$GCS_OBJECT_JSON" + +gcloud sql import sql "$INSTANCE" "$GCS_IMPORT_URI" \ + --project="$PROJECT_ID" \ + --database="$DATABASE" \ + --quiet \ + --format=json \ + > "$IMPORT_START_JSON" + +python3 - "$PROOF_JSON" "$GCS_OBJECT_JSON" "$IMPORT_START_JSON" <<'PY' +import json +import sys +from pathlib import Path + +proof_path = Path(sys.argv[1]) +object_path = Path(sys.argv[2]) +operation_path = Path(sys.argv[3]) +proof = json.loads(proof_path.read_text()) +gcs_object = json.loads(object_path.read_text()) +operation = json.loads(operation_path.read_text() or "{}") +proof["status"] = "import_started" +proof["gcs_object"] = { + "name": gcs_object.get("name"), + "bucket": gcs_object.get("bucket"), + "generation": gcs_object.get("generation"), + "size": gcs_object.get("size"), + "crc32c": gcs_object.get("crc32c"), +} +proof["cloudsql_import_operation"] = { + "name": operation.get("name"), + "status": operation.get("status"), + "operationType": operation.get("operationType"), + "targetId": operation.get("targetId"), +} +proof_path.write_text(json.dumps(proof, indent=2, sort_keys=True) + "\n") +print(json.dumps(proof["cloudsql_import_operation"], indent=2, sort_keys=True)) +PY + +operation_name="$(python3 - "$IMPORT_START_JSON" <<'PY' +import json, sys +print(json.loads(open(sys.argv[1]).read()).get("name", "")) +PY +)" + +if [[ -n "$operation_name" ]]; then + gcloud sql operations wait "$operation_name" \ + --project="$PROJECT_ID" \ + --timeout=3600 \ + --format=json \ + > "$IMPORT_WAIT_JSON" + + python3 - "$PROOF_JSON" "$IMPORT_WAIT_JSON" <<'PY' +import json +import sys +from pathlib import Path + +proof_path = Path(sys.argv[1]) +wait_path = Path(sys.argv[2]) +proof = json.loads(proof_path.read_text()) +operation = json.loads(wait_path.read_text() or "{}") +proof["cloudsql_import_wait"] = { + "name": operation.get("name"), + "status": operation.get("status"), + "operationType": operation.get("operationType"), + "targetId": operation.get("targetId"), + "error": operation.get("error"), +} +proof["status"] = "import_operation_done" if operation.get("status") == "DONE" and not operation.get("error") else "import_operation_failed" +proof_path.write_text(json.dumps(proof, indent=2, sort_keys=True) + "\n") +print(json.dumps({"status": proof["status"], "proof": str(proof_path)}, indent=2, sort_keys=True)) +PY +fi diff --git a/tests/test_gcp_cloudsql_restore_drill.py b/tests/test_gcp_cloudsql_restore_drill.py new file mode 100644 index 0000000..9540879 --- /dev/null +++ b/tests/test_gcp_cloudsql_restore_drill.py @@ -0,0 +1,44 @@ +import json +import os +import sqlite3 +import subprocess + + +def test_gcp_cloudsql_restore_drill_dry_run(tmp_path): + db_path = tmp_path / "source.db" + conn = sqlite3.connect(db_path) + conn.execute("create table response_audit (id integer primary key, query text)") + conn.execute("insert into response_audit (query) values (?)", ("hello",)) + conn.commit() + conn.close() + + proof_root = tmp_path / "proofs" + output_root = tmp_path / "private" + env = { + **os.environ, + "SQLITE_BACKUP": str(db_path), + "OUTPUT_ROOT": str(output_root), + "PROOF_ROOT": str(proof_root), + "TS": "20260707Tdryrun", + "PROJECT_ID": "teleo-501523", + "INSTANCE": "teleo-pgvector-standby", + "DATABASE": "teleo_kb", + } + result = subprocess.run( + ["bash", "ops/run_gcp_cloudsql_restore_drill.sh"], + check=True, + capture_output=True, + text=True, + env=env, + ) + + payload = json.loads(result.stdout) + assert payload["mode"] == "dry_run" + assert payload["source_table_count"] == 1 + assert payload["source_total_rows"] == 1 + + 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 "gcloud sql import sql" in "\n".join(proof["planned_commands"]) + assert (output_root / "gcp-cloudsql-restore-drill-20260707Tdryrun" / "target-counts.sql").exists()