import copy import unittest from authority_gate import ( ACTION_RISK, ActionIntent, ApprovalAssertion, AuthorityGate, AuthorityGrant, intent_digest, verify_receipt, ) NOW = 1_800_000_000 def grant(**overrides): values = dict( grant_id="grant-01", tenant_id="tenant-A", principal_id="agent-7", task_id="task-9", audience="tool-gateway", issued_at=NOW-10, expires_at=NOW+300, tools=("calendar.get", "mail.send", "files.write"), capabilities=("calendar.read", "mail.send", "files.write"), actions=("read", "draft", "write", "external_send"), resources=("calendar:user-A", "mail:draft-1", "files:workspace-A"), revoked=False, ) values.update(overrides) return AuthorityGrant(**values) def intent(**overrides): values = dict( intent_id="intent-001", tenant_id="tenant-A", principal_id="agent-7", task_id="task-9", audience="tool-gateway", tool_id="calendar.get", effect_digest="a" * 64, capability="calendar.read", action="read", resource_id="calendar:user-A", idempotency_key=None, ) values.update(overrides) return ActionIntent(**values) def approval_for(request, **overrides): values = dict( approval_id="approval-01", tenant_id=request.tenant_id, principal_id=request.principal_id, task_id=request.task_id, intent_digest=intent_digest(request), approver_id="user-42", issued_at=NOW-2, expires_at=NOW+120, ) values.update(overrides) return ApprovalAssertion(**values) class AuthorityGateTests(unittest.TestCase): def test_scoped_read_is_allowed(self): result = AuthorityGate().evaluate(grant(), intent(), now=NOW) self.assertEqual((result["decision"], result["reason"]), ("allow", "scope_match")) self.assertTrue(verify_receipt(result)) def test_draft_is_allowed_without_human_approval(self): req = intent(action="draft", tool_id="mail.send", capability="mail.send", resource_id="mail:draft-1") result = AuthorityGate().evaluate(grant(), req, now=NOW) self.assertEqual(result["decision"], "allow") self.assertEqual(result["risk"], "low") def test_cross_tenant_is_denied(self): result = AuthorityGate().evaluate(grant(), intent(tenant_id="tenant-B"), now=NOW) self.assertEqual(result["reason"], "tenant_mismatch") def test_cross_principal_is_denied(self): result = AuthorityGate().evaluate(grant(), intent(principal_id="agent-other"), now=NOW) self.assertEqual(result["reason"], "principal_mismatch") def test_cross_task_is_denied(self): result = AuthorityGate().evaluate(grant(), intent(task_id="task-other"), now=NOW) self.assertEqual(result["reason"], "task_mismatch") def test_wrong_audience_is_denied(self): result = AuthorityGate().evaluate(grant(), intent(audience="other-service"), now=NOW) self.assertEqual(result["reason"], "audience_mismatch") def test_expired_grant_is_denied(self): result = AuthorityGate().evaluate(grant(expires_at=NOW), intent(), now=NOW) self.assertEqual(result["reason"], "grant_expired") def test_future_grant_is_denied(self): result = AuthorityGate().evaluate(grant(issued_at=NOW+1, expires_at=NOW+30), intent(), now=NOW) self.assertEqual(result["reason"], "grant_not_yet_valid") def test_oversized_grant_ttl_is_denied(self): result = AuthorityGate().evaluate(grant(expires_at=NOW+3601), intent(), now=NOW) self.assertEqual(result["reason"], "grant_ttl_invalid") def test_revoked_grant_is_denied(self): result = AuthorityGate().evaluate(grant(revoked=True), intent(), now=NOW) self.assertEqual(result["reason"], "grant_revoked") def test_missing_capability_is_denied(self): result = AuthorityGate().evaluate(grant(capabilities=("mail.send",)), intent(), now=NOW) self.assertEqual(result["reason"], "capability_not_granted") def test_missing_action_is_denied(self): result = AuthorityGate().evaluate(grant(actions=("draft",)), intent(), now=NOW) self.assertEqual(result["reason"], "action_not_granted") def test_wrong_resource_is_denied(self): result = AuthorityGate().evaluate(grant(), intent(resource_id="calendar:user-B"), now=NOW) self.assertEqual(result["reason"], "resource_not_granted") def test_unknown_action_is_denied(self): result = AuthorityGate().evaluate(grant(), intent(action="run_shell"), now=NOW) self.assertEqual(result["reason"], "action_unknown") def test_high_risk_write_without_approval_is_denied(self): req = intent(action="external_send", tool_id="mail.send", capability="mail.send", resource_id="mail:draft-1", idempotency_key="idem-000001") result = AuthorityGate().evaluate(grant(), req, now=NOW) self.assertEqual(result["reason"], "human_approval_required") def test_mutating_action_requires_idempotency_key(self): req = intent(action="external_send", tool_id="mail.send", capability="mail.send", resource_id="mail:draft-1") result = AuthorityGate().evaluate(grant(), req, now=NOW) self.assertEqual(result["reason"], "idempotency_key_required") def test_high_risk_approval_is_bound_to_exact_intent(self): req = intent(action="external_send", tool_id="mail.send", capability="mail.send", resource_id="mail:draft-1", idempotency_key="idem-000001") wrong = approval_for(req, intent_digest="0"*64) result = AuthorityGate(lambda _: True).evaluate(grant(), req, now=NOW, approval=wrong) self.assertEqual(result["reason"], "approval_intent_mismatch") def test_expired_approval_is_denied(self): req = intent(action="external_send", tool_id="mail.send", capability="mail.send", resource_id="mail:draft-1", idempotency_key="idem-000001") approval = approval_for(req, expires_at=NOW) result = AuthorityGate(lambda _: True).evaluate(grant(), req, now=NOW, approval=approval) self.assertEqual(result["reason"], "approval_expired_or_ttl_invalid") def test_missing_approval_verifier_fails_closed(self): req = intent(action="external_send", tool_id="mail.send", capability="mail.send", resource_id="mail:draft-1", idempotency_key="idem-000001") result = AuthorityGate().evaluate(grant(), req, now=NOW, approval=approval_for(req)) self.assertEqual(result["reason"], "approval_verifier_unavailable") def test_invalid_approval_signature_is_denied(self): req = intent(action="external_send", tool_id="mail.send", capability="mail.send", resource_id="mail:draft-1", idempotency_key="idem-000001") result = AuthorityGate(lambda _: False).evaluate(grant(), req, now=NOW, approval=approval_for(req)) self.assertEqual(result["reason"], "approval_authentication_failed") def test_approval_exception_fails_closed(self): def fail(_): raise RuntimeError("signature service unavailable") req = intent(action="external_send", tool_id="mail.send", capability="mail.send", resource_id="mail:draft-1", idempotency_key="idem-000001") result = AuthorityGate(fail).evaluate(grant(), req, now=NOW, approval=approval_for(req)) self.assertEqual(result["reason"], "approval_authentication_failed") def test_valid_approval_allows_once(self): req = intent(action="external_send", tool_id="mail.send", capability="mail.send", resource_id="mail:draft-1", idempotency_key="idem-000001") approval = approval_for(req) gate = AuthorityGate(lambda candidate: candidate == approval) allowed = gate.evaluate(grant(), req, now=NOW, approval=approval) replay = gate.evaluate(grant(), req, now=NOW, approval=approval) self.assertEqual(allowed["decision"], "allow") self.assertEqual(replay["reason"], "approval_replayed") def test_invalid_clock_fails_closed(self): result = AuthorityGate().evaluate(grant(), intent(), now=True) self.assertEqual(result["reason"], "clock_invalid") def test_receipt_tampering_is_detected(self): result = AuthorityGate().evaluate(grant(), intent(), now=NOW) changed = copy.deepcopy(result) changed["decision"] = "allow" if result["decision"] == "deny" else "deny" self.assertTrue(verify_receipt(result)) self.assertFalse(verify_receipt(changed)) def test_receipt_shape_is_strict(self): result = AuthorityGate().evaluate(grant(), intent(), now=NOW) changed = copy.deepcopy(result) changed["extra"] = "untrusted" self.assertFalse(verify_receipt(changed)) def test_action_risk_is_policy_derived(self): for action in ("payment", "credential_change", "network_change", "delete", "external_send"): with self.subTest(action=action): req = intent(action=action, tool_id="mail.send", capability="mail.send", resource_id="mail:draft-1", idempotency_key="idem-000001") result = AuthorityGate().evaluate(grant(), req, now=NOW) self.assertEqual(result["risk"], ACTION_RISK[action]) self.assertNotEqual(result["decision"], "allow") def test_resource_wildcards_are_not_accepted_in_grants(self): bad = grant(resources=("*",)) result = AuthorityGate().evaluate(bad, intent(), now=NOW) self.assertEqual(result["reason"], "grant_scope_invalid") def test_mutating_idempotency_key_is_one_time(self): req = intent(action="write", tool_id="files.write", capability="files.write", resource_id="files:workspace-A", idempotency_key="idem-write-0001") gate = AuthorityGate() first = gate.evaluate(grant(), req, now=NOW) second = gate.evaluate(grant(), req, now=NOW) self.assertEqual(first["decision"], "allow") self.assertEqual(second["reason"], "idempotency_replayed") def test_idempotency_keys_are_scoped_per_tenant_principal_and_tool(self): shared_key = "idem-shared-0001" req_a = intent(action="write", tool_id="files.write", capability="files.write", resource_id="files:workspace-A", idempotency_key=shared_key) grant_b = grant(tenant_id="tenant-B", principal_id="agent-B", task_id="task-B", resources=("files:workspace-B",)) req_b = intent(tenant_id="tenant-B", principal_id="agent-B", task_id="task-B", action="write", tool_id="files.write", capability="files.write", resource_id="files:workspace-B", idempotency_key=shared_key) gate = AuthorityGate() result_a = gate.evaluate(grant(), req_a, now=NOW) result_b = gate.evaluate(grant_b, req_b, now=NOW) self.assertEqual(result_a["decision"], "allow") self.assertEqual(result_b["decision"], "allow") def test_non_string_action_fails_closed(self): result = AuthorityGate().evaluate(grant(), intent(action=["payment"]), now=NOW) self.assertEqual(result["reason"], "action_unknown") def test_tool_must_be_explicitly_granted(self): result = AuthorityGate().evaluate(grant(tools=("mail.send",)), intent(), now=NOW) self.assertEqual(result["reason"], "tool_not_granted") def test_effect_digest_must_be_sha256_hex(self): result = AuthorityGate().evaluate(grant(), intent(effect_digest="not-a-digest"), now=NOW) self.assertEqual(result["reason"], "effect_digest_invalid") def test_approval_binds_exact_tool_effect_payload(self): req = intent(action="external_send", tool_id="mail.send", capability="mail.send", resource_id="mail:draft-1", idempotency_key="idem-000001") approval = approval_for(req) substituted = intent(action="external_send", tool_id="mail.send", capability="mail.send", resource_id="mail:draft-1", effect_digest="b" * 64, idempotency_key="idem-000001") result = AuthorityGate(lambda _: True).evaluate(grant(), substituted, now=NOW, approval=approval) self.assertEqual(result["reason"], "approval_intent_mismatch") def test_tool_wildcards_are_not_accepted_in_grants(self): result = AuthorityGate().evaluate(grant(tools=("*",)), intent(), now=NOW) self.assertEqual(result["reason"], "grant_scope_invalid") def test_duplicate_tool_scopes_are_denied(self): result = AuthorityGate().evaluate(grant(tools=("calendar.get", "calendar.get")), intent(), now=NOW) self.assertEqual(result["reason"], "grant_scope_duplicate") def test_concurrent_replay_allows_only_one_use(self): from concurrent.futures import ThreadPoolExecutor req = intent(action="external_send", tool_id="mail.send", capability="mail.send", resource_id="mail:draft-1", idempotency_key="idem-000001") approval = approval_for(req) gate = AuthorityGate(lambda _: True) with ThreadPoolExecutor(max_workers=12) as pool: results = list(pool.map(lambda _: gate.evaluate(grant(), req, now=NOW, approval=approval), range(24))) self.assertEqual(sum(r["decision"] == "allow" for r in results), 1) self.assertEqual(sum(r["reason"] == "approval_replayed" for r in results), 23) def test_duplicate_scope_entries_are_denied(self): bad = grant(resources=("calendar:user-A", "calendar:user-A")) result = AuthorityGate().evaluate(bad, intent(), now=NOW) self.assertEqual(result["reason"], "grant_scope_duplicate") if __name__ == "__main__": unittest.main(verbosity=2)