49 lines
2.4 KiB
Python
49 lines
2.4 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")
|
||
|
|
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
|