Refuse explicit policy authentication and binding denials before side effects

Assistant: codex
Assistant-Model: gpt-5.6-luna
Assistant-Session: 01a07ff8-19d0-7820-b4d0-1353833cb7fc

tests/test_policy_http_refusal.py

@@ -0,0 +1,48 @@
+"""An explicit caller refusal cannot authorize a CA or credential side effect."""
+from unittest.mock import patch
+
+import httpx
+import pytest
+
+from warden.ca import CAError
+from warden.config import PolicyConfig
+from warden.models import ActorType, CertSpec
+from warden.policy import check_fetch_policy, check_sign_policy
+
+
+@pytest.mark.parametrize("status", [401, 403])
+@pytest.mark.parametrize("failure_mode", ["fail_open", "fail_closed"])
+@pytest.mark.parametrize("operation", ["sign", "fetch"])
+def test_explicit_refusal_blocks_under_every_outage_mode(tmp_path, status, failure_mode, operation):
+    cfg = PolicyConfig(flex_auth_url="http://pdp.test")
+    cfg.failure_modes["unknown"] = failure_mode
+    public_key = tmp_path / "id.pub"
+    public_key.write_text("ssh-ed25519 AAAA test\n")
+    spec = CertSpec(actor_name="agt-example", actor_type=ActorType.AGT,
+                    pubkey_path=public_key, ttl_hours=1, principals=["agt"])
+    response = httpx.Response(status, request=httpx.Request("POST", "http://pdp.test/v1/check"),
+                              text="untrusted response body must not be exposed")
+    with patch("warden.policy.httpx.post", return_value=response):
+        with pytest.raises(CAError, match=f"HTTP {status}") as error:
+            if operation == "sign":
+                check_sign_policy(cfg, spec)
+            else:
+                check_fetch_policy(cfg, need_id="forgejo-admin-api-token",
+                                   owner_repo="railiance-platform", domain=None)
+    assert "untrusted response" not in str(error.value)
+    if operation == "sign":
+        assert spec.policy_outcome == "deny"
+
+
+@pytest.mark.parametrize("status", [500, 503])
+@pytest.mark.parametrize("failure_mode", ["fail_open", "fail_closed"])
+def test_actual_evaluator_failure_retains_declared_outage_mode(status, failure_mode):
+    cfg = PolicyConfig(flex_auth_url="http://pdp.test")
+    cfg.failure_modes["unknown"] = failure_mode
+    response = httpx.Response(status, request=httpx.Request("POST", "http://pdp.test/v1/check"))
+    with patch("warden.policy.httpx.post", return_value=response):
+        if failure_mode == "fail_closed":
+            with pytest.raises(CAError, match=f"HTTP {status}"):
+                check_fetch_policy(cfg, need_id="example", owner_repo="example", domain=None)
+        else:
+            assert check_fetch_policy(cfg, need_id="example", owner_repo="example", domain=None) is None

tests/test_proxy.py

@@ -234,6 +234,28 @@ def test_cli_proxy_requires_caller_auth(monkeypatch, tmp_path):
     assert r.exit_code == 3
 
 
+@pytest.mark.parametrize("status", [401, 403])
+def test_cli_explicit_policy_refusal_never_fetches_or_starts_child(monkeypatch, tmp_path, status):
+    import httpx
+
+    _proxy_env(monkeypatch, tmp_path)
+    cfg = tmp_path / "warden.yaml"
+    cfg.write_text(cfg.read_text() + "policy:\n  flex_auth_url: http://pdp.test\n")
+    monkeypatch.setenv("VAULT_TOKEN", "caller-test-value")
+    monkeypatch.setattr(
+        "warden.policy.httpx.post",
+        lambda *a, **k: httpx.Response(status, request=httpx.Request("POST", "http://pdp.test/v1/check")),
+    )
+    calls = []
+    for name in ("proxy_exec", "proxy_fetch", "proxy_fetch_to_file", "proxy_fetch_wrapped"):
+        monkeypatch.setattr("warden.proxy." + name, lambda *a, **k: calls.append(True))
+    result = runner.invoke(app, ["access", "forgejo-admin-api-token", "--exec", "--field", "API_TOKEN", "--", "true"])
+    assert result.exit_code == 4
+    assert f"HTTP {status}" in result.output
+    assert "fail_open applied" not in result.output
+    assert calls == []
+
+
 def test_cli_proxy_rejects_retired_no_policy_bypass(monkeypatch, tmp_path):
     _proxy_env(monkeypatch, tmp_path)
     monkeypatch.setenv("VAULT_TOKEN", "caller")