#!/usr/bin/env python3
"""Converge the reviewed GitHub OIDC/WIF and GCP IAP operator bootstrap.

The default mode is a no-network dry run. ``--execute`` performs read-before-
write discovery, fails closed on security-sensitive drift, repairs only absent
resources or narrow settings, verifies the resulting state, and installs the
reviewed dispatcher. It deliberately does not dispatch GitHub Actions and does
not disable or otherwise mutate the existing public SSH firewall rule.
"""

from __future__ import annotations

import argparse
import hashlib
import json
import subprocess
import tempfile
from collections.abc import Callable
from dataclasses import asdict, dataclass
from datetime import UTC, datetime
from pathlib import Path
from typing import Any

from plan_gcp_iap_operator_access import (
    ATTRIBUTE_MAPPING,
    CHANGING_PUBLIC_RULE,
    CLONE_SERVICE_ACCOUNT,
    CLONE_SERVICE_ACCOUNT_ID,
    CUSTOM_PERMISSIONS,
    CUSTOM_ROLE,
    CUSTOM_ROLE_ID,
    DISPATCHER_VERSION,
    FORBIDDEN_OPERATOR_ROLES,
    GITHUB_REPOSITORY,
    IAP_CONDITION,
    IAP_FIREWALL_RULE,
    IAP_ROLE,
    IAP_SOURCE_RANGE,
    INSTANCE,
    INSTANCE_INTERNAL_IP,
    NETWORK,
    POOL,
    PROJECT,
    PROVIDER,
    PROVIDER_CONDITION,
    STATUS_SERVICE_ACCOUNT,
    STATUS_SERVICE_ACCOUNT_ID,
    TARGET_TAG,
    VM_SERVICE_ACCOUNT,
    WIF_PRINCIPAL,
    WORKFLOW_REF,
    ZONE,
)

Runner = Callable[..., subprocess.CompletedProcess[str]]

REQUIRED_APIS = (
    "compute.googleapis.com",
    "iap.googleapis.com",
    "iamcredentials.googleapis.com",
    "sts.googleapis.com",
)
PROVIDER_ISSUER = "https://token.actions.githubusercontent.com"
PROVIDER_DISPLAY_NAME = "Teleo fixed IAP operator"
POOL_DISPLAY_NAME = "GitHub Actions"
CUSTOM_ROLE_TITLE = "Teleo exact-instance SSH discovery"
CUSTOM_ROLE_DESCRIPTION = "Only the instance and project reads required by gcloud compute ssh"
IAP_CONDITION_TITLE = "teleo-prod-1-ssh-only"
REMOTE_DISPATCHER = "/usr/local/sbin/teleo-gcp-iap-operator"
REMOTE_RUN_ROOT = "/var/lib/teleo-gcp-iap-operator/runs"
DEFAULT_OUTPUT = "outputs/gcp-infra-hardening-20260707/proofs/gcp-iap-operator-apply.json"
DEFAULT_DISPATCHER = Path(__file__).resolve().parents[1] / "scripts/gcp_iap_operator.sh"
EXPECTED_ATTRIBUTE_MAPPING = dict(item.split("=", 1) for item in ATTRIBUTE_MAPPING.split(","))
SERVICE_ACCOUNTS = (
    (STATUS_SERVICE_ACCOUNT_ID, STATUS_SERVICE_ACCOUNT, "Teleo IAP status canary"),
    (CLONE_SERVICE_ACCOUNT_ID, CLONE_SERVICE_ACCOUNT, "Teleo IAP clone operator"),
)
OPERATOR_MEMBERS = tuple(f"serviceAccount:{email}" for _, email, _ in SERVICE_ACCOUNTS)


class ConvergenceError(RuntimeError):
    """A fail-closed discovery, command, or verification error."""


@dataclass
class Operation:
    name: str
    action: str
    command: list[str]
    status: str
    exitcode: int | None = None
    error_class: str | None = None


@dataclass(frozen=True)
class Mutation:
    name: str
    action: str
    command: list[str]


def _redact_command(command: list[str]) -> list[str]:
    redacted: list[str] = []
    redact_next = False
    for value in command:
        lowered = value.lower()
        if redact_next:
            redacted.append("<redacted>")
            redact_next = False
            continue
        redacted.append(value)
        if any(marker in lowered for marker in ("password", "credential", "access-token", "refresh-token")):
            redact_next = True
    return redacted


def _error_class(stderr: str) -> str:
    lowered = stderr.lower()
    if "reauthentication failed" in lowered or "cannot prompt during non-interactive" in lowered:
        return "reauthentication_required"
    if "permission_denied" in lowered or "permission denied" in lowered or "forbidden" in lowered:
        return "permission_denied"
    if "not_found" in lowered or "not found" in lowered or "does not exist" in lowered:
        return "not_found"
    return "command_failed"


class Gcloud:
    def __init__(self, runner: Runner | None = None) -> None:
        self.runner = runner or subprocess.run
        self.operations: list[Operation] = []

    def run(
        self,
        name: str,
        action: str,
        command: list[str],
        *,
        recorded_command: list[str] | None = None,
        allow_not_found: bool = False,
    ) -> subprocess.CompletedProcess[str] | None:
        _validate_command(command, mutating=action not in {"probe", "verify"})
        completed = self.runner(command, text=True, capture_output=True, check=False)
        safe_command = _redact_command(recorded_command or command)
        if completed.returncode == 0:
            self.operations.append(Operation(name, action, safe_command, "pass", 0))
            return completed
        classification = _error_class(completed.stderr or "")
        if allow_not_found and classification == "not_found":
            self.operations.append(
                Operation(name, action, safe_command, "absent", completed.returncode, classification)
            )
            return None
        self.operations.append(Operation(name, action, safe_command, "fail", completed.returncode, classification))
        raise ConvergenceError(f"{name} failed ({classification})")

    def json(
        self,
        name: str,
        command: list[str],
        *,
        allow_not_found: bool = False,
        action: str = "probe",
    ) -> Any | None:
        completed = self.run(name, action, command, allow_not_found=allow_not_found)
        if completed is None:
            return None
        try:
            return json.loads(completed.stdout)
        except (TypeError, json.JSONDecodeError) as exc:
            self.operations[-1].status = "fail"
            self.operations[-1].error_class = "invalid_json"
            raise ConvergenceError(f"{name} returned invalid JSON") from exc


def _validate_command(command: list[str], *, mutating: bool) -> None:
    if not command or command[0] != "gcloud":
        raise ConvergenceError("only structured gcloud commands are permitted")
    if any(value in {"workflow", "workflows"} for value in command):
        raise ConvergenceError("workflow dispatch is outside this bootstrap")
    if mutating and CHANGING_PUBLIC_RULE in command:
        raise ConvergenceError("the public SSH firewall rule is read-only in this bootstrap")
    if mutating and "--disabled" in command:
        raise ConvergenceError("this bootstrap never disables access")


def _project_args() -> list[str]:
    return ["--project", PROJECT]


def _json_args() -> list[str]:
    return ["--format=json"]


def _policy_bindings(policy: dict[str, Any] | None) -> list[dict[str, Any]]:
    if not policy:
        return []
    bindings = policy.get("bindings", [])
    if not isinstance(bindings, list):
        raise ConvergenceError("IAM policy bindings are not a list")
    return [binding for binding in bindings if isinstance(binding, dict)]


def _condition(binding: dict[str, Any]) -> tuple[str, str] | None:
    condition = binding.get("condition")
    if not condition:
        return None
    if not isinstance(condition, dict):
        return ("<invalid>", "<invalid>")
    return (str(condition.get("title", "")), str(condition.get("expression", "")))


def _binding_matches(
    binding: dict[str, Any],
    *,
    member: str,
    role: str,
    condition: tuple[str, str] | None,
) -> bool:
    members = binding.get("members", [])
    return binding.get("role") == role and member in members and _condition(binding) == condition


def _ensure_binding(
    *,
    policy: dict[str, Any] | None,
    member: str,
    role: str,
    condition: tuple[str, str] | None,
    mutation: Mutation,
    scope: str,
) -> Mutation | None:
    bindings = _policy_bindings(policy)
    if any(_binding_matches(binding, member=member, role=role, condition=condition) for binding in bindings):
        return None
    conflicts = [
        binding
        for binding in bindings
        if binding.get("role") == role and member in binding.get("members", []) and _condition(binding) != condition
    ]
    if conflicts:
        raise ConvergenceError(f"{scope} has a conflicting conditional binding for {member} and {role}")
    return mutation


def _assert_exact_member_roles(
    policy: dict[str, Any] | None,
    *,
    members: tuple[str, ...],
    allowed_roles: set[str],
    scope: str,
) -> None:
    observed: dict[str, set[str]] = {member: set() for member in members}
    for binding in _policy_bindings(policy):
        role = str(binding.get("role", ""))
        for member in members:
            if member in binding.get("members", []):
                observed[member].add(role)
    for member, roles in observed.items():
        unexpected = roles - allowed_roles
        if unexpected:
            raise ConvergenceError(f"{scope} grants unexpected roles to {member}: {sorted(unexpected)}")
        forbidden = roles & set(FORBIDDEN_OPERATOR_ROLES)
        if forbidden:
            raise ConvergenceError(f"{scope} grants forbidden roles to {member}: {sorted(forbidden)}")


def _metadata(instance: dict[str, Any]) -> dict[str, str]:
    items = instance.get("metadata", {}).get("items", [])
    if not isinstance(items, list):
        raise ConvergenceError("instance metadata items are invalid")
    return {str(item.get("key")): str(item.get("value")) for item in items if isinstance(item, dict)}


def _find_resource(resources: Any, *, suffix: str, label: str) -> dict[str, Any] | None:
    if not isinstance(resources, list):
        raise ConvergenceError(f"{label} deleted-resource readback is invalid")
    matches = [
        item
        for item in resources
        if isinstance(item, dict) and str(item.get("name", "")).endswith(suffix)
    ]
    if len(matches) > 1:
        raise ConvergenceError(f"{label} deleted-resource readback is ambiguous")
    return matches[0] if matches else None


def _firewall_is_structurally_exact(firewall: dict[str, Any]) -> bool:
    allowed = firewall.get("allowed", [])
    return (
        firewall.get("direction") == "INGRESS"
        and str(firewall.get("network", "")).endswith(f"/networks/{NETWORK}")
        and allowed == [{"IPProtocol": "tcp", "ports": ["22"]}]
        and set(firewall.get("sourceRanges", [])) == {IAP_SOURCE_RANGE}
    )


def discover(client: Gcloud, *, require_public_rule: bool = True) -> dict[str, Any]:
    client.json(
        "probe_project",
        ["gcloud", "projects", "describe", PROJECT, *_json_args()],
    )
    services = client.json(
        "probe_enabled_services",
        ["gcloud", "services", "list", "--enabled", *_project_args(), "--format=json(config.name)"],
    )
    if not isinstance(services, list):
        raise ConvergenceError("enabled service readback is invalid")
    enabled_services = {
        str(item.get("config", {}).get("name"))
        for item in services
        if isinstance(item, dict) and isinstance(item.get("config"), dict)
    }

    pool = client.json(
        "probe_wif_pool",
        [
            "gcloud",
            "iam",
            "workload-identity-pools",
            "describe",
            POOL,
            "--location",
            "global",
            *_project_args(),
            *_json_args(),
        ],
        allow_not_found=True,
    )
    if pool is None:
        pool = _find_resource(
            client.json(
                "probe_deleted_wif_pools",
                [
                    "gcloud",
                    "iam",
                    "workload-identity-pools",
                    "list",
                    "--location",
                    "global",
                    *_project_args(),
                    "--show-deleted",
                    *_json_args(),
                ],
            ),
            suffix=f"/workloadIdentityPools/{POOL}",
            label="WIF pool",
        )
    provider = client.json(
        "probe_wif_provider",
        [
            "gcloud",
            "iam",
            "workload-identity-pools",
            "providers",
            "describe",
            PROVIDER,
            "--location",
            "global",
            "--workload-identity-pool",
            POOL,
            *_project_args(),
            *_json_args(),
        ],
        allow_not_found=True,
    )
    if provider is None and pool is not None:
        deleted_providers = client.json(
            "probe_deleted_wif_providers",
            [
                "gcloud",
                "iam",
                "workload-identity-pools",
                "providers",
                "list",
                "--location",
                "global",
                "--workload-identity-pool",
                POOL,
                *_project_args(),
                "--show-deleted",
                *_json_args(),
            ],
            allow_not_found=True,
        )
        if deleted_providers is not None:
            provider = _find_resource(
                deleted_providers,
                suffix=f"/workloadIdentityPools/{POOL}/providers/{PROVIDER}",
                label="WIF provider",
            )

    accounts: dict[str, Any | None] = {}
    account_policies: dict[str, Any | None] = {}
    for _, email, _ in SERVICE_ACCOUNTS:
        account = client.json(
            f"probe_service_account:{email}",
            ["gcloud", "iam", "service-accounts", "describe", email, *_project_args(), *_json_args()],
            allow_not_found=True,
        )
        accounts[email] = account
        account_policies[email] = (
            client.json(
                f"probe_service_account_policy:{email}",
                ["gcloud", "iam", "service-accounts", "get-iam-policy", email, *_project_args(), *_json_args()],
            )
            if account is not None
            else None
        )

    custom_role = client.json(
        "probe_custom_role",
        ["gcloud", "iam", "roles", "describe", CUSTOM_ROLE_ID, *_project_args(), *_json_args()],
        allow_not_found=True,
    )
    project_policy = client.json(
        "probe_project_policy",
        ["gcloud", "projects", "get-iam-policy", PROJECT, *_json_args()],
    )
    vm_service_account_policy = client.json(
        "probe_vm_service_account_policy",
        ["gcloud", "iam", "service-accounts", "get-iam-policy", VM_SERVICE_ACCOUNT, *_project_args(), *_json_args()],
    )
    instance = client.json(
        "probe_instance",
        ["gcloud", "compute", "instances", "describe", INSTANCE, "--zone", ZONE, *_project_args(), *_json_args()],
    )
    instance_policy = client.json(
        "probe_instance_policy",
        [
            "gcloud",
            "compute",
            "instances",
            "get-iam-policy",
            INSTANCE,
            "--zone",
            ZONE,
            *_project_args(),
            *_json_args(),
        ],
    )
    iap_firewall = client.json(
        "probe_iap_firewall",
        ["gcloud", "compute", "firewall-rules", "describe", IAP_FIREWALL_RULE, *_project_args(), *_json_args()],
        allow_not_found=True,
    )
    public_firewall = client.json(
        "probe_public_ssh_firewall",
        ["gcloud", "compute", "firewall-rules", "describe", CHANGING_PUBLIC_RULE, *_project_args(), *_json_args()],
        allow_not_found=not require_public_rule,
    )
    if require_public_rule and (public_firewall is None or public_firewall.get("disabled") is True):
        raise ConvergenceError("the public SSH rollback rule is absent or already disabled")

    return {
        "enabled_services": enabled_services,
        "pool": pool,
        "provider": provider,
        "accounts": accounts,
        "account_policies": account_policies,
        "custom_role": custom_role,
        "project_policy": project_policy,
        "vm_service_account_policy": vm_service_account_policy,
        "instance": instance,
        "instance_policy": instance_policy,
        "iap_firewall": iap_firewall,
        "public_firewall": public_firewall,
    }


def _service_account_mutations(state: dict[str, Any]) -> list[Mutation]:
    mutations: list[Mutation] = []
    for account_id, email, display_name in SERVICE_ACCOUNTS:
        account = state["accounts"][email]
        if account is None:
            mutations.append(
                Mutation(
                    f"create_service_account:{email}",
                    "create",
                    [
                        "gcloud",
                        "iam",
                        "service-accounts",
                        "create",
                        account_id,
                        *_project_args(),
                        "--display-name",
                        display_name,
                        "--quiet",
                    ],
                )
            )
            continue
        if account.get("disabled") is True:
            raise ConvergenceError(f"service account is disabled: {email}")
        if account.get("email") not in (None, email):
            raise ConvergenceError(f"service account identity drifted: {email}")
        if account.get("displayName") != display_name:
            mutations.append(
                Mutation(
                    f"update_service_account:{email}",
                    "update",
                    [
                        "gcloud",
                        "iam",
                        "service-accounts",
                        "update",
                        email,
                        *_project_args(),
                        "--display-name",
                        display_name,
                        "--quiet",
                    ],
                )
            )
    return mutations


def _provider_mutations(state: dict[str, Any]) -> list[Mutation]:
    mutations: list[Mutation] = []
    pool = state["pool"]
    if pool is None:
        mutations.append(
            Mutation(
                "create_wif_pool",
                "create",
                [
                    "gcloud",
                    "iam",
                    "workload-identity-pools",
                    "create",
                    POOL,
                    "--location",
                    "global",
                    *_project_args(),
                    "--display-name",
                    POOL_DISPLAY_NAME,
                    "--quiet",
                ],
            )
        )
    elif pool.get("state") == "DELETED":
        mutations.append(
            Mutation(
                "undelete_wif_pool",
                "undelete",
                [
                    "gcloud",
                    "iam",
                    "workload-identity-pools",
                    "undelete",
                    POOL,
                    "--location",
                    "global",
                    *_project_args(),
                    "--quiet",
                ],
            )
        )
        return mutations
    elif pool.get("state") not in (None, "ACTIVE"):
        raise ConvergenceError("the shared GitHub Actions workload identity pool is not active")
    elif pool.get("disabled") is True:
        mutations.append(
            Mutation(
                "enable_wif_pool",
                "update",
                [
                    "gcloud",
                    "iam",
                    "workload-identity-pools",
                    "update",
                    POOL,
                    "--location",
                    "global",
                    *_project_args(),
                    "--no-disabled",
                    "--quiet",
                ],
            )
        )

    provider = state["provider"]
    if provider is None:
        mutations.append(
            Mutation(
                "create_wif_provider",
                "create",
                [
                    "gcloud",
                    "iam",
                    "workload-identity-pools",
                    "providers",
                    "create-oidc",
                    PROVIDER,
                    "--location",
                    "global",
                    "--workload-identity-pool",
                    POOL,
                    *_project_args(),
                    "--issuer-uri",
                    PROVIDER_ISSUER,
                    "--attribute-mapping",
                    ATTRIBUTE_MAPPING,
                    "--attribute-condition",
                    PROVIDER_CONDITION,
                    "--display-name",
                    PROVIDER_DISPLAY_NAME,
                    "--quiet",
                ],
            )
        )
        return mutations

    observed_mapping = provider.get("attributeMapping", {})
    security_fields_exact = (
        provider.get("oidc", {}).get("issuerUri") == PROVIDER_ISSUER
        and observed_mapping == EXPECTED_ATTRIBUTE_MAPPING
        and provider.get("attributeCondition") == PROVIDER_CONDITION
    )
    if provider.get("state") == "DELETED":
        if not security_fields_exact:
            raise ConvergenceError("the soft-deleted WIF provider has security-sensitive drift")
        mutations.append(
            Mutation(
                "undelete_wif_provider",
                "undelete",
                [
                    "gcloud",
                    "iam",
                    "workload-identity-pools",
                    "providers",
                    "undelete",
                    PROVIDER,
                    "--location",
                    "global",
                    "--workload-identity-pool",
                    POOL,
                    *_project_args(),
                    "--quiet",
                ],
            )
        )
        return mutations
    security_exact = provider.get("state") in (None, "ACTIVE") and security_fields_exact
    if not security_exact:
        raise ConvergenceError("the existing WIF provider has security-sensitive drift")
    if provider.get("disabled") is True or provider.get("displayName") != PROVIDER_DISPLAY_NAME:
        mutations.append(
            Mutation(
                "update_wif_provider",
                "update",
                [
                    "gcloud",
                    "iam",
                    "workload-identity-pools",
                    "providers",
                    "update-oidc",
                    PROVIDER,
                    "--location",
                    "global",
                    "--workload-identity-pool",
                    POOL,
                    *_project_args(),
                    "--issuer-uri",
                    PROVIDER_ISSUER,
                    "--attribute-mapping",
                    ATTRIBUTE_MAPPING,
                    "--attribute-condition",
                    PROVIDER_CONDITION,
                    "--display-name",
                    PROVIDER_DISPLAY_NAME,
                    "--no-disabled",
                    "--quiet",
                ],
            )
        )
    return mutations


def _custom_role_mutations(state: dict[str, Any]) -> list[Mutation]:
    role = state["custom_role"]
    create = [
        "gcloud",
        "iam",
        "roles",
        "create",
        CUSTOM_ROLE_ID,
        *_project_args(),
        "--title",
        CUSTOM_ROLE_TITLE,
        "--description",
        CUSTOM_ROLE_DESCRIPTION,
        "--permissions",
        ",".join(CUSTOM_PERMISSIONS),
        "--stage",
        "GA",
        "--quiet",
    ]
    if role is None:
        return [Mutation("create_custom_role", "create", create)]
    observed = set(role.get("includedPermissions", []))
    expected = set(CUSTOM_PERMISSIONS)
    if observed - expected:
        raise ConvergenceError("the custom discovery role has unexpected permissions")
    if role.get("stage") == "DISABLED":
        raise ConvergenceError("the custom discovery role is disabled")
    if (
        observed == expected
        and role.get("title") == CUSTOM_ROLE_TITLE
        and role.get("description") == CUSTOM_ROLE_DESCRIPTION
    ):
        return []
    update = create.copy()
    update[3] = "update"
    return [Mutation("update_custom_role", "update", update)]


def _binding_mutations(state: dict[str, Any]) -> list[Mutation]:
    mutations: list[Mutation] = []
    project_policy = state["project_policy"]
    vm_policy = state["vm_service_account_policy"]
    instance_policy = state["instance_policy"]
    _assert_exact_member_roles(
        project_policy,
        members=OPERATOR_MEMBERS,
        allowed_roles={CUSTOM_ROLE, IAP_ROLE},
        scope="project policy",
    )
    _assert_exact_member_roles(
        vm_policy,
        members=OPERATOR_MEMBERS,
        allowed_roles={"roles/iam.serviceAccountUser"},
        scope="VM service account policy",
    )
    expected_instance_roles = {
        OPERATOR_MEMBERS[0]: "roles/compute.osLogin",
        OPERATOR_MEMBERS[1]: "roles/compute.osAdminLogin",
    }
    _assert_exact_member_roles(
        instance_policy,
        members=OPERATOR_MEMBERS,
        allowed_roles=set(expected_instance_roles.values()),
        scope="instance policy",
    )

    for _, email, _ in SERVICE_ACCOUNTS:
        if state["accounts"][email] is None:
            policy: dict[str, Any] = {"bindings": []}
        else:
            policy = state["account_policies"][email]
        _assert_exact_member_roles(
            policy,
            members=(WIF_PRINCIPAL,),
            allowed_roles={"roles/iam.workloadIdentityUser"},
            scope=f"service account policy {email}",
        )
        binding = _ensure_binding(
            policy=policy,
            member=WIF_PRINCIPAL,
            role="roles/iam.workloadIdentityUser",
            condition=None,
            scope=f"service account policy {email}",
            mutation=Mutation(
                f"bind_wif:{email}",
                "ensure_binding",
                [
                    "gcloud",
                    "iam",
                    "service-accounts",
                    "add-iam-policy-binding",
                    email,
                    *_project_args(),
                    "--member",
                    WIF_PRINCIPAL,
                    "--role",
                    "roles/iam.workloadIdentityUser",
                    "--quiet",
                ],
            ),
        )
        if binding:
            mutations.append(binding)

    for member in OPERATOR_MEMBERS:
        custom = _ensure_binding(
            policy=project_policy,
            member=member,
            role=CUSTOM_ROLE,
            condition=None,
            scope="project policy",
            mutation=Mutation(
                f"bind_custom_discovery:{member}",
                "ensure_binding",
                [
                    "gcloud",
                    "projects",
                    "add-iam-policy-binding",
                    PROJECT,
                    "--member",
                    member,
                    "--role",
                    CUSTOM_ROLE,
                    "--quiet",
                ],
            ),
        )
        if custom:
            mutations.append(custom)
        iap = _ensure_binding(
            policy=project_policy,
            member=member,
            role=IAP_ROLE,
            condition=(IAP_CONDITION_TITLE, IAP_CONDITION),
            scope="project policy",
            mutation=Mutation(
                f"bind_iap_tunnel:{member}",
                "ensure_binding",
                [
                    "gcloud",
                    "projects",
                    "add-iam-policy-binding",
                    PROJECT,
                    "--member",
                    member,
                    "--role",
                    IAP_ROLE,
                    "--condition",
                    f"expression={IAP_CONDITION},title={IAP_CONDITION_TITLE}",
                    "--quiet",
                ],
            ),
        )
        if iap:
            mutations.append(iap)
        service_account_user = _ensure_binding(
            policy=vm_policy,
            member=member,
            role="roles/iam.serviceAccountUser",
            condition=None,
            scope="VM service account policy",
            mutation=Mutation(
                f"bind_vm_service_account_user:{member}",
                "ensure_binding",
                [
                    "gcloud",
                    "iam",
                    "service-accounts",
                    "add-iam-policy-binding",
                    VM_SERVICE_ACCOUNT,
                    *_project_args(),
                    "--member",
                    member,
                    "--role",
                    "roles/iam.serviceAccountUser",
                    "--quiet",
                ],
            ),
        )
        if service_account_user:
            mutations.append(service_account_user)

    for member, role in expected_instance_roles.items():
        os_login = _ensure_binding(
            policy=instance_policy,
            member=member,
            role=role,
            condition=None,
            scope="instance policy",
            mutation=Mutation(
                f"bind_instance_os_login:{member}",
                "ensure_binding",
                [
                    "gcloud",
                    "compute",
                    "instances",
                    "add-iam-policy-binding",
                    INSTANCE,
                    "--zone",
                    ZONE,
                    *_project_args(),
                    "--member",
                    member,
                    "--role",
                    role,
                    "--quiet",
                ],
            ),
        )
        if os_login:
            mutations.append(os_login)
    return mutations


def _instance_and_firewall_mutations(state: dict[str, Any]) -> list[Mutation]:
    mutations: list[Mutation] = []
    instance = state["instance"]
    if instance.get("status") != "RUNNING":
        raise ConvergenceError("the target instance is not running")
    interfaces = instance.get("networkInterfaces", [])
    exact_interfaces = [
        item
        for item in interfaces
        if isinstance(item, dict)
        and item.get("networkIP") == INSTANCE_INTERNAL_IP
        and str(item.get("network", "")).endswith(f"/networks/{NETWORK}")
    ]
    if len(exact_interfaces) != 1:
        raise ConvergenceError("the target instance network or internal IP drifted")
    service_accounts = {item.get("email") for item in instance.get("serviceAccounts", []) if isinstance(item, dict)}
    if VM_SERVICE_ACCOUNT not in service_accounts:
        raise ConvergenceError("the target instance service account drifted")
    metadata = _metadata(instance)
    if metadata.get("enable-oslogin", "").upper() != "TRUE":
        raise ConvergenceError(
            "instance OS Login is not already enabled; refusing to change the SSH authentication mode in bootstrap"
        )
    tags = set(instance.get("tags", {}).get("items", []))
    if TARGET_TAG not in tags:
        public_target_tags = set((state.get("public_firewall") or {}).get("targetTags", []))
        if TARGET_TAG in public_target_tags:
            raise ConvergenceError(
                "adding the IAP target tag would newly match the existing public SSH firewall rule"
            )
        mutations.append(
            Mutation(
                "add_instance_iap_target_tag",
                "update",
                [
                    "gcloud",
                    "compute",
                    "instances",
                    "add-tags",
                    INSTANCE,
                    "--zone",
                    ZONE,
                    *_project_args(),
                    "--tags",
                    TARGET_TAG,
                    "--quiet",
                ],
            )
        )

    firewall = state["iap_firewall"]
    if firewall is None:
        mutations.append(
            Mutation(
                "create_iap_firewall",
                "create",
                [
                    "gcloud",
                    "compute",
                    "firewall-rules",
                    "create",
                    IAP_FIREWALL_RULE,
                    *_project_args(),
                    "--network",
                    NETWORK,
                    "--direction",
                    "INGRESS",
                    "--action",
                    "ALLOW",
                    "--rules",
                    "tcp:22",
                    "--source-ranges",
                    IAP_SOURCE_RANGE,
                    "--target-tags",
                    TARGET_TAG,
                    "--enable-logging",
                    "--quiet",
                ],
            )
        )
    else:
        if not _firewall_is_structurally_exact(firewall):
            raise ConvergenceError("the existing IAP firewall has security-sensitive drift")
        target_tags = set(firewall.get("targetTags", []))
        if target_tags not in ({TARGET_TAG}, set()):
            raise ConvergenceError("the existing IAP firewall target tags broaden access")
        if (
            target_tags != {TARGET_TAG}
            or firewall.get("disabled") is True
            or not firewall.get("logConfig", {}).get("enable")
        ):
            mutations.append(
                Mutation(
                    "update_iap_firewall",
                    "update",
                    [
                        "gcloud",
                        "compute",
                        "firewall-rules",
                        "update",
                        IAP_FIREWALL_RULE,
                        *_project_args(),
                        "--source-ranges",
                        IAP_SOURCE_RANGE,
                        "--target-tags",
                        TARGET_TAG,
                        "--allow",
                        "tcp:22",
                        "--enable-logging",
                        "--no-disabled",
                        "--quiet",
                    ],
                )
            )
    return mutations


def plan_mutations(state: dict[str, Any]) -> list[Mutation]:
    mutations: list[Mutation] = []
    missing_services = sorted(set(REQUIRED_APIS) - state["enabled_services"])
    if missing_services:
        mutations.append(
            Mutation(
                "enable_required_apis",
                "update",
                ["gcloud", "services", "enable", *missing_services, *_project_args(), "--quiet"],
            )
        )
    mutations.extend(_provider_mutations(state))
    mutations.extend(_service_account_mutations(state))
    mutations.extend(_custom_role_mutations(state))
    mutations.extend(_binding_mutations(state))
    mutations.extend(_instance_and_firewall_mutations(state))
    return mutations


def _run_mutations(client: Gcloud, mutations: list[Mutation]) -> None:
    for mutation in mutations:
        client.run(mutation.name, mutation.action, mutation.command)


def _dispatcher_probe_command(ssh_key: str) -> list[str]:
    remote = (
        f"if sudo test -f {REMOTE_DISPATCHER}; then "
        f"sudo stat -c 'FILE=%U:%G:%a' {REMOTE_DISPATCHER}; "
        f"sudo sha256sum {REMOTE_DISPATCHER} | awk '{{print \"SHA=\"$1}}'; "
        "else echo FILE=MISSING; fi; "
        f"if sudo test -d {REMOTE_RUN_ROOT}; then sudo stat -c 'RUN=%U:%G:%a' {REMOTE_RUN_ROOT}; "
        "else echo RUN=MISSING; fi"
    )
    return [
        "gcloud",
        "compute",
        "ssh",
        INSTANCE,
        "--zone",
        ZONE,
        *_project_args(),
        "--tunnel-through-iap",
        "--ssh-key-file",
        ssh_key,
        "--ssh-key-expire-after=5m",
        "--quiet",
        "--command",
        remote,
    ]


def _parse_dispatcher_probe(stdout: str) -> dict[str, str]:
    values: dict[str, str] = {}
    for line in stdout.splitlines():
        key, separator, value = line.partition("=")
        if separator and key in {"FILE", "SHA", "RUN"}:
            values[key] = value.strip()
    if "FILE" not in values or "RUN" not in values:
        raise ConvergenceError("remote dispatcher probe returned an invalid readback")
    return values


def converge_dispatcher(client: Gcloud, dispatcher_path: Path) -> dict[str, Any]:
    if not dispatcher_path.is_file() or dispatcher_path.is_symlink():
        raise ConvergenceError("the reviewed local dispatcher is missing or unsafe")
    dispatcher_bytes = dispatcher_path.read_bytes()
    marker = f'readonly DISPATCHER_VERSION="{DISPATCHER_VERSION}"'.encode()
    if marker not in dispatcher_bytes:
        raise ConvergenceError("the reviewed local dispatcher version marker drifted")
    expected_hash = hashlib.sha256(dispatcher_bytes).hexdigest()

    with tempfile.TemporaryDirectory(prefix="teleo-iap-bootstrap-") as temporary:
        ssh_key = str(Path(temporary) / "ssh_key")
        recorded_key = "<temporary-ssh-key>"
        probe = client.run(
            "probe_remote_dispatcher",
            "probe",
            _dispatcher_probe_command(ssh_key),
            recorded_command=_dispatcher_probe_command(recorded_key),
        )
        assert probe is not None
        observed = _parse_dispatcher_probe(probe.stdout)
        exact = observed == {"FILE": "root:root:755", "SHA": expected_hash, "RUN": "root:root:700"}
        if not exact:
            remote_temporary = f"/tmp/teleo-gcp-iap-operator-{expected_hash[:16]}"
            scp = [
                "gcloud",
                "compute",
                "scp",
                str(dispatcher_path),
                f"{INSTANCE}:{remote_temporary}",
                "--zone",
                ZONE,
                *_project_args(),
                "--tunnel-through-iap",
                "--ssh-key-file",
                ssh_key,
                "--ssh-key-expire-after=5m",
                "--quiet",
            ]
            recorded_scp = scp.copy()
            recorded_scp[recorded_scp.index(str(dispatcher_path))] = "scripts/gcp_iap_operator.sh"
            recorded_scp[recorded_scp.index(ssh_key)] = recorded_key
            client.run("upload_remote_dispatcher", "update", scp, recorded_command=recorded_scp)
            remote_install = (
                f"test \"$(sha256sum {remote_temporary} | awk '{{print $1}}')\" = {expected_hash} && "
                f"grep -F '{marker.decode()}' {remote_temporary} >/dev/null && "
                f"sudo install -o root -g root -m 0755 {remote_temporary} {REMOTE_DISPATCHER} && "
                f"sudo install -d -o root -g root -m 0700 {REMOTE_RUN_ROOT} && "
                f"rm -f {remote_temporary}"
            )
            install = [
                "gcloud",
                "compute",
                "ssh",
                INSTANCE,
                "--zone",
                ZONE,
                *_project_args(),
                "--tunnel-through-iap",
                "--ssh-key-file",
                ssh_key,
                "--ssh-key-expire-after=5m",
                "--quiet",
                "--command",
                remote_install,
            ]
            recorded_install = install.copy()
            recorded_install[recorded_install.index(ssh_key)] = recorded_key
            client.run("install_remote_dispatcher", "update", install, recorded_command=recorded_install)

        verify = client.run(
            "verify_remote_dispatcher",
            "verify",
            _dispatcher_probe_command(ssh_key),
            recorded_command=_dispatcher_probe_command(recorded_key),
        )
        assert verify is not None
        verified = _parse_dispatcher_probe(verify.stdout)
        if verified != {"FILE": "root:root:755", "SHA": expected_hash, "RUN": "root:root:700"}:
            raise ConvergenceError("remote dispatcher verification did not converge")
    return {"sha256": expected_hash, "owner_group_mode": verified["FILE"], "run_root_owner_group_mode": verified["RUN"]}


def _dry_run_mutations() -> list[Mutation]:
    empty_policy = {"bindings": []}
    empty_state: dict[str, Any] = {
        "enabled_services": set(),
        "pool": None,
        "provider": None,
        "accounts": {email: None for _, email, _ in SERVICE_ACCOUNTS},
        "account_policies": {email: None for _, email, _ in SERVICE_ACCOUNTS},
        "custom_role": None,
        "project_policy": empty_policy,
        "vm_service_account_policy": empty_policy,
        "instance": {
            "status": "RUNNING",
            "networkInterfaces": [
                {
                    "networkIP": INSTANCE_INTERNAL_IP,
                    "network": f"projects/{PROJECT}/global/networks/{NETWORK}",
                }
            ],
            "serviceAccounts": [{"email": VM_SERVICE_ACCOUNT}],
            "metadata": {"items": [{"key": "enable-oslogin", "value": "TRUE"}]},
            "tags": {"items": []},
        },
        "instance_policy": empty_policy,
        "iap_firewall": None,
        "public_firewall": {
            "name": CHANGING_PUBLIC_RULE,
            "disabled": False,
            "targetTags": ["teleo-public-ssh-bootstrap"],
        },
    }
    return plan_mutations(empty_state)


def _receipt_base(*, execute: bool) -> dict[str, Any]:
    return {
        "schema": "livingip.gcpIapOperatorBootstrapReceipt.v1",
        "generated_at_utc": datetime.now(UTC).isoformat(),
        "mode": "execute" if execute else "dry_run",
        "project": PROJECT,
        "github": {
            "repository": GITHUB_REPOSITORY,
            "ref": "refs/heads/main",
            "workflow_ref": WORKFLOW_REF,
        },
        "public_ssh_rule": {
            "name": CHANGING_PUBLIC_RULE,
            "mutated": False,
            "disable_command_present": False,
        },
        "workflow_dispatched": False,
        "credentials_logged": False,
        "least_privilege": {
            "status_service_account": STATUS_SERVICE_ACCOUNT,
            "clone_service_account": CLONE_SERVICE_ACCOUNT,
            "custom_role_permissions": sorted(CUSTOM_PERMISSIONS),
            "forbidden_operator_roles": sorted(FORBIDDEN_OPERATOR_ROLES),
        },
    }


def apply_operator_access(
    *,
    execute: bool,
    output: Path,
    runner: Runner | None = None,
    dispatcher_path: Path = DEFAULT_DISPATCHER,
) -> tuple[dict[str, Any], int]:
    receipt = _receipt_base(execute=execute)
    client = Gcloud(runner)
    exitcode = 0
    try:
        if not execute:
            mutations = _dry_run_mutations()
            receipt.update(
                {
                    "status": "planned",
                    "operations": [
                        asdict(Operation(item.name, item.action, _redact_command(item.command), "planned"))
                        for item in mutations
                    ],
                    "exact_next_action": f"python ops/{Path(__file__).name} --execute --output {output}",
                    "claim_ceiling": (
                        "Dry-run plan only: no GCP state was read or changed, no workflow was dispatched, "
                        "and the public SSH firewall rule was not touched."
                    ),
                    "dispatcher_plan": {
                        "source": "scripts/gcp_iap_operator.sh",
                        "action": "probe_then_install_if_hash_or_ownership_drifted",
                        "transport": "gcloud_compute_ssh_and_scp_through_iap",
                    },
                }
            )
        else:
            before = discover(client)
            current = before
            for _ in range(6):
                mutations = plan_mutations(current)
                if not mutations:
                    break
                _run_mutations(client, mutations)
                current = discover(client)
            after = current
            remaining = plan_mutations(after)
            if remaining:
                raise ConvergenceError(
                    f"control-plane convergence exceeded six mutation rounds; {len(remaining)} mutation(s) remain"
                )
            dispatcher = converge_dispatcher(client, dispatcher_path)
            receipt.update(
                {
                    "status": "applied",
                    "operations": [asdict(operation) for operation in client.operations],
                    "mutation_count": len(
                        [operation for operation in client.operations if operation.action not in {"probe", "verify"}]
                    ),
                    "dispatcher": dispatcher,
                    "public_ssh_rule": {
                        "name": CHANGING_PUBLIC_RULE,
                        "observed_disabled_before": bool(before["public_firewall"].get("disabled", False)),
                        "observed_disabled_after": bool(after["public_firewall"].get("disabled", False)),
                        "target_tags_before": sorted(before["public_firewall"].get("targetTags", [])),
                        "target_tags_after": sorted(after["public_firewall"].get("targetTags", [])),
                        "effective_target_tag_expansion": False,
                        "mutated": False,
                        "disable_command_present": False,
                    },
                    "exact_next_action": (
                        "Dispatch the main-branch gcp-iap-operator status operation, inspect its sanitized artifact, "
                        "and keep teleo-prod-allow-ssh-current-ip enabled until that independent canary passes."
                    ),
                    "claim_ceiling": (
                        "The GCP OIDC/WIF, least-privilege IAP identities, target IAM, IAP firewall, OS Login, "
                        "and reviewed dispatcher converged and were read back. No GitHub workflow was dispatched, "
                        "operator behavior was not live-tested, and the public SSH rule remains enabled."
                    ),
                }
            )
    except ConvergenceError as exc:
        exitcode = 1
        receipt.update(
            {
                "status": "failed",
                "error": str(exc),
                "operations": [asdict(operation) for operation in client.operations],
                "exact_next_action": (
                    f"Resolve the reported fail-closed drift or authentication gate, then rerun: "
                    f"python ops/{Path(__file__).name} --execute --output {output}"
                ),
                "claim_ceiling": (
                    "Bootstrap did not prove convergence. Some idempotent operations may have completed before "
                    "the failure; rerun is required. No workflow was dispatched and the public SSH rule was not mutated."
                ),
            }
        )
    output.parent.mkdir(parents=True, exist_ok=True)
    output.write_text(json.dumps(receipt, indent=2, sort_keys=True) + "\n", encoding="utf-8")
    return receipt, exitcode


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--execute", action="store_true", help="converge GCP state; default is a no-network dry run")
    parser.add_argument("--output", default=DEFAULT_OUTPUT, help="sanitized JSON receipt path")
    parser.add_argument("--dispatcher", type=Path, default=DEFAULT_DISPATCHER, help="reviewed dispatcher source")
    args = parser.parse_args()

    receipt, exitcode = apply_operator_access(
        execute=args.execute,
        output=Path(args.output),
        dispatcher_path=args.dispatcher,
    )
    print(json.dumps({"receipt": str(args.output), "status": receipt["status"]}, sort_keys=True))
    return exitcode


if __name__ == "__main__":
    raise SystemExit(main())