From 607baefc524d4ef545ff51fbd9104ba604d92af6 Mon Sep 17 00:00:00 2001 From: twentyOne2x Date: Tue, 7 Jul 2026 13:05:08 +0200 Subject: [PATCH] Add idempotent GCP IAM apply runner --- .github/workflows/ci.yml | 2 + docs/gcp-infra-hardening-20260707.md | 15 ++ ops/apply_gcp_iam_split.py | 318 +++++++++++++++++++++++++++ tests/test_gcp_iam_split_apply.py | 42 ++++ 4 files changed, 377 insertions(+) create mode 100644 ops/apply_gcp_iam_split.py create mode 100644 tests/test_gcp_iam_split_apply.py diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index dbebfbd..1db7ccd 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -41,6 +41,7 @@ jobs: lib/evaluate.py \ lib/llm.py \ lib/post_extract.py \ + ops/apply_gcp_iam_split.py \ ops/check_gcp_infra_readiness.py \ ops/plan_gcp_iam_split.py \ ops/sqlite_to_postgres_dump.py \ @@ -53,6 +54,7 @@ jobs: tests/test_decision_engine_replay.py \ tests/test_evaluate_agent_routing.py \ tests/test_gcp_cloudsql_restore_drill.py \ + tests/test_gcp_iam_split_apply.py \ tests/test_gcp_iam_split_plan.py \ tests/test_gcp_readiness_workflow.py \ tests/test_phase1b_end_to_end.py \ diff --git a/docs/gcp-infra-hardening-20260707.md b/docs/gcp-infra-hardening-20260707.md index 923d985..e7bce59 100644 --- a/docs/gcp-infra-hardening-20260707.md +++ b/docs/gcp-infra-hardening-20260707.md @@ -149,6 +149,21 @@ python3 ops/plan_gcp_iam_split.py --format json python3 ops/plan_gcp_iam_split.py --format shell ``` +For an idempotent retained apply attempt, use: + +```bash +python3 ops/apply_gcp_iam_split.py \ + --output outputs/gcp-infra-hardening-20260707/proofs/gcp-iam-split-apply-dry-run.json + +python3 ops/apply_gcp_iam_split.py \ + --execute \ + --output outputs/gcp-infra-hardening-20260707/proofs/gcp-iam-split-apply-execute.json +``` + +The first command is dry-run only. The second command mutates IAM and must run +from an authenticated GCP admin shell. It is safe to re-run: existing service +accounts are skipped, and IAM binding commands are additive/idempotent. + The plan creates two separate accounts: - `sa-teleo-readiness@teleo-501523.iam.gserviceaccount.com` diff --git a/ops/apply_gcp_iam_split.py b/ops/apply_gcp_iam_split.py new file mode 100644 index 0000000..dc7663e --- /dev/null +++ b/ops/apply_gcp_iam_split.py @@ -0,0 +1,318 @@ +#!/usr/bin/env python3 +"""Apply the Teleo GCP IAM split with retained proof. + +Default mode is dry-run. Execute mode is intentionally idempotent where GCP +supports it: existing service accounts are skipped, IAM bindings are added with +Google's policy-binding commands, and every command result is retained without +printing secret values. +""" + +from __future__ import annotations + +import argparse +import json +import subprocess +from dataclasses import asdict, dataclass +from datetime import UTC, datetime +from pathlib import Path + +from plan_gcp_iam_split import ( + BACKUP_BUCKET, + CLOUDSQL_INSTANCE, + PROJECT, + READINESS_PROJECT_ROLES, + READINESS_SERVICE_ACCOUNT, + READINESS_SERVICE_ACCOUNT_ID, + RESTORE_BUCKET_ROLES, + RESTORE_PROJECT_ROLES, + RESTORE_SERVICE_ACCOUNT, + RESTORE_SERVICE_ACCOUNT_ID, + build_plan, + github_wif_member, +) + + +@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) -> OperationResult: + if not execute: + return OperationResult(name=name, command=command, action=action) + completed = subprocess.run(command, text=True, capture_output=True, check=False) + return OperationResult( + name=name, + command=command, + action=action, + exitcode=completed.returncode, + stdout_tail=tail(completed.stdout), + stderr_tail=tail(completed.stderr), + ) + + +def service_account_exists(email: str) -> OperationResult: + return run_command( + [ + "gcloud", + "iam", + "service-accounts", + "describe", + email, + "--project", + PROJECT, + "--format=json", + ], + execute=True, + name=f"describe_service_account:{email}", + action="probe", + ) + + +def create_service_account(account_id: str, email: str, display_name: str, *, execute: bool) -> list[OperationResult]: + if execute: + probe = service_account_exists(email) + if probe.exitcode == 0: + probe.action = "already_exists" + return [probe] + if "Reauthentication failed" in probe.stderr_tail: + return [probe] + return [ + run_command( + [ + "gcloud", + "iam", + "service-accounts", + "create", + account_id, + "--project", + PROJECT, + "--display-name", + display_name, + ], + execute=execute, + name=f"create_service_account:{email}", + action="create_if_missing", + ) + ] + + +def add_service_account_iam_binding( + email: str, + member: str, + role: str, + *, + execute: bool, + name: str, +) -> OperationResult: + return run_command( + [ + "gcloud", + "iam", + "service-accounts", + "add-iam-policy-binding", + email, + "--project", + PROJECT, + "--member", + member, + "--role", + role, + "--quiet", + ], + execute=execute, + name=name, + action="ensure_binding", + ) + + +def add_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 add_bucket_binding(member: str, role: str, *, execute: bool, prefix: str) -> OperationResult: + return run_command( + [ + "gcloud", + "storage", + "buckets", + "add-iam-policy-binding", + f"gs://{BACKUP_BUCKET}", + "--member", + member, + "--role", + role, + "--quiet", + ], + execute=execute, + name=f"{prefix}:{role}", + action="ensure_binding", + ) + + +def read_cloudsql_instance_service_account(*, execute: bool) -> tuple[str | None, OperationResult]: + command = [ + "gcloud", + "sql", + "instances", + "describe", + CLOUDSQL_INSTANCE, + "--project", + PROJECT, + "--format=value(serviceAccountEmailAddress)", + ] + result = run_command( + command, + execute=execute, + name="describe_cloudsql_instance_service_account", + action="probe", + ) + if not execute or result.exitcode != 0: + return None, result + value = result.stdout_tail.strip() + return value or None, result + + +def build_operations(*, execute: bool) -> list[OperationResult]: + operations: list[OperationResult] = [] + operations.extend( + create_service_account( + READINESS_SERVICE_ACCOUNT_ID, + READINESS_SERVICE_ACCOUNT, + "Teleo read-only infra readiness", + execute=execute, + ) + ) + if operations[-1].exitcode not in (None, 0): + return operations + + operations.append( + add_service_account_iam_binding( + READINESS_SERVICE_ACCOUNT, + github_wif_member(), + "roles/iam.workloadIdentityUser", + execute=execute, + name="bind_github_wif_to_readiness_service_account", + ) + ) + if operations[-1].exitcode not in (None, 0): + return operations + + for role in READINESS_PROJECT_ROLES: + operations.append(add_project_binding(READINESS_SERVICE_ACCOUNT, role, execute=execute, prefix="readiness_project_role")) + if operations[-1].exitcode not in (None, 0): + return operations + + operations.extend( + create_service_account( + RESTORE_SERVICE_ACCOUNT_ID, + RESTORE_SERVICE_ACCOUNT, + "Teleo Cloud SQL restore drill", + execute=execute, + ) + ) + if operations[-1].exitcode not in (None, 0): + return operations + + for role in RESTORE_PROJECT_ROLES: + operations.append(add_project_binding(RESTORE_SERVICE_ACCOUNT, role, execute=execute, prefix="restore_project_role")) + if operations[-1].exitcode not in (None, 0): + return operations + + for role in RESTORE_BUCKET_ROLES: + operations.append( + add_bucket_binding(f"serviceAccount:{RESTORE_SERVICE_ACCOUNT}", role, execute=execute, prefix="restore_bucket_role") + ) + if operations[-1].exitcode not in (None, 0): + return operations + + instance_service_account, probe = read_cloudsql_instance_service_account(execute=execute) + operations.append(probe) + if probe.exitcode not in (None, 0): + return operations + + if execute and not instance_service_account: + operations.append( + OperationResult( + name="cloudsql_instance_service_account_missing", + command=[], + action="fail", + exitcode=1, + stderr_tail=f"Cloud SQL instance {CLOUDSQL_INSTANCE} did not return serviceAccountEmailAddress", + ) + ) + return operations + + cloudsql_member = ( + f"serviceAccount:{instance_service_account}" + if instance_service_account + else f"serviceAccount:$(gcloud sql instances describe {CLOUDSQL_INSTANCE} --project {PROJECT} --format=value(serviceAccountEmailAddress))" + ) + operations.append( + add_bucket_binding(cloudsql_member, "roles/storage.objectAdmin", execute=execute, prefix="cloudsql_instance_bucket_role") + ) + 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 IAM bindings; default is dry-run") + parser.add_argument( + "--output", + default="outputs/gcp-infra-hardening-20260707/proofs/gcp-iam-split-apply.json", + help="proof JSON path", + ) + args = parser.parse_args() + + operations = build_operations(execute=args.execute) + failed = [operation for operation in operations if operation.exitcode not in (None, 0)] + payload = { + "artifact": "teleo_gcp_iam_split_apply", + "generated_at_utc": datetime.now(UTC).isoformat(), + "mode": "execute" if args.execute else "dry_run", + "project": PROJECT, + "plan": build_plan(), + "operations": [asdict(operation) for operation in operations], + "status": "failed" if failed else ("applied" if args.execute else "planned"), + "failed_operation_count": len(failed), + "next_checks": [ + f"gh workflow run gcp-readiness.yml --repo living-ip/teleo-infrastructure --ref main -f service_account={READINESS_SERVICE_ACCOUNT}", + "EXECUTE=1 ops/run_gcp_cloudsql_restore_drill.sh", + ], + } + 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/tests/test_gcp_iam_split_apply.py b/tests/test_gcp_iam_split_apply.py new file mode 100644 index 0000000..a2ebee0 --- /dev/null +++ b/tests/test_gcp_iam_split_apply.py @@ -0,0 +1,42 @@ +import json +import subprocess +from pathlib import Path + +REPO_ROOT = Path(__file__).resolve().parents[1] + + +def run_apply(*args: str) -> subprocess.CompletedProcess[str]: + return subprocess.run( + ["python3", "ops/apply_gcp_iam_split.py", *args], + cwd=REPO_ROOT, + text=True, + capture_output=True, + check=False, + ) + + +def test_gcp_iam_apply_dry_run_writes_planned_operations(tmp_path: Path) -> None: + output = tmp_path / "iam-apply.json" + completed = run_apply("--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 + assert payload["plan"]["readiness_service_account"] == "sa-teleo-readiness@teleo-501523.iam.gserviceaccount.com" + assert payload["plan"]["restore_service_account"] == "sa-teleo-restore-drill@teleo-501523.iam.gserviceaccount.com" + assert any(operation["name"].startswith("readiness_project_role:") for operation in payload["operations"]) + assert any(operation["name"].startswith("restore_bucket_role:") for operation in payload["operations"]) + assert any(operation["name"] == "describe_cloudsql_instance_service_account" for operation in payload["operations"]) + + +def test_gcp_iam_apply_dry_run_does_not_shell_out_to_gcloud(tmp_path: Path) -> None: + output = tmp_path / "iam-apply.json" + completed = run_apply("--output", str(output)) + + assert completed.returncode == 0, completed.stderr + payload = json.loads(output.read_text()) + assert all(operation["exitcode"] is None for operation in payload["operations"]) + assert all(operation["stdout_tail"] == "" for operation in payload["operations"]) + assert all(operation["stderr_tail"] == "" for operation in payload["operations"])