import copy import os from pathlib import Path import pytest from secrets_engine.apply import apply_plan from secrets_engine.catalog import load_catalog, validate_entry from secrets_engine.config import repo_root from secrets_engine.errors import PolicyGuardError, ProvisioningError from secrets_engine.handoff import write_approle_handoff from secrets_engine.plan import build_plan from secrets_engine.provision import provision_from_file from secrets_engine.routing import route_lane from secrets_engine.verify import run_verification AUTH = { "id": "warden-sign", "kind": "auth-capability", "org": "netkingdom", "repo": "ops-warden", "stage": "prod", "mount": "ssh", "path": "sign", "consumers": [ {"name": "ops-warden-policy-smoke", "auth": "approle", "claim": "agent:agt"} ], "delivery_modes": ["approle-login"], "approval": {"model": "decision", "decision_ref": "SECRETS-WP-0004"}, "verification": {"positive": "allowlisted sign", "negative": "deny probes"}, "auth_capability": { "policy_name": "warden-sign", "role_name": "warden-sign", "token_ttl": "15m", "token_max_ttl": "15m", "secret_id_ttl": "30m", "secret_id_num_uses": 1, "token_num_uses": 0, "allowed_paths": [ {"path": "ssh/sign/agt-role", "capabilities": ["update"]}, {"path": "ssh/sign/adm-role", "capabilities": ["update"]}, {"path": "ssh/sign/atm-role", "capabilities": ["update"]}, ], "denied_probe_paths": ["ssh/sign/unlisted-role", "ssh/roles/agt-role"], }, "rotation": {"expectation": "single-use handoff", "ttl": "15m"}, "deactivation": {"expectation": "delete approle and policy"}, "audit": {"evidence": "non-secret pointers only"}, } def _auth_entry(**over): data = copy.deepcopy(AUTH) for key, value in over.items(): if key == "auth_capability": data[key].update(value) else: data[key] = value return validate_entry(data) def test_auth_capability_entry_parses_without_fields(): entry = _auth_entry() assert entry.kind == "auth-capability" assert entry.fields == [] assert entry.policy_name == "warden-sign" assert entry.role_name == "warden-sign" assert sorted(entry.auth_allowed_paths) == [ "ssh/sign/adm-role", "ssh/sign/agt-role", "ssh/sign/atm-role", ] def test_repo_catalog_loads_warden_sign_lane(): entries = load_catalog(repo_root() / "catalog") entry = entries["warden-sign"] assert entry.kind == "auth-capability" assert entry.mount == "ssh" assert entry.token_ttl == "15m" assert entry.secret_id_num_uses == 1 def test_auth_capability_rejects_kv_fields(): data = copy.deepcopy(AUTH) data["fields"] = ["token"] with pytest.raises(Exception): validate_entry(data) def test_auth_capability_plan_has_no_kv_mount(): entry = _auth_entry() plan = build_plan(entry, "prod", decision_id="d1") assert not any(action.kind == "kv-mount" for action in plan.actions) assert any(action.kind == "policy" and action.target == "warden-sign" for action in plan.actions) assert "ssh/sign/agt-role" in plan.policy_hcl assert "ssh/sign/adm-role" in plan.policy_hcl assert "ssh/sign/atm-role" in plan.policy_hcl assert "secret/data" not in plan.policy_hcl @pytest.mark.parametrize( "path", [ "ssh/*", "ssh/sign/*", "ssh/roles/agt-role", "sys/policies/acl/warden-sign", "auth/token/create", "identity/entity/id", ], ) def test_auth_capability_broad_or_admin_paths_refused(path): entry = _auth_entry( auth_capability={ "allowed_paths": [{"path": path, "capabilities": ["update"]}], } ) with pytest.raises(PolicyGuardError): build_plan(entry, "prod") def test_auth_capability_root_like_policy_name_refused(): entry = _auth_entry(auth_capability={"policy_name": "root-warden-sign"}) with pytest.raises(PolicyGuardError): build_plan(entry, "prod") def test_auth_capability_provision_rejected(tmp_path): value_file = tmp_path / "value" value_file.write_text("not-real-secret") os.chmod(value_file, 0o600) with pytest.raises(ProvisioningError): provision_from_file(object(), _auth_entry(), "", value_file) class FakeApplyClient: def __init__(self): self.policy = "" self.approle = None def kv_mount_exists(self, mount): raise AssertionError("auth-capability apply must not inspect KV mounts") def ensure_kv_mount(self, mount): raise AssertionError("auth-capability apply must not create KV mounts") def read_policy(self, name): return None def write_policy(self, name, hcl): self.policy = hcl def ensure_approle_enabled(self): pass def write_approle(self, role_name, policies, ttl="30m", **kwargs): self.approle = (role_name, policies, ttl, kwargs) def test_apply_auth_capability_bypasses_kv_and_writes_ttl_options(): entry = _auth_entry() plan = build_plan(entry, "prod", decision_id="d1") client = FakeApplyClient() result = apply_plan(client, entry, plan) assert any("not applicable" in item for item in result.skipped) assert "ssh/sign/agt-role" in client.policy assert client.approle[0] == "warden-sign" assert client.approle[2] == "15m" assert client.approle[3]["secret_id_num_uses"] == 1 assert client.approle[3]["token_num_uses"] == 0 class FakeVerifyClient: def approle_login_token(self, role_name): assert role_name == "warden-sign" return "test-token" def token_capabilities(self, path, *, token): assert token == "test-token" if path.startswith("ssh/sign/") and path != "ssh/sign/unlisted-role": return ["update"] return ["deny"] def test_auth_capability_verification_uses_capability_probes(): results = run_verification( FakeVerifyClient(), _auth_entry(), "", positive=True, negative=True ) assert [result.passed for result in results] == [True, True] class FakeHandoffClient: def read_approle_role_id(self, role_name): assert role_name == "warden-sign" return "role-id-test" def create_approle_secret_id(self, role_name): assert role_name == "warden-sign" return "secret-id-test" def test_handoff_rejects_repo_paths(): with pytest.raises(ProvisioningError): write_approle_handoff( FakeHandoffClient(), _auth_entry(), role_id_file=Path("role-id.out"), secret_id_file=Path("secret-id.out"), ) def test_handoff_writes_mode_0600_files(tmp_path): role_file = tmp_path / "role-id" secret_file = tmp_path / "secret-id" result = write_approle_handoff( FakeHandoffClient(), _auth_entry(), role_id_file=role_file, secret_id_file=secret_file, ) assert result.role_name == "warden-sign" assert role_file.read_text().strip() == "role-id-test" assert secret_file.read_text().strip() == "secret-id-test" assert (role_file.stat().st_mode & 0o077) == 0 assert (secret_file.stat().st_mode & 0o077) == 0 def test_route_missing_decision_blocks_auth_capability(tmp_path): result = route_lane(_auth_entry(), hub_url="", repo_root=tmp_path, client=None) assert result.kind == "auth-capability" assert result.decision_status == "missing" assert result.ready is False assert result.missing == "approved decision for 'SECRETS-WP-0004'" assert result.next_command == "secrets-engine decision inspect SECRETS-WP-0004"