tegwick
7d4381e887
check_fetch_policy now posts only to policy.fetch_flex_auth_url (access-engine pin flex-auth-ops-warden-fetch, workstation 127.0.0.1:19091). Signing keeps policy.flex_auth_url. Unset or unreachable fetch evaluator fails closed with a message naming it; no fallback to the signing pin or to allow. warden access advisory output names the fetch evaluator. Recorded under WARDEN-WP-0039-T03. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Assistant: claude-code Assistant-Model: opus Assistant-Process: 3775@bnt-lap001 Assistant-Session: cfacb6a5-ad5b-4f79-ac67-053e3b0d34a3
48 lines
2.5 KiB
Python
48 lines
2.5 KiB
Python
"""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", fetch_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://sign.test", fetch_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
|