ops-warden/tests/test_proxy.py

"""Tests for the access proxy lane (WP-0014 T3) and its three guardrails."""
from __future__ import annotations

import json
import subprocess
from pathlib import Path

import pytest
from typer.testing import CliRunner

from warden.cli import app
from warden.proxy import (
    ProxyError,
    ResolvedFetch,
    caller_auth_present,
    proxy_attended_login_exec,
    proxy_exec,
    proxy_fetch,
    resolve_fetch_command,
    write_audit,
)
from warden.routing.models import RouteEntry

runner = CliRunner()


def _entry(**over) -> RouteEntry:
    base = dict(
        id="openbao-api-key",
        title="API key",
        need_keywords=["npm", "token"],
        owner_repo="railiance-platform",
        subsystem="OpenBao",
        warden_executes=False,
        wiki_ref="w",
        canon_ref="c",
        reviewed="2026-06-27",
        status="active",
        path_template="platform/workloads/<domain>/<workload>/<bundle>",
        fetch_command="bao kv get -field=<FIELD> <path_template>",
        exec_capable=True,
    )
    base.update(over)
    return RouteEntry(**base)


# --- resolve_fetch_command -------------------------------------------------

def test_resolve_builds_argv():
    resolved = resolve_fetch_command(
        _entry(), domain="coulomb_social", field="NPM_AUTH_TOKEN", path="platform/x/y/z"
    )
    assert resolved.argv == ["bao", "kv", "get", "-field=NPM_AUTH_TOKEN", "platform/x/y/z"]
    assert resolved.shell_cmd is None


def test_resolve_refuses_unresolved_placeholder():
    # no --field / --path → <FIELD>, <workload>, <bundle> remain
    with pytest.raises(ProxyError, match="unresolved placeholder"):
        resolve_fetch_command(_entry(), domain="coulomb_social")


def test_resolve_refuses_non_exec_capable():
    with pytest.raises(ProxyError, match="not exec_capable"):
        resolve_fetch_command(_entry(exec_capable=False, fetch_command=None))


def test_resolve_bao_fetch_uses_argv():
    from warden.routing import load_catalog

    catalog = load_catalog(Path(__file__).resolve().parents[1] / "registry" / "routing" / "catalog.yaml")
    entry = catalog.get("reuse-surface-hub-write-token")
    resolved = resolve_fetch_command(entry)
    assert resolved.argv is not None
    assert resolved.shell_cmd is None
    assert resolved.argv[0] == "bao"
    assert "platform/workloads/reuse/reuse-surface/runtime-secrets" in resolved.argv


# --- G2: transit-only fetch (inherited stdout) -----------------------------

def test_proxy_fetch_inherits_stdout_never_pipes(monkeypatch):
    calls = {}

    def fake_run(argv, **kw):
        calls.update(kw)
        return subprocess.CompletedProcess(argv, 0)

    monkeypatch.setattr("warden.proxy.subprocess.run", fake_run)
    rc = proxy_fetch(ResolvedFetch(argv=["bao", "kv", "get", "x"]))
    assert rc == 0
    # The value must never enter warden's memory — stdout is inherited, not piped.
    assert calls["stdout"] is None
    assert calls.get("stderr") is None


def test_proxy_fetch_shell_pipeline_inherits_stdio(monkeypatch):
    calls = {}

    def fake_run(cmd, **kw):
        calls.update(kw)
        return subprocess.CompletedProcess(cmd, 0)

    monkeypatch.setattr("warden.proxy.subprocess.run", fake_run)
    rc = proxy_fetch(ResolvedFetch(shell_cmd="kubectl get x | base64 -d"))
    assert rc == 0
    assert calls["shell"] is True
    assert calls["stdout"] is None


# --- G1 + inject: exec injects value into child env, adds no warden token ---

def test_proxy_exec_injects_only_into_child_env(monkeypatch):
    seen_env = {}

    def fake_run(argv, **kw):
        if argv[0] == "bao":
            return subprocess.CompletedProcess(argv, 0, stdout="SECRETVAL\n")
        seen_env.update(kw["env"])
        return subprocess.CompletedProcess(argv, 0)

    monkeypatch.setattr("warden.proxy.subprocess.run", fake_run)
    monkeypatch.delenv("NPM_AUTH_TOKEN", raising=False)
    rc = proxy_exec(
        ResolvedFetch(argv=["bao", "kv", "get", "x"]),
        env_var="NPM_AUTH_TOKEN",
        child_argv=["true"],
    )
    assert rc == 0
    # Value injected into child env (trailing newline stripped)…
    assert seen_env["NPM_AUTH_TOKEN"] == "SECRETVAL"
    # …and warden added no credential of its own beyond the caller's environment.
    assert "VAULT_TOKEN" not in {k for k in seen_env if k not in __import__("os").environ}


def test_proxy_exec_shell_pipeline_captures_stdout(monkeypatch):
    seen_env = {}

    def fake_run(cmd, **kw):
        if kw.get("shell"):
            return subprocess.CompletedProcess(cmd, 0, stdout="PIPEVAL\n")
        seen_env.update(kw["env"])
        return subprocess.CompletedProcess(cmd, 0)

    monkeypatch.setattr("warden.proxy.subprocess.run", fake_run)
    rc = proxy_exec(
        ResolvedFetch(shell_cmd="kubectl get x | base64 -d"),
        env_var="REUSE_SURFACE_TOKEN",
        child_argv=["true"],
    )
    assert rc == 0
    assert seen_env["REUSE_SURFACE_TOKEN"] == "PIPEVAL"


def test_proxy_exec_requires_env_var():
    with pytest.raises(ProxyError, match="requires --field"):
        proxy_exec(ResolvedFetch(argv=["bao"]), env_var="", child_argv=["true"])


# --- G1 caller auth detection ----------------------------------------------

def test_caller_auth_present_from_env(monkeypatch):
    monkeypatch.setenv("VAULT_TOKEN", "x")
    assert caller_auth_present() is True


def test_caller_auth_absent(monkeypatch, tmp_path):
    monkeypatch.delenv("VAULT_TOKEN", raising=False)
    monkeypatch.delenv("BAO_TOKEN", raising=False)
    monkeypatch.setattr(Path, "home", lambda: tmp_path)  # no ~/.vault-token
    assert caller_auth_present() is False


# --- audit metadata only ---------------------------------------------------

def test_write_audit_has_no_value_field(tmp_path):
    p = write_audit(
        tmp_path, need_id="openbao-api-key", owner_repo="railiance-platform",
        domain="coulomb_social", action="fetch", decision_id=None,
    )
    rec = json.loads(p.read_text().strip())
    assert rec["need_id"] == "openbao-api-key"
    assert "value" not in rec and "secret" not in rec


# --- CLI guardrail wiring ---------------------------------------------------

def _repo_catalog() -> Path:
    return Path(__file__).resolve().parents[1] / "registry" / "routing" / "catalog.yaml"


def _warden_yaml(tmp_path: Path) -> Path:
    cfg = tmp_path / "warden.yaml"
    (tmp_path / "ca").write_text("")
    cfg.write_text(
        f"backend: local\nca_key: {tmp_path/'ca'}\nstate_dir: {tmp_path/'state'}\n"
    )
    return cfg


def _proxy_env(monkeypatch, tmp_path):
    monkeypatch.setenv("WARDEN_ROUTING_CATALOG", str(_repo_catalog()))
    monkeypatch.setenv("WARDEN_CONFIG", str(_warden_yaml(tmp_path)))


def test_cli_proxy_unknown_zone_fail_open_reaches_transport_guard(monkeypatch, tmp_path):
    _proxy_env(monkeypatch, tmp_path)
    monkeypatch.setenv("VAULT_TOKEN", "caller")
    # The unknown-zone profile proceeds when no evaluator is configured, then
    # the independent safe-transport boundary still refuses captured stdout.
    monkeypatch.setattr(
        "warden.proxy.subprocess.run",
        lambda *a, **k: (_ for _ in ()).throw(AssertionError("fetch ran despite gate")),
    )
    r = runner.invoke(
        app,
        ["access", "npm", "--domain", "coulomb_social", "--field", "NPM_AUTH_TOKEN",
         "--path", "platform/x/y/z", "--fetch"],
    )
    assert r.exit_code == 6
    assert "unknown-zone fail_open" in r.output


def test_cli_proxy_requires_caller_auth(monkeypatch, tmp_path):
    _proxy_env(monkeypatch, tmp_path)
    monkeypatch.delenv("VAULT_TOKEN", raising=False)
    monkeypatch.delenv("BAO_TOKEN", raising=False)
    monkeypatch.setattr(Path, "home", lambda: tmp_path)
    r = runner.invoke(
        app,
        ["access", "npm", "--domain", "coulomb_social", "--field", "NPM_AUTH_TOKEN",
         "--path", "platform/x/y/z", "--fetch"],
    )
    assert r.exit_code == 3


def test_cli_proxy_rejects_retired_no_policy_bypass(monkeypatch, tmp_path):
    _proxy_env(monkeypatch, tmp_path)
    monkeypatch.setenv("VAULT_TOKEN", "caller")
    monkeypatch.setattr(
        "warden.proxy.subprocess.run",
        lambda *a, **k: (_ for _ in ()).throw(AssertionError("fetch ran despite retired flag")),
    )
    r = runner.invoke(
        app,
        ["access", "npm", "--domain", "coulomb_social", "--field", "NPM_AUTH_TOKEN",
         "--path", "platform/x/y/z", "--fetch", "--no-policy"],
    )
    assert r.exit_code == 2
    assert "--no-policy is retired" in r.output


# --- T4: login lane --------------------------------------------------------

def test_cli_login_lane_contains_login_handoff_and_revocation(monkeypatch, tmp_path):
    """Login and its reviewed child share a private, silent helper session."""
    _proxy_env(monkeypatch, tmp_path)
    monkeypatch.delenv("VAULT_TOKEN", raising=False)
    monkeypatch.delenv("BAO_TOKEN", raising=False)
    monkeypatch.setattr(Path, "home", lambda: tmp_path)  # no ~/.vault-token

    calls = []

    def fake_run(argv, **kw):
        calls.append((argv, kw))
        assert kw["stdout"] is subprocess.PIPE
        assert kw["stderr"] is subprocess.PIPE
        private_home = Path(kw["env"]["HOME"])
        assert private_home != tmp_path
        assert oct(private_home.stat().st_mode & 0o777) == "0o700"
        helper = private_home / ".vault-token"
        assert oct(helper.stat().st_mode & 0o777) == "0o600"
        if argv[:2] == ["bao", "login"]:
            helper.write_bytes(b"non-production-test-double")
            helper.chmod(0o600)
        return subprocess.CompletedProcess(argv, 0, stdout=b"", stderr=b"")

    monkeypatch.setattr("warden.proxy.subprocess.run", fake_run)
    r = runner.invoke(
        app,
        [
            "access", "login oidc", "--domain", "coulomb_social",
            "--exec", "--", "true",
        ],
    )
    assert r.exit_code == 0
    assert [call[0][:2] for call in calls] == [
        ["bao", "login"],
        ["true"],
        ["bao", "token"],
    ]
    assert not (tmp_path / ".warden-attended-login").exists()
    assert "non-production-test-double" not in r.output
    audit = (tmp_path / "state" / "access-audit.log").read_text()
    assert "non-production-test-double" not in audit


def test_cli_login_lane_rejects_persistent_fetch(monkeypatch, tmp_path):
    _proxy_env(monkeypatch, tmp_path)
    monkeypatch.setattr(
        "warden.proxy.subprocess.run",
        lambda *a, **k: (_ for _ in ()).throw(AssertionError("should not run")),
    )
    r = runner.invoke(
        app, ["access", "login oidc", "--domain", "coulomb_social", "--fetch"]
    )
    assert r.exit_code == 2
    assert "requires --exec" in r.output


def test_attended_login_refuses_read_only_home_before_auth(monkeypatch, tmp_path):
    monkeypatch.setattr(Path, "home", lambda: tmp_path)
    monkeypatch.setattr(
        "warden.proxy.subprocess.run",
        lambda *a, **k: (_ for _ in ()).throw(AssertionError("OIDC started")),
    )
    tmp_path.chmod(0o555)
    try:
        with pytest.raises(ProxyError, match="writable default home"):
            proxy_attended_login_exec(
                ResolvedFetch(argv=["bao", "login", "-no-print"]),
                child_argv=["true"],
            )
    finally:
        tmp_path.chmod(0o700)


def test_attended_login_persistence_failure_revokes_and_cleans(monkeypatch, tmp_path):
    monkeypatch.setattr(Path, "home", lambda: tmp_path)
    calls = []

    def fake_run(argv, **kw):
        calls.append(argv)
        # Login succeeds but the pre-created helper remains empty: persistence failed.
        return subprocess.CompletedProcess(argv, 0, stdout=b"", stderr=b"")

    monkeypatch.setattr("warden.proxy.subprocess.run", fake_run)
    with pytest.raises(ProxyError, match="failed closed before command handoff"):
        proxy_attended_login_exec(
            ResolvedFetch(argv=["bao", "login", "-no-print"]),
            child_argv=["should-not-run"],
        )
    assert calls == [
        ["bao", "login", "-no-print", "-format=json"],
        ["bao", "token", "revoke", "-self"],
    ]
    assert not (tmp_path / ".warden-attended-login").exists()


@pytest.mark.parametrize("stream", ["stdout", "stderr"])
def test_attended_login_unexpected_output_is_contained_revoked_and_cleaned(
    monkeypatch, tmp_path, capsys, stream
):
    monkeypatch.setattr(Path, "home", lambda: tmp_path)
    sentinel = "hvs.NONPRODUCTION_SENTINEL"
    calls = []

    def fake_run(argv, **kw):
        calls.append((argv, dict(kw["env"])))
        if argv[:2] == ["bao", "login"]:
            output = sentinel.encode()
            return subprocess.CompletedProcess(
                argv,
                0,
                stdout=output if stream == "stdout" else b"",
                stderr=output if stream == "stderr" else b"",
            )
        if argv[:3] == ["bao", "token", "revoke"]:
            # The helper is empty. The second contained attempt uses the captured
            # value only through BAO_TOKEN, never argv or visible output.
            return subprocess.CompletedProcess(
                argv,
                0 if kw["env"].get("BAO_TOKEN") == sentinel else 1,
                stdout=b"",
                stderr=b"",
            )
        raise AssertionError("reviewed child ran after unexpected login output")

    monkeypatch.setattr("warden.proxy.subprocess.run", fake_run)
    with pytest.raises(ProxyError, match="failed closed before command handoff") as exc:
        proxy_attended_login_exec(
            ResolvedFetch(argv=["bao", "login", "-no-print"]),
            child_argv=["should-not-run"],
        )
    captured = capsys.readouterr()
    assert sentinel not in str(exc.value)
    assert sentinel not in captured.out
    assert sentinel not in captured.err
    assert all(sentinel not in " ".join(argv) for argv, _ in calls)
    assert calls[-1][1]["BAO_TOKEN"] == sentinel
    assert not (tmp_path / ".warden-attended-login").exists()


def test_attended_login_contained_success_output_never_escapes(monkeypatch, tmp_path, capsys):
    monkeypatch.setattr(Path, "home", lambda: tmp_path)
    sentinel = "hvs.NONPRODUCTION_CONTAINED_LOGIN"
    child_ran = False

    def fake_run(argv, **kw):
        nonlocal child_ran
        helper = Path(kw["env"]["HOME"]) / ".vault-token"
        if argv[:2] == ["bao", "login"]:
            helper.write_text(sentinel)
            helper.chmod(0o600)
            return subprocess.CompletedProcess(
                argv,
                0,
                stdout=json.dumps({"auth": {"client_token": sentinel}}).encode(),
                stderr=b"",
            )
        if argv == ["reviewed-child"]:
            child_ran = True
            return subprocess.CompletedProcess(argv, 0, stdout=b"", stderr=b"")
        if argv[:3] == ["bao", "token", "revoke"]:
            return subprocess.CompletedProcess(argv, 0, stdout=b"", stderr=b"")
        raise AssertionError(argv)

    monkeypatch.setattr("warden.proxy.subprocess.run", fake_run)
    assert proxy_attended_login_exec(
        ResolvedFetch(argv=["bao", "login", "-no-print"]),
        child_argv=["reviewed-child"],
    ) == 0
    captured = capsys.readouterr()
    assert child_ran is True
    assert sentinel not in captured.out
    assert sentinel not in captured.err
    assert not (tmp_path / ".warden-attended-login").exists()


def test_real_catalog_login_entry_is_login_lane():
    from warden.routing import load_catalog
    e = load_catalog(_repo_catalog()).get("key-cape-oidc-login")
    assert e is not None and e.lane == "login" and e.exec_capable


def test_invalid_lane_rejected(tmp_path):
    import yaml
    from warden.routing import CatalogError, load_catalog
    entry = dict(
        id="x", title="t", need_keywords=["k"], owner_repo="o", subsystem="s",
        warden_executes=False, wiki_ref="w", canon_ref="c", reviewed="2026-06-27",
        status="active", lane="bogus",
        workload_ref={"applicability": "not-applicable", "reason": "fixture"},
    )
    p = tmp_path / "c.yaml"
    p.write_text(yaml.dump({"version": 1, "entries": [entry]}))
    import pytest
    with pytest.raises(CatalogError, match="invalid lane"):
        load_catalog(p)


# ---------------------------------------------------------------------------
# Safe access transports (WARDEN-WP-0026 T02) — no secret values on stdout
# ---------------------------------------------------------------------------

from warden.proxy import (  # noqa: E402
    build_wrapped_fetch,
    is_bao_kv_fetch,
    proxy_fetch_to_file,
    proxy_fetch_wrapped,
)


def test_fetch_to_file_writes_mode_0600_and_no_stdout(tmp_path, capsys):
    out = tmp_path / "secret.out"
    rc = proxy_fetch_to_file(ResolvedFetch(shell_cmd="printf 'sekret'"), out)
    assert rc == 0
    assert out.read_text() == "sekret"
    assert oct(out.stat().st_mode & 0o777) == "0o600"
    # nothing printed to stdout/stderr by the transport itself
    captured = capsys.readouterr()
    assert "sekret" not in captured.out and "sekret" not in captured.err


def test_fetch_to_file_forces_0600_on_preexisting_loose_file(tmp_path):
    out = tmp_path / "pre.out"
    out.write_text("old")
    out.chmod(0o644)
    proxy_fetch_to_file(ResolvedFetch(shell_cmd="printf 'new'"), out)
    assert out.read_text() == "new"
    assert oct(out.stat().st_mode & 0o777) == "0o600"


def test_wrapped_fetch_returns_token_not_value():
    payload = '{"wrap_info":{"token":"hvs.WRAP"}}'
    token = proxy_fetch_wrapped(ResolvedFetch(shell_cmd=f"printf '%s' '{payload}'"))
    assert token == "hvs.WRAP"


def test_wrapped_fetch_bad_output_raises():
    with pytest.raises(ProxyError, match="wrapping token"):
        proxy_fetch_wrapped(ResolvedFetch(shell_cmd="printf 'not-json'"))


def test_build_wrapped_fetch_only_for_bao_kv():
    bao = _entry(fetch_command="bao kv get -field=API_TOKEN platform/x", path_template="platform/x")
    assert is_bao_kv_fetch(bao)
    argv = build_wrapped_fetch(bao, ttl="9m").argv
    assert argv == ["bao", "kv", "get", "-wrap-ttl=9m", "-format=json", "platform/x"]

    piped = _entry(fetch_command="kubectl get secret x -o json | base64 -d", path_template="x")
    assert not is_bao_kv_fetch(piped)
    with pytest.raises(ProxyError, match="response wrapping"):
        build_wrapped_fetch(piped)


def test_build_wrapped_fetch_refuses_placeholder_path():
    e = _entry(fetch_command="bao kv get -field=<FIELD> <path_template>",
               path_template="platform/workloads/<domain>/x")
    with pytest.raises(ProxyError, match="concrete path"):
        build_wrapped_fetch(e)


def test_access_fetch_to_nonterminal_stdout_is_refused(tmp_path, monkeypatch):
    """The anti-pattern: streaming a value to captured stdout is refused (exit 6)."""
    _proxy_env(monkeypatch, tmp_path)
    monkeypatch.setenv("VAULT_TOKEN", "caller-token")  # G1 caller-auth precheck
    # The guard trips before the fetch runs; make a real bao call fail loudly if reached.
    monkeypatch.setattr(
        "warden.proxy.subprocess.run",
        lambda *a, **k: (_ for _ in ()).throw(AssertionError("fetch ran despite stdout guard")),
    )
    # CliRunner captures stdout (not a tty), so the guard trips without --unsafe-stdout.
    r = runner.invoke(app, ["access", "whynot-design-npm-publish", "--fetch"])
    assert r.exit_code == 6
    assert "sanctioned transport" in r.output.lower() or "refusing" in r.output.lower()


def test_access_fingerprint_masks_and_bypasses_stdout_guard(monkeypatch, tmp_path):
    """--fingerprint prints a masked fingerprint (never the value) even to captured stdout."""
    _proxy_env(monkeypatch, tmp_path)
    monkeypatch.setenv("VAULT_TOKEN", "caller-token")

    class _Fake:
        returncode = 0
        stdout = "top-secret-token-value"

    monkeypatch.setattr("warden.proxy.subprocess.run", lambda *a, **k: _Fake())
    r = runner.invoke(
        app,
        ["access", "whynot-design-npm-publish", "--fingerprint"],
    )
    assert r.exit_code == 0
    assert "top-secret-token-value" not in r.output   # value never shown
    assert "hidden" in r.output and "sha256:" in r.output


def test_access_agent_high_risk_raw_stream_refused(tmp_path, monkeypatch):
    """WP-0026 T04: WARDEN_AGENT_ID + risk=high refuses raw value stream (exit 7)."""
    _proxy_env(monkeypatch, tmp_path)
    monkeypatch.setenv("VAULT_TOKEN", "caller-token")
    monkeypatch.setenv("WARDEN_AGENT_ID", "grok")
    # Prefer high-risk lane; use --unsafe-stdout so T02 would allow if T04 failed.
    r = runner.invoke(
        app,
        [
            "access", "railiance-backup-offsite-lane",
            "--fetch", "--unsafe-stdout",
        ],
    )
    assert r.exit_code == 7, r.output
    assert "agent read-boundary" in r.output.lower() or "risk=high" in r.output.lower()


def test_access_agent_high_risk_fingerprint_allowed(tmp_path, monkeypatch):
    """Agents may use --fingerprint on high-risk lanes (no raw value)."""
    _proxy_env(monkeypatch, tmp_path)
    monkeypatch.setenv("VAULT_TOKEN", "caller-token")
    monkeypatch.setenv("WARDEN_AGENT_ID", "grok")

    class _Fake:
        returncode = 0
        stdout = "should-not-appear"

    monkeypatch.setattr("warden.proxy.subprocess.run", lambda *a, **k: _Fake())
    r = runner.invoke(
        app,
        ["access", "railiance-backup-offsite-lane", "--fingerprint"],
    )
    assert r.exit_code == 0, r.output
    assert "should-not-appear" not in r.output