"""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 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_disabled_returns_none(tmp_path): pubkey = tmp_path / "key.pub" pubkey.write_text("ssh-ed25519 AAAA\n") cfg = PolicyConfig(enabled=False) assert check_sign_policy(cfg, _spec(pubkey)) is None def test_allow_returns_decision_id(tmp_path): pubkey = tmp_path / "key.pub" pubkey.write_text("ssh-ed25519 AAAA\n") cfg = PolicyConfig(enabled=True, 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() with patch("warden.policy.httpx.post", return_value=mock_response) as post: result = check_sign_policy(cfg, _spec(pubkey)) assert result == "dec-123" 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(enabled=True) 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(enabled=True, fail_closed=True) 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(enabled=True, fail_closed=False) 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(enabled=True, fail_closed=True) 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(enabled=True) 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(enabled=True, 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( enabled=True, 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( enabled=True, fail_closed=True, 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)