from __future__ import annotations import unittest from uuid import UUID, uuid4 from connectors.unraid import FakeBrokerClient, RegisteredContainer, UnraidConnector from packages.connector_kit import ConnectorContext from packages.hub_policy import ( Decision, Governance, InMemoryAuditLog, InvocationContext, PolicyDenied, PolicyEngine, PolicyRule, PolicySimulator, SlidingWindowRateLimiter, ) PROJECT_ID = UUID("33333333-3333-3333-3333-333333333333") CLIENT_ID = UUID("55555555-5555-5555-5555-555555555555") def invocation( *, project: str = "ludarium", environment: str = "dev", binding: bool = True, tool: str = "unraid.host.status", arguments: dict[str, object] | None = None, ) -> InvocationContext: return InvocationContext( workspace="itworx", project=project, environment=environment, principal="svc-ludarium", client_id=CLIENT_ID, client_enabled=binding, binding_present=binding, tool=tool, connector="unraid", risk_class="R1", arguments=arguments or {}, ) class WP110GovernanceTests(unittest.TestCase): def test_merge_result_is_retained_in_append_only_audit(self) -> None: audit = InMemoryAuditLog() engine = PolicyEngine( [ PolicyRule( slug="allow-merge", project="ludarium", priority=10, description="allow merge", match={"tool": "gitea.pullrequest.merge"}, decision=Decision.ALLOW, ) ] ) governance = Governance(engine, audit) evidence = { "repositoryRef": "test-repo", "pr": 7, "headSha": "e" * 40, "baseSha": "a" * 40, "requiredContext": "Managed validation / full (pull_request)", "validationRunId": 42, "validationEvent": "pull_request", "validationConclusion": "success", "mergeMethod": "squash", "resultingMainSha": "6" * 40, } result = governance.invoke( invocation(tool="gitea.pullrequest.merge"), lambda: {"data": {"auditEvidence": evidence}}, ) self.assertEqual(result["data"]["auditEvidence"], evidence) self.assertEqual(audit.events[0].sanitized_arguments["_result"], evidence) def test_global_terminal_deny_cannot_be_overridden(self) -> None: engine = PolicyEngine( [ PolicyRule( slug="global-block", priority=1, description="block", match={"toolPatterns": ["unraid.*"]}, decision=Decision.DENY, terminal=True, ), PolicyRule( slug="project-allow", project="ludarium", priority=100, description="allow", match={"project": "ludarium"}, decision=Decision.ALLOW, ), ] ) result = engine.evaluate(invocation()) self.assertEqual(result.decision, Decision.DENY) self.assertEqual(result.reason_code, "GLOBAL_DENY") def test_missing_binding_and_default_deny_are_explainable(self) -> None: engine = PolicyEngine( [ PolicyRule( slug="allow", project="ludarium", priority=10, description="allow", match={"project": "ludarium"}, decision=Decision.ALLOW, ) ] ) missing = engine.evaluate(invocation(binding=False)) self.assertEqual(missing.reason_code, "MISSING_BINDING") unknown = engine.evaluate(invocation(project="other")) self.assertEqual(unknown.reason_code, "DEFAULT_DENY") def test_non_terminal_global_default_deny_allows_a_specific_bound_rule(self) -> None: engine = PolicyEngine( [ PolicyRule( slug="project-allow", project="ludarium", priority=10, description="allow a bound project read", match={"riskClass": "R1"}, conditions={"requireBinding": True, "requireEnabledClient": True}, decision=Decision.ALLOW_IF_CONDITIONS, ), PolicyRule( slug="global-deny", priority=0, description="default deny fallback", decision=Decision.DENY, terminal=False, ), ] ) self.assertEqual(engine.evaluate(invocation()).decision, Decision.ALLOW_IF_CONDITIONS) def test_dev_policy_does_not_imply_prod(self) -> None: engine = PolicyEngine( [ PolicyRule( slug="dev-only", project="ludarium", priority=10, description="dev", match={"environment": "dev"}, decision=Decision.ALLOW, ) ] ) self.assertEqual(engine.evaluate(invocation(environment="dev")).decision, Decision.ALLOW) self.assertEqual(engine.evaluate(invocation(environment="prod")).decision, Decision.DENY) def test_simulator_rejects_unknown_entities_and_runtime_matches(self) -> None: engine = PolicyEngine( [ PolicyRule( slug="allow", project="ludarium", priority=10, description="allow", match={"tool": "unraid.host.status"}, decision=Decision.ALLOW, ) ] ) simulator = PolicySimulator( engine, known_projects=("ludarium",), known_tools=("unraid.host.status",) ) self.assertEqual( simulator.simulate(invocation()).decision, engine.evaluate(invocation()).decision ) self.assertEqual( simulator.simulate(invocation(project="unknown")).reason_code, "UNKNOWN_PROJECT" ) def test_audit_is_sanitized_append_only_and_paginated(self) -> None: audit = InMemoryAuditLog() engine = PolicyEngine( [ PolicyRule( slug="allow", project="ludarium", priority=10, description="allow", match={"project": "ludarium"}, decision=Decision.ALLOW, ) ] ) governance = Governance(engine, audit) governance.invoke( invocation( arguments={ "containerRef": "web", "apiToken": "do-not-store", "note": "Bearer do-not-store", } ), lambda: {"ok": True}, ) event = audit.search(workspace="itworx", project="ludarium", limit=1).items[0] self.assertEqual(event.sanitized_arguments["apiToken"], "") self.assertNotIn("do-not-store", event.model_dump_json()) self.assertEqual(event.policy_trace[0]["phase"], "match") self.assertEqual( audit.export_metadata(workspace="itworx", project="ludarium")["eventCount"], 1 ) def test_rate_limit_key_includes_client_project_and_tool(self) -> None: limiter = SlidingWindowRateLimiter(limit=1) self.assertTrue(limiter.allow(client_id=CLIENT_ID, project="a", tool="tool")) self.assertFalse(limiter.allow(client_id=CLIENT_ID, project="a", tool="tool")) self.assertTrue(limiter.allow(client_id=CLIENT_ID, project="b", tool="tool")) self.assertTrue(limiter.allow(client_id=CLIENT_ID, project="a", tool="other")) def test_connector_policy_denies_before_target_and_emits_one_event(self) -> None: audit = InMemoryAuditLog() engine = PolicyEngine( [ PolicyRule( slug="allow", project="ludarium", priority=10, description="allow", match={"tool": "unraid.host.status"}, decision=Decision.ALLOW, ) ] ) governance = Governance(engine, audit) connector = UnraidConnector( FakeBrokerClient(), (RegisteredContainer("ludarium-web", PROJECT_ID, "prod"),), governance, ) context = ConnectorContext( workspace_id=uuid4(), project_id=PROJECT_ID, project_slug="ludarium", environment="prod", client_id=CLIENT_ID, ) result = connector.host_status(context) self.assertEqual(result["data"]["hostRef"], "unraid") self.assertEqual(len(audit.events), 1) with self.assertRaises(PolicyDenied): connector.invoke("unraid.container.inspect", context, container_ref="ludarium-web") self.assertEqual(len(audit.events), 2) if __name__ == "__main__": unittest.main()