From 464d1b3fd935f9ac37c5aa116f33ea0b7db42478 Mon Sep 17 00:00:00 2001 From: twentyOne2x Date: Tue, 7 Jul 2026 14:34:11 +0200 Subject: [PATCH] Add GCP infra execute canary runner (#56) --- .github/workflows/ci.yml | 2 + ops/run_gcp_infra_execute_canary.py | 269 +++++++++++++++++++++++++ tests/test_gcp_infra_execute_canary.py | 147 ++++++++++++++ 3 files changed, 418 insertions(+) create mode 100644 ops/run_gcp_infra_execute_canary.py create mode 100644 tests/test_gcp_infra_execute_canary.py diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index d9d0f32..993d505 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -43,6 +43,7 @@ jobs: lib/post_extract.py \ ops/apply_gcp_iam_split.py \ ops/check_gcp_infra_readiness.py \ + ops/run_gcp_infra_execute_canary.py \ ops/apply_gcp_runtime_baseline.py \ ops/check_gcp_service_communications.py \ ops/plan_gcp_iam_split.py \ @@ -58,6 +59,7 @@ jobs: tests/test_decision_engine_replay.py \ tests/test_evaluate_agent_routing.py \ tests/test_gcp_artifact_workflow.py \ + tests/test_gcp_infra_execute_canary.py \ tests/test_gcp_infra_readiness_checker.py \ tests/test_gcp_runtime_baseline_apply.py \ tests/test_gcp_service_communications.py \ diff --git a/ops/run_gcp_infra_execute_canary.py b/ops/run_gcp_infra_execute_canary.py new file mode 100644 index 0000000..ea37491 --- /dev/null +++ b/ops/run_gcp_infra_execute_canary.py @@ -0,0 +1,269 @@ +#!/usr/bin/env python3 +"""Run or plan the post-reauth GCP infra execute canary. + +Dry-run is the default: it writes the exact command sequence without mutating +GCP. Execute mode runs the IAM split, runtime baseline, readiness workflow, and +optionally the Cloud SQL restore drill. +""" + +from __future__ import annotations + +import argparse +import base64 +import ipaddress +import json +import os +import subprocess +from dataclasses import asdict, dataclass +from datetime import UTC, datetime +from pathlib import Path + +PROJECT = "teleo-501523" +GITHUB_REPO = "living-ip/teleo-infrastructure" +DEFAULT_PROOF_ROOT = Path("outputs/gcp-infra-hardening-20260707/proofs") +DEFAULT_PRIVATE_OUTPUT_ROOT = Path("outputs/gcp-infra-hardening-20260707/private-cloudsql-restore-drills") + + +@dataclass +class Operation: + name: str + action: str + command: list[str] + exitcode: int | None = None + stdout_tail: str = "" + stderr_tail: str = "" + proof: str | None = None + + +def utc_stamp() -> str: + return datetime.now(UTC).strftime("%Y%m%dT%H%M%SZ") + + +def tail(text: str, limit: int = 1200) -> str: + return text[-limit:] + + +def validate_admin_ssh_cidr(value: str | None) -> str | None: + if not value: + return "--admin-ssh-cidr is required in execute mode" + try: + network = ipaddress.ip_network(value, strict=True) + except ValueError as exc: + return f"--admin-ssh-cidr must be a valid single-host IPv4 /32: {exc}" + if network.version != 4 or network.prefixlen != 32: + return "--admin-ssh-cidr must be a single trusted IPv4 /32" + return None + + +def read_capsule_b64(path: Path | None) -> str | None: + if not path: + return None + return base64.b64encode(path.read_bytes()).decode("ascii") + + +def command_for_report(command: list[str], *, redact_capsule: bool = False) -> list[str]: + if not redact_capsule: + return command + redacted: list[str] = [] + for part in command: + if part.startswith("restore_canary_capsule_b64="): + redacted.append("restore_canary_capsule_b64=") + else: + redacted.append(part) + return redacted + + +def operations_for_report(operations: list[Operation], *, redact_capsule: bool = False) -> list[dict[str, object]]: + rows = [] + for operation in operations: + row = asdict(operation) + if operation.name == "trigger_gcp_readiness": + row["command"] = command_for_report(operation.command, redact_capsule=redact_capsule) + rows.append(row) + return rows + + +def run_operation(operation: Operation, *, env: dict[str, str] | None = None, redact_stdout: bool = False) -> Operation: + completed = subprocess.run( + operation.command, + text=True, + capture_output=True, + check=False, + env=env, + ) + operation.exitcode = completed.returncode + operation.stdout_tail = "" if redact_stdout and completed.stdout else tail(completed.stdout) + operation.stderr_tail = tail(completed.stderr) + return operation + + +def build_operations(args: argparse.Namespace, ts: str) -> list[Operation]: + proof_root = args.proof_root + iam_proof = proof_root / f"gcp-iam-split-execute-canary-{ts}.json" + runtime_proof = proof_root / f"gcp-runtime-baseline-execute-canary-{ts}.json" + restore_proof = proof_root / f"gcp-cloudsql-restore-drill-{ts}.json" + + operations = [ + Operation( + name="gcloud_token_preflight", + action="preflight", + command=["gcloud", "auth", "print-access-token", "--account", args.gcloud_account], + ), + Operation( + name="apply_gcp_iam_split", + action="execute" if args.execute else "plan", + command=[ + "python3", + "ops/apply_gcp_iam_split.py", + *(["--execute"] if args.execute else []), + "--output", + str(iam_proof), + ], + proof=str(iam_proof), + ), + Operation( + name="apply_gcp_runtime_baseline", + action="execute" if args.execute else "plan", + command=[ + "python3", + "ops/apply_gcp_runtime_baseline.py", + *(["--execute"] if args.execute else []), + "--admin-ssh-cidr", + args.admin_ssh_cidr or "127.0.0.1/32", + "--output", + str(runtime_proof), + ], + proof=str(runtime_proof), + ), + ] + + readiness_command = [ + "gh", + "workflow", + "run", + "gcp-readiness.yml", + "--repo", + GITHUB_REPO, + "--ref", + args.ref, + ] + if args.restore_canary_capsule: + capsule_b64 = ( + read_capsule_b64(args.restore_canary_capsule) + if args.execute and args.restore_canary_capsule.exists() + else "" + ) + readiness_command.extend(["-f", f"restore_canary_capsule_b64={capsule_b64}"]) + operations.append( + Operation( + name="trigger_gcp_readiness", + action="trigger", + command=readiness_command, + ) + ) + + if args.execute_restore: + operations.append( + Operation( + name="run_gcp_cloudsql_restore_drill", + action="execute", + command=["bash", "ops/run_gcp_cloudsql_restore_drill.sh"], + proof=str(restore_proof), + ) + ) + return operations + + +def parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--execute", action="store_true", help="run the mutable GCP apply path") + parser.add_argument("--execute-restore", action="store_true", help="also run the Cloud SQL restore drill") + parser.add_argument("--admin-ssh-cidr", help="single trusted operator IPv4 /32 for SSH ingress") + parser.add_argument("--restore-canary-capsule", type=Path, help="redacted restore canary capsule JSON") + parser.add_argument("--sqlite-backup", type=Path, help="SQLite backup for --execute-restore") + parser.add_argument("--gcloud-account", default="billy@livingip.xyz") + parser.add_argument("--ref", default="main", help="Git ref for gcp-readiness workflow") + parser.add_argument("--proof-root", type=Path, default=DEFAULT_PROOF_ROOT) + parser.add_argument("--private-output-root", type=Path, default=DEFAULT_PRIVATE_OUTPUT_ROOT) + parser.add_argument("--output", type=Path, default=None) + return parser.parse_args() + + +def main() -> int: + args = parse_args() + ts = utc_stamp() + args.proof_root.mkdir(parents=True, exist_ok=True) + args.private_output_root.mkdir(parents=True, exist_ok=True) + output = args.output or args.proof_root / f"gcp-infra-execute-canary-{ts}.json" + + problems = [] + if args.execute and (problem := validate_admin_ssh_cidr(args.admin_ssh_cidr)): + problems.append(problem) + if args.restore_canary_capsule and not args.restore_canary_capsule.exists(): + problems.append(f"restore canary capsule not found: {args.restore_canary_capsule}") + if args.execute_restore and not args.execute: + problems.append("--execute-restore requires --execute") + if args.execute_restore and not args.sqlite_backup: + problems.append("--execute-restore requires --sqlite-backup") + if args.sqlite_backup and not args.sqlite_backup.exists(): + problems.append(f"SQLite backup not found: {args.sqlite_backup}") + + operations = build_operations(args, ts) + if problems: + payload = { + "artifact": "gcp_infra_execute_canary", + "generated_at_utc": datetime.now(UTC).isoformat(), + "mode": "execute" if args.execute else "dry_run", + "status": "failed_validation", + "problems": problems, + "operations": operations_for_report(operations, redact_capsule=bool(args.restore_canary_capsule)), + } + output.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n") + print(json.dumps(payload, indent=2, sort_keys=True)) + return 1 + + if args.execute: + env = os.environ.copy() + for operation in operations: + if operation.name == "gcloud_token_preflight": + run_operation(operation, redact_stdout=True) + elif operation.name == "run_gcp_cloudsql_restore_drill": + env.update( + { + "EXECUTE": "1", + "SQLITE_BACKUP": str(args.sqlite_backup), + "OUTPUT_ROOT": str(args.private_output_root), + "PROOF_ROOT": str(args.proof_root), + "TS": ts, + } + ) + run_operation(operation, env=env) + else: + run_operation(operation) + if operation.exitcode: + break + + failed = [operation for operation in operations if operation.exitcode not in (None, 0)] + payload = { + "artifact": "gcp_infra_execute_canary", + "generated_at_utc": datetime.now(UTC).isoformat(), + "mode": "execute" if args.execute else "dry_run", + "status": "failed" if failed else ("executed" if args.execute else "planned"), + "project": PROJECT, + "admin_ssh_cidr": args.admin_ssh_cidr, + "restore_canary_capsule": str(args.restore_canary_capsule) if args.restore_canary_capsule else None, + "execute_restore": args.execute_restore, + "operations": operations_for_report(operations, redact_capsule=bool(args.restore_canary_capsule)), + "not_proven_by_this_artifact": [ + "GCP readiness green until the triggered readiness workflow passes", + "Cloud SQL row-count parity until restore drill and readback verifier pass", + "Telegram/Leo parity on GCP", + ], + } + output.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n") + print(json.dumps(payload, indent=2, sort_keys=True)) + return 1 if failed else 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/test_gcp_infra_execute_canary.py b/tests/test_gcp_infra_execute_canary.py new file mode 100644 index 0000000..e1e78ae --- /dev/null +++ b/tests/test_gcp_infra_execute_canary.py @@ -0,0 +1,147 @@ +from __future__ import annotations + +import json +import os +import subprocess +from pathlib import Path + +REPO_ROOT = Path(__file__).resolve().parents[1] +RUNNER = "ops/run_gcp_infra_execute_canary.py" + + +def run_runner(args: list[str], *, env: dict[str, str] | None = None) -> subprocess.CompletedProcess[str]: + merged_env = os.environ.copy() + if env: + merged_env.update(env) + return subprocess.run( + ["python3", RUNNER, *args], + cwd=REPO_ROOT, + text=True, + capture_output=True, + check=False, + env=merged_env, + ) + + +def write_capsule(path: Path) -> None: + path.write_text( + json.dumps( + { + "artifact": "sqlite_postgres_restore_canary_capsule", + "status": "pass", + "source_total_rows": 805745, + "target_total_rows": 805745, + "secret_marker": "capsule-secret-marker", + } + ) + ) + + +def load_json(path: Path) -> dict: + return json.loads(path.read_text()) + + +def test_dry_run_writes_full_plan_and_redacts_capsule(tmp_path: Path) -> None: + capsule = tmp_path / "capsule.json" + output = tmp_path / "execute-canary.json" + write_capsule(capsule) + + completed = run_runner( + [ + "--restore-canary-capsule", + str(capsule), + "--output", + str(output), + ], + env={"TELEO_CLOUDSQL_POSTGRES_PASSWORD": "do-not-record-this"}, + ) + + assert completed.returncode == 0, completed.stderr + payload = load_json(output) + rendered = output.read_text() + assert payload["mode"] == "dry_run" + assert payload["status"] == "planned" + assert [operation["name"] for operation in payload["operations"]] == [ + "gcloud_token_preflight", + "apply_gcp_iam_split", + "apply_gcp_runtime_baseline", + "trigger_gcp_readiness", + ] + assert all(operation["exitcode"] is None for operation in payload["operations"]) + assert "restore_canary_capsule_b64=" in rendered + assert "capsule-secret-marker" not in rendered + assert "do-not-record-this" not in rendered + + +def test_validation_failure_also_redacts_capsule(tmp_path: Path) -> None: + capsule = tmp_path / "capsule.json" + output = tmp_path / "failed-validation.json" + write_capsule(capsule) + + completed = run_runner( + [ + "--execute", + "--restore-canary-capsule", + str(capsule), + "--output", + str(output), + ], + ) + + assert completed.returncode == 1 + payload = load_json(output) + rendered = output.read_text() + assert payload["status"] == "failed_validation" + assert "--admin-ssh-cidr is required in execute mode" in payload["problems"] + assert "restore_canary_capsule_b64=" in rendered + assert "capsule-secret-marker" not in rendered + + +def test_execute_restore_requires_execute_and_sqlite_backup(tmp_path: Path) -> None: + output = tmp_path / "failed-validation.json" + + completed = run_runner(["--execute-restore", "--output", str(output)]) + + assert completed.returncode == 1 + payload = load_json(output) + assert payload["status"] == "failed_validation" + assert "--execute-restore requires --execute" in payload["problems"] + assert "--execute-restore requires --sqlite-backup" in payload["problems"] + + +def test_execute_requires_single_host_admin_cidr(tmp_path: Path) -> None: + output = tmp_path / "failed-validation.json" + + completed = run_runner( + [ + "--execute", + "--admin-ssh-cidr", + "10.0.0.0/24", + "--output", + str(output), + ], + ) + + assert completed.returncode == 1 + payload = load_json(output) + assert payload["status"] == "failed_validation" + assert "--admin-ssh-cidr must be a single trusted IPv4 /32" in payload["problems"] + + +def test_invalid_capsule_path_fails_validation(tmp_path: Path) -> None: + output = tmp_path / "failed-validation.json" + missing = tmp_path / "missing-capsule.json" + + completed = run_runner( + [ + "--restore-canary-capsule", + str(missing), + "--output", + str(output), + ], + ) + + assert completed.returncode == 1 + payload = load_json(output) + assert payload["status"] == "failed_validation" + assert f"restore canary capsule not found: {missing}" in payload["problems"]