diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 23d482f..6f5b730 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/apply_gcp_runtime_baseline.py \ ops/plan_gcp_iam_split.py \ ops/sqlite_to_postgres_dump.py \ ops/verify_gcp_cloudsql_restore_readback.py \ @@ -55,6 +56,7 @@ jobs: tests/test_decision_engine_replay.py \ tests/test_evaluate_agent_routing.py \ tests/test_gcp_artifact_workflow.py \ + tests/test_gcp_runtime_baseline_apply.py \ tests/test_gcp_cloudsql_restore_drill.py \ tests/test_gcp_cloudsql_restore_readback.py \ tests/test_gcp_iam_split_apply.py \ diff --git a/docs/gcp-infra-hardening-20260707.md b/docs/gcp-infra-hardening-20260707.md index f16dc79..266d0f5 100644 --- a/docs/gcp-infra-hardening-20260707.md +++ b/docs/gcp-infra-hardening-20260707.md @@ -2,7 +2,14 @@ This records the current GCP hardening contract for `teleo-501523`. -## What Is Live +## Current-State Rule + +The latest retained `gcp-readiness` artifact is authoritative for what is live +right now. The resource lists below define the target runtime contract and the +checks this lane enforces; do not treat them as current production proof unless +the current readiness run passes. + +## Target Runtime Contract - Artifact Registry Docker repositories exist in `europe-west6`: - `teleo` @@ -196,6 +203,37 @@ This plan is not itself a completed redundancy proof. DB redundancy is complete only after the restore drill imports source data into Cloud SQL and the retained `target-counts.sql` readback matches the source/local restore proof. +## Runtime Baseline Runner + +Do not create or repair the GCP runtime baseline manually in the console if an +audited runner can do it. The runtime baseline runner is dry-run by default: + +```bash +python3 ops/apply_gcp_runtime_baseline.py \ + --admin-ssh-cidr /32 \ + --output outputs/gcp-infra-hardening-20260707/proofs/gcp-runtime-baseline-dry-run.json +``` + +The dry-run proof records the exact service accounts, network, firewall, VM, +snapshot, backup bucket, secret, and Cloud SQL operations needed for the +readiness checker. It does not prove that those resources exist. + +To apply after an authenticated GCP admin session is available: + +```bash +export TELEO_CLOUDSQL_POSTGRES_PASSWORD='' +python3 ops/apply_gcp_runtime_baseline.py \ + --execute \ + --admin-ssh-cidr /32 \ + --output outputs/gcp-infra-hardening-20260707/proofs/gcp-runtime-baseline-execute.json +``` + +`--execute` refuses to run without a single trusted IPv4 `/32` SSH CIDR. The +Cloud SQL password is passed through the environment and redacted from retained +operation commands. After execute mode succeeds, rerun `gcp-readiness.yml` with +the dedicated readiness service account and then run the Cloud SQL restore +drill/readback verifier before claiming database redundancy. + ## Communication Posture Current GCP communication hardening is limited to the surfaces already read back: diff --git a/ops/apply_gcp_runtime_baseline.py b/ops/apply_gcp_runtime_baseline.py new file mode 100644 index 0000000..3d0bfd0 --- /dev/null +++ b/ops/apply_gcp_runtime_baseline.py @@ -0,0 +1,800 @@ +#!/usr/bin/env python3 +"""Plan or apply the Teleo GCP runtime baseline. + +Default mode is dry-run. Execute mode is intentionally explicit and proof +oriented: it records every non-secret operation, redacts password-bearing +commands, requires a /32 admin SSH CIDR, and stops before Cloud SQL password +work unless the password is present in the configured environment variable. +""" + +from __future__ import annotations + +import argparse +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" +REGION = "europe-west6" +ZONE = "europe-west6-a" +NETWORK = "teleo-staging-net" +SUBNET = "teleo-staging-subnet" +SUBNET_RANGE = "10.70.0.0/24" +PRIVATE_RANGE = "teleo-cloudsql-private-range" +SNAPSHOT_POLICY = "teleo-daily-7d-snapshots" +BACKUP_BUCKETS = ("teleo-501523-prod-backups", "teleo-501523-leoclean-backups") +CLOUDSQL_INSTANCE = "teleo-pgvector-standby" +CLOUDSQL_DATABASE = "teleo_kb" +CLOUDSQL_PASSWORD_SECRET = "gcp-teleo-pgvector-standby-postgres-password" + +SERVICE_ACCOUNTS = { + "sa-teleo-cloudbuild": { + "email": f"sa-teleo-cloudbuild@{PROJECT}.iam.gserviceaccount.com", + "display_name": "Teleo Cloud Build image publisher", + "roles": ("roles/artifactregistry.writer", "roles/logging.logWriter", "roles/storage.objectViewer"), + }, + "sa-teleo-prod-vm": { + "email": f"sa-teleo-prod-vm@{PROJECT}.iam.gserviceaccount.com", + "display_name": "Teleo production VM runtime", + "roles": ("roles/artifactregistry.reader", "roles/logging.logWriter", "roles/monitoring.metricWriter"), + }, + "sa-teleo-staging-vm": { + "email": f"sa-teleo-staging-vm@{PROJECT}.iam.gserviceaccount.com", + "display_name": "Teleo staging VM runtime", + "roles": ("roles/artifactregistry.reader", "roles/logging.logWriter", "roles/monitoring.metricWriter"), + }, +} + +VMS = { + "teleo-prod-1": { + "service_account": SERVICE_ACCOUNTS["sa-teleo-prod-vm"]["email"], + "tag": "teleo-prod-ssh", + "machine_type": "e2-standard-2", + "boot_disk_size": "80GB", + }, + "teleo-staging-1": { + "service_account": SERVICE_ACCOUNTS["sa-teleo-staging-vm"]["email"], + "tag": "teleo-staging-ssh", + "machine_type": "e2-standard-2", + "boot_disk_size": "80GB", + }, +} + +SSH_FIREWALL_RULES = { + "teleo-prod-allow-ssh-current-ip": "teleo-prod-ssh", + "teleo-staging-allow-ssh-current-ip": "teleo-staging-ssh", +} + +REQUIRED_APIS = ( + "artifactregistry.googleapis.com", + "cloudbuild.googleapis.com", + "compute.googleapis.com", + "logging.googleapis.com", + "monitoring.googleapis.com", + "secretmanager.googleapis.com", + "servicenetworking.googleapis.com", + "sqladmin.googleapis.com", +) + + +@dataclass +class OperationResult: + name: str + command: list[str] + action: str + exitcode: int | None = None + stdout_tail: str = "" + stderr_tail: str = "" + + +def tail(value: str, limit: int = 2000) -> str: + return value[-limit:] + + +def run_command( + command: list[str], + *, + execute: bool, + name: str, + action: str, + recorded_command: list[str] | None = None, + stdin: str | None = None, +) -> OperationResult: + safe_command = recorded_command or command + if not execute: + return OperationResult(name=name, command=safe_command, action=action) + completed = subprocess.run(command, input=stdin, text=True, capture_output=True, check=False) + return OperationResult( + name=name, + command=safe_command, + action=action, + exitcode=completed.returncode, + stdout_tail=tail(completed.stdout), + stderr_tail=tail(completed.stderr), + ) + + +def is_reauth_failure(result: OperationResult) -> bool: + return "Reauthentication failed" in result.stderr_tail or "cannot prompt during non-interactive execution" in result.stderr_tail + + +def validate_admin_ssh_cidr(value: str | None, *, execute: bool) -> str: + if not value: + if execute: + raise ValueError("--admin-ssh-cidr is required in execute mode") + return "/32" + network = ipaddress.ip_network(value, strict=False) + if network.version != 4 or network.prefixlen != 32: + raise ValueError("--admin-ssh-cidr must be a single IPv4 /32") + return str(network) + + +def create_if_missing( + *, + probe: list[str], + create: list[str], + execute: bool, + name: str, + recorded_create: list[str] | None = None, +) -> list[OperationResult]: + if execute: + probe_result = run_command(probe, execute=True, name=f"probe:{name}", action="probe") + if probe_result.exitcode == 0: + probe_result.action = "already_exists" + return [probe_result] + if is_reauth_failure(probe_result): + return [probe_result] + return [ + run_command( + create, + execute=execute, + name=name, + action="create_if_missing", + recorded_command=recorded_create, + ) + ] + + +def ensure_service_account(account_id: str, email: str, display_name: str, *, execute: bool) -> list[OperationResult]: + return create_if_missing( + probe=["gcloud", "iam", "service-accounts", "describe", email, "--project", PROJECT, "--format=json"], + create=[ + "gcloud", + "iam", + "service-accounts", + "create", + account_id, + "--project", + PROJECT, + "--display-name", + display_name, + ], + execute=execute, + name=f"create_service_account:{email}", + ) + + +def ensure_project_binding(email: str, role: str, *, execute: bool, prefix: str) -> OperationResult: + return run_command( + [ + "gcloud", + "projects", + "add-iam-policy-binding", + PROJECT, + "--member", + f"serviceAccount:{email}", + "--role", + role, + "--quiet", + ], + execute=execute, + name=f"{prefix}:{role}", + action="ensure_binding", + ) + + +def ensure_network(*, execute: bool) -> list[OperationResult]: + return create_if_missing( + probe=["gcloud", "compute", "networks", "describe", NETWORK, "--project", PROJECT, "--format=json"], + create=[ + "gcloud", + "compute", + "networks", + "create", + NETWORK, + "--project", + PROJECT, + "--subnet-mode=custom", + "--bgp-routing-mode=regional", + ], + execute=execute, + name=f"create_network:{NETWORK}", + ) + + +def ensure_subnet(*, execute: bool) -> list[OperationResult]: + operations = create_if_missing( + probe=[ + "gcloud", + "compute", + "networks", + "subnets", + "describe", + SUBNET, + "--region", + REGION, + "--project", + PROJECT, + "--format=json", + ], + create=[ + "gcloud", + "compute", + "networks", + "subnets", + "create", + SUBNET, + "--project", + PROJECT, + "--network", + NETWORK, + "--region", + REGION, + "--range", + SUBNET_RANGE, + "--enable-private-ip-google-access", + ], + execute=execute, + name=f"create_subnet:{SUBNET}", + ) + operations.append( + run_command( + [ + "gcloud", + "compute", + "networks", + "subnets", + "update", + SUBNET, + "--project", + PROJECT, + "--region", + REGION, + "--enable-private-ip-google-access", + ], + execute=execute, + name=f"enable_private_google_access:{SUBNET}", + action="ensure_setting", + ) + ) + return operations + + +def ensure_private_services_access(*, execute: bool) -> list[OperationResult]: + operations = create_if_missing( + probe=[ + "gcloud", + "compute", + "addresses", + "describe", + PRIVATE_RANGE, + "--project", + PROJECT, + "--global", + "--format=json", + ], + create=[ + "gcloud", + "compute", + "addresses", + "create", + PRIVATE_RANGE, + "--project", + PROJECT, + "--global", + "--prefix-length=16", + "--purpose=VPC_PEERING", + "--network", + NETWORK, + ], + execute=execute, + name=f"create_private_service_range:{PRIVATE_RANGE}", + ) + operations.append( + run_command( + [ + "gcloud", + "services", + "vpc-peerings", + "connect", + "--project", + PROJECT, + "--service=servicenetworking.googleapis.com", + "--network", + NETWORK, + "--ranges", + PRIVATE_RANGE, + ], + execute=execute, + name=f"ensure_private_services_access:{NETWORK}", + action="ensure_vpc_peering", + ) + ) + return operations + + +def ensure_firewall_rule(name: str, target_tag: str, admin_ssh_cidr: str, *, execute: bool) -> list[OperationResult]: + return [ + *create_if_missing( + probe=["gcloud", "compute", "firewall-rules", "describe", name, "--project", PROJECT, "--format=json"], + create=[ + "gcloud", + "compute", + "firewall-rules", + "create", + name, + "--project", + PROJECT, + "--network", + NETWORK, + "--direction=INGRESS", + "--priority=1000", + "--allow=tcp:22", + "--source-ranges", + admin_ssh_cidr, + "--target-tags", + target_tag, + "--enable-logging", + ], + execute=execute, + name=f"create_firewall_rule:{name}", + ), + run_command( + [ + "gcloud", + "compute", + "firewall-rules", + "update", + name, + "--project", + PROJECT, + "--source-ranges", + admin_ssh_cidr, + "--target-tags", + target_tag, + "--allow=tcp:22", + "--enable-logging", + ], + execute=execute, + name=f"update_firewall_rule:{name}", + action="ensure_setting", + ), + ] + + +def ensure_vm(name: str, *, execute: bool) -> list[OperationResult]: + vm = VMS[name] + return create_if_missing( + probe=["gcloud", "compute", "instances", "describe", name, "--zone", ZONE, "--project", PROJECT, "--format=json"], + create=[ + "gcloud", + "compute", + "instances", + "create", + name, + "--project", + PROJECT, + "--zone", + ZONE, + "--machine-type", + vm["machine_type"], + "--image-family=debian-12", + "--image-project=debian-cloud", + "--boot-disk-size", + vm["boot_disk_size"], + "--boot-disk-type=pd-balanced", + "--network", + NETWORK, + "--subnet", + SUBNET, + "--no-address", + "--tags", + vm["tag"], + "--service-account", + vm["service_account"], + "--scopes=cloud-platform", + "--metadata=enable-oslogin=TRUE", + ], + execute=execute, + name=f"create_vm:{name}", + ) + + +def ensure_snapshot_policy(*, execute: bool) -> list[OperationResult]: + operations = create_if_missing( + probe=[ + "gcloud", + "compute", + "resource-policies", + "describe", + SNAPSHOT_POLICY, + "--region", + REGION, + "--project", + PROJECT, + "--format=json", + ], + create=[ + "gcloud", + "compute", + "resource-policies", + "create", + "snapshot-schedule", + SNAPSHOT_POLICY, + "--project", + PROJECT, + "--region", + REGION, + "--start-time=03:00", + "--daily-schedule", + "--max-retention-days=7", + "--on-source-disk-delete=apply-retention-policy", + "--storage-location", + REGION, + ], + execute=execute, + name=f"create_snapshot_policy:{SNAPSHOT_POLICY}", + ) + for disk in VMS: + operations.append( + run_command( + [ + "gcloud", + "compute", + "disks", + "add-resource-policies", + disk, + "--project", + PROJECT, + "--zone", + ZONE, + "--resource-policies", + SNAPSHOT_POLICY, + ], + execute=execute, + name=f"attach_snapshot_policy:{disk}", + action="ensure_binding", + ) + ) + return operations + + +def ensure_bucket(bucket: str, *, execute: bool) -> list[OperationResult]: + operations = create_if_missing( + probe=["gcloud", "storage", "buckets", "describe", f"gs://{bucket}", "--format=json"], + create=[ + "gcloud", + "storage", + "buckets", + "create", + f"gs://{bucket}", + "--project", + PROJECT, + "--location", + REGION, + "--uniform-bucket-level-access", + "--public-access-prevention", + "--soft-delete-duration=7d", + ], + execute=execute, + name=f"create_bucket:{bucket}", + ) + operations.append( + run_command( + [ + "gcloud", + "storage", + "buckets", + "update", + f"gs://{bucket}", + "--versioning", + "--uniform-bucket-level-access", + "--public-access-prevention", + ], + execute=execute, + name=f"harden_bucket:{bucket}", + action="ensure_setting", + ) + ) + return operations + + +def ensure_secret(password: str | None, *, execute: bool, password_env: str) -> list[OperationResult]: + operations = create_if_missing( + probe=["gcloud", "secrets", "describe", CLOUDSQL_PASSWORD_SECRET, "--project", PROJECT, "--format=json"], + create=[ + "gcloud", + "secrets", + "create", + CLOUDSQL_PASSWORD_SECRET, + "--project", + PROJECT, + "--replication-policy=automatic", + ], + execute=execute, + name=f"create_secret:{CLOUDSQL_PASSWORD_SECRET}", + ) + if execute and password is None: + operations.append( + OperationResult( + name=f"secret_version_missing_env:{password_env}", + command=[], + action="blocked_missing_secret_env", + exitcode=1, + stderr_tail=f"{password_env} is required to create the Cloud SQL password secret version", + ) + ) + return operations + operations.append( + run_command( + [ + "gcloud", + "secrets", + "versions", + "add", + CLOUDSQL_PASSWORD_SECRET, + "--project", + PROJECT, + "--data-file=-", + ], + execute=execute, + name=f"add_secret_version:{CLOUDSQL_PASSWORD_SECRET}", + action="ensure_secret_version", + recorded_command=[ + "gcloud", + "secrets", + "versions", + "add", + CLOUDSQL_PASSWORD_SECRET, + "--project", + PROJECT, + "--data-file=-", + f"# stdin from ${password_env}", + ], + stdin=password, + ) + ) + return operations + + +def ensure_cloudsql(password: str | None, *, execute: bool, password_env: str) -> list[OperationResult]: + recorded_create = [ + "gcloud", + "sql", + "instances", + "create", + CLOUDSQL_INSTANCE, + "--project", + PROJECT, + "--database-version=POSTGRES_16", + "--region", + REGION, + "--cpu=2", + "--memory=8GiB", + "--storage-type=SSD", + "--storage-size=50GB", + "--availability-type=zonal", + "--network", + f"projects/{PROJECT}/global/networks/{NETWORK}", + "--no-assign-ip", + "--ssl-mode=ENCRYPTED_ONLY", + "--database-flags=cloudsql.iam_authentication=on", + "--deletion-protection", + "--backup-start-time=03:00", + "--enable-point-in-time-recovery", + "--retained-backups-count=7", + "--retained-transaction-log-days=7", + f"--root-password=${password_env}", + ] + if execute and password is None: + return [ + OperationResult( + name=f"cloudsql_password_missing_env:{password_env}", + command=recorded_create, + action="blocked_missing_secret_env", + exitcode=1, + stderr_tail=f"{password_env} is required to create {CLOUDSQL_INSTANCE}", + ) + ] + create = [part if part != f"--root-password=${password_env}" else f"--root-password={password}" for part in recorded_create] + operations = create_if_missing( + probe=[ + "gcloud", + "sql", + "instances", + "describe", + CLOUDSQL_INSTANCE, + "--project", + PROJECT, + "--format=json", + ], + create=create, + execute=execute, + name=f"create_cloudsql_instance:{CLOUDSQL_INSTANCE}", + recorded_create=recorded_create, + ) + operations.append( + run_command( + [ + "gcloud", + "sql", + "databases", + "create", + CLOUDSQL_DATABASE, + "--instance", + CLOUDSQL_INSTANCE, + "--project", + PROJECT, + ], + execute=execute, + name=f"create_cloudsql_database:{CLOUDSQL_DATABASE}", + action="create_if_missing", + ) + ) + return operations + + +def build_operations(*, execute: bool, admin_ssh_cidr: str | None, password_env: str) -> list[OperationResult]: + try: + ssh_cidr = validate_admin_ssh_cidr(admin_ssh_cidr, execute=execute) + except ValueError as exc: + return [OperationResult(name="validate_admin_ssh_cidr", command=[], action="validate", exitcode=1, stderr_tail=str(exc))] + + password = os.environ.get(password_env) + operations: list[OperationResult] = [] + for api in REQUIRED_APIS: + operations.append( + run_command( + ["gcloud", "services", "enable", api, "--project", PROJECT], + execute=execute, + name=f"enable_api:{api}", + action="ensure_enabled", + ) + ) + if operations[-1].exitcode not in (None, 0): + return operations + + for account_id, account in SERVICE_ACCOUNTS.items(): + operations.extend( + ensure_service_account( + account_id, + str(account["email"]), + str(account["display_name"]), + execute=execute, + ) + ) + if operations[-1].exitcode not in (None, 0): + return operations + for role in account["roles"]: + operations.append(ensure_project_binding(str(account["email"]), str(role), execute=execute, prefix=f"{account_id}_project_role")) + if operations[-1].exitcode not in (None, 0): + return operations + + operations.extend(ensure_network(execute=execute)) + if operations[-1].exitcode not in (None, 0): + return operations + operations.extend(ensure_subnet(execute=execute)) + if operations[-1].exitcode not in (None, 0): + return operations + operations.extend(ensure_private_services_access(execute=execute)) + if operations[-1].exitcode not in (None, 0): + return operations + + for broad_rule in ("default-allow-ssh", "default-allow-rdp"): + operations.append( + run_command( + ["gcloud", "compute", "firewall-rules", "update", broad_rule, "--project", PROJECT, "--disabled"], + execute=execute, + name=f"disable_broad_firewall_rule:{broad_rule}", + action="disable_if_present", + ) + ) + if operations[-1].exitcode not in (None, 0) and "not found" not in operations[-1].stderr_tail.lower(): + return operations + + for name, tag in SSH_FIREWALL_RULES.items(): + operations.extend(ensure_firewall_rule(name, tag, ssh_cidr, execute=execute)) + if operations[-1].exitcode not in (None, 0): + return operations + + for name in VMS: + operations.extend(ensure_vm(name, execute=execute)) + if operations[-1].exitcode not in (None, 0): + return operations + + operations.extend(ensure_snapshot_policy(execute=execute)) + if operations[-1].exitcode not in (None, 0): + return operations + + for bucket in BACKUP_BUCKETS: + operations.extend(ensure_bucket(bucket, execute=execute)) + if operations[-1].exitcode not in (None, 0): + return operations + + operations.extend(ensure_secret(password, execute=execute, password_env=password_env)) + if operations[-1].exitcode not in (None, 0): + return operations + operations.extend(ensure_cloudsql(password, execute=execute, password_env=password_env)) + return operations + + +def write_output(path: Path, payload: dict[str, object]) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n") + + +def main() -> int: + parser = argparse.ArgumentParser() + parser.add_argument("--execute", action="store_true", help="apply resources; default is dry-run") + parser.add_argument("--admin-ssh-cidr", help="single trusted operator IPv4 /32 allowed to SSH") + parser.add_argument("--postgres-password-env", default="TELEO_CLOUDSQL_POSTGRES_PASSWORD") + parser.add_argument( + "--output", + default="outputs/gcp-infra-hardening-20260707/proofs/gcp-runtime-baseline-apply.json", + help="proof JSON path", + ) + args = parser.parse_args() + + operations = build_operations( + execute=args.execute, + admin_ssh_cidr=args.admin_ssh_cidr, + password_env=args.postgres_password_env, + ) + failed = [operation for operation in operations if operation.exitcode not in (None, 0)] + payload = { + "artifact": "teleo_gcp_runtime_baseline_apply", + "generated_at_utc": datetime.now(UTC).isoformat(), + "mode": "execute" if args.execute else "dry_run", + "project": PROJECT, + "region": REGION, + "zone": ZONE, + "network": NETWORK, + "subnet": SUBNET, + "admin_ssh_cidr": args.admin_ssh_cidr if args.admin_ssh_cidr else ("required_in_execute" if args.execute else "/32"), + "password_env": args.postgres_password_env, + "resources": { + "service_accounts": {account_id: account["email"] for account_id, account in SERVICE_ACCOUNTS.items()}, + "vms": VMS, + "backup_buckets": BACKUP_BUCKETS, + "snapshot_policy": SNAPSHOT_POLICY, + "cloudsql_instance": CLOUDSQL_INSTANCE, + "cloudsql_database": CLOUDSQL_DATABASE, + "cloudsql_password_secret": CLOUDSQL_PASSWORD_SECRET, + }, + "operations": [asdict(operation) for operation in operations], + "status": "failed" if failed else ("applied" if args.execute else "planned"), + "failed_operation_count": len(failed), + "not_proven_by_this_artifact": [ + "GCP runtime readiness until execute mode succeeds and gcp-readiness passes", + "Cloud SQL row-count parity until restore/import/readback verifier passes", + "application cutover or Telegram bot parity", + ], + "next_checks": [ + "python3 ops/apply_gcp_iam_split.py --execute", + "python3 ops/apply_gcp_runtime_baseline.py --execute --admin-ssh-cidr /32", + "gh workflow run gcp-readiness.yml --repo living-ip/teleo-infrastructure --ref main -f service_account=sa-teleo-readiness@teleo-501523.iam.gserviceaccount.com", + "EXECUTE=1 ops/run_gcp_cloudsql_restore_drill.sh", + "python3 ops/verify_gcp_cloudsql_restore_readback.py --drill-proof --target-counts-csv --output ", + ], + } + write_output(Path(args.output), payload) + print(json.dumps(payload, indent=2, sort_keys=True)) + return 1 if failed else 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/ops/check_gcp_infra_readiness.py b/ops/check_gcp_infra_readiness.py index 68fdac9..e9b1c84 100755 --- a/ops/check_gcp_infra_readiness.py +++ b/ops/check_gcp_infra_readiness.py @@ -649,6 +649,52 @@ def check_gcp_cloudsql_restore_drill_contract() -> Check: ) +def check_gcp_runtime_baseline_contract() -> Check: + script = Path("ops/apply_gcp_runtime_baseline.py") + runbook = Path("docs/gcp-infra-hardening-20260707.md") + missing = [str(path) for path in (script, runbook) if not path.exists()] + if missing: + return Check( + "gcp_runtime_baseline_contract", + "blocked", + f"missing={missing}", + required_tier="T2_runtime_baseline_contract", + current_tier="T1_foundation", + ) + text = script.read_text() + runbook_text = runbook.read_text() + required = [ + "--execute", + "--admin-ssh-cidr", + "teleo-prod-1", + "teleo-staging-1", + "teleo-501523-prod-backups", + "teleo-501523-leoclean-backups", + "teleo-pgvector-standby", + "teleo-daily-7d-snapshots", + "TELEO_CLOUDSQL_POSTGRES_PASSWORD", + "not_proven_by_this_artifact", + ] + absent = [item for item in required if item not in text] + if "apply_gcp_runtime_baseline.py" not in runbook_text: + absent.append("runbook_reference") + if absent: + return Check( + "gcp_runtime_baseline_contract", + "fail", + f"missing_contract={absent}", + required_tier="T2_runtime_baseline_contract", + current_tier="T1_foundation", + ) + return Check( + "gcp_runtime_baseline_contract", + "pass", + "dry-run-by-default runtime baseline runner covers service accounts, network, firewall, VMs, snapshots, buckets, and Cloud SQL target", + required_tier="T2_runtime_baseline_contract", + current_tier="T2_runtime_baseline_contract", + ) + + def check_kb_restore_or_replication() -> Check: return Check( "kb_restore_or_replication", @@ -674,6 +720,7 @@ def main() -> int: 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("gcp_runtime_baseline_contract", check_gcp_runtime_baseline_contract), safe_check("kb_restore_or_replication", check_kb_restore_or_replication), ] payload = { diff --git a/tests/test_gcp_runtime_baseline_apply.py b/tests/test_gcp_runtime_baseline_apply.py new file mode 100644 index 0000000..6a483b1 --- /dev/null +++ b/tests/test_gcp_runtime_baseline_apply.py @@ -0,0 +1,74 @@ +import json +import subprocess +from pathlib import Path + +REPO_ROOT = Path(__file__).resolve().parents[1] + + +def run_runtime_baseline(*args: str) -> subprocess.CompletedProcess[str]: + return subprocess.run( + ["python3", "ops/apply_gcp_runtime_baseline.py", *args], + cwd=REPO_ROOT, + text=True, + capture_output=True, + check=False, + ) + + +def test_gcp_runtime_baseline_dry_run_writes_planned_operations(tmp_path: Path) -> None: + output = tmp_path / "runtime-baseline.json" + completed = run_runtime_baseline("--admin-ssh-cidr", "198.51.100.10/32", "--output", str(output)) + + assert completed.returncode == 0, completed.stderr + payload = json.loads(output.read_text()) + assert payload["mode"] == "dry_run" + assert payload["status"] == "planned" + assert payload["failed_operation_count"] == 0 + operation_names = {operation["name"] for operation in payload["operations"]} + assert "create_vm:teleo-prod-1" in operation_names + assert "create_vm:teleo-staging-1" in operation_names + assert "create_bucket:teleo-501523-prod-backups" in operation_names + assert "create_cloudsql_instance:teleo-pgvector-standby" in operation_names + assert "add_secret_version:gcp-teleo-pgvector-standby-postgres-password" in operation_names + assert all(operation["exitcode"] is None for operation in payload["operations"]) + + +def test_gcp_runtime_baseline_dry_run_does_not_record_raw_password(tmp_path: Path, monkeypatch) -> None: + monkeypatch.setenv("TELEO_CLOUDSQL_POSTGRES_PASSWORD", "do-not-record-this") + output = tmp_path / "runtime-baseline.json" + completed = run_runtime_baseline("--admin-ssh-cidr", "198.51.100.10/32", "--output", str(output)) + + assert completed.returncode == 0, completed.stderr + text = output.read_text() + assert "do-not-record-this" not in text + assert "--root-password=$TELEO_CLOUDSQL_POSTGRES_PASSWORD" in text + + +def test_gcp_runtime_baseline_execute_requires_admin_ssh_cidr_without_shelling_out(tmp_path: Path) -> None: + output = tmp_path / "runtime-baseline.json" + completed = run_runtime_baseline("--execute", "--output", str(output)) + + assert completed.returncode == 1 + payload = json.loads(output.read_text()) + assert payload["mode"] == "execute" + assert payload["status"] == "failed" + assert payload["operations"] == [ + { + "action": "validate", + "command": [], + "exitcode": 1, + "name": "validate_admin_ssh_cidr", + "stderr_tail": "--admin-ssh-cidr is required in execute mode", + "stdout_tail": "", + } + ] + + +def test_gcp_runtime_baseline_contract_is_part_of_readiness_checker() -> None: + checker = (REPO_ROOT / "ops/check_gcp_infra_readiness.py").read_text() + docs = (REPO_ROOT / "docs/gcp-infra-hardening-20260707.md").read_text() + + assert "check_gcp_runtime_baseline_contract" in checker + assert "apply_gcp_runtime_baseline.py" in checker + assert "Current-State Rule" in docs + assert "apply_gcp_runtime_baseline.py" in docs