teleo-infrastructure/tests/test_gcp_operator_reauth.py

import os
import re
import shutil
import subprocess
import sys
from pathlib import Path

import pytest

REPO_ROOT = Path(__file__).resolve().parents[1]
SCRIPT = REPO_ROOT / "scripts" / "gcp-operator-reauth.sh"
DOC = REPO_ROOT / "docs" / "gcp-operator-reauth.md"
FAKE_SECRET = "fake password%value"


def write_executable(path: Path, source: str) -> None:
    path.write_text(source, encoding="utf-8")
    path.chmod(0o755)


def fake_operator_env(tmp_path: Path, *, token_ready: bool = True) -> tuple[dict[str, str], Path, Path, Path]:
    bin_dir = tmp_path / "bin"
    bin_dir.mkdir()
    calls = tmp_path / "gcloud-calls.txt"
    helper_calls = tmp_path / "keychain-helper-calls.txt"
    keychain_state = tmp_path / "keychain-state"

    write_executable(
        bin_dir / "gcloud",
        """#!/usr/bin/env bash
set -eu
printf '%s\\n' "$*" >> "$FAKE_GCLOUD_CALLS"
case "$1:$2:${3:-}" in
  config:get-value:account) printf '%s\\n' 'billy@livingip.xyz' ;;
  config:get-value:project) printf '%s\\n' 'teleo-501523' ;;
  auth:print-access-token:*)
    [[ "${FAKE_GCLOUD_TOKEN_READY:-0}" == "1" ]] || exit 1
    printf '%s\\n' 'fake-token-that-must-be-discarded'
    ;;
  compute:instances:describe) printf 'teleo-prod-1\\tRUNNING\\t10.60.0.3\\n' ;;
  compute:ssh:teleo-prod-1) exit 0 ;;
  *) exit 90 ;;
esac
""",
    )
    write_executable(bin_dir / "ssh", "#!/usr/bin/env bash\nexit 0\n")
    write_executable(
        bin_dir / "pinentry-mac",
        """#!/usr/bin/env bash
set -eu
cat >/dev/null
printf '%s\\n' 'OK fake-pinentry' 'D fake%20password%25value' 'OK'
""",
    )
    write_executable(
        bin_dir / "keychain-helper",
        """#!/usr/bin/env bash
set -eu
printf '%s\\n' "$*" >> "$FAKE_KEYCHAIN_HELPER_CALLS"
action="$1"
case "$action" in
  status)
    if [[ -s "$FAKE_KEYCHAIN_STATE" ]]; then printf 'present\\n'; else printf 'absent\\n'; fi
    ;;
  store-pinentry)
    response="$(cat)"
    [[ "$response" == *'D fake%20password%25value'* ]]
    printf '%s' 'fake password%value' > "$FAKE_KEYCHAIN_STATE"
    printf 'stored\\n'
    ;;
  prompt-store)
    printf '%s' 'fake AppKit password' > "$FAKE_KEYCHAIN_STATE"
    printf 'stored\\n'
    ;;
  *) exit 91 ;;
esac
""",
    )

    env = {
        **os.environ,
        "PATH": f"{bin_dir}:{os.environ['PATH']}",
        "FAKE_GCLOUD_CALLS": str(calls),
        "FAKE_GCLOUD_TOKEN_READY": "1" if token_ready else "0",
        "FAKE_KEYCHAIN_HELPER_CALLS": str(helper_calls),
        "FAKE_KEYCHAIN_STATE": str(keychain_state),
        "TELEO_GCP_REAUTH_TESTING": "1",
        "TELEO_GCP_REAUTH_KEYCHAIN_HELPER": str(bin_dir / "keychain-helper"),
        "TELEO_GCP_REAUTH_PINENTRY": str(bin_dir / "pinentry-mac"),
    }
    return env, calls, helper_calls, keychain_state


def run_script(env: dict[str, str], *args: str, input_text: str | None = None) -> subprocess.CompletedProcess[str]:
    return subprocess.run(
        ["bash", str(SCRIPT), *args],
        cwd=REPO_ROOT,
        env=env,
        input=input_text,
        text=True,
        capture_output=True,
        check=False,
    )


def test_fake_pinentry_to_keychain_to_ready_status_lifecycle(tmp_path: Path) -> None:
    env, calls_path, helper_calls_path, state_path = fake_operator_env(tmp_path)

    completed = run_script(env, "--store-password")

    assert completed.returncode == 0, completed.stderr
    assert completed.stdout.count("security_truth=") == 1
    assert "secure_prompt=pinentry-mac" in completed.stdout
    assert "keychain_password=stored" in completed.stdout
    assert "keychain_password=present" in completed.stdout
    assert "oauth_state=ready" in completed.stdout
    assert "instance_readback=teleo-prod-1 RUNNING 10.60.0.3" in completed.stdout
    assert "iap_ssh_preflight=ready_dry_run_only" in completed.stdout
    assert "iap_connection=not_attempted" in completed.stdout
    assert "operator_readiness=ready_for_explicit_operator_action" in completed.stdout
    assert state_path.read_text(encoding="utf-8") == FAKE_SECRET

    combined_output = completed.stdout + completed.stderr
    calls = calls_path.read_text(encoding="utf-8")
    helper_calls = helper_calls_path.read_text(encoding="utf-8")
    assert FAKE_SECRET not in combined_output
    assert "fake-token-that-must-be-discarded" not in combined_output
    assert FAKE_SECRET not in calls
    assert FAKE_SECRET not in helper_calls
    assert "auth print-access-token --account=billy@livingip.xyz" in calls
    assert "compute instances describe teleo-prod-1" in calls
    assert "compute ssh teleo-prod-1" in calls
    assert "--tunnel-through-iap" in calls
    assert "--ssh-key-expire-after=5m" in calls
    assert "--dry-run" in calls


def test_default_interactive_mode_uses_the_secure_popup_path(tmp_path: Path) -> None:
    env, _, _, state_path = fake_operator_env(tmp_path)

    completed = run_script(env, input_text="yes\n")

    assert completed.returncode == 0, completed.stderr
    assert "Store or update" in completed.stderr
    assert "secure_prompt=pinentry-mac" in completed.stdout
    assert state_path.read_text(encoding="utf-8") == FAKE_SECRET
    assert FAKE_SECRET not in completed.stdout + completed.stderr


def test_appkit_secure_field_fallback_runs_when_pinentry_is_absent(tmp_path: Path) -> None:
    env, _, _, state_path = fake_operator_env(tmp_path)
    pinentry = Path(env.pop("TELEO_GCP_REAUTH_PINENTRY"))
    pinentry.unlink()
    env["PATH"] = f"{pinentry.parent}:/usr/bin:/bin"

    completed = run_script(env, "--store-password")

    assert completed.returncode == 0, completed.stderr
    assert "secure_prompt=AppKit NSSecureTextField" in completed.stdout
    assert state_path.read_text(encoding="utf-8") == "fake AppKit password"
    assert "fake AppKit password" not in completed.stdout + completed.stderr


def test_status_does_not_open_a_popup_or_read_a_password(tmp_path: Path) -> None:
    env, _, helper_calls_path, state_path = fake_operator_env(tmp_path)
    state_path.write_text(FAKE_SECRET, encoding="utf-8")

    completed = run_script(env, "--status")

    assert completed.returncode == 0, completed.stderr
    assert "keychain_password=present" in completed.stdout
    assert "secure_prompt=" not in completed.stdout
    assert helper_calls_path.read_text(encoding="utf-8").splitlines() == [
        "status com.livingip.teleo.gcp-operator-reauth billy@livingip.xyz Teleo GCP operator emergency password"
    ]
    assert FAKE_SECRET not in completed.stdout + completed.stderr


def test_stale_oauth_emits_one_exact_nonlaunching_cta_and_stops(tmp_path: Path) -> None:
    env, calls_path, _, _ = fake_operator_env(tmp_path, token_ready=False)

    completed = run_script(env, "--status")

    assert completed.returncode == 1
    assert "oauth_state=stale_or_unavailable" in completed.stdout
    assert completed.stdout.count("clear_CTA=") == 1
    assert "gcloud auth login billy@livingip.xyz --force --no-launch-browser" in completed.stdout
    assert "dedicated Chrome GCP session" in completed.stdout
    assert "scripts/gcp-operator-reauth.sh --status" in completed.stdout
    assert "operator_readiness=not_ready" in completed.stdout
    calls = calls_path.read_text(encoding="utf-8")
    assert "compute instances describe" not in calls
    assert "compute ssh" not in calls


def test_source_enforces_secret_and_gui_boundaries() -> None:
    source = SCRIPT.read_text(encoding="utf-8")

    assert "NSAlert" in source
    assert "NSSecureTextField" in source
    assert "app.activate(ignoringOtherApps: true)" in source
    assert 'action: #selector(NSText.paste(_:)), keyEquivalent: "v"' in source
    assert "field.menu = fieldMenu" in source
    assert "Empty values are never stored." in source
    assert "SecItemAdd" in source
    assert "SecItemUpdate" in source
    assert "kSecAttrAccessibleWhenUnlockedThisDeviceOnly" in source
    assert "kSecAttrSynchronizable" in source
    assert "store-pinentry" in source
    assert "FileHandle.standardInput.readDataToEndOfFile()" in source
    assert "cannot satisfy or automate gcloud OAuth" in source
    assert "ready_dry_run_only" in source
    assert "iap_connection=not_attempted" in source
    for forbidden in (
        "osascript",
        "AppleScript",
        "System Events",
        "NSPasteboard",
        "pbcopy",
        "open -a",
        "open -n",
        "security add-generic-password",
    ):
        assert forbidden not in source
    assert not re.search(r"gcloud auth login.*(?:^|\n)\s*gcloud auth login", source)


def test_shell_syntax_and_operator_documentation() -> None:
    subprocess.run(["bash", "-n", str(SCRIPT)], cwd=REPO_ROOT, check=True)
    documentation = DOC.read_text(encoding="utf-8")
    assert "scripts/gcp-operator-reauth.sh --status" in documentation
    assert "scripts/gcp-operator-reauth.sh --store-password" in documentation
    assert "cannot refresh or satisfy gcloud OAuth by itself" in documentation
    assert ".github/workflows/gcp-iap-operator.yml" in documentation
    assert "ready_dry_run_only" in documentation


@pytest.mark.skipif(sys.platform != "darwin" or shutil.which("swiftc") is None, reason="macOS Swift toolchain required")
def test_embedded_swift_helper_typechecks(tmp_path: Path) -> None:
    source = SCRIPT.read_text(encoding="utf-8")
    match = re.search(r"<<'SWIFT'\n(?P<swift>.*?)\nSWIFT", source, flags=re.DOTALL)
    assert match is not None
    swift_source = tmp_path / "GcpOperatorKeychain.swift"
    swift_source.write_text(match.group("swift"), encoding="utf-8")

    subprocess.run(
        ["swiftc", "-typecheck", "-framework", "AppKit", "-framework", "Security", str(swift_source)],
        cwd=REPO_ROOT,
        check=True,
    )