From cb49ce0e6aa424252c971257e98483983c9b5009 Mon Sep 17 00:00:00 2001 From: twentyOne2x Date: Mon, 13 Jul 2026 05:24:25 +0200 Subject: [PATCH] Make GCP replay bootstrap recoverable --- .github/workflows/gcp-iap-operator.yml | 4 +- ops/apply_gcp_iap_operator_access.py | 157 +++++++++++++++--- scripts/gcp_iap_operator.sh | 135 ++++++++++----- .../run_leo_clone_bound_handler_checkpoint.py | 7 +- tests/test_apply_gcp_iap_operator_access.py | 131 +++++++++++++-- tests/test_gcp_iap_operator_access.py | 43 +++-- ..._run_leo_clone_bound_handler_checkpoint.py | 23 +++ 7 files changed, 404 insertions(+), 96 deletions(-) diff --git a/.github/workflows/gcp-iap-operator.yml b/.github/workflows/gcp-iap-operator.yml index f72fa77..49c58c6 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 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 diff --git a/ops/apply_gcp_iap_operator_access.py b/ops/apply_gcp_iap_operator_access.py index 61a8fc2..883b281 100644 --- a/ops/apply_gcp_iap_operator_access.py +++ b/ops/apply_gcp_iap_operator_access.py @@ -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, }, diff --git a/scripts/gcp_iap_operator.sh b/scripts/gcp_iap_operator.sh index 28d7a41..7f0e816 100755 --- a/scripts/gcp_iap_operator.sh +++ b/scripts/gcp_iap_operator.sh @@ -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 diff --git a/scripts/run_leo_clone_bound_handler_checkpoint.py b/scripts/run_leo_clone_bound_handler_checkpoint.py index 4e00db8..ad0483b 100644 --- a/scripts/run_leo_clone_bound_handler_checkpoint.py +++ b/scripts/run_leo_clone_bound_handler_checkpoint.py @@ -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"]: diff --git a/tests/test_apply_gcp_iap_operator_access.py b/tests/test_apply_gcp_iap_operator_access.py index 56c94f9..e2f3f27 100644 --- a/tests/test_apply_gcp_iap_operator_access.py +++ b/tests/test_apply_gcp_iap_operator_access.py @@ -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 == [] diff --git a/tests/test_gcp_iap_operator_access.py b/tests/test_gcp_iap_operator_access.py index 1a2f2f8..ef01fb4 100644 --- a/tests/test_gcp_iap_operator_access.py +++ b/tests/test_gcp_iap_operator_access.py @@ -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: diff --git a/tests/test_run_leo_clone_bound_handler_checkpoint.py b/tests/test_run_leo_clone_bound_handler_checkpoint.py index 36d7930..291f508 100644 --- a/tests/test_run_leo_clone_bound_handler_checkpoint.py +++ b/tests/test_run_leo_clone_bound_handler_checkpoint.py @@ -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",