ops-warden/tests/test_policy_fetch_evaluator.py

"""Secret-read checks use their own evaluator pin, never the SSH signing pin (CUST-WP-0075)."""
from unittest.mock import patch

import httpx
import pytest

from warden.access import policy_gate_status
from warden.ca import CAError
from warden.config import PolicyConfig, load_config
from warden.models import ActorType, CertSpec
from warden.policy import check_fetch_policy, check_sign_policy

SIGN_URL = "http://127.0.0.1:19090"
FETCH_URL = "http://127.0.0.1:19091"


def _allow(url, json=None, headers=None, timeout=None):
    return httpx.Response(
        200,
        json={"effect": "allow", "id": "decision:fetch", "reason": "secret_read_lane_matched"},
        request=httpx.Request("POST", url),
    )


def _fetch(cfg):
    return check_fetch_policy(
        cfg, need_id="forgejo-admin-api-token", owner_repo="railiance-platform", domain=None
    )


def test_fetch_check_goes_to_fetch_pin_never_sign_pin():
    cfg = PolicyConfig(flex_auth_url=SIGN_URL, fetch_flex_auth_url=FETCH_URL)
    with patch("warden.policy.httpx.post", side_effect=_allow) as post:
        assert _fetch(cfg) == "decision:fetch"
    urls = [c.args[0] for c in post.call_args_list]
    assert urls == [f"{FETCH_URL}/v1/check"]
    body = post.call_args.kwargs["json"]
    assert body["resource"]["system"] == "railiance-platform"
    assert body["action"] == "read"


@pytest.mark.parametrize("failure_mode", ["fail_open", "fail_closed"])
def test_unset_fetch_url_fails_closed_without_using_sign_pin(failure_mode):
    cfg = PolicyConfig(flex_auth_url=SIGN_URL)
    cfg.failure_modes["unknown"] = failure_mode
    with patch("warden.policy.httpx.post") as post:
        with pytest.raises(CAError, match="policy.fetch_flex_auth_url") as err:
            _fetch(cfg)
    assert "secret-read evaluator not configured" in str(err.value)
    post.assert_not_called()


@pytest.mark.parametrize("failure_mode", ["fail_open", "fail_closed"])
def test_unreachable_fetch_url_fails_closed(failure_mode):
    cfg = PolicyConfig(flex_auth_url=SIGN_URL, fetch_flex_auth_url=FETCH_URL)
    cfg.failure_modes["unknown"] = failure_mode
    calls = []

    def boom(url, **kwargs):
        calls.append(url)
        raise httpx.ConnectError("refused", request=httpx.Request("POST", url))

    with patch("warden.policy.httpx.post", side_effect=boom):
        with pytest.raises(CAError, match="flex-auth-ops-warden-fetch\\) unreachable"):
            _fetch(cfg)
    assert calls == [f"{FETCH_URL}/v1/check"]


def test_fetch_403_keeps_existing_refusal_text():
    cfg = PolicyConfig(flex_auth_url=SIGN_URL, fetch_flex_auth_url=FETCH_URL)
    response = httpx.Response(403, request=httpx.Request("POST", f"{FETCH_URL}/v1/check"))
    with patch("warden.policy.httpx.post", return_value=response):
        with pytest.raises(CAError) as err:
            _fetch(cfg)
    assert str(err.value) == (
        "flex-auth refused fetch policy check (HTTP 403); "
        "caller authentication or system binding was rejected"
    )


def test_sign_check_still_uses_sign_pin(tmp_path):
    cfg = PolicyConfig(flex_auth_url=SIGN_URL, fetch_flex_auth_url=FETCH_URL)
    pub = tmp_path / "id.pub"
    pub.write_text("ssh-ed25519 AAAA test\n")
    spec = CertSpec(actor_name="agt-example", actor_type=ActorType.AGT,
                    pubkey_path=pub, ttl_hours=1, principals=["agt"])
    with patch("warden.policy.httpx.post", side_effect=_allow) as post:
        assert check_sign_policy(cfg, spec) == "decision:fetch"
    assert [c.args[0] for c in post.call_args_list] == [f"{SIGN_URL}/v1/check"]
    assert post.call_args.kwargs["json"]["resource"]["system"] == "ops-warden"


def _write_cfg(tmp_path, monkeypatch, policy_block):
    cfg = tmp_path / "warden.yaml"
    cfg.write_text(
        "backend: local\nca_key: /tmp/none\n" + policy_block
    )
    monkeypatch.setenv("WARDEN_CONFIG", str(cfg))
    return cfg


def test_config_loads_fetch_url(tmp_path, monkeypatch):
    _write_cfg(tmp_path, monkeypatch,
               f"policy:\n  flex_auth_url: {SIGN_URL}\n  fetch_flex_auth_url: {FETCH_URL}\n")
    cfg = load_config()
    assert cfg.policy.flex_auth_url == SIGN_URL
    assert cfg.policy.fetch_flex_auth_url == FETCH_URL


def test_gate_status_names_fetch_evaluator(tmp_path, monkeypatch):
    _write_cfg(tmp_path, monkeypatch,
               f"policy:\n  flex_auth_url: {SIGN_URL}\n  fetch_flex_auth_url: {FETCH_URL}\n")
    status = policy_gate_status()
    assert "flex-auth-ops-warden-fetch" in status and FETCH_URL in status
    assert SIGN_URL not in status


def test_gate_status_unset_fetch_url_says_refused(tmp_path, monkeypatch):
    _write_cfg(tmp_path, monkeypatch, f"policy:\n  flex_auth_url: {SIGN_URL}\n")
    status = policy_gate_status()
    assert "fetch_flex_auth_url unset" in status and "refused" in status


def test_access_advisory_prints_fetch_evaluator(tmp_path, monkeypatch):
    from pathlib import Path

    from typer.testing import CliRunner

    from warden.cli import app

    catalog = Path(__file__).resolve().parents[1] / "registry" / "routing" / "catalog.yaml"
    monkeypatch.setenv("WARDEN_ROUTING_CATALOG", str(catalog))
    _write_cfg(tmp_path, monkeypatch,
               f"policy:\n  flex_auth_url: {SIGN_URL}\n  fetch_flex_auth_url: {FETCH_URL}\n")
    result = CliRunner().invoke(app, ["access", "forgejo-admin-api-token"])
    assert result.exit_code == 0, result.output
    flat = " ".join(result.output.split())
    assert "flex-auth-ops-warden-fetch" in flat and FETCH_URL in flat