From ff5fb56a366f7e6c5bf0f27b1667924c8f869714 Mon Sep 17 00:00:00 2001 From: twentyOne2x Date: Mon, 13 Jul 2026 04:58:16 +0200 Subject: [PATCH] Harden GCP replay and durable IAP bootstrap --- .github/workflows/gcp-iap-operator.yml | 4 +- ops/apply_gcp_iap_operator_access.py | 1230 +++++++++++++++++++ ops/plan_gcp_iap_operator_access.py | 17 +- scripts/gcp_iap_operator.sh | 272 +++- tests/test_apply_gcp_iap_operator_access.py | 409 ++++++ tests/test_gcp_iap_operator_access.py | 114 +- 6 files changed, 2028 insertions(+), 18 deletions(-) create mode 100644 ops/apply_gcp_iap_operator_access.py create mode 100644 tests/test_apply_gcp_iap_operator_access.py diff --git a/.github/workflows/gcp-iap-operator.yml b/.github/workflows/gcp-iap-operator.yml index 157063b..f72fa77 100644 --- a/.github/workflows/gcp-iap-operator.yml +++ b/.github/workflows/gcp-iap-operator.yml @@ -240,11 +240,11 @@ jobs: path.chmod(0o600) PY - - name: Upload sanitized result + - name: Upload verified result receipts if: always() uses: actions/upload-artifact@v4 with: name: gcp-iap-operator-${{ github.run_id }} - path: ${{ env.RESULT_DIR }}/result.json + path: ${{ env.RESULT_DIR }}/ if-no-files-found: error retention-days: 7 diff --git a/ops/apply_gcp_iap_operator_access.py b/ops/apply_gcp_iap_operator_access.py new file mode 100644 index 0000000..61a8fc2 --- /dev/null +++ b/ops/apply_gcp_iap_operator_access.py @@ -0,0 +1,1230 @@ +#!/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("") + 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 ("", "") + 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 _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, + ) + 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, + ) + + 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") 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_exact = ( + provider.get("state") in (None, "ACTIVE") + and provider.get("oidc", {}).get("issuerUri") == PROVIDER_ISSUER + and observed_mapping == EXPECTED_ATTRIBUTE_MAPPING + and provider.get("attributeCondition") == PROVIDER_CONDITION + ) + 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": + mutations.append( + Mutation( + "enable_instance_os_login", + "update", + [ + "gcloud", + "compute", + "instances", + "add-metadata", + INSTANCE, + "--zone", + ZONE, + *_project_args(), + "--metadata", + "enable-oslogin=TRUE", + "--quiet", + ], + ) + ) + tags = set(instance.get("tags", {}).get("items", [])) + if TARGET_TAG not in tags: + 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 = "" + 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": []}, + "tags": {"items": []}, + }, + "instance_policy": empty_policy, + "iap_firewall": None, + "public_firewall": {"name": CHANGING_PUBLIC_RULE, "disabled": False}, + } + 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) + mutations = plan_mutations(before) + _run_mutations(client, mutations) + after = discover(client) + remaining = plan_mutations(after) + if remaining: + raise ConvergenceError(f"control-plane verification still requires {len(remaining)} mutation(s)") + 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)), + "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()) diff --git a/ops/plan_gcp_iap_operator_access.py b/ops/plan_gcp_iap_operator_access.py index d9dae94..715176b 100755 --- a/ops/plan_gcp_iap_operator_access.py +++ b/ops/plan_gcp_iap_operator_access.py @@ -134,7 +134,7 @@ def bootstrap_commands() -> list[str]: "--attribute-condition", PROVIDER_CONDITION, "--display-name", - "Teleo fixed IAP operator workflow", + "Teleo fixed IAP operator", ] ), *[ @@ -288,6 +288,21 @@ def bootstrap_commands() -> list[str]: "enable-oslogin=TRUE", ] ), + shell_join( + [ + "gcloud", + "compute", + "instances", + "add-tags", + INSTANCE, + "--zone", + ZONE, + "--project", + PROJECT, + "--tags", + TARGET_TAG, + ] + ), shell_join( [ "gcloud", diff --git a/scripts/gcp_iap_operator.sh b/scripts/gcp_iap_operator.sh index a0eafba..28d7a41 100755 --- a/scripts/gcp_iap_operator.sh +++ b/scripts/gcp_iap_operator.sh @@ -15,6 +15,10 @@ readonly REMOTE_RUN_ROOT="/var/lib/teleo-gcp-iap-operator/runs" readonly LIVE_PROFILE="/home/teleo/.hermes/profiles/leoclean" readonly LIVE_SERVICE="leoclean-gcp-prod-parallel.service" readonly STATUS_TARGET="teleo_clone_status" +readonly CORY_REPLAY_DB="teleo_clone_cory_20260712t1940z" +readonly CORY_REPLAY_FINGERPRINT="48ea5332c4f303dea02d503447f1ac8e1d8aa00e60d198a566d628b37dbdac1c" +readonly CORY_REPLAY_LEGACY_DIR="/home/teleo/gcp-cory-model-replay-20260712t1940z" +readonly DISABLED_ROLLBACK_DB="teleo_canonical_pre_20260712t1905z" readonly REQUEST_ID_RE='^iap-[a-z0-9]{12,32}$' readonly TARGET_DB_RE='^teleo_clone_[a-z0-9][a-z0-9_]{0,50}$' @@ -75,9 +79,24 @@ for pair in pairs: key, separator, value = pair.partition("=") if not separator: raise SystemExit("invalid result field") - if key in {"nrestarts", "result_bytes"}: + if key in { + "nrestarts", + "result_bytes", + "clone_database_remaining", + "run_directory_remaining", + "legacy_directory_remaining", + "export_file_remaining", + "leftover_temp_directory_count", + "rollback_database_connections", + }: payload[key] = int(value) - elif key in {"no_message_send", "live_service_unchanged", "live_profile_unchanged"}: + elif key in { + "no_message_send", + "live_service_unchanged", + "live_profile_unchanged", + "full_result_export_ready", + "rollback_database_datallowconn", + }: payload[key] = value == "true" else: payload[key] = value @@ -295,6 +314,7 @@ remote_direct_claim_replay() { local request_id="$1" local target_db="$2" local receipt_relative receipt_path run_dir output_path stdout_path stderr_path report_sha result_bytes + local caller_home caller_group export_path fixed_baseline_validation receive_and_validate_bundle "direct-claim-replay" "$request_id" "$target_db" receipt_relative="$(parity_receipt_relative "$target_db")" @@ -330,13 +350,15 @@ remote_direct_claim_replay() { fi [[ -s "$output_path" && ! -L "$output_path" ]] || die "six-turn replay did not retain its private receipt" - python3 - "$output_path" "$target_db" <<'PY' + python3 - "$output_path" "$target_db" "$CORY_REPLAY_DB" "$CORY_REPLAY_FINGERPRINT" <<'PY' import json import sys from pathlib import Path payload = json.loads(Path(sys.argv[1]).read_text(encoding="utf-8")) target_db = sys.argv[2] +expected_db = sys.argv[3] +expected_fingerprint = sys.argv[4] checks = payload.get("checks") or {} required = { "status": payload.get("status") == "pass", @@ -348,6 +370,28 @@ required = { "database_unchanged": checks.get("generated_database_unchanged") is True, "temporary_profile_removed": checks.get("temporary_profile_removed") is True, } +if target_db == expected_db: + expected_counts = { + "claims": 1837, + "sources": 4145, + "claim_edges": 4916, + "claim_evidence": 4670, + "kb_proposals": 26, + } + before_fingerprint = payload.get("database_fingerprint_before") or {} + after_fingerprint = payload.get("database_fingerprint_after") or {} + before_counts = ((payload.get("canonical_status_before") or {}).get("high_signal_rows") or {}) + after_counts = ((payload.get("canonical_status_after") or {}).get("high_signal_rows") or {}) + required.update( + { + "fixed_fingerprint_before": before_fingerprint.get("sha256") == expected_fingerprint, + "fixed_fingerprint_after": after_fingerprint.get("sha256") == expected_fingerprint, + "fixed_counts_before": all(before_counts.get(key) == value for key, value in expected_counts.items()), + "fixed_counts_after": all(after_counts.get(key) == value for key, value in expected_counts.items()), + "strict_six_of_six": (payload.get("score") or {}).get("passes") == 6 + and (payload.get("score") or {}).get("expected_prompt_count") == 6, + } + ) failed = sorted(name for name, ok in required.items() if not ok) if failed: raise SystemExit("replay receipt failed fixed checks: " + ", ".join(failed)) @@ -355,12 +399,30 @@ PY chmod 0600 "$output_path" report_sha="$(sha256sum "$output_path" | awk '{print $1}')" result_bytes="$(wc -c <"$output_path" | tr -d ' ')" + caller_home="$(getent passwd "$SUDO_USER" | cut -d: -f6)" + caller_group="$(id -gn "$SUDO_USER")" + [[ "$caller_home" == /* && -d "$caller_home" && ! -L "$caller_home" ]] || { + die "OS Login caller home is unsafe for replay receipt export" + } + [[ "$(realpath "$caller_home")" == "$caller_home" ]] || die "OS Login caller home path drifted" + export_path="$caller_home/.teleo-iap-result-$request_id.json" + [[ ! -e "$export_path" ]] || die "request-derived replay receipt export already exists" + install -o "$SUDO_USER" -g "$caller_group" -m 0600 "$output_path" "$export_path" + [[ "$(sha256sum "$export_path" | awk '{print $1}')" == "$report_sha" ]] || { + die "exported replay receipt hash mismatch" + } + fixed_baseline_validation="not_applicable" + if [[ "$target_db" == "$CORY_REPLAY_DB" ]]; then + fixed_baseline_validation="pass" + fi emit_json \ "direct-claim-replay" "$request_id" "$target_db" "pass" \ "bundle_commit=$VALIDATED_BUNDLE_COMMIT" \ "bundle_sha256=$VALIDATED_BUNDLE_SHA256" \ "result_sha256=$report_sha" \ "result_bytes=$result_bytes" \ + "full_result_export_ready=true" \ + "fixed_baseline_validation=$fixed_baseline_validation" \ "no_message_send=true" \ "live_service_unchanged=true" \ "live_profile_unchanged=true" @@ -369,11 +431,33 @@ PY remote_cleanup_clone() { local request_id="$1" local target_db="$2" - local run_dir password exists remaining + local run_dir password exists remaining legacy_dir legacy_resolved + local caller_home export_path export_state + local rollback_exists rollback_allowconn rollback_connections leftover_temp_count receive_and_validate_bundle "cleanup-clone" "$request_id" "$target_db" run_dir="$REMOTE_RUN_ROOT/$request_id" validate_run_dir "$run_dir" "$request_id" "$target_db" + legacy_dir="" + if [[ "$target_db" == "$CORY_REPLAY_DB" ]]; then + legacy_dir="$CORY_REPLAY_LEGACY_DIR" + if [[ -e "$legacy_dir" ]]; then + [[ -d "$legacy_dir" && ! -L "$legacy_dir" ]] || die "fixed legacy replay path is unsafe" + legacy_resolved="$(realpath "$legacy_dir")" + [[ "$legacy_resolved" == "$CORY_REPLAY_LEGACY_DIR" ]] || die "fixed legacy replay path drifted" + fi + fi + caller_home="$(getent passwd "$SUDO_USER" | cut -d: -f6)" + [[ "$caller_home" == /* && -d "$caller_home" && ! -L "$caller_home" ]] || { + die "OS Login caller home is unsafe for cleanup" + } + [[ "$(realpath "$caller_home")" == "$caller_home" ]] || die "OS Login caller home path drifted" + export_path="$caller_home/.teleo-iap-result-$request_id.json" + if [[ -e "$export_path" ]]; then + [[ -f "$export_path" && ! -L "$export_path" ]] || die "request-derived replay export is unsafe" + export_state="$(stat -c '%U:%a' "$export_path")" + [[ "$export_state" == "$SUDO_USER:600" ]] || die "request-derived replay export ownership or mode drifted" + fi command -v gcloud >/dev/null 2>&1 || die "gcloud is unavailable for secret-backed clone cleanup" command -v psql >/dev/null 2>&1 || die "psql is unavailable for clone cleanup" @@ -387,6 +471,24 @@ remote_cleanup_clone() { -c "select exists(select 1 from pg_database where datname = :'target_db');" )" [[ "$exists" == "t" ]] || die "marked clone database is absent; refusing directory-only cleanup" + rollback_exists="$( + psql "host=$CLOUDSQL_HOST port=5432 dbname=postgres user=postgres sslmode=require connect_timeout=8" \ + -X -Atq -v ON_ERROR_STOP=1 -v "rollback_db=$DISABLED_ROLLBACK_DB" \ + -c "select count(*) from pg_database where datname = :'rollback_db';" + )" + rollback_allowconn="$( + psql "host=$CLOUDSQL_HOST port=5432 dbname=postgres user=postgres sslmode=require connect_timeout=8" \ + -X -Atq -v ON_ERROR_STOP=1 -v "rollback_db=$DISABLED_ROLLBACK_DB" \ + -c "select datallowconn from pg_database where datname = :'rollback_db';" + )" + rollback_connections="$( + psql "host=$CLOUDSQL_HOST port=5432 dbname=postgres user=postgres sslmode=require connect_timeout=8" \ + -X -Atq -v ON_ERROR_STOP=1 -v "rollback_db=$DISABLED_ROLLBACK_DB" \ + -c "select count(*) from pg_stat_activity where datname = :'rollback_db';" + )" + [[ "$rollback_exists" == "1" ]] || die "disabled rollback database is absent" + [[ "$rollback_allowconn" == "f" ]] || die "rollback database unexpectedly allows connections" + [[ "$rollback_connections" == "0" ]] || die "rollback database has active connections" psql "host=$CLOUDSQL_HOST port=5432 dbname=postgres user=postgres sslmode=require connect_timeout=8" \ -X -q -v ON_ERROR_STOP=1 -v "target_db=$target_db" <<'SQL' @@ -407,11 +509,34 @@ SQL validate_run_dir "$run_dir" "$request_id" "$target_db" rm -rf --one-file-system -- "$run_dir" [[ ! -e "$run_dir" ]] || die "exact run-owned directory remained after cleanup" + if [[ -n "$legacy_dir" && -e "$legacy_dir" ]]; then + [[ -d "$legacy_dir" && ! -L "$legacy_dir" ]] || die "fixed legacy replay path became unsafe" + [[ "$(realpath "$legacy_dir")" == "$CORY_REPLAY_LEGACY_DIR" ]] || { + die "fixed legacy replay path changed before cleanup" + } + rm -rf --one-file-system -- "$legacy_dir" + fi + [[ -z "$legacy_dir" || ! -e "$legacy_dir" ]] || die "fixed legacy replay directory remained" + if [[ -e "$export_path" ]]; then + rm -f -- "$export_path" + fi + [[ ! -e "$export_path" ]] || die "request-derived replay export remained" + leftover_temp_count="$( + find /tmp -mindepth 1 -maxdepth 1 -type d \ + -name "leo-clone-checkpoint-$request_id-gcp-direct-claim-*" -print | wc -l | tr -d ' ' + )" + [[ "$leftover_temp_count" == "0" ]] || die "request-derived temporary replay directories remained" emit_json \ "cleanup-clone" "$request_id" "$target_db" "pass" \ "bundle_commit=$VALIDATED_BUNDLE_COMMIT" \ "bundle_sha256=$VALIDATED_BUNDLE_SHA256" \ - "clone_database_remaining=0" + "clone_database_remaining=0" \ + "run_directory_remaining=0" \ + "legacy_directory_remaining=0" \ + "export_file_remaining=0" \ + "leftover_temp_directory_count=$leftover_temp_count" \ + "rollback_database_datallowconn=false" \ + "rollback_database_connections=$rollback_connections" } remote_main() { @@ -463,10 +588,18 @@ if exit_code == 0: "bundle_sha256", "result_sha256", "result_bytes", + "full_result_export_ready", + "fixed_baseline_validation", "no_message_send", "live_service_unchanged", "live_profile_unchanged", "clone_database_remaining", + "run_directory_remaining", + "legacy_directory_remaining", + "export_file_remaining", + "leftover_temp_directory_count", + "rollback_database_datallowconn", + "rollback_database_connections", } remote = {key: value for key, value in candidate.items() if key in allowed} except (UnicodeDecodeError, json.JSONDecodeError, IndexError): @@ -507,13 +640,91 @@ validate_local_bundle() { printf '%s\n' "$path" } +validate_downloaded_replay() { + local result_path="$1" + local full_result="$2" + local target_db="$3" + RESULT_PATH="$result_path" FULL_RESULT="$full_result" TARGET_DB="$target_db" \ + EXPECTED_CORY_REPLAY_DB="$CORY_REPLAY_DB" \ + EXPECTED_CORY_REPLAY_FINGERPRINT="$CORY_REPLAY_FINGERPRINT" \ + python3 - <<'PY' +import hashlib +import json +import os +from pathlib import Path + +result_path = Path(os.environ["RESULT_PATH"]) +full_path = Path(os.environ["FULL_RESULT"]) +result = json.loads(result_path.read_text(encoding="utf-8")) +full_bytes = full_path.read_bytes() +full = json.loads(full_bytes) +remote = result.get("remote_result") or {} +checks = full.get("checks") or {} +required = { + "result_hash": hashlib.sha256(full_bytes).hexdigest() == remote.get("result_sha256"), + "result_size": len(full_bytes) == remote.get("result_bytes"), + "status": full.get("status") == "pass", + "target": full.get("target_database") == os.environ["TARGET_DB"], + "strict_six_of_six": (full.get("score") or {}).get("passes") == 6 + and (full.get("score") or {}).get("expected_prompt_count") == 6, + "all_checks": bool(checks) and all(checks.values()), +} +if os.environ["TARGET_DB"] == os.environ["EXPECTED_CORY_REPLAY_DB"]: + expected_counts = { + "claims": 1837, + "sources": 4145, + "claim_edges": 4916, + "claim_evidence": 4670, + "kb_proposals": 26, + } + before_counts = ((full.get("canonical_status_before") or {}).get("high_signal_rows") or {}) + after_counts = ((full.get("canonical_status_after") or {}).get("high_signal_rows") or {}) + required.update( + { + "fixed_fingerprint_before": (full.get("database_fingerprint_before") or {}).get("sha256") + == os.environ["EXPECTED_CORY_REPLAY_FINGERPRINT"], + "fixed_fingerprint_after": (full.get("database_fingerprint_after") or {}).get("sha256") + == os.environ["EXPECTED_CORY_REPLAY_FINGERPRINT"], + "fixed_counts_before": all(before_counts.get(key) == value for key, value in expected_counts.items()), + "fixed_counts_after": all(after_counts.get(key) == value for key, value in expected_counts.items()), + } + ) +failed = sorted(name for name, passed in required.items() if not passed) +if failed: + raise SystemExit("downloaded replay receipt failed: " + ", ".join(failed)) +result["full_result_exported"] = True +result["full_result_sha256"] = hashlib.sha256(full_bytes).hexdigest() +result["full_result_bytes"] = len(full_bytes) +result["full_result_fixed_checks"] = required +result_path.write_text(json.dumps(result, indent=2, sort_keys=True) + "\n", encoding="utf-8") +PY +} + +mark_result_export_failure() { + local result_path="$1" + local exit_code="$2" + RESULT_PATH="$result_path" EXIT_CODE="$exit_code" python3 - <<'PY' +import json +import os +from pathlib import Path + +path = Path(os.environ["RESULT_PATH"]) +payload = json.loads(path.read_text(encoding="utf-8")) +payload["status"] = "fail" +payload["full_result_exported"] = False +payload["full_result_export_failure"] = "receipt_download_validation_or_remote_export_cleanup_failed" +payload["ssh_exit_code"] = int(os.environ["EXIT_CODE"]) +path.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n", encoding="utf-8") +PY +} + runner_main() { [[ "$#" -eq 3 ]] || die "runner mode requires operation, request_id, and target_db" local operation="$1" local request_id="$2" local target_db="$3" local temp_dir result_dir result_path raw_stdout raw_stderr scp_stdout ssh_key remote_command exit_code - local sanitized_status bundle_path remote_upload + local sanitized_status bundle_path remote_upload full_result remote_result_export export_cleanup_command local -a remote_argv validate_inputs "$operation" "$request_id" "$target_db" @@ -591,6 +802,55 @@ runner_main() { if [[ "$exit_code" -eq 0 && "$sanitized_status" != "pass" ]]; then exit_code=90 fi + if [[ "$exit_code" -eq 0 && "$operation" == "direct-claim-replay" ]]; then + full_result="$result_dir/direct-claim-result.json" + remote_result_export="$INSTANCE:.teleo-iap-result-$request_id.json" + set +e + gcloud compute scp "$remote_result_export" "$full_result" \ + --project="$PROJECT_ID" \ + --zone="$ZONE" \ + --tunnel-through-iap \ + --ssh-key-file="$ssh_key" \ + --ssh-key-expire-after=5m \ + --quiet >>"$scp_stdout" 2>>"$raw_stderr" + exit_code=$? + set -e + if [[ "$exit_code" -eq 0 ]]; then + chmod 0600 "$full_result" + set +e + validate_downloaded_replay "$result_path" "$full_result" "$target_db" + exit_code=$? + set -e + fi + if [[ "$exit_code" -eq 0 ]]; then + printf -v export_cleanup_command '%q ' rm -f -- ".teleo-iap-result-$request_id.json" + set +e + gcloud compute ssh "$INSTANCE" \ + --project="$PROJECT_ID" \ + --zone="$ZONE" \ + --tunnel-through-iap \ + --ssh-key-file="$ssh_key" \ + --ssh-key-expire-after=5m \ + --command="$export_cleanup_command" \ + --quiet >>"$scp_stdout" 2>>"$raw_stderr" + exit_code=$? + set -e + fi + if [[ "$exit_code" -ne 0 ]]; then + mark_result_export_failure "$result_path" "$exit_code" + else + RESULT_PATH="$result_path" python3 - <<'PY' +import json +import os +from pathlib import Path + +path = Path(os.environ["RESULT_PATH"]) +payload = json.loads(path.read_text(encoding="utf-8")) +payload["remote_full_result_export_removed"] = True +path.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n", encoding="utf-8") +PY + fi + fi printf 'sanitized_result=%s\n' "$result_path" return "$exit_code" } diff --git a/tests/test_apply_gcp_iap_operator_access.py b/tests/test_apply_gcp_iap_operator_access.py new file mode 100644 index 0000000..56c94f9 --- /dev/null +++ b/tests/test_apply_gcp_iap_operator_access.py @@ -0,0 +1,409 @@ +from __future__ import annotations + +import copy +import hashlib +import importlib +import json +import subprocess +import sys +from pathlib import Path +from typing import Any + +ROOT = Path(__file__).resolve().parents[1] +OPS = ROOT / "ops" +sys.path.insert(0, str(OPS)) +apply = importlib.import_module("apply_gcp_iap_operator_access") +sys.path.pop(0) + + +def _binding(role: str, member: str, condition: tuple[str, str] | None = None) -> dict[str, Any]: + result: dict[str, Any] = {"role": role, "members": [member]} + if condition: + result["condition"] = {"title": condition[0], "expression": condition[1]} + return result + + +def _full_state() -> dict[str, Any]: + operator_members = [f"serviceAccount:{email}" for _, email, _ in apply.SERVICE_ACCOUNTS] + account_policies = { + email: {"bindings": [_binding("roles/iam.workloadIdentityUser", apply.WIF_PRINCIPAL)]} + for _, email, _ in apply.SERVICE_ACCOUNTS + } + project_bindings: list[dict[str, Any]] = [] + for member in operator_members: + project_bindings.extend( + [ + _binding(apply.CUSTOM_ROLE, member), + _binding(apply.IAP_ROLE, member, (apply.IAP_CONDITION_TITLE, apply.IAP_CONDITION)), + ] + ) + return { + "services": set(apply.REQUIRED_APIS), + "pool": {"state": "ACTIVE", "disabled": False}, + "provider": { + "state": "ACTIVE", + "disabled": False, + "displayName": apply.PROVIDER_DISPLAY_NAME, + "oidc": {"issuerUri": apply.PROVIDER_ISSUER}, + "attributeMapping": apply.EXPECTED_ATTRIBUTE_MAPPING, + "attributeCondition": apply.PROVIDER_CONDITION, + }, + "accounts": { + email: {"email": email, "displayName": display_name, "disabled": False} + for _, email, display_name in apply.SERVICE_ACCOUNTS + }, + "account_policies": account_policies, + "custom_role": { + "title": apply.CUSTOM_ROLE_TITLE, + "description": apply.CUSTOM_ROLE_DESCRIPTION, + "includedPermissions": list(apply.CUSTOM_PERMISSIONS), + "stage": "GA", + }, + "project_policy": {"bindings": project_bindings}, + "vm_policy": {"bindings": [_binding("roles/iam.serviceAccountUser", member) for member in operator_members]}, + "instance": { + "status": "RUNNING", + "networkInterfaces": [ + { + "networkIP": apply.INSTANCE_INTERNAL_IP, + "network": f"projects/{apply.PROJECT}/global/networks/{apply.NETWORK}", + } + ], + "serviceAccounts": [{"email": apply.VM_SERVICE_ACCOUNT}], + "metadata": {"items": [{"key": "enable-oslogin", "value": "TRUE"}]}, + "tags": {"items": [apply.TARGET_TAG]}, + }, + "instance_policy": { + "bindings": [ + _binding("roles/compute.osLogin", operator_members[0]), + _binding("roles/compute.osAdminLogin", operator_members[1]), + ] + }, + "iap_firewall": { + "direction": "INGRESS", + "network": f"projects/{apply.PROJECT}/global/networks/{apply.NETWORK}", + "allowed": [{"IPProtocol": "tcp", "ports": ["22"]}], + "sourceRanges": [apply.IAP_SOURCE_RANGE], + "targetTags": [apply.TARGET_TAG], + "disabled": False, + "logConfig": {"enable": True}, + }, + "public_firewall": {"name": apply.CHANGING_PUBLIC_RULE, "disabled": False}, + "dispatcher_hash": hashlib.sha256(apply.DEFAULT_DISPATCHER.read_bytes()).hexdigest(), + "dispatcher_file_mode": "root:root:755", + "dispatcher_run_mode": "root:root:700", + } + + +def _clean_state() -> dict[str, Any]: + state = _full_state() + state.update( + { + "services": set(), + "pool": None, + "provider": None, + "accounts": {email: None for _, email, _ in apply.SERVICE_ACCOUNTS}, + "account_policies": {email: None for _, email, _ in apply.SERVICE_ACCOUNTS}, + "custom_role": None, + "project_policy": {"bindings": []}, + "vm_policy": {"bindings": []}, + "instance_policy": {"bindings": []}, + "iap_firewall": None, + "dispatcher_hash": None, + "dispatcher_file_mode": None, + "dispatcher_run_mode": None, + } + ) + state["instance"]["metadata"] = {"items": []} + state["instance"]["tags"] = {"items": []} + return state + + +class FakeGcloud: + def __init__(self, state: dict[str, Any]) -> None: + self.state = copy.deepcopy(state) + self.commands: list[list[str]] = [] + self.mutations: list[list[str]] = [] + + @staticmethod + def _value(command: list[str], flag: str) -> str: + return command[command.index(flag) + 1] + + @staticmethod + def _completed(command: list[str], payload: Any = None, *, returncode: int = 0, stderr: str = ""): + stdout = "" if payload is None else json.dumps(payload) + return subprocess.CompletedProcess(command, returncode, stdout, stderr) + + @classmethod + def _missing(cls, command: list[str]): + return cls._completed(command, returncode=1, stderr="ERROR: NOT_FOUND: resource does not exist") + + @staticmethod + def _add_binding(policy: dict[str, Any], role: str, member: str, condition: tuple[str, str] | None) -> None: + candidate = _binding(role, member, condition) + if candidate not in policy["bindings"]: + policy["bindings"].append(candidate) + + def _read(self, command: list[str]): + words = command[1:] + if words[:2] == ["projects", "describe"]: + return self._completed(command, {"projectId": apply.PROJECT}) + if words[:2] == ["services", "list"]: + return self._completed(command, [{"config": {"name": name}} for name in sorted(self.state["services"])]) + if words[:3] == ["iam", "workload-identity-pools", "describe"]: + value = self.state["pool"] + return self._missing(command) if value is None else self._completed(command, value) + if words[:4] == ["iam", "workload-identity-pools", "providers", "describe"]: + value = self.state["provider"] + return self._missing(command) if value is None else self._completed(command, value) + if words[:3] == ["iam", "service-accounts", "describe"]: + email = words[3] + value = self.state["accounts"].get(email) + return self._missing(command) if value is None else self._completed(command, value) + if words[:3] == ["iam", "service-accounts", "get-iam-policy"]: + email = words[3] + value = ( + self.state["vm_policy"] if email == apply.VM_SERVICE_ACCOUNT else self.state["account_policies"][email] + ) + return self._completed(command, value) + if words[:3] == ["iam", "roles", "describe"]: + value = self.state["custom_role"] + return self._missing(command) if value is None else self._completed(command, value) + if words[:2] == ["projects", "get-iam-policy"]: + return self._completed(command, self.state["project_policy"]) + if words[:3] == ["compute", "instances", "describe"]: + return self._completed(command, self.state["instance"]) + if words[:3] == ["compute", "instances", "get-iam-policy"]: + return self._completed(command, self.state["instance_policy"]) + if words[:3] == ["compute", "firewall-rules", "describe"]: + rule = words[3] + value = self.state["iap_firewall"] if rule == apply.IAP_FIREWALL_RULE else self.state["public_firewall"] + return self._missing(command) if value is None else self._completed(command, value) + return None + + def _mutate(self, command: list[str]): + words = command[1:] + if words[:2] == ["services", "enable"]: + self.state["services"].update(words[2 : words.index("--project")]) + elif words[:3] == ["iam", "workload-identity-pools", "create"]: + self.state["pool"] = {"state": "ACTIVE", "disabled": False} + elif words[:3] == ["iam", "workload-identity-pools", "update"]: + self.state["pool"]["disabled"] = False + elif words[:4] == ["iam", "workload-identity-pools", "providers", "create-oidc"]: + self.state["provider"] = { + "state": "ACTIVE", + "disabled": False, + "displayName": self._value(command, "--display-name"), + "oidc": {"issuerUri": self._value(command, "--issuer-uri")}, + "attributeMapping": apply.EXPECTED_ATTRIBUTE_MAPPING, + "attributeCondition": self._value(command, "--attribute-condition"), + } + elif words[:4] == ["iam", "workload-identity-pools", "providers", "update-oidc"]: + self.state["provider"].update( + { + "state": "ACTIVE", + "disabled": False, + "displayName": self._value(command, "--display-name"), + "oidc": {"issuerUri": self._value(command, "--issuer-uri")}, + "attributeMapping": apply.EXPECTED_ATTRIBUTE_MAPPING, + "attributeCondition": self._value(command, "--attribute-condition"), + } + ) + elif words[:3] == ["iam", "service-accounts", "create"]: + account_id = words[3] + email = f"{account_id}@{apply.PROJECT}.iam.gserviceaccount.com" + self.state["accounts"][email] = { + "email": email, + "displayName": self._value(command, "--display-name"), + "disabled": False, + } + self.state["account_policies"][email] = {"bindings": []} + elif words[:3] == ["iam", "service-accounts", "update"]: + self.state["accounts"][words[3]]["displayName"] = self._value(command, "--display-name") + elif words[:3] == ["iam", "roles", "create"] or words[:3] == ["iam", "roles", "update"]: + self.state["custom_role"] = { + "title": self._value(command, "--title"), + "description": self._value(command, "--description"), + "includedPermissions": self._value(command, "--permissions").split(","), + "stage": self._value(command, "--stage"), + } + elif words[:3] == ["iam", "service-accounts", "add-iam-policy-binding"]: + email = words[3] + policy = ( + self.state["vm_policy"] if email == apply.VM_SERVICE_ACCOUNT else self.state["account_policies"][email] + ) + self._add_binding(policy, self._value(command, "--role"), self._value(command, "--member"), None) + elif words[:2] == ["projects", "add-iam-policy-binding"]: + condition = None + if "--condition" in command: + expression, title = self._value(command, "--condition").rsplit(",title=", 1) + condition = (title, expression.removeprefix("expression=")) + self._add_binding( + self.state["project_policy"], + self._value(command, "--role"), + self._value(command, "--member"), + condition, + ) + elif words[:3] == ["compute", "instances", "add-iam-policy-binding"]: + self._add_binding( + self.state["instance_policy"], + self._value(command, "--role"), + self._value(command, "--member"), + None, + ) + elif words[:3] == ["compute", "instances", "add-metadata"]: + self.state["instance"]["metadata"] = {"items": [{"key": "enable-oslogin", "value": "TRUE"}]} + elif words[:3] == ["compute", "instances", "add-tags"]: + self.state["instance"]["tags"] = {"items": [apply.TARGET_TAG]} + elif words[:3] == ["compute", "firewall-rules", "create"]: + self.state["iap_firewall"] = _full_state()["iap_firewall"] + elif words[:3] == ["compute", "firewall-rules", "update"]: + self.state["iap_firewall"].update( + {"targetTags": [apply.TARGET_TAG], "disabled": False, "logConfig": {"enable": True}} + ) + elif words[:2] == ["compute", "scp"]: + pass + elif words[:2] == ["compute", "ssh"] and "sudo install -o root" in self._value(command, "--command"): + self.state["dispatcher_hash"] = hashlib.sha256(apply.DEFAULT_DISPATCHER.read_bytes()).hexdigest() + self.state["dispatcher_file_mode"] = "root:root:755" + self.state["dispatcher_run_mode"] = "root:root:700" + else: + return None + self.mutations.append(command) + return self._completed(command) + + def __call__(self, command: list[str], **_: Any): + self.commands.append(command) + read = self._read(command) + if read is not None: + return read + if command[1:3] == ["compute", "ssh"]: + remote = self._value(command, "--command") + if "sudo install -o root" in remote: + return self._mutate(command) + file_mode = self.state["dispatcher_file_mode"] or "MISSING" + run_mode = self.state["dispatcher_run_mode"] or "MISSING" + lines = [f"FILE={file_mode}"] + if self.state["dispatcher_hash"]: + lines.append(f"SHA={self.state['dispatcher_hash']}") + lines.append(f"RUN={run_mode}") + return subprocess.CompletedProcess(command, 0, "\n".join(lines) + "\n", "") + mutated = self._mutate(command) + if mutated is not None: + return mutated + return subprocess.CompletedProcess(command, 2, "", "unsupported fake gcloud command") + + +def _run(tmp_path: Path, state: dict[str, Any]): + fake = FakeGcloud(state) + receipt, exitcode = apply.apply_operator_access( + execute=True, + output=tmp_path / "receipt.json", + runner=fake, + ) + return fake, receipt, exitcode + + +def test_clean_create_converges_and_retains_sanitized_receipt(tmp_path: Path) -> None: + fake, receipt, exitcode = _run(tmp_path, _clean_state()) + + assert exitcode == 0 + assert receipt["status"] == "applied" + assert receipt["mutation_count"] > 0 + assert fake.state["provider"]["attributeCondition"] == apply.PROVIDER_CONDITION + assert fake.state["dispatcher_hash"] == receipt["dispatcher"]["sha256"] + assert receipt["credentials_logged"] is False + assert len(apply.PROVIDER_DISPLAY_NAME) <= 32 + assert apply.PROVIDER_DISPLAY_NAME == "Teleo fixed IAP operator" + assert json.loads((tmp_path / "receipt.json").read_text())["status"] == "applied" + + +def test_fully_existing_state_is_control_plane_noop(tmp_path: Path) -> None: + fake, receipt, exitcode = _run(tmp_path, _full_state()) + + assert exitcode == 0 + assert receipt["status"] == "applied" + assert receipt["mutation_count"] == 0 + assert fake.mutations == [] + + +def test_partial_state_recovers_only_missing_clone_identity_and_bindings(tmp_path: Path) -> None: + state = _full_state() + clone_member = f"serviceAccount:{apply.CLONE_SERVICE_ACCOUNT}" + state["accounts"][apply.CLONE_SERVICE_ACCOUNT] = None + state["account_policies"][apply.CLONE_SERVICE_ACCOUNT] = None + for policy_name in ("project_policy", "vm_policy", "instance_policy"): + state[policy_name]["bindings"] = [ + binding for binding in state[policy_name]["bindings"] if clone_member not in binding["members"] + ] + + fake, receipt, exitcode = _run(tmp_path, state) + mutation_names = [ + operation["name"] for operation in receipt["operations"] if operation["action"] not in {"probe", "verify"} + ] + + assert exitcode == 0 + assert f"create_service_account:{apply.CLONE_SERVICE_ACCOUNT}" in mutation_names + assert f"create_service_account:{apply.STATUS_SERVICE_ACCOUNT}" not in mutation_names + assert all(apply.STATUS_SERVICE_ACCOUNT not in " ".join(command) for command in fake.mutations) + + +def test_security_sensitive_provider_drift_fails_before_mutation(tmp_path: Path) -> None: + state = _full_state() + state["provider"]["attributeCondition"] = "assertion.repository == 'attacker/repository'" + + fake, receipt, exitcode = _run(tmp_path, state) + + assert exitcode == 1 + assert receipt["status"] == "failed" + assert "security-sensitive drift" in receipt["error"] + assert fake.mutations == [] + + +def test_disabled_pool_and_provider_are_narrowly_reenabled(tmp_path: Path) -> None: + state = _full_state() + state["pool"]["disabled"] = True + state["provider"]["disabled"] = True + + fake, receipt, exitcode = _run(tmp_path, state) + mutation_names = [ + operation["name"] for operation in receipt["operations"] if operation["action"] not in {"probe", "verify"} + ] + + assert exitcode == 0 + assert "enable_wif_pool" in mutation_names + assert "update_wif_provider" in mutation_names + assert fake.state["pool"]["disabled"] is False + assert fake.state["provider"]["disabled"] is False + + +def test_wrong_or_stopped_instance_fails_before_mutation(tmp_path: Path) -> None: + state = _full_state() + state["instance"]["status"] = "TERMINATED" + + fake, receipt, exitcode = _run(tmp_path, state) + + assert exitcode == 1 + assert receipt["status"] == "failed" + assert "not running" in receipt["error"] + assert fake.mutations == [] + + +def test_bootstrap_never_disables_or_mutates_public_ssh_rule(tmp_path: Path) -> None: + fake, receipt, exitcode = _run(tmp_path, _clean_state()) + + assert exitcode == 0 + assert receipt["public_ssh_rule"] == { + "name": apply.CHANGING_PUBLIC_RULE, + "observed_disabled_before": False, + "observed_disabled_after": False, + "mutated": False, + "disable_command_present": False, + } + assert not any("--disabled" in command for command in fake.commands) + assert not any( + apply.CHANGING_PUBLIC_RULE in command + and command[1:3] == ["compute", "firewall-rules"] + and command[3] != "describe" + for command in fake.commands + ) diff --git a/tests/test_gcp_iap_operator_access.py b/tests/test_gcp_iap_operator_access.py index 49f488a..1a2f2f8 100644 --- a/tests/test_gcp_iap_operator_access.py +++ b/tests/test_gcp_iap_operator_access.py @@ -56,6 +56,10 @@ def test_plan_uses_exact_iap_firewall_and_least_privilege_split() -> None: "cleanup-clone", ] assert "roles/iam.serviceAccountUser" in commands + assert "--display-name 'Teleo fixed IAP operator'" in commands + assert "Teleo fixed IAP operator workflow" not in commands + assert "instances add-tags teleo-prod-1" in commands + assert "--tags teleo-prod-ssh" in commands assert "--ssh-key-expire-after=5m" in commands assert "--ssh-key-expiration" not in commands assert "sha256sum -c dispatcher.sha256" in commands @@ -116,7 +120,7 @@ def test_workflow_accepts_only_fixed_operation_and_bounded_ids() -> None: assert forbidden_input not in inputs -def test_workflow_uses_wif_official_actions_and_sanitized_short_retention() -> None: +def test_workflow_uses_wif_official_actions_and_verified_short_retention() -> None: workflow = load_workflow() assert workflow["permissions"] == {"contents": "read", "id-token": "write"} steps = workflow["jobs"]["operate"]["steps"] @@ -132,7 +136,7 @@ def test_workflow_uses_wif_official_actions_and_sanitized_short_retention() -> N assert auth["with"]["create_credentials_file"] == "true" assert auth["with"]["cleanup_credentials"] == "true" upload = next(step for step in steps if step.get("uses") == "actions/upload-artifact@v4") - assert upload["with"]["path"] == "${{ env.RESULT_DIR }}/result.json" + assert upload["with"]["path"] == "${{ env.RESULT_DIR }}/" assert upload["with"]["retention-days"] == "7" validate_index = next(i for i, step in enumerate(steps) if step.get("id") == "validate") auth_index = next(i for i, step in enumerate(steps) if step.get("uses") == "google-github-actions/auth@v3") @@ -221,6 +225,27 @@ def test_operator_cleanup_is_bound_to_marker_clone_prefix_and_exact_run_dir() -> assert 'drop database :"target_db";' in source assert '[[ "$remaining" == "0" ]]' in source assert 'rm -rf --one-file-system -- "$run_dir"' in source + assert 'CORY_REPLAY_LEGACY_DIR="/home/teleo/gcp-cory-model-replay-20260712t1940z"' in source + assert '[[ "$(realpath "$legacy_dir")" == "$CORY_REPLAY_LEGACY_DIR" ]]' in source + assert 'rm -rf --one-file-system -- "$legacy_dir"' in source + assert 'export_path="$caller_home/.teleo-iap-result-$request_id.json"' in source + assert '[[ "$export_state" == "$SUDO_USER:600" ]]' in source + assert 'rm -f -- "$export_path"' in source + assert 'DISABLED_ROLLBACK_DB="teleo_canonical_pre_20260712t1905z"' in source + assert "rollback database unexpectedly allows connections" in source + + +def test_operator_exports_and_fixed_validates_full_replay_before_cleanup() -> None: + source = OPERATOR.read_text(encoding="utf-8") + + assert 'CORY_REPLAY_FINGERPRINT="48ea5332c4f303dea02d503447f1ac8e1d8aa00e60d198a566d628b37dbdac1c"' in source + assert 'export_path="$caller_home/.teleo-iap-result-$request_id.json"' in source + assert 'remote_result_export="$INSTANCE:.teleo-iap-result-$request_id.json"' in source + assert 'full_result="$result_dir/direct-claim-result.json"' in source + assert '"strict_six_of_six"' in source + assert '"fixed_fingerprint_before"' in source + assert '"fixed_counts_after"' in source + assert 'payload["remote_full_result_export_removed"] = True' in source def test_operator_uses_five_minute_key_private_modes_and_cleanup_trap() -> None: @@ -339,24 +364,75 @@ def test_clone_runner_scps_only_request_derived_bundle_over_iap(tmp_path: Path) result_dir = runner_temp / "result" bundle_dir = runner_temp / "gcp-iap-operator-bundle" invocation = tmp_path / "gcloud-argv.txt" + counter = tmp_path / "gcloud-counter.txt" + full_receipt = tmp_path / "full-receipt.json" bin_dir.mkdir() runner_temp.mkdir() bundle_dir.mkdir(mode=0o700) bundle = bundle_dir / "iap-abcdefghijkl.tar.gz" bundle.write_bytes(b"reviewed-bundle") bundle.chmod(0o600) + full_receipt.write_text( + json.dumps( + { + "status": "pass", + "target_database": "teleo_clone_cory_20260712t1940z", + "score": {"passes": 6, "expected_prompt_count": 6}, + "checks": {"all_runtime_checks": True}, + "database_fingerprint_before": { + "sha256": "48ea5332c4f303dea02d503447f1ac8e1d8aa00e60d198a566d628b37dbdac1c" + }, + "database_fingerprint_after": { + "sha256": "48ea5332c4f303dea02d503447f1ac8e1d8aa00e60d198a566d628b37dbdac1c" + }, + "canonical_status_before": { + "high_signal_rows": { + "claims": 1837, + "sources": 4145, + "claim_edges": 4916, + "claim_evidence": 4670, + "kb_proposals": 26, + } + }, + "canonical_status_after": { + "high_signal_rows": { + "claims": 1837, + "sources": 4145, + "claim_edges": 4916, + "claim_evidence": 4670, + "kb_proposals": 26, + } + }, + }, + sort_keys=True, + ) + + "\n", + encoding="utf-8", + ) fake_gcloud = bin_dir / "gcloud" fake_gcloud.write_text( "#!/usr/bin/env bash\n" 'printf \'%s \' "$@" >> "${FAKE_GCLOUD_ARGV}"\n' "printf '\\n' >> \"${FAKE_GCLOUD_ARGV}\"\n" - 'if [[ "$2" == "ssh" ]]; then\n' - " printf '%s\\n' " + 'count="$(cat "${FAKE_GCLOUD_COUNTER}" 2>/dev/null || printf 0)"\n' + 'count="$((count + 1))"\n' + 'printf \'%s\\n\' "$count" > "${FAKE_GCLOUD_COUNTER}"\n' + 'case "$count" in\n' + " 2)\n" + ' result_sha="$(sha256sum "${FAKE_FULL_RECEIPT}" | awk \'{print $1}\')"\n' + ' result_bytes="$(wc -c <"${FAKE_FULL_RECEIPT}" | tr -d \' \')"\n' + " printf " '\'{"operation":"direct-claim-replay","request_id":"iap-abcdefghijkl",\'' - '\'"target_db":"teleo_clone_test","status":"pass",\'' + '\'"target_db":"teleo_clone_cory_20260712t1940z","status":"pass",\'' '\'"bundle_commit":"aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",\'' - '\'"bundle_sha256":"bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb"}\'\n' - "fi\n", + '\'"bundle_sha256":"bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb",\'' + '\'"result_sha256":"%s","result_bytes":%s,"full_result_export_ready":true}\n\' ' + '"$result_sha" "$result_bytes"\n' + " ;;\n" + " 3)\n" + ' cp "${FAKE_FULL_RECEIPT}" "$4"\n' + " ;;\n" + "esac\n", encoding="utf-8", ) fake_gcloud.chmod(0o755) @@ -366,10 +442,18 @@ def test_clone_runner_scps_only_request_derived_bundle_over_iap(tmp_path: Path) "RUNNER_TEMP": str(runner_temp), "RESULT_DIR": str(result_dir), "FAKE_GCLOUD_ARGV": str(invocation), + "FAKE_GCLOUD_COUNTER": str(counter), + "FAKE_FULL_RECEIPT": str(full_receipt), } completed = subprocess.run( - ["bash", str(OPERATOR), "direct-claim-replay", "iap-abcdefghijkl", "teleo_clone_test"], + [ + "bash", + str(OPERATOR), + "direct-claim-replay", + "iap-abcdefghijkl", + "teleo_clone_cory_20260712t1940z", + ], cwd=REPO_ROOT, env=env, text=True, @@ -379,7 +463,7 @@ def test_clone_runner_scps_only_request_derived_bundle_over_iap(tmp_path: Path) assert completed.returncode == 0, completed.stderr calls = invocation.read_text(encoding="utf-8").splitlines() - assert len(calls) == 2 + assert len(calls) == 4 assert "compute scp" in calls[0] assert str(bundle) in calls[0] assert "teleo-prod-1:.teleo-iap-upload-iap-abcdefghijkl.tar.gz" in calls[0] @@ -387,6 +471,18 @@ def test_clone_runner_scps_only_request_derived_bundle_over_iap(tmp_path: Path) assert "--ssh-key-expire-after=5m" in calls[0] assert "compute ssh" in calls[1] assert "--tunnel-through-iap" in calls[1] + assert "compute scp" in calls[2] + assert "teleo-prod-1:.teleo-iap-result-iap-abcdefghijkl.json" in calls[2] + assert str(result_dir / "direct-claim-result.json") in calls[2] + assert "compute ssh" in calls[3] + assert ".teleo-iap-result-iap-abcdefghijkl.json" in calls[3] + result = json.loads((result_dir / "result.json").read_text(encoding="utf-8")) + assert result["status"] == "pass" + assert result["full_result_exported"] is True + assert result["remote_full_result_export_removed"] is True + assert result["full_result_fixed_checks"]["strict_six_of_six"] is True + assert result["full_result_fixed_checks"]["fixed_fingerprint_before"] is True + assert result["full_result_fixed_checks"]["fixed_counts_after"] is True def test_clone_runner_rejects_bundle_symlink_before_gcloud(tmp_path: Path) -> None: