Add idempotent GCP IAM apply runner

.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 \

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`

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())

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"])