tegwick
7ce58ae638
Assistant: codex Assistant-Model: gpt-5.6-sol Assistant-Session: 01a0291a-1e87-7151-9934-fcbfe3f65eb1
309 lines
10 KiB
Python
309 lines
10 KiB
Python
"""Tests for warden.policy — flex-auth gate."""
|
|
import subprocess
|
|
from pathlib import Path
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import httpx
|
|
import pytest
|
|
|
|
from warden.ca import CAError
|
|
from warden.caller_identity import (
|
|
CallerIdentityError,
|
|
caller_auth_headers,
|
|
resolve_caller_token,
|
|
)
|
|
from warden.config import CallerAuthConfig, PolicyConfig
|
|
from warden.models import ActorType, CertSpec
|
|
from warden.policy import check_sign_policy, pubkey_fingerprint
|
|
|
|
|
|
def _spec(pubkey_path: Path) -> CertSpec:
|
|
return CertSpec(
|
|
actor_name="agt-state-hub-bridge",
|
|
actor_type=ActorType.AGT,
|
|
pubkey_path=pubkey_path,
|
|
ttl_hours=24,
|
|
principals=["agt-task-bridge"],
|
|
)
|
|
|
|
|
|
def _zone_registry(tmp_path: Path, zone: str) -> Path:
|
|
path = tmp_path / "registry.json"
|
|
path.write_text(
|
|
'{"resource_manifests":[{"resources":[{"id":'
|
|
'"ssh-cert:actor/agt-state-hub-bridge","attributes":{'
|
|
f'"security_zone":"{zone}","security_zone_admission":"satisfied"'
|
|
'}}]}]}'
|
|
)
|
|
return path
|
|
|
|
|
|
def test_pubkey_fingerprint(tmp_path):
|
|
pubkey = tmp_path / "key.pub"
|
|
pubkey.write_text("ssh-ed25519 AAAA test\n")
|
|
fp = pubkey_fingerprint(pubkey)
|
|
assert fp.startswith("sha256:")
|
|
assert len(fp) == 7 + 64
|
|
|
|
|
|
def test_unconfigured_evaluator_uses_unknown_fail_open_profile(tmp_path):
|
|
pubkey = tmp_path / "key.pub"
|
|
pubkey.write_text("ssh-ed25519 AAAA\n")
|
|
cfg = PolicyConfig()
|
|
spec = _spec(pubkey)
|
|
assert check_sign_policy(cfg, spec) is None
|
|
assert spec.policy_zone == "unknown"
|
|
assert spec.policy_failure_mode == "fail_open"
|
|
assert spec.policy_outcome == "fail_open"
|
|
|
|
|
|
def test_allow_returns_decision_id(tmp_path):
|
|
pubkey = tmp_path / "key.pub"
|
|
pubkey.write_text("ssh-ed25519 AAAA\n")
|
|
cfg = PolicyConfig(flex_auth_url="http://flex-auth.test")
|
|
|
|
mock_response = MagicMock()
|
|
mock_response.json.return_value = {"effect": "allow", "id": "dec-123"}
|
|
mock_response.raise_for_status = MagicMock()
|
|
|
|
spec = _spec(pubkey)
|
|
with patch("warden.policy.httpx.post", return_value=mock_response) as post:
|
|
result = check_sign_policy(cfg, spec)
|
|
|
|
assert result == "dec-123"
|
|
assert spec.policy_outcome == "allow"
|
|
post.assert_called_once()
|
|
call_kwargs = post.call_args
|
|
assert call_kwargs[0][0] == "http://flex-auth.test/v1/check"
|
|
body = call_kwargs[1]["json"]
|
|
assert body["action"] == "sign"
|
|
assert body["resource"]["type"] == "ssh-certificate"
|
|
assert body["context"]["actor_name"] == "agt-state-hub-bridge"
|
|
|
|
|
|
def test_deny_raises_ca_error(tmp_path):
|
|
pubkey = tmp_path / "key.pub"
|
|
pubkey.write_text("ssh-ed25519 AAAA\n")
|
|
cfg = PolicyConfig(flex_auth_url="http://flex-auth.test")
|
|
|
|
mock_response = MagicMock()
|
|
mock_response.json.return_value = {
|
|
"effect": "deny",
|
|
"reason": "actor not authorized",
|
|
}
|
|
mock_response.raise_for_status = MagicMock()
|
|
|
|
with patch("warden.policy.httpx.post", return_value=mock_response):
|
|
with pytest.raises(CAError, match="denied SSH sign"):
|
|
check_sign_policy(cfg, _spec(pubkey))
|
|
|
|
|
|
def test_unreachable_fail_closed_raises(tmp_path):
|
|
pubkey = tmp_path / "key.pub"
|
|
pubkey.write_text("ssh-ed25519 AAAA\n")
|
|
cfg = PolicyConfig(
|
|
flex_auth_url="http://flex-auth.test",
|
|
zone_registry_path=_zone_registry(tmp_path, "z3-critical"),
|
|
)
|
|
|
|
with patch(
|
|
"warden.policy.httpx.post",
|
|
side_effect=httpx.ConnectError("connection refused"),
|
|
):
|
|
with pytest.raises(CAError, match="unreachable"):
|
|
check_sign_policy(cfg, _spec(pubkey))
|
|
|
|
|
|
def test_unreachable_fail_open_returns_none(tmp_path):
|
|
pubkey = tmp_path / "key.pub"
|
|
pubkey.write_text("ssh-ed25519 AAAA\n")
|
|
cfg = PolicyConfig(flex_auth_url="http://flex-auth.test")
|
|
|
|
with patch(
|
|
"warden.policy.httpx.post",
|
|
side_effect=httpx.ConnectError("connection refused"),
|
|
):
|
|
assert check_sign_policy(cfg, _spec(pubkey)) is None
|
|
|
|
|
|
def test_http_error_fail_closed_raises(tmp_path):
|
|
pubkey = tmp_path / "key.pub"
|
|
pubkey.write_text("ssh-ed25519 AAAA\n")
|
|
cfg = PolicyConfig(
|
|
flex_auth_url="http://flex-auth.test",
|
|
zone_registry_path=_zone_registry(tmp_path, "z3-critical"),
|
|
)
|
|
|
|
mock_response = MagicMock()
|
|
mock_response.status_code = 403
|
|
error = httpx.HTTPStatusError(
|
|
"forbidden", request=MagicMock(), response=mock_response
|
|
)
|
|
|
|
with patch("warden.policy.httpx.post", side_effect=error):
|
|
with pytest.raises(CAError, match="HTTP 403"):
|
|
check_sign_policy(cfg, _spec(pubkey))
|
|
|
|
|
|
def test_missing_pubkey_raises(tmp_path):
|
|
cfg = PolicyConfig(flex_auth_url="http://flex-auth.test")
|
|
spec = _spec(tmp_path / "missing.pub")
|
|
with pytest.raises(CAError, match="Public key not found"):
|
|
check_sign_policy(cfg, spec)
|
|
|
|
|
|
def test_subject_from_env(tmp_path, monkeypatch):
|
|
pubkey = tmp_path / "key.pub"
|
|
pubkey.write_text("ssh-ed25519 AAAA\n")
|
|
cfg = PolicyConfig(
|
|
flex_auth_url="http://flex-auth.test",
|
|
subject_env="WARDEN_POLICY_SUBJECT",
|
|
)
|
|
monkeypatch.setenv("WARDEN_POLICY_SUBJECT", "iam:bernd")
|
|
|
|
mock_response = MagicMock()
|
|
mock_response.json.return_value = {"effect": "allow", "id": "dec-456"}
|
|
mock_response.raise_for_status = MagicMock()
|
|
|
|
with patch("warden.policy.httpx.post", return_value=mock_response) as post:
|
|
check_sign_policy(cfg, _spec(pubkey))
|
|
|
|
body = post.call_args[1]["json"]
|
|
assert body["subject"]["id"] == "iam:bernd"
|
|
|
|
# --- caller identity (FLEX-WP-0016 / WARDEN-WP-0031) -----------------------
|
|
|
|
def test_caller_auth_none_sends_no_header():
|
|
assert caller_auth_headers(CallerAuthConfig()) == {}
|
|
|
|
|
|
def test_caller_auth_file_reads_projected_token(tmp_path):
|
|
token_file = tmp_path / "token"
|
|
token_file.write_text("sa-token-value\n")
|
|
cfg = CallerAuthConfig(mode="file", token_path=token_file)
|
|
assert caller_auth_headers(cfg) == {"Authorization": "Bearer sa-token-value"}
|
|
|
|
|
|
def test_caller_auth_file_missing_raises(tmp_path):
|
|
cfg = CallerAuthConfig(mode="file", token_path=tmp_path / "absent")
|
|
with pytest.raises(CallerIdentityError, match="unreadable"):
|
|
resolve_caller_token(cfg)
|
|
|
|
|
|
def test_caller_auth_env_mode(monkeypatch):
|
|
monkeypatch.setenv("WARDEN_POLICY_CALLER_TOKEN", " env-token ")
|
|
assert resolve_caller_token(CallerAuthConfig(mode="env")) == "env-token"
|
|
monkeypatch.setenv("WARDEN_POLICY_CALLER_TOKEN", "")
|
|
with pytest.raises(CallerIdentityError, match="unset or empty"):
|
|
resolve_caller_token(CallerAuthConfig(mode="env"))
|
|
|
|
|
|
def test_caller_auth_command_mode_uses_stdout(monkeypatch):
|
|
cfg = CallerAuthConfig(mode="command", command=["kubectl", "create", "token"])
|
|
|
|
def fake_run(cmd, **kwargs):
|
|
assert cmd == cfg.command
|
|
return subprocess.CompletedProcess(cmd, 0, stdout="minted-token\n", stderr="")
|
|
|
|
monkeypatch.setattr(subprocess, "run", fake_run)
|
|
assert resolve_caller_token(cfg) == "minted-token"
|
|
|
|
|
|
def test_caller_auth_command_failure_message_excludes_token(monkeypatch):
|
|
cfg = CallerAuthConfig(mode="command", command=["kubectl", "create", "token"])
|
|
|
|
def fake_run(cmd, **kwargs):
|
|
return subprocess.CompletedProcess(cmd, 1, stdout="", stderr="error: forbidden\n")
|
|
|
|
monkeypatch.setattr(subprocess, "run", fake_run)
|
|
with pytest.raises(CallerIdentityError, match="error: forbidden"):
|
|
resolve_caller_token(cfg)
|
|
|
|
|
|
def test_caller_auth_rejects_whitespace_token(tmp_path):
|
|
token_file = tmp_path / "token"
|
|
token_file.write_text("two words")
|
|
cfg = CallerAuthConfig(mode="file", token_path=token_file)
|
|
with pytest.raises(CallerIdentityError, match="whitespace"):
|
|
resolve_caller_token(cfg)
|
|
|
|
|
|
def test_sign_policy_sends_authorization_header(tmp_path, monkeypatch):
|
|
"""The header flex-auth's ops-warden pin needs to leave warn mode."""
|
|
from warden import policy as policy_mod
|
|
|
|
token_file = tmp_path / "token"
|
|
token_file.write_text("sa-token-value")
|
|
pubkey = tmp_path / "id.pub"
|
|
pubkey.write_text("ssh-ed25519 AAAA test\n")
|
|
|
|
cfg = PolicyConfig(
|
|
flex_auth_url="http://flex-auth.test",
|
|
caller_auth=CallerAuthConfig(mode="file", token_path=token_file),
|
|
)
|
|
spec = CertSpec(
|
|
actor_name="agt-state-hub-bridge",
|
|
actor_type=ActorType.AGT,
|
|
principals=["agt"],
|
|
ttl_hours=24,
|
|
pubkey_path=pubkey,
|
|
)
|
|
|
|
seen = {}
|
|
|
|
class _Response:
|
|
status_code = 200
|
|
|
|
def raise_for_status(self):
|
|
return None
|
|
|
|
def json(self):
|
|
return {"effect": "allow", "id": "decision:49350f1064f674d7"}
|
|
|
|
def fake_post(url, json=None, headers=None, timeout=None):
|
|
seen["headers"] = headers
|
|
return _Response()
|
|
|
|
monkeypatch.setattr(policy_mod.httpx, "post", fake_post)
|
|
assert policy_mod.check_sign_policy(cfg, spec) == "decision:49350f1064f674d7"
|
|
assert seen["headers"] == {"Authorization": "Bearer sa-token-value"}
|
|
|
|
|
|
def test_sign_policy_fail_closed_when_caller_token_unavailable(tmp_path):
|
|
from warden.ca import CAError
|
|
from warden import policy as policy_mod
|
|
|
|
pubkey = tmp_path / "id.pub"
|
|
pubkey.write_text("ssh-ed25519 AAAA test\n")
|
|
cfg = PolicyConfig(
|
|
flex_auth_url="http://flex-auth.test",
|
|
zone_registry_path=_zone_registry(tmp_path, "z3-critical"),
|
|
caller_auth=CallerAuthConfig(mode="file", token_path=tmp_path / "absent"),
|
|
)
|
|
spec = CertSpec(
|
|
actor_name="agt-state-hub-bridge",
|
|
actor_type=ActorType.AGT,
|
|
principals=["agt"],
|
|
ttl_hours=24,
|
|
pubkey_path=pubkey,
|
|
)
|
|
with pytest.raises(CAError, match="caller identity unavailable"):
|
|
policy_mod.check_sign_policy(cfg, spec)
|
|
|
|
|
|
def test_advisory_decision_is_recorded_and_does_not_block(tmp_path):
|
|
pubkey = tmp_path / "id.pub"
|
|
pubkey.write_text("ssh-ed25519 AAAA test\n")
|
|
cfg = PolicyConfig(flex_auth_url="http://flex-auth.test")
|
|
response = MagicMock()
|
|
response.json.return_value = {
|
|
"effect": "audit_only",
|
|
"reason": "advisory_would_deny_disallowed_principal",
|
|
"id": "decision:advisory",
|
|
}
|
|
response.raise_for_status = MagicMock()
|
|
spec = _spec(pubkey)
|
|
with patch("warden.policy.httpx.post", return_value=response):
|
|
assert check_sign_policy(cfg, spec) == "decision:advisory"
|
|
assert spec.policy_zone == "unknown"
|
|
assert spec.policy_outcome == "audit_only"
|