teleo/teleo-infrastructure
8
0
Fork 0
Code Issues Pull requests 1 Projects Releases Packages Wiki Activity Actions 1

Harden GCP replay and durable IAP bootstrap

Browse source
This commit is contained in:
twentyOne2x 2026-07-13 04:58:16 +02:00
parent 63689589d7
commit ff5fb56a36
6 changed files with 2028 additions and 18 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
.github/workflows/gcp-iap-operator.yml vendored
View file

@ -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

1230
ops/apply_gcp_iap_operator_access.py Normal file
View file

File diff suppressed because it is too large Load diff

17
ops/plan_gcp_iap_operator_access.py
View file

@ -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",

272
scripts/gcp_iap_operator.sh
View file

@ -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"
}

409
tests/test_apply_gcp_iap_operator_access.py Normal file
View file

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

114
tests/test_gcp_iap_operator_access.py
View file

@ -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: