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

Make GCP replay bootstrap recoverable

Browse source
This commit is contained in:
twentyOne2x 2026-07-13 05:24:25 +02:00
parent ff5fb56a36
commit cb49ce0e6a
7 changed files with 404 additions and 96 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 verified result receipts
- name: Upload sanitized result receipt
if: always()
uses: actions/upload-artifact@v4
with:
name: gcp-iap-operator-${{ github.run_id }}
path: ${{ env.RESULT_DIR }}/
path: ${{ env.RESULT_DIR }}/result.json
if-no-files-found: error
retention-days: 7

157
ops/apply_gcp_iap_operator_access.py
View file

@ -272,6 +272,19 @@ def _metadata(instance: dict[str, Any]) -> dict[str, str]:
return {str(item.get("key")): str(item.get("value")) for item in items if isinstance(item, dict)}
def _find_resource(resources: Any, *, suffix: str, label: str) -> dict[str, Any] | None:
if not isinstance(resources, list):
raise ConvergenceError(f"{label} deleted-resource readback is invalid")
matches = [
item
for item in resources
if isinstance(item, dict) and str(item.get("name", "")).endswith(suffix)
]
if len(matches) > 1:
raise ConvergenceError(f"{label} deleted-resource readback is ambiguous")
return matches[0] if matches else None
def _firewall_is_structurally_exact(firewall: dict[str, Any]) -> bool:
allowed = firewall.get("allowed", [])
return (
@ -314,6 +327,25 @@ def discover(client: Gcloud, *, require_public_rule: bool = True) -> dict[str, A
],
allow_not_found=True,
)
if pool is None:
pool = _find_resource(
client.json(
"probe_deleted_wif_pools",
[
"gcloud",
"iam",
"workload-identity-pools",
"list",
"--location",
"global",
*_project_args(),
"--show-deleted",
*_json_args(),
],
),
suffix=f"/workloadIdentityPools/{POOL}",
label="WIF pool",
)
provider = client.json(
"probe_wif_provider",
[
@ -332,6 +364,31 @@ def discover(client: Gcloud, *, require_public_rule: bool = True) -> dict[str, A
],
allow_not_found=True,
)
if provider is None and pool is not None:
deleted_providers = client.json(
"probe_deleted_wif_providers",
[
"gcloud",
"iam",
"workload-identity-pools",
"providers",
"list",
"--location",
"global",
"--workload-identity-pool",
POOL,
*_project_args(),
"--show-deleted",
*_json_args(),
],
allow_not_found=True,
)
if deleted_providers is not None:
provider = _find_resource(
deleted_providers,
suffix=f"/workloadIdentityPools/{POOL}/providers/{PROVIDER}",
label="WIF provider",
)
accounts: dict[str, Any | None] = {}
account_policies: dict[str, Any | None] = {}
@ -482,6 +539,25 @@ def _provider_mutations(state: dict[str, Any]) -> list[Mutation]:
],
)
)
elif pool.get("state") == "DELETED":
mutations.append(
Mutation(
"undelete_wif_pool",
"undelete",
[
"gcloud",
"iam",
"workload-identity-pools",
"undelete",
POOL,
"--location",
"global",
*_project_args(),
"--quiet",
],
)
)
return mutations
elif pool.get("state") not in (None, "ACTIVE"):
raise ConvergenceError("the shared GitHub Actions workload identity pool is not active")
elif pool.get("disabled") is True:
@ -537,12 +613,36 @@ def _provider_mutations(state: dict[str, Any]) -> list[Mutation]:
return mutations
observed_mapping = provider.get("attributeMapping", {})
security_exact = (
provider.get("state") in (None, "ACTIVE")
and provider.get("oidc", {}).get("issuerUri") == PROVIDER_ISSUER
security_fields_exact = (
provider.get("oidc", {}).get("issuerUri") == PROVIDER_ISSUER
and observed_mapping == EXPECTED_ATTRIBUTE_MAPPING
and provider.get("attributeCondition") == PROVIDER_CONDITION
)
if provider.get("state") == "DELETED":
if not security_fields_exact:
raise ConvergenceError("the soft-deleted WIF provider has security-sensitive drift")
mutations.append(
Mutation(
"undelete_wif_provider",
"undelete",
[
"gcloud",
"iam",
"workload-identity-pools",
"providers",
"undelete",
PROVIDER,
"--location",
"global",
"--workload-identity-pool",
POOL,
*_project_args(),
"--quiet",
],
)
)
return mutations
security_exact = provider.get("state") in (None, "ACTIVE") and security_fields_exact
if not security_exact:
raise ConvergenceError("the existing WIF provider has security-sensitive drift")
if provider.get("disabled") is True or provider.get("displayName") != PROVIDER_DISPLAY_NAME:
@ -812,27 +912,16 @@ def _instance_and_firewall_mutations(state: dict[str, Any]) -> list[Mutation]:
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",
],
)
raise ConvergenceError(
"instance OS Login is not already enabled; refusing to change the SSH authentication mode in bootstrap"
)
tags = set(instance.get("tags", {}).get("items", []))
if TARGET_TAG not in tags:
public_target_tags = set((state.get("public_firewall") or {}).get("targetTags", []))
if TARGET_TAG in public_target_tags:
raise ConvergenceError(
"adding the IAP target tag would newly match the existing public SSH firewall rule"
)
mutations.append(
Mutation(
"add_instance_iap_target_tag",
@ -1084,12 +1173,16 @@ def _dry_run_mutations() -> list[Mutation]:
}
],
"serviceAccounts": [{"email": VM_SERVICE_ACCOUNT}],
"metadata": {"items": []},
"metadata": {"items": [{"key": "enable-oslogin", "value": "TRUE"}]},
"tags": {"items": []},
},
"instance_policy": empty_policy,
"iap_firewall": None,
"public_firewall": {"name": CHANGING_PUBLIC_RULE, "disabled": False},
"public_firewall": {
"name": CHANGING_PUBLIC_RULE,
"disabled": False,
"targetTags": ["teleo-public-ssh-bootstrap"],
},
}
return plan_mutations(empty_state)
@ -1155,12 +1248,19 @@ def apply_operator_access(
)
else:
before = discover(client)
mutations = plan_mutations(before)
_run_mutations(client, mutations)
after = discover(client)
current = before
for _ in range(6):
mutations = plan_mutations(current)
if not mutations:
break
_run_mutations(client, mutations)
current = discover(client)
after = current
remaining = plan_mutations(after)
if remaining:
raise ConvergenceError(f"control-plane verification still requires {len(remaining)} mutation(s)")
raise ConvergenceError(
f"control-plane convergence exceeded six mutation rounds; {len(remaining)} mutation(s) remain"
)
dispatcher = converge_dispatcher(client, dispatcher_path)
receipt.update(
{
@ -1174,6 +1274,9 @@ def apply_operator_access(
"name": CHANGING_PUBLIC_RULE,
"observed_disabled_before": bool(before["public_firewall"].get("disabled", False)),
"observed_disabled_after": bool(after["public_firewall"].get("disabled", False)),
"target_tags_before": sorted(before["public_firewall"].get("targetTags", [])),
"target_tags_after": sorted(after["public_firewall"].get("targetTags", [])),
"effective_target_tag_expansion": False,
"mutated": False,
"disable_command_present": False,
},

135
scripts/gcp_iap_operator.sh
View file

@ -168,6 +168,28 @@ validate_run_dir() {
}
}
validate_replay_temp_root() {
local temp_path="$1"
local request_id="$2"
local temp_resolved temp_state temp_marker temp_marker_state temp_marker_content
[[ -d "$temp_path" && ! -L "$temp_path" ]] || die "request-derived temporary replay path is unsafe"
temp_resolved="$(realpath "$temp_path")"
[[ "$temp_resolved" == "/tmp/leo-clone-checkpoint-$request_id-gcp-direct-claim-"* ]] || {
die "request-derived temporary replay path drifted"
}
temp_state="$(stat -c '%U:%G:%a' "$temp_path")"
[[ "$temp_state" == "root:root:700" ]] || die "request-derived temporary replay path ownership or mode drifted"
temp_marker="$temp_path/.teleo-temp-owner"
[[ -f "$temp_marker" && ! -L "$temp_marker" ]] || die "temporary replay ownership marker is absent or unsafe"
temp_marker_state="$(stat -c '%U:%G:%a' "$temp_marker")"
[[ "$temp_marker_state" == "root:root:600" ]] || die "temporary replay ownership marker state drifted"
temp_marker_content="$(cat "$temp_marker")"
[[ "$temp_marker_content" == "$request_id"$'\n'"gcp-direct-claim" ]] || {
die "temporary replay ownership marker does not match request_id and prompt"
}
}
remote_upload_path() {
local request_id="$1"
local passwd_entry home resolved upload
@ -433,7 +455,7 @@ remote_cleanup_clone() {
local target_db="$2"
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
local rollback_exists rollback_allowconn rollback_connections leftover_temp_count temp_path
receive_and_validate_bundle "cleanup-clone" "$request_id" "$target_db"
run_dir="$REMOTE_RUN_ROOT/$request_id"
@ -441,7 +463,7 @@ remote_cleanup_clone() {
legacy_dir=""
if [[ "$target_db" == "$CORY_REPLAY_DB" ]]; then
legacy_dir="$CORY_REPLAY_LEGACY_DIR"
if [[ -e "$legacy_dir" ]]; then
if [[ -e "$legacy_dir" || -L "$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"
@ -453,11 +475,17 @@ remote_cleanup_clone() {
}
[[ "$(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
if [[ -e "$export_path" || -L "$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
while IFS= read -r -d '' temp_path; do
validate_replay_temp_root "$temp_path" "$request_id"
done < <(
find /tmp -mindepth 1 -maxdepth 1 \
-name "leo-clone-checkpoint-$request_id-gcp-direct-claim-*" -print0
)
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"
@ -470,7 +498,7 @@ remote_cleanup_clone() {
-X -Atq -v ON_ERROR_STOP=1 -v "target_db=$target_db" \
-c "select exists(select 1 from pg_database where datname = :'target_db');"
)"
[[ "$exists" == "t" ]] || die "marked clone database is absent; refusing directory-only cleanup"
[[ "$exists" == "t" || "$exists" == "f" ]] || die "clone existence readback was invalid"
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" \
@ -490,13 +518,43 @@ remote_cleanup_clone() {
[[ "$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'
if [[ -n "$legacy_dir" && ( -e "$legacy_dir" || -L "$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" && ! -L "$legacy_dir" ) ]] || {
die "fixed legacy replay directory remained"
}
if [[ -e "$export_path" || -L "$export_path" ]]; then
[[ -f "$export_path" && ! -L "$export_path" ]] || die "request-derived replay export became unsafe"
rm -f -- "$export_path"
fi
[[ ! -e "$export_path" && ! -L "$export_path" ]] || die "request-derived replay export remained"
while IFS= read -r -d '' temp_path; do
validate_replay_temp_root "$temp_path" "$request_id"
rm -rf --one-file-system -- "$temp_path"
done < <(
find /tmp -mindepth 1 -maxdepth 1 \
-name "leo-clone-checkpoint-$request_id-gcp-direct-claim-*" -print0
)
leftover_temp_count="$(
find /tmp -mindepth 1 -maxdepth 1 \
-name "leo-clone-checkpoint-$request_id-gcp-direct-claim-*" -print | wc -l | tr -d ' '
)"
[[ "$leftover_temp_count" == "0" ]] || die "request-derived temporary replay paths remained"
if [[ "$exists" == "t" ]]; then
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'
select pg_terminate_backend(pid)
from pg_stat_activity
where datname = :'target_db' and pid <> pg_backend_pid();
drop database :"target_db";
SQL
fi
remaining="$(
psql "host=$CLOUDSQL_HOST port=5432 dbname=postgres user=postgres sslmode=require connect_timeout=8" \
-X -Atq -v ON_ERROR_STOP=1 -v "target_db=$target_db" \
@ -508,24 +566,7 @@ 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"
[[ ! -e "$run_dir" && ! -L "$run_dir" ]] || die "exact run-owned directory remained after cleanup"
emit_json \
"cleanup-clone" "$request_id" "$target_db" "pass" \
"bundle_commit=$VALIDATED_BUNDLE_COMMIT" \
@ -692,18 +733,21 @@ if os.environ["TARGET_DB"] == os.environ["EXPECTED_CORY_REPLAY_DB"]:
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_exported_and_validated"] = True
result["full_result_retained"] = False
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")
full_path.unlink()
PY
}
mark_result_export_failure() {
local result_path="$1"
local exit_code="$2"
RESULT_PATH="$result_path" EXIT_CODE="$exit_code" python3 - <<'PY'
local cleanup_exit_code="$3"
RESULT_PATH="$result_path" EXIT_CODE="$exit_code" CLEANUP_EXIT_CODE="$cleanup_exit_code" python3 - <<'PY'
import json
import os
from pathlib import Path
@ -711,9 +755,11 @@ 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_exported_and_validated"] = False
payload["full_result_retained"] = False
payload["full_result_export_failure"] = "receipt_download_validation_or_remote_export_cleanup_failed"
payload["ssh_exit_code"] = int(os.environ["EXIT_CODE"])
payload["remote_full_result_export_removed"] = int(os.environ["CLEANUP_EXIT_CODE"]) == 0
path.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n", encoding="utf-8")
PY
}
@ -725,6 +771,7 @@ runner_main() {
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 full_result remote_result_export export_cleanup_command
local replay_exit_code export_cleanup_exit_code
local -a remote_argv
validate_inputs "$operation" "$request_id" "$target_db"
@ -804,6 +851,7 @@ runner_main() {
fi
if [[ "$exit_code" -eq 0 && "$operation" == "direct-claim-replay" ]]; then
full_result="$result_dir/direct-claim-result.json"
[[ ! -e "$full_result" && ! -L "$full_result" ]] || die "local full replay result path was not empty"
remote_result_export="$INSTANCE:.teleo-iap-result-$request_id.json"
set +e
gcloud compute scp "$remote_result_export" "$full_result" \
@ -822,22 +870,27 @@ runner_main() {
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
rm -f -- "$full_result"
replay_exit_code="$exit_code"
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"
export_cleanup_exit_code=$?
set -e
if [[ "$replay_exit_code" -ne 0 ]]; then
exit_code="$replay_exit_code"
else
exit_code="$export_cleanup_exit_code"
fi
if [[ "$exit_code" -ne 0 ]]; then
mark_result_export_failure "$result_path" "$exit_code"
mark_result_export_failure "$result_path" "$exit_code" "$export_cleanup_exit_code"
else
RESULT_PATH="$result_path" python3 - <<'PY'
import json

7
scripts/run_leo_clone_bound_handler_checkpoint.py
View file

@ -631,13 +631,18 @@ def copy_profile(
if not live_profile.is_dir():
raise CheckpointError(f"live leoclean profile does not exist: {live_profile}")
unique = uuid.uuid4().hex[:12]
prefix = f"leo-clone-checkpoint-{_safe_name(run_id, 'run')}-{_safe_name(prompt_id, 'prompt')}-{unique}-"
safe_run_id = _safe_name(run_id, "run")
safe_prompt_id = _safe_name(prompt_id, "prompt")
prefix = f"leo-clone-checkpoint-{safe_run_id}-{safe_prompt_id}-{unique}-"
temp_root = Path(tempfile.mkdtemp(prefix=prefix))
register_temp_root(temp_root)
temp_root.chmod(0o700)
target = temp_root / "profile"
policy = ProfileCopyPolicy()
try:
owner_marker = temp_root / ".teleo-temp-owner"
owner_marker.write_text(f"{safe_run_id}\n{safe_prompt_id}\n", encoding="utf-8")
owner_marker.chmod(0o600)
shutil.copytree(live_profile, target, symlinks=False, ignore=policy)
audit = audit_temp_profile(target, policy)
if not audit["passes"]:

131
tests/test_apply_gcp_iap_operator_access.py
View file

@ -39,8 +39,16 @@ def _full_state() -> dict[str, Any]:
)
return {
"services": set(apply.REQUIRED_APIS),
"pool": {"state": "ACTIVE", "disabled": False},
"pool": {
"name": f"projects/123456789/locations/global/workloadIdentityPools/{apply.POOL}",
"state": "ACTIVE",
"disabled": False,
},
"provider": {
"name": (
f"projects/123456789/locations/global/workloadIdentityPools/{apply.POOL}"
f"/providers/{apply.PROVIDER}"
),
"state": "ACTIVE",
"disabled": False,
"displayName": apply.PROVIDER_DISPLAY_NAME,
@ -88,7 +96,11 @@ def _full_state() -> dict[str, Any]:
"disabled": False,
"logConfig": {"enable": True},
},
"public_firewall": {"name": apply.CHANGING_PUBLIC_RULE, "disabled": False},
"public_firewall": {
"name": apply.CHANGING_PUBLIC_RULE,
"disabled": False,
"targetTags": ["teleo-public-ssh-bootstrap"],
},
"dispatcher_hash": hashlib.sha256(apply.DEFAULT_DISPATCHER.read_bytes()).hexdigest(),
"dispatcher_file_mode": "root:root:755",
"dispatcher_run_mode": "root:root:700",
@ -114,7 +126,6 @@ def _clean_state() -> dict[str, Any]:
"dispatcher_run_mode": None,
}
)
state["instance"]["metadata"] = {"items": []}
state["instance"]["tags"] = {"items": []}
return state
@ -152,10 +163,16 @@ class FakeGcloud:
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)
return self._missing(command) if value is None or value.get("state") == "DELETED" else self._completed(command, value)
if words[:3] == ["iam", "workload-identity-pools", "list"]:
value = self.state["pool"]
return self._completed(command, [] if value is None else [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)
return self._missing(command) if value is None or value.get("state") == "DELETED" else self._completed(command, value)
if words[:4] == ["iam", "workload-identity-pools", "providers", "list"]:
value = self.state["provider"]
return self._completed(command, [] if value is None else [value])
if words[:3] == ["iam", "service-accounts", "describe"]:
email = words[3]
value = self.state["accounts"].get(email)
@ -186,18 +203,15 @@ class FakeGcloud:
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}
self.state["pool"] = _full_state()["pool"]
elif words[:3] == ["iam", "workload-identity-pools", "undelete"]:
self.state["pool"]["state"] = "ACTIVE"
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"),
}
self.state["provider"] = _full_state()["provider"]
elif words[:4] == ["iam", "workload-identity-pools", "providers", "undelete"]:
self.state["provider"]["state"] = "ACTIVE"
elif words[:4] == ["iam", "workload-identity-pools", "providers", "update-oidc"]:
self.state["provider"].update(
{
@ -315,6 +329,12 @@ def test_clean_create_converges_and_retains_sanitized_receipt(tmp_path: Path) ->
assert receipt["credentials_logged"] is False
assert len(apply.PROVIDER_DISPLAY_NAME) <= 32
assert apply.PROVIDER_DISPLAY_NAME == "Teleo fixed IAP operator"
provider_lists = [
command
for command in fake.commands
if command[1:5] == ["iam", "workload-identity-pools", "providers", "list"]
]
assert provider_lists == []
assert json.loads((tmp_path / "receipt.json").read_text())["status"] == "applied"
@ -327,6 +347,43 @@ def test_fully_existing_state_is_control_plane_noop(tmp_path: Path) -> None:
assert fake.mutations == []
def test_exact_sixth_mutation_round_can_converge(tmp_path: Path, monkeypatch) -> None:
discovery_round = 0
def fake_discover(_client):
nonlocal discovery_round
state = {
"round": discovery_round,
"public_firewall": {
"name": apply.CHANGING_PUBLIC_RULE,
"disabled": False,
"targetTags": ["teleo-public-ssh-bootstrap"],
},
}
discovery_round += 1
return state
def fake_plan(state):
if state["round"] >= 6:
return []
return [apply.Mutation(f"round_{state['round']}", "update", ["gcloud", "noop"])]
monkeypatch.setattr(apply, "discover", fake_discover)
monkeypatch.setattr(apply, "plan_mutations", fake_plan)
monkeypatch.setattr(apply, "_run_mutations", lambda _client, _mutations: None)
monkeypatch.setattr(apply, "converge_dispatcher", lambda _client, _path: {"status": "unchanged"})
receipt, exitcode = apply.apply_operator_access(
execute=True,
output=tmp_path / "receipt.json",
runner=lambda command, **_: subprocess.CompletedProcess(command, 0, "", ""),
)
assert exitcode == 0
assert receipt["status"] == "applied"
assert discovery_round == 7
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}"
@ -377,6 +434,25 @@ def test_disabled_pool_and_provider_are_narrowly_reenabled(tmp_path: Path) -> No
assert fake.state["provider"]["disabled"] is False
def test_soft_deleted_pool_and_provider_are_undeleted_and_converged(tmp_path: Path) -> None:
state = _full_state()
state["pool"]["state"] = "DELETED"
state["provider"]["state"] = "DELETED"
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 "undelete_wif_pool" in mutation_names
assert "undelete_wif_provider" in mutation_names
assert "create_wif_pool" not in mutation_names
assert "create_wif_provider" not in mutation_names
assert fake.state["pool"]["state"] == "ACTIVE"
assert fake.state["provider"]["state"] == "ACTIVE"
def test_wrong_or_stopped_instance_fails_before_mutation(tmp_path: Path) -> None:
state = _full_state()
state["instance"]["status"] = "TERMINATED"
@ -389,6 +465,18 @@ def test_wrong_or_stopped_instance_fails_before_mutation(tmp_path: Path) -> None
assert fake.mutations == []
def test_missing_os_login_fails_before_mutation(tmp_path: Path) -> None:
state = _full_state()
state["instance"]["metadata"] = {"items": []}
fake, receipt, exitcode = _run(tmp_path, state)
assert exitcode == 1
assert receipt["status"] == "failed"
assert "refusing to change the SSH authentication mode" 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())
@ -397,6 +485,9 @@ def test_bootstrap_never_disables_or_mutates_public_ssh_rule(tmp_path: Path) ->
"name": apply.CHANGING_PUBLIC_RULE,
"observed_disabled_before": False,
"observed_disabled_after": False,
"target_tags_before": ["teleo-public-ssh-bootstrap"],
"target_tags_after": ["teleo-public-ssh-bootstrap"],
"effective_target_tag_expansion": False,
"mutated": False,
"disable_command_present": False,
}
@ -407,3 +498,15 @@ def test_bootstrap_never_disables_or_mutates_public_ssh_rule(tmp_path: Path) ->
and command[3] != "describe"
for command in fake.commands
)
def test_public_rule_tag_collision_fails_before_mutation(tmp_path: Path) -> None:
state = _clean_state()
state["public_firewall"]["targetTags"] = [apply.TARGET_TAG]
fake, receipt, exitcode = _run(tmp_path, state)
assert exitcode == 1
assert receipt["status"] == "failed"
assert "would newly match the existing public SSH firewall rule" in receipt["error"]
assert fake.mutations == []

43
tests/test_gcp_iap_operator_access.py
View file

@ -3,6 +3,7 @@ import os
import subprocess
from pathlib import Path
import pytest
import yaml
REPO_ROOT = Path(__file__).resolve().parents[1]
@ -120,7 +121,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_verified_short_retention() -> None:
def test_workflow_uses_wif_official_actions_and_sanitized_short_retention() -> None:
workflow = load_workflow()
assert workflow["permissions"] == {"contents": "read", "id-token": "write"}
steps = workflow["jobs"]["operate"]["steps"]
@ -136,7 +137,7 @@ def test_workflow_uses_wif_official_actions_and_verified_short_retention() -> No
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 }}/"
assert upload["with"]["path"] == "${{ env.RESULT_DIR }}/result.json"
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")
@ -225,6 +226,13 @@ 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 '[[ "$exists" == "t" || "$exists" == "f" ]]' in source
assert '[[ ! -e "$run_dir" && ! -L "$run_dir" ]]' in source
assert 'temp_marker="$temp_path/.teleo-temp-owner"' in source
assert '[[ "$temp_marker_state" == "root:root:600" ]]' in source
assert '[[ "$temp_marker_content" == "$request_id"$\'\\n\'"gcp-direct-claim" ]]' in source
assert source.index('rm -rf --one-file-system -- "$legacy_dir"') < source.index('drop database :"target_db";')
assert source.index('drop database :"target_db";') < source.index('rm -rf --one-file-system -- "$run_dir"')
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
@ -242,6 +250,11 @@ def test_operator_exports_and_fixed_validates_full_replay_before_cleanup() -> No
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 '[[ ! -e "$full_result" && ! -L "$full_result" ]]' in source
assert 'rm -f -- "$full_result"' in source
assert source.index('validate_downloaded_replay "$result_path" "$full_result" "$target_db"') < source.index(
'rm -f -- "$full_result"'
)
assert '"strict_six_of_six"' in source
assert '"fixed_fingerprint_before"' in source
assert '"fixed_counts_after"' in source
@ -358,7 +371,8 @@ def test_runner_fails_closed_when_remote_output_is_not_valid_json(tmp_path: Path
assert not list(runner_temp.glob("teleo-iap-operator.*"))
def test_clone_runner_scps_only_request_derived_bundle_over_iap(tmp_path: Path) -> None:
@pytest.mark.parametrize("valid_receipt", [True, False], ids=["valid", "invalid"])
def test_clone_runner_scps_only_request_derived_bundle_over_iap(tmp_path: Path, valid_receipt: bool) -> None:
bin_dir = tmp_path / "bin"
runner_temp = tmp_path / "runner"
result_dir = runner_temp / "result"
@ -372,6 +386,8 @@ def test_clone_runner_scps_only_request_derived_bundle_over_iap(tmp_path: Path)
bundle = bundle_dir / "iap-abcdefghijkl.tar.gz"
bundle.write_bytes(b"reviewed-bundle")
bundle.chmod(0o600)
expected_fingerprint = "48ea5332c4f303dea02d503447f1ac8e1d8aa00e60d198a566d628b37dbdac1c"
observed_fingerprint = expected_fingerprint if valid_receipt else "0" * 64
full_receipt.write_text(
json.dumps(
{
@ -380,10 +396,10 @@ def test_clone_runner_scps_only_request_derived_bundle_over_iap(tmp_path: Path)
"score": {"passes": 6, "expected_prompt_count": 6},
"checks": {"all_runtime_checks": True},
"database_fingerprint_before": {
"sha256": "48ea5332c4f303dea02d503447f1ac8e1d8aa00e60d198a566d628b37dbdac1c"
"sha256": observed_fingerprint
},
"database_fingerprint_after": {
"sha256": "48ea5332c4f303dea02d503447f1ac8e1d8aa00e60d198a566d628b37dbdac1c"
"sha256": observed_fingerprint
},
"canonical_status_before": {
"high_signal_rows": {
@ -461,7 +477,7 @@ def test_clone_runner_scps_only_request_derived_bundle_over_iap(tmp_path: Path)
check=False,
)
assert completed.returncode == 0, completed.stderr
assert completed.returncode == (0 if valid_receipt else 1), completed.stderr
calls = invocation.read_text(encoding="utf-8").splitlines()
assert len(calls) == 4
assert "compute scp" in calls[0]
@ -477,12 +493,17 @@ def test_clone_runner_scps_only_request_derived_bundle_over_iap(tmp_path: Path)
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["status"] == ("pass" if valid_receipt else "fail")
assert result["full_result_exported_and_validated"] is valid_receipt
assert result["full_result_retained"] is False
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
if valid_receipt:
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
else:
assert result["full_result_export_failure"] == "receipt_download_validation_or_remote_export_cleanup_failed"
assert not (result_dir / "direct-claim-result.json").exists()
def test_clone_runner_rejects_bundle_symlink_before_gcloud(tmp_path: Path) -> None:

23
tests/test_run_leo_clone_bound_handler_checkpoint.py
View file

@ -695,6 +695,9 @@ def test_temp_copy_excludes_sensitive_and_state_artifacts_without_readback(live_
)
root = temp_profile.parent
try:
owner_marker = root / ".teleo-temp-owner"
assert owner_marker.read_text(encoding="utf-8") == "copy-test\nCB-COPY\n"
assert owner_marker.stat().st_mode & 0o777 == 0o600
assert audit["passes"] is True
assert audit["secret_contents_read_or_recorded"] is False
assert audit["omitted_counts_by_category"]
@ -709,6 +712,26 @@ def test_temp_copy_excludes_sensitive_and_state_artifacts_without_readback(live_
assert checkpoint.temp_path_absent(root) is True
def test_temp_owner_marker_failure_removes_registered_root(
live_profile: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
original_write_text = Path.write_text
def fail_owner_marker(path: Path, *args: Any, **kwargs: Any) -> int:
if path.name == ".teleo-temp-owner":
raise OSError("marker write failed")
return original_write_text(path, *args, **kwargs)
monkeypatch.setattr(Path, "write_text", fail_owner_marker)
with pytest.raises(OSError, match="marker write failed"):
checkpoint.copy_profile(run_id="copy-fail", prompt_id="CB-COPY", live_profile=live_profile)
assert not list((live_profile.parent / "temp-roots").iterdir())
assert not checkpoint._ACTIVE_TEMP_ROOTS
def test_patch_temp_bridge_rejects_preexisting_tool_log(live_profile: Path) -> None:
temp_profile, _audit = checkpoint.copy_profile(
run_id="stale-log-test",