Coverage for src/lexigram/auth/policies/engine.py: 76%
154 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-26 00:58 +0800
1"""Policy Engine for ABAC evaluation."""
3from __future__ import annotations
5import re
6import threading
7from typing import TYPE_CHECKING, Literal, Protocol
9from lexigram.auth.policies.evaluator import ConditionEvaluator
10from lexigram.auth.policies.types import (
11 AuthorizationDecision,
12 AuthorizationRequest,
13 DecisionOutcome,
14 Policy,
15 PolicyEffect,
16)
17from lexigram.logging import get_logger
19if TYPE_CHECKING:
20 from lexigram.contracts.auth.policy import PolicyStoreProtocol
22logger = get_logger(__name__)
25# =============================================================================
26# Pattern Matcher Registry
27# =============================================================================
30class PatternMatcher(Protocol):
31 """Protocol for pattern matchers."""
33 def can_match(self, pattern: str) -> bool:
34 """Check if this matcher can handle the given pattern."""
35 ...
37 def matches(self, pattern: str, target: str) -> bool:
38 """Check if the target matches the pattern."""
39 ...
42class ExactPatternMatcher:
43 """Matches exact string patterns without wildcards."""
45 def can_match(self, pattern: str) -> bool:
46 return "*" not in pattern
48 def matches(self, pattern: str, target: str) -> bool:
49 return pattern == target
52class WildcardPatternMatcher:
53 """Matches wildcard patterns using regex."""
55 def can_match(self, pattern: str) -> bool:
56 return "*" in pattern
58 def matches(self, pattern: str, target: str) -> bool:
59 regex = "^" + pattern.replace("*", ".*") + "$"
60 return bool(re.match(regex, target))
63class GlobPatternMatcher:
64 """Matches glob-style patterns (e.g., 'user.*' matches 'user.read')."""
66 def can_match(self, pattern: str) -> bool:
67 return "." in pattern and "*" in pattern
69 def matches(self, pattern: str, target: str) -> bool:
70 # Convert glob pattern to regex
71 regex_pattern = pattern.replace(".", r"\.").replace("*", ".*")
72 regex = f"^{regex_pattern}$"
73 return bool(re.match(regex, target))
76class PatternMatcherRegistry:
77 """Registry for pattern matchers.
79 Provides extensible pattern matching for policy evaluation.
80 """
82 def __init__(self) -> None:
83 self._lock = threading.Lock()
84 self._matchers: list[PatternMatcher] = []
86 @classmethod
87 def with_defaults(cls) -> PatternMatcherRegistry:
88 """Create a registry pre-loaded with the standard pattern matchers."""
89 instance = cls()
90 instance._register_default_matchers()
91 return instance
93 def _register_default_matchers(self) -> None:
94 """Register the default pattern matchers."""
95 self._matchers = [
96 ExactPatternMatcher(),
97 WildcardPatternMatcher(),
98 ]
100 def register_matcher(self, matcher: PatternMatcher) -> None:
101 """Register a custom pattern matcher."""
102 with self._lock:
103 self._matchers.insert(0, matcher)
105 def matches(self, pattern: str, target: str) -> bool:
106 """Check if the target matches the pattern using registered matchers."""
107 with self._lock:
108 matchers = list(self._matchers)
109 for matcher in matchers:
110 if matcher.can_match(pattern):
111 return matcher.matches(pattern, target)
112 return False
115class PolicyEngine:
116 """Evaluates authorization requests against a collection of policies.
118 Three evaluation strategies are supported, selected at construction time via
119 the ``strategy`` parameter:
121 * ``"deny_first"`` *(default)* — the first **DENY** match short-circuits
122 evaluation and the request is immediately rejected. Any subsequent
123 ALLOW policies are never reached. This is the most secure default and
124 the one mandated by most enterprise security frameworks.
126 * ``"allow_first"`` — the first **ALLOW** match short-circuits evaluation
127 and the request is immediately granted. Useful for additive permission
128 models where policies are non-conflicting.
130 * ``"unanimous"`` — every matching policy must be an ALLOW; a single DENY
131 (or INDETERMINATE, i.e. no matching ALLOW) causes the request to be
132 denied. Suitable for high-security contexts where all gates must pass.
133 """
135 def __init__(
136 self,
137 policies: list[Policy] | None = None,
138 *,
139 store: PolicyStoreProtocol | None = None,
140 strategy: Literal["deny_first", "allow_first", "unanimous"] = "deny_first",
141 ) -> None:
142 """Initialise the policy engine with a static list and/or a store.
144 Args:
145 policies: Optional list of in-memory policies. Sorted by
146 priority (descending) so that the highest-priority policy
147 wins on conflict.
148 store: Optional :class:`~lexigram.contracts.auth.policy.PolicyStoreProtocol`
149 used to load/persist policies asynchronously. When provided,
150 call :meth:`load_from_store` during application boot to merge
151 the stored policies with any static ones.
152 strategy: Evaluation strategy controlling short-circuit behaviour.
154 * ``"deny_first"`` — first DENY wins (most secure, default).
155 * ``"allow_first"`` — first ALLOW wins (additive model).
156 * ``"unanimous"`` — all matching policies must be ALLOW.
157 """
158 self.policies: list[Policy] = sorted(
159 policies or [],
160 key=lambda p: p.priority,
161 reverse=True,
162 )
163 self._store: PolicyStoreProtocol | None = store
164 self._strategy: Literal["deny_first", "allow_first", "unanimous"] = strategy
165 self.evaluator = ConditionEvaluator()
166 self._pattern_matchers = PatternMatcherRegistry.with_defaults()
167 # Resource-pattern index for O(1) candidate lookup in evaluate().
168 # Keys are exact resource patterns; the special key "" covers catch-all
169 # policies (empty resources list). Wildcard-containing patterns are
170 # stored separately in _wildcard_resource_policies so they are always
171 # checked via pattern matching.
172 self._resource_index: dict[str, list[Policy]] = {}
173 self._wildcard_resource_policies: list[Policy] = []
174 self._rebuild_resource_index()
176 def __repr__(self) -> str:
177 """Return developer-friendly string representation."""
178 return f"PolicyEngine(policies={len(self.policies)})"
180 def _rebuild_resource_index(self) -> None:
181 """Rebuild the resource-pattern index from ``self.policies``.
183 Exact resource patterns are indexed for O(1) lookup. Wildcard
184 patterns (containing ``*``) are gathered in
185 ``_wildcard_resource_policies`` so they are always evaluated via
186 the full pattern-match path. Catch-all policies (empty
187 ``resources`` list) are stored under the ``""`` key.
188 """
189 index: dict[str, list[Policy]] = {}
190 wildcard: list[Policy] = []
191 for policy in self.policies:
192 if not policy.resources:
193 index.setdefault("", []).append(policy)
194 else:
195 has_wildcard = False
196 for pattern in policy.resources:
197 if "*" in pattern:
198 has_wildcard = True
199 else:
200 index.setdefault(pattern, []).append(policy)
201 if has_wildcard:
202 wildcard.append(policy)
203 self._resource_index = index
204 self._wildcard_resource_policies = wildcard
206 async def load_from_store(self) -> None:
207 """Load policies from the configured :class:`PolicyStoreProtocol`.
209 Fetches all policies from the store and merges them with any
210 statically registered policies, re-sorting the combined list by
211 priority. A no-op if no store was provided at construction time.
213 Raises:
214 RuntimeError: If no store was provided when this method is called.
215 """
216 if self._store is None:
217 return
218 stored = await self._store.load_policies()
219 merged = {**{p.name: p for p in self.policies}, **{p.name: p for p in stored}}
220 self.policies = sorted(merged.values(), key=lambda p: p.priority, reverse=True)
221 self._rebuild_resource_index()
222 logger.info("policy_engine.loaded_from_store", count=len(stored))
224 async def save_policy(self, policy: Policy) -> None:
225 """Persist a policy to the store and add it to the in-memory list.
227 Args:
228 policy: The policy to persist.
230 Raises:
231 RuntimeError: If no store was configured.
232 """
233 if self._store is None:
234 msg = "Cannot persist policy: PolicyEngine has no store configured"
235 raise RuntimeError(msg)
236 await self._store.save_policy(policy)
237 # Merge into in-memory list (replace any existing policy with same name)
238 self.policies = sorted(
239 [p for p in self.policies if p.name != policy.name] + [policy],
240 key=lambda p: p.priority,
241 reverse=True,
242 )
243 self._rebuild_resource_index()
244 logger.info("policy_engine.policy_saved", policy=policy.name)
246 def evaluate(self, request: AuthorizationRequest) -> AuthorizationDecision:
247 """Evaluate an authorization request against loaded policies.
249 Short-circuit behaviour is governed by ``self._strategy``:
251 * ``deny_first`` — first DENY match returns immediately.
252 * ``allow_first`` — first ALLOW match returns immediately.
253 * ``unanimous`` — every matching policy must be ALLOW; a single DENY
254 or the absence of any ALLOW yields DENY/INDETERMINATE respectively.
255 """
256 # Build candidate set using the resource index to avoid a full scan.
257 # Priority order is preserved via sorted self.policies insertion order.
258 seen: set[str] = set()
259 candidates: list[Policy] = []
261 # 1. Catch-all policies (empty resources list)
262 for p in self._resource_index.get("", []):
263 if p.policy_id not in seen:
264 seen.add(p.policy_id)
265 candidates.append(p)
267 # 2. Exact resource-pattern match
268 for p in self._resource_index.get(request.resource, []):
269 if p.policy_id not in seen:
270 seen.add(p.policy_id)
271 candidates.append(p)
273 # 3. Wildcard resource patterns (must still be pattern-matched below)
274 for p in self._wildcard_resource_policies:
275 if p.policy_id not in seen:
276 seen.add(p.policy_id)
277 candidates.append(p)
279 # Re-sort the smaller candidate set by priority (descending)
280 candidates.sort(key=lambda p: p.priority, reverse=True)
282 matched_policies: list[str] = []
283 allow_found = False
285 for policy in candidates:
286 if not self._matches(policy, request):
287 continue
289 matched_policies.append(policy.policy_id)
291 if self._strategy == "deny_first":
292 # First DENY wins — most secure default.
293 if policy.effect == PolicyEffect.DENY:
294 logger.info("Access DENIED by policy (deny_first): %s", policy.name)
295 return AuthorizationDecision(
296 decision=DecisionOutcome.DENY,
297 reason=f"Denied by policy: {policy.name}",
298 applied_policies=matched_policies,
299 )
300 if policy.effect == PolicyEffect.ALLOW:
301 allow_found = True
303 elif self._strategy == "allow_first":
304 # First ALLOW wins — additive model, stops on first grant.
305 if policy.effect == PolicyEffect.ALLOW:
306 logger.info(
307 "Access ALLOWED by policy (allow_first): %s", policy.name
308 )
309 return AuthorizationDecision(
310 decision=DecisionOutcome.ALLOW,
311 reason=f"Allowed by policy: {policy.name}",
312 applied_policies=matched_policies,
313 )
314 if policy.effect == PolicyEffect.DENY:
315 allow_found = False # keep scanning but track deny seen
317 else: # unanimous
318 # Every matching policy must be ALLOW; a single DENY short-circuits.
319 if policy.effect == PolicyEffect.DENY:
320 logger.info("Access DENIED by policy (unanimous): %s", policy.name)
321 return AuthorizationDecision(
322 decision=DecisionOutcome.DENY,
323 reason=f"Denied by policy: {policy.name}",
324 applied_policies=matched_policies,
325 )
326 if policy.effect == PolicyEffect.ALLOW:
327 allow_found = True
329 if allow_found:
330 return AuthorizationDecision(
331 decision=DecisionOutcome.ALLOW,
332 applied_policies=matched_policies,
333 )
335 return AuthorizationDecision(
336 decision=DecisionOutcome.INDETERMINATE,
337 reason="No matching policies found",
338 applied_policies=matched_policies,
339 )
341 def _matches(self, policy: Policy, request: AuthorizationRequest) -> bool:
342 """Check if a policy applies to the given request."""
343 # 1. Match Action
344 if not self._pattern_match(policy.actions, request.action):
345 return False
347 # 2. Match Resource
348 if not self._pattern_match(policy.resources, request.resource):
349 return False
351 # 3. Match Principal
352 if not self._pattern_match(policy.principals, request.principal):
353 return False
355 # 4. Evaluate Conditions
356 for cond in policy.conditions:
357 if not self.evaluator.evaluate(cond, request.context):
358 return False
360 return True
362 def _pattern_match(self, patterns: list[str], target: str) -> bool:
363 """Check if target matches any of the patterns using the registry."""
364 if not patterns:
365 return True # Empty means matches all
367 for pattern in patterns:
368 if self._pattern_matchers.matches(pattern, target):
369 return True
371 return False
374__all__ = [
375 "ExactPatternMatcher",
376 "GlobPatternMatcher",
377 "PatternMatcher",
378 "PatternMatcherRegistry",
379 "PolicyEngine",
380 "WildcardPatternMatcher",
381 "logger",
382]