1333 lines
47 KiB
Python
1333 lines
47 KiB
Python
#!/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())
|