550 lines
24 KiB
Python
550 lines
24 KiB
Python
import json
|
|
import os
|
|
import subprocess
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
import yaml
|
|
|
|
REPO_ROOT = Path(__file__).resolve().parents[1]
|
|
PLAN = REPO_ROOT / "ops" / "plan_gcp_iap_operator_access.py"
|
|
WORKFLOW = REPO_ROOT / ".github" / "workflows" / "gcp-iap-operator.yml"
|
|
OPERATOR = REPO_ROOT / "scripts" / "gcp_iap_operator.sh"
|
|
RESULT_HELPER = REPO_ROOT / "scripts" / "gcp_iap_operator_result.py"
|
|
|
|
|
|
def run_plan(*args: str) -> str:
|
|
return subprocess.check_output(["python3", str(PLAN), *args], cwd=REPO_ROOT, text=True)
|
|
|
|
|
|
def test_plan_is_emit_only_and_pins_exact_github_identity() -> None:
|
|
plan = json.loads(run_plan())
|
|
|
|
assert plan["mode"] == "dry_run_emit_only"
|
|
assert plan["executes_commands"] is False
|
|
condition = plan["wif"]["provider_condition"]
|
|
assert "assertion.repository == 'living-ip/teleo-infrastructure'" in condition
|
|
assert "assertion.ref == 'refs/heads/main'" in condition
|
|
assert (
|
|
"assertion.workflow_ref == "
|
|
"'living-ip/teleo-infrastructure/.github/workflows/gcp-iap-operator.yml@refs/heads/main'"
|
|
) in condition
|
|
assert "assertion.event_name == 'workflow_dispatch'" in condition
|
|
assert plan["wif"]["long_lived_google_credentials"] is False
|
|
|
|
|
|
def test_plan_uses_exact_iap_firewall_and_least_privilege_split() -> None:
|
|
plan = json.loads(run_plan())
|
|
commands = "\n".join(plan["bootstrap_commands"])
|
|
|
|
assert plan["target"]["internal_ip"] == "10.60.0.3"
|
|
assert plan["iap"]["condition"] == 'destination.ip == "10.60.0.3" && destination.port == 22'
|
|
assert plan["iap"]["firewall"] == {
|
|
"protocol_ports": "tcp:22",
|
|
"rule": "teleo-prod-allow-ssh-iap",
|
|
"source_range": "35.235.240.0/20",
|
|
"target_tag": "teleo-prod-ssh",
|
|
}
|
|
assert plan["gcloud_ssh_custom_role"]["permissions"] == [
|
|
"compute.instances.get",
|
|
"compute.instances.list",
|
|
"compute.projects.get",
|
|
]
|
|
assert plan["identities"]["status"]["os_login_role"] == "roles/compute.osLogin"
|
|
assert plan["identities"]["status"]["sudo"] is False
|
|
assert plan["identities"]["clone_operator"]["os_login_role"] == "roles/compute.osAdminLogin"
|
|
assert plan["identities"]["clone_operator"]["operations"] == [
|
|
"direct-claim-replay",
|
|
"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
|
|
assert "teleo-gcp-iap-operator-v1" in commands
|
|
for role in plan["explicitly_absent"]["roles"]:
|
|
assert f"--role {role}" not in commands
|
|
assert "compute.instances.setMetadata" not in plan["gcloud_ssh_custom_role"]["permissions"]
|
|
|
|
|
|
def test_public_rule_is_disabled_only_after_workflow_status_verification() -> None:
|
|
plan = json.loads(run_plan())
|
|
verification = plan["post_bootstrap_verification"]
|
|
joined = "\n".join(verification)
|
|
|
|
dispatch_index = next(i for i, command in enumerate(verification) if "gh workflow run" in command)
|
|
disable_index = next(
|
|
i
|
|
for i, command in enumerate(verification)
|
|
if "teleo-prod-allow-ssh-current-ip" in command and "--disabled" in command
|
|
)
|
|
assert dispatch_index < disable_index
|
|
assert "operation=status" in joined
|
|
assert "target_db=teleo_clone_status" in joined
|
|
assert "dispatcher_sha256" in joined
|
|
assert 'Path("scripts/gcp_iap_operator.sh")' in joined
|
|
assert plan["public_ingress_cutover"]["automatic_fallback_to_public_ssh"] is False
|
|
assert "DATA_READ for iap.googleapis.com" in plan["audit_logging"]["guidance"]
|
|
assert plan["revocation_commands"]
|
|
|
|
|
|
def test_shell_plan_is_clearly_dry_run() -> None:
|
|
shell = run_plan("--format", "shell")
|
|
|
|
assert shell.startswith("# DRY RUN:")
|
|
assert "# bootstrap_commands" in shell
|
|
assert "# post_bootstrap_verification" in shell
|
|
assert "# revocation_commands" in shell
|
|
assert "subprocess" not in PLAN.read_text(encoding="utf-8")
|
|
|
|
|
|
def load_workflow() -> dict[str, object]:
|
|
return yaml.load(WORKFLOW.read_text(encoding="utf-8"), Loader=yaml.BaseLoader)
|
|
|
|
|
|
def test_workflow_accepts_only_fixed_operation_and_bounded_ids() -> None:
|
|
workflow = load_workflow()
|
|
inputs = workflow["on"]["workflow_dispatch"]["inputs"]
|
|
|
|
assert set(inputs) == {"operation", "request_id", "target_db"}
|
|
assert inputs["operation"]["options"] == ["status", "direct-claim-replay", "cleanup-clone"]
|
|
assert inputs["target_db"]["default"] == "teleo_clone_status"
|
|
source = WORKFLOW.read_text(encoding="utf-8")
|
|
assert '[[ "${GITHUB_REF}" == "refs/heads/main" ]]' in source
|
|
assert '[[ "${GITHUB_WORKFLOW_REF}" == "${EXPECTED_WORKFLOW_REF}" ]]' in source
|
|
assert "request_id_re='^iap-[a-z0-9]{12,32}$'" in source
|
|
assert "target_db_re='^teleo_clone_[a-z0-9][a-z0-9_]{0,50}$'" in source
|
|
for forbidden_input in ("command:", "path:", "host:", "project:", "zone:", "ip:", "ref:", "service_account:"):
|
|
assert forbidden_input not in inputs
|
|
|
|
|
|
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"]
|
|
actions = [step["uses"] for step in steps if "uses" in step]
|
|
|
|
assert actions == [
|
|
"actions/checkout@v4",
|
|
"google-github-actions/auth@v3",
|
|
"google-github-actions/setup-gcloud@v3",
|
|
"actions/upload-artifact@v4",
|
|
]
|
|
auth = next(step for step in steps if step.get("uses") == "google-github-actions/auth@v3")
|
|
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"]["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")
|
|
assert validate_index < auth_index
|
|
|
|
|
|
def test_workflow_initializes_runner_paths_at_step_runtime() -> None:
|
|
workflow = load_workflow()
|
|
assert "BUNDLE_DIR" not in workflow["env"]
|
|
assert "RESULT_DIR" not in workflow["env"]
|
|
steps = workflow["jobs"]["operate"]["steps"]
|
|
paths_index = next(i for i, step in enumerate(steps) if step.get("id") == "paths")
|
|
validate_index = next(i for i, step in enumerate(steps) if step.get("id") == "validate")
|
|
paths_source = steps[paths_index]["run"]
|
|
assert paths_index < validate_index
|
|
assert '"${RUNNER_TEMP}/gcp-iap-operator-bundle" >> "${GITHUB_ENV}"' in paths_source
|
|
assert '"${RUNNER_TEMP}/gcp-iap-operator-result" >> "${GITHUB_ENV}"' in paths_source
|
|
assert "runner.temp" not in WORKFLOW.read_text(encoding="utf-8")
|
|
|
|
|
|
def test_workflow_retains_sanitized_failure_receipt_before_upload() -> None:
|
|
workflow = load_workflow()
|
|
steps = workflow["jobs"]["operate"]["steps"]
|
|
receipt_index = next(i for i, step in enumerate(steps) if step.get("name") == "Ensure sanitized result receipt")
|
|
upload_index = next(i for i, step in enumerate(steps) if step.get("uses") == "actions/upload-artifact@v4")
|
|
receipt = steps[receipt_index]
|
|
source = receipt["run"]
|
|
assert receipt_index < upload_index
|
|
assert receipt["if"] == "always()"
|
|
assert receipt["env"]["AUTH_OUTCOME"] == "${{ steps.auth.outcome }}"
|
|
assert receipt["env"]["OPERATE_OUTCOME"] == "${{ steps.operate.outcome }}"
|
|
assert "workflow_stopped_before_valid_operator_result" in source
|
|
assert '"credential_values_logged": False' in source
|
|
assert '"raw_stdout_retained": False' in source
|
|
assert '"raw_stderr_retained": False' in source
|
|
|
|
|
|
def test_workflow_builds_exact_main_bundle_from_tracked_files_and_hashes() -> None:
|
|
source = WORKFLOW.read_text(encoding="utf-8")
|
|
|
|
expected_files = [
|
|
"scripts/gcp_iap_operator.sh",
|
|
"scripts/run_gcp_generated_db_direct_claim_suite.py",
|
|
"scripts/run_leo_clone_bound_handler_checkpoint.py",
|
|
"scripts/working_leo_open_ended_benchmark.py",
|
|
"hermes-agent/leoclean-bin/cloudsql_memory_tool.py",
|
|
"ops/postgres_parity_manifest.sql",
|
|
]
|
|
for relative in expected_files:
|
|
assert f'"{relative}"' in source
|
|
assert 'f"docs/reports/leo-working-state-20260709/{target_db}-canonical-parity-receipt.json"' in source
|
|
assert '["git", "ls-files", "--error-unmatch", "--", relative]' in source
|
|
assert 'if head != os.environ["GITHUB_SHA"]' in source
|
|
assert "hashlib.sha256(data).hexdigest()" in source
|
|
assert '"schema": "livingip.gcpIapOperatorBundle.v1"' in source
|
|
assert 'bundle_path = bundle_dir / f"{request_id}.tar.gz"' in source
|
|
|
|
|
|
def test_operator_allowlists_commands_and_has_no_live_mutation_surface() -> None:
|
|
source = OPERATOR.read_text(encoding="utf-8")
|
|
|
|
assert "status | direct-claim-replay | cleanup-clone" in source
|
|
assert "scripts/run_gcp_generated_db_direct_claim_suite.py" in source
|
|
assert "six_replies_returned" in source
|
|
assert "posted_to_telegram" in source
|
|
assert "live_service_unchanged" in source
|
|
assert "live_profile_unchanged" in source
|
|
assert "%s-canonical-parity-receipt.json" in source
|
|
assert "/opt/teleo-eval" not in source
|
|
assert "systemctl restart" not in source
|
|
assert "systemctl stop" not in source
|
|
assert "systemctl start" not in source
|
|
assert "curl " not in source
|
|
assert "--profile" not in source
|
|
assert "eval " not in source
|
|
|
|
|
|
def test_operator_cleanup_is_bound_to_marker_clone_prefix_and_exact_run_dir() -> None:
|
|
source = OPERATOR.read_text(encoding="utf-8")
|
|
|
|
assert "TARGET_DB_RE='^teleo_clone_" in source
|
|
assert 'expected="$REMOTE_RUN_ROOT/$request_id"' in source
|
|
assert '[[ "$resolved" == "$expected" ]]' in source
|
|
assert ".run-owner" in source
|
|
assert "run ownership marker does not match request_id and target_db" in source
|
|
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
|
|
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")
|
|
result_helper = RESULT_HELPER.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 '[[ ! -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 result_helper
|
|
assert '"fixed_fingerprint_before"' in result_helper
|
|
assert '"fixed_counts_after"' in result_helper
|
|
assert 'payload["remote_full_result_export_removed"] = True' in result_helper
|
|
|
|
|
|
def test_operator_uses_five_minute_key_private_modes_and_cleanup_trap() -> None:
|
|
source = OPERATOR.read_text(encoding="utf-8")
|
|
|
|
assert '--ssh-key-file="$ssh_key"' in source
|
|
assert "--ssh-key-expire-after=5m" in source
|
|
assert "--ssh-key-expiration" not in source
|
|
assert "chmod 0700" in source
|
|
assert "chmod 0600" in source
|
|
assert "trap cleanup_runner_temp EXIT" in source
|
|
assert "--tunnel-through-iap" in source
|
|
assert "gcloud compute scp" in source
|
|
assert "--verbosity=debug" not in source
|
|
|
|
|
|
def test_dispatcher_validates_exact_members_hashes_context_and_self_hash() -> None:
|
|
source = OPERATOR.read_text(encoding="utf-8")
|
|
|
|
assert 'expected_members = sorted([*expected, "bundle-manifest.json"])' in source
|
|
assert "bundle member allowlist mismatch" in source
|
|
assert "bundle manifest file allowlist mismatch" in source
|
|
assert 'receipt.get("sha256") != hashlib.sha256(data).hexdigest()' in source
|
|
assert '"workflow_ref": expected_workflow_ref' in source
|
|
assert "installed dispatcher does not match the reviewed main bundle" in source
|
|
assert 'dispatcher_sha != files["scripts/gcp_iap_operator.sh"]["sha256"]' in source
|
|
|
|
|
|
def test_runner_fake_gcloud_smoke_retains_only_sanitized_result(tmp_path: Path) -> None:
|
|
bin_dir = tmp_path / "bin"
|
|
runner_temp = tmp_path / "runner"
|
|
result_dir = runner_temp / "result"
|
|
invocation = tmp_path / "gcloud-argv.txt"
|
|
bin_dir.mkdir()
|
|
runner_temp.mkdir()
|
|
fake_gcloud = bin_dir / "gcloud"
|
|
fake_gcloud.write_text(
|
|
"#!/usr/bin/env bash\n"
|
|
'printf \'%s\\n\' "$@" > "${FAKE_GCLOUD_ARGV}"\n'
|
|
"printf '%s\\n' "
|
|
'\'{"operation":"status","request_id":"iap-abcdefghijkl",\''
|
|
'\'"target_db":"teleo_clone_status","status":"pass",\''
|
|
'\'"instance":"teleo-prod-1","expected_internal_ip":"10.60.0.3",\''
|
|
'\'"active_state":"active","sub_state":"running","nrestarts":0,\''
|
|
'\'"dispatcher_version":"teleo-gcp-iap-operator-v1",\''
|
|
'\'"dispatcher_sha256":"aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"}\'\n',
|
|
encoding="utf-8",
|
|
)
|
|
fake_gcloud.chmod(0o755)
|
|
env = {
|
|
**os.environ,
|
|
"PATH": f"{bin_dir}:{os.environ['PATH']}",
|
|
"RUNNER_TEMP": str(runner_temp),
|
|
"RESULT_DIR": str(result_dir),
|
|
"FAKE_GCLOUD_ARGV": str(invocation),
|
|
}
|
|
|
|
completed = subprocess.run(
|
|
["bash", str(OPERATOR), "status", "iap-abcdefghijkl", "teleo_clone_status"],
|
|
cwd=REPO_ROOT,
|
|
env=env,
|
|
text=True,
|
|
capture_output=True,
|
|
check=False,
|
|
)
|
|
|
|
assert completed.returncode == 0, completed.stderr
|
|
result = json.loads((result_dir / "result.json").read_text(encoding="utf-8"))
|
|
assert result["status"] == "pass"
|
|
assert result["credential_values_logged"] is False
|
|
assert result["raw_stdout_retained"] is False
|
|
assert result["raw_stderr_retained"] is False
|
|
assert result["remote_result"]["active_state"] == "active"
|
|
argv = invocation.read_text(encoding="utf-8")
|
|
assert "compute\nssh\nteleo-prod-1\n" in argv
|
|
assert "--tunnel-through-iap" in argv
|
|
assert "--ssh-key-expire-after=5m" in argv
|
|
assert not list(runner_temp.glob("teleo-iap-operator.*"))
|
|
|
|
|
|
def test_runner_fails_closed_when_remote_output_is_not_valid_json(tmp_path: Path) -> None:
|
|
bin_dir = tmp_path / "bin"
|
|
runner_temp = tmp_path / "runner"
|
|
result_dir = runner_temp / "result"
|
|
bin_dir.mkdir()
|
|
runner_temp.mkdir()
|
|
fake_gcloud = bin_dir / "gcloud"
|
|
fake_gcloud.write_text("#!/usr/bin/env bash\nprintf 'not-json\\n'\n", encoding="utf-8")
|
|
fake_gcloud.chmod(0o755)
|
|
env = {
|
|
**os.environ,
|
|
"PATH": f"{bin_dir}:{os.environ['PATH']}",
|
|
"RUNNER_TEMP": str(runner_temp),
|
|
"RESULT_DIR": str(result_dir),
|
|
}
|
|
|
|
completed = subprocess.run(
|
|
["bash", str(OPERATOR), "status", "iap-abcdefghijkl", "teleo_clone_status"],
|
|
cwd=REPO_ROOT,
|
|
env=env,
|
|
text=True,
|
|
capture_output=True,
|
|
check=False,
|
|
)
|
|
|
|
assert completed.returncode == 90
|
|
result = json.loads((result_dir / "result.json").read_text(encoding="utf-8"))
|
|
assert result["status"] == "fail"
|
|
assert result["remote_result"] == {}
|
|
assert not list(runner_temp.glob("teleo-iap-operator.*"))
|
|
|
|
|
|
@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"
|
|
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)
|
|
expected_fingerprint = "48ea5332c4f303dea02d503447f1ac8e1d8aa00e60d198a566d628b37dbdac1c"
|
|
observed_fingerprint = expected_fingerprint if valid_receipt else "0" * 64
|
|
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": observed_fingerprint},
|
|
"database_fingerprint_after": {"sha256": observed_fingerprint},
|
|
"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"
|
|
'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_cory_20260712t1940z","status":"pass",\''
|
|
'\'"bundle_commit":"aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",\''
|
|
'\'"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)
|
|
env = {
|
|
**os.environ,
|
|
"PATH": f"{bin_dir}:{os.environ['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_cory_20260712t1940z",
|
|
],
|
|
cwd=REPO_ROOT,
|
|
env=env,
|
|
text=True,
|
|
capture_output=True,
|
|
check=False,
|
|
)
|
|
|
|
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]
|
|
assert str(bundle) in calls[0]
|
|
assert "teleo-prod-1:.teleo-iap-upload-iap-abcdefghijkl.tar.gz" in calls[0]
|
|
assert "--tunnel-through-iap" in calls[0]
|
|
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" 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
|
|
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:
|
|
bin_dir = tmp_path / "bin"
|
|
runner_temp = tmp_path / "runner"
|
|
result_dir = runner_temp / "result"
|
|
bundle_dir = runner_temp / "gcp-iap-operator-bundle"
|
|
outside = tmp_path / "outside.tar.gz"
|
|
marker = tmp_path / "gcloud-called"
|
|
bin_dir.mkdir()
|
|
runner_temp.mkdir()
|
|
bundle_dir.mkdir(mode=0o700)
|
|
outside.write_bytes(b"outside")
|
|
(bundle_dir / "iap-abcdefghijkl.tar.gz").symlink_to(outside)
|
|
fake_gcloud = bin_dir / "gcloud"
|
|
fake_gcloud.write_text(
|
|
'#!/usr/bin/env bash\ntouch "${FAKE_GCLOUD_MARKER}"\n',
|
|
encoding="utf-8",
|
|
)
|
|
fake_gcloud.chmod(0o755)
|
|
env = {
|
|
**os.environ,
|
|
"PATH": f"{bin_dir}:{os.environ['PATH']}",
|
|
"RUNNER_TEMP": str(runner_temp),
|
|
"RESULT_DIR": str(result_dir),
|
|
"FAKE_GCLOUD_MARKER": str(marker),
|
|
}
|
|
|
|
completed = subprocess.run(
|
|
["bash", str(OPERATOR), "cleanup-clone", "iap-abcdefghijkl", "teleo_clone_test"],
|
|
cwd=REPO_ROOT,
|
|
env=env,
|
|
text=True,
|
|
capture_output=True,
|
|
check=False,
|
|
)
|
|
|
|
assert completed.returncode == 2
|
|
assert "fixed workflow bundle is absent or unsafe" in completed.stderr
|
|
assert not marker.exists()
|
|
|
|
|
|
def test_google_auth_credential_file_pattern_is_ignored_exactly() -> None:
|
|
ignore_lines = (REPO_ROOT / ".gitignore").read_text(encoding="utf-8").splitlines()
|
|
|
|
assert "gha-creds-*.json" in ignore_lines
|