Coverage for src/lexigram/auth/authz/_check_mixin.py: 22%
208 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-26 00:58 +0800
1"""Authorization check mixin for AuthorizationService."""
3from __future__ import annotations
5from typing import TYPE_CHECKING, Any
7from lexigram.contracts.audit import AuditEntry, AuditLoggerProtocol
8from lexigram.logging import get_logger
9from lexigram.primitives import clock as ambient_clock
10from lexigram.result import Ok, Result
12if TYPE_CHECKING:
13 from lexigram.auth.authz._parsers import ValueParserRegistry
14 from lexigram.auth.exceptions import AuthorizationError
16logger = get_logger(__name__)
19class _AuthCheckMixin:
20 """Mixin providing authorization check methods for AuthorizationService."""
22 # Type annotations for attributes provided by AuthorizationService.__init__.
23 # Declared here so type checkers understand the mixin's assumptions.
24 _roles: dict[str, Any]
25 _policy_engine: Any | None
26 _permission_cache: dict[str, tuple[float, set[str]]]
27 _permission_cache_ttl: float
28 _max_cache_entries: int
29 _audit_logger: AuditLoggerProtocol | None
30 _role_flatten_cache: dict[frozenset[str], tuple[float, set[str]]]
31 _value_parsers: ValueParserRegistry
33 def _parse_list(self, val: Any) -> list[str]:
34 """Parse a value into a list of strings using the registry."""
35 return self._value_parsers.parse(val)
37 async def check_access(
38 self,
39 user: Any,
40 allowed_roles: set[str],
41 resource: str | None = None,
42 action: str | None = None,
43 ) -> bool:
44 """Core access check combining roles, inheritance, and permission patterns."""
45 if "*" in allowed_roles:
46 return True
48 if not user:
49 return False
51 # Gather user roles
52 user_roles = set()
54 if hasattr(user, "roles") and user.roles:
55 user_roles.update(self._parse_list(user.roles))
56 if hasattr(user, "role") and user.role:
57 user_roles.add(str(user.role))
58 # Check superuser/admin bypass
59 if (
60 getattr(user, "is_superuser", False)
61 or "admin" in user_roles
62 or "superuser" in user_roles
63 ):
64 return True
66 # 1. Direct role check
67 if bool(user_roles & allowed_roles):
68 return True
70 # 2. Inheritance check
71 effective_roles = self._flatten_roles(user_roles)
72 if bool(effective_roles & allowed_roles):
73 return True
75 # 3. Permission Pattern check (if resource/action provided)
76 if resource and action:
77 # 1. Gather context for ABAC evaluation
78 context = {
79 "user": user,
80 "resource": resource,
81 "action": action,
82 "resource_id": resource.rsplit(":", maxsplit=1)[-1]
83 if ":" in resource
84 else None,
85 "request": getattr(user, "_request_metadata", {}),
86 "delegations": getattr(user, "delegations", []),
87 }
89 # 2. ABAC Policy evaluation
90 if hasattr(self, "_policy_engine") and self._policy_engine:
91 from lexigram.auth.policies.types import (
92 AuthorizationRequest,
93 DecisionOutcome,
94 )
96 # Principal identity e.g., "user:123"
97 principal = f"user:{getattr(user, 'user_id', 'unknown')}"
99 req = AuthorizationRequest(
100 principal=principal,
101 action=action,
102 resource=resource,
103 context=context,
104 )
106 abac_decision = self._policy_engine.evaluate(req)
107 if abac_decision.decision == DecisionOutcome.DENY:
108 logger.info("ABAC: Access DENIED by policy for %s", principal)
109 return False
110 if abac_decision.decision == DecisionOutcome.ALLOW:
111 logger.info("ABAC: Access ALLOWED by policy for %s", principal)
112 return True
114 # 3. Permission Pattern check (Fallback if no ABAC policy matched)
115 user_id = getattr(user, "user_id", None) or getattr(user, "id", None)
116 now = ambient_clock.monotonic()
118 user_perms = None
119 if user_id:
120 cache_entry = self._permission_cache.get(str(user_id))
121 if cache_entry:
122 expiry, cached_perms = cache_entry
123 if expiry > now:
124 user_perms = cached_perms
126 if user_perms is None:
127 user_perms = self._get_user_permissions(effective_roles)
128 # Mix in direct user permissions
129 if hasattr(user, "permissions") and user.permissions:
130 user_perms.update(self._parse_list(user.permissions))
132 if user_id:
133 # Bounded cache: evict the oldest entry (by insertion order)
134 # when the cache is at capacity. Python 3.7+ dicts preserve
135 # insertion order, so ``next(iter(...))`` is always the entry
136 # that was added earliest. The effective eviction policy is
137 # FIFO (first-in, first-out), not true recency-based LRU.
138 # ``max_cache_entries`` (default 10 000) caps memory usage.
139 if len(self._permission_cache) >= self._max_cache_entries:
140 oldest_key = next(iter(self._permission_cache))
141 del self._permission_cache[oldest_key]
142 self._permission_cache[str(user_id)] = (
143 now + self._permission_cache_ttl,
144 user_perms,
145 )
147 required_perm = f"{resource}.{action}"
148 if self._has_permission(user_perms, required_perm):
149 return True
151 # 4. Permission Delegation check
152 delegations = context.get("delegations")
153 # Also check user.delegations if injected there
154 if not delegations and hasattr(user, "delegations"):
155 delegations = user.delegations
157 if delegations:
158 for delegation in delegations:
159 # Match Action and Resource
160 if (
161 action in delegation.permissions
162 or "*" in delegation.permissions
163 ) and self._has_permission(set(delegation.resources), resource):
164 logger.info(
165 "Access ALLOWED via delegation %s (from %s)",
166 delegation.delegation_id,
167 delegation.delegator_id,
168 )
169 return True
171 return False
173 def has_any_role(self, user: Any, roles: list[str]) -> bool:
174 """Check if user has any of the given roles."""
175 if not user:
176 return False
178 user_roles = set()
179 user_roles_attr = (
180 user.get("roles")
181 if isinstance(user, dict)
182 else getattr(user, "roles", None)
183 )
184 if user_roles_attr:
185 user_roles.update(self._parse_list(user_roles_attr))
186 user_role_attr = (
187 user.get("role") if isinstance(user, dict) else getattr(user, "role", None)
188 )
189 if user_role_attr:
190 user_roles.add(str(user_role_attr))
192 # Check superuser/admin bypass
193 is_super = (
194 user.get("is_superuser", False)
195 if isinstance(user, dict)
196 else getattr(user, "is_superuser", False)
197 )
198 if is_super or "admin" in user_roles or "superuser" in user_roles:
199 return True
201 effective_roles = self._flatten_roles(user_roles)
202 return bool(effective_roles.intersection(roles))
204 def has_any_permission(self, user: Any, permissions: list[str]) -> bool:
205 """Return True if user has at least one of the given permissions."""
206 if not user:
207 return False
209 user_roles = set()
210 user_roles_attr = (
211 user.get("roles")
212 if isinstance(user, dict)
213 else getattr(user, "roles", None)
214 )
215 if user_roles_attr:
216 user_roles.update(self._parse_list(user_roles_attr))
217 user_role_attr = (
218 user.get("role") if isinstance(user, dict) else getattr(user, "role", None)
219 )
220 if user_role_attr:
221 user_roles.add(str(user_role_attr))
223 # Check superuser/admin bypass
224 is_super = (
225 user.get("is_superuser", False)
226 if isinstance(user, dict)
227 else getattr(user, "is_superuser", False)
228 )
229 if is_super or "admin" in user_roles or "superuser" in user_roles:
230 return True
232 effective_roles = self._flatten_roles(user_roles)
233 user_perms = self._get_user_permissions(effective_roles)
235 user_perms_attr = (
236 user.get("permissions")
237 if isinstance(user, dict)
238 else getattr(user, "permissions", None)
239 )
240 if user_perms_attr:
241 user_perms.update(self._parse_list(user_perms_attr))
243 for req_perm in permissions:
244 if self._has_permission(user_perms, req_perm):
245 return True
247 return False
249 async def can(self, user: Any, action: str, resource: str) -> bool:
250 """Convenience alias: return True if user can perform action on resource.
252 Delegates to :meth:`authorize` and unwraps the result, returning
253 ``False`` on ``Err`` (i.e. if the authorization check itself fails).
254 """
255 result = await self.authorize(user, action, resource)
256 return result.unwrap_or(False)
258 async def can_view(self, user: Any, resource: str, record: Any = None) -> bool:
259 """Grant view when the user holds ``<resource>.view`` (record ignored for now)."""
260 return self.has_any_permission(user, [f"{resource}.view"])
262 async def can_create(self, user: Any, resource: str) -> bool:
263 """Grant create when the user holds ``<resource>.create``."""
264 return self.has_any_permission(user, [f"{resource}.create"])
266 async def can_update(self, user: Any, resource: str, record: Any = None) -> bool:
267 """Grant update when the user holds ``<resource>.update`` (record ignored for now)."""
268 return self.has_any_permission(user, [f"{resource}.update"])
270 async def can_delete(self, user: Any, resource: str, record: Any = None) -> bool:
271 """Grant delete when the user holds ``<resource>.delete`` (record ignored for now)."""
272 return self.has_any_permission(user, [f"{resource}.delete"])
274 async def can_execute_action(
275 self, user: Any, resource: str, action: str, record: Any | None = None
276 ) -> bool:
277 """Grant action execution when the user holds ``<resource>.<action>``."""
278 return self.has_any_permission(user, [f"{resource}.{action}"])
280 async def authorize(
281 self, user: Any, action: str, resource: Any
282 ) -> Result[bool, AuthorizationError]:
283 """Check whether *user* is allowed to perform *action* on *resource*.
285 This is the primary authorization method. Satisfies the
286 :class:`~lexigram.contracts.auth.guard.AuthorizerProtocol` protocol (note:
287 the contracts protocol declares ``-> bool``; this implementation
288 returns ``Result[bool, AuthorizationError]`` to make failures
289 explicit. Use :meth:`can` when a plain ``bool`` is required).
291 ``Ok(True)`` — access granted.
292 ``Ok(False)`` — access denied.
293 ``Err(AuthorizationError)`` — the authorization check itself failed
294 (e.g. policy engine error); callers should treat this as denied.
296 Args:
297 user: Authenticated user object.
298 action: Action identifier (e.g. ``"read"``).
299 resource: Target resource; coerced to :class:`str` for matching.
301 Returns:
302 ``Result[bool, AuthorizationError]``
303 """
304 granted = await self.check_access(user, set(), str(resource), action)
306 if self._audit_logger is not None:
307 actor_id = str(
308 getattr(user, "user_id", None) or getattr(user, "id", None) or "unknown"
309 )
310 await self._audit_logger.log(
311 AuditEntry(
312 action="authz.decision",
313 actor_id=actor_id,
314 resource_type=type(resource).__name__,
315 resource_id=str(resource),
316 outcome="granted" if granted else "denied",
317 metadata={"authz_action": action},
318 )
319 )
321 return Ok(granted)
323 # --- Internal Helpers ---
325 def _get_effective_roles(self, user_roles: set[str]) -> set[str]:
326 """Resolve all roles including inherited ones."""
327 effective = set(user_roles)
328 to_process = list(user_roles)
329 processed = set()
331 while to_process:
332 role_name = to_process.pop()
333 if role_name in processed:
334 continue
335 processed.add(role_name)
337 role_def = self._roles.get(role_name)
338 if role_def:
339 inherits = getattr(role_def, "inherits", [])
340 if not inherits and isinstance(role_def, dict):
341 inherits = role_def.get("inherits", [])
343 for parent in inherits:
344 effective.add(parent)
345 to_process.append(parent)
346 return effective
348 def _get_user_permissions(self, effective_roles: set[str]) -> set[str]:
349 """Flatten all permissions from effective roles."""
350 permissions = set()
351 for role_name in effective_roles:
352 role_def = self._roles.get(role_name)
353 if role_def is not None:
354 perms = []
355 if hasattr(role_def, "permissions"):
356 perms = role_def.permissions
357 elif isinstance(role_def, dict):
358 perms = role_def.get("permissions", [])
360 if perms:
361 permissions.update(perms)
362 return permissions
364 def _has_permission(self, user_permissions: set[str], required: str) -> bool:
365 """Check if required permission matches any user permission patterns.
367 Supports bidirectional wildcard matching:
368 1. User has wildcard: user='admin.*', required='admin.users' -> True
369 2. Requirement has wildcard: user='community.list', required='community.*' -> True
370 """
371 if "*" in user_permissions:
372 return True
374 if required in user_permissions:
375 return True
377 # Helper to check if a pattern matches a string
378 def matches(pattern: str, string: str) -> bool:
379 if "*" not in pattern:
380 return pattern == string
381 parts = pattern.split("*")
382 # For simplicity, we only support a single wildcard for now (standard for the framework)
383 prefix, suffix = parts[0], parts[1] if len(parts) > 1 else ""
384 return string.startswith(prefix) and string.endswith(suffix)
386 # Check if any user permission pattern matches the required string
387 for user_perm in user_permissions:
388 if "*" in user_perm and matches(user_perm, required):
389 return True
391 # Check if the required string is a pattern that matches any user permission
392 if "*" in required:
393 for user_perm in user_permissions:
394 if matches(required, user_perm):
395 return True
397 return False
399 def _flatten_roles(self, role_names: set[str]) -> set[str]:
400 """Recursively collect all parent roles for the given set of role names.
402 This helper walks the role inheritance graph stored in ``self._roles`` and
403 returns a set containing the original roles plus any roles they inherit
404 from, directly or indirectly.
406 Results are memoized for performance.
407 """
408 # OPT-AUTH-1: Use memoization cache
409 cache_key = frozenset(role_names)
410 now = ambient_clock.monotonic()
412 if cache_key in self._role_flatten_cache:
413 expiry, cached_roles = self._role_flatten_cache[cache_key]
414 if expiry > now:
415 return cached_roles
417 result: set[str] = set(role_names)
418 to_process: list[str] = list(role_names)
419 while to_process:
420 name = to_process.pop()
421 role_def = self._roles.get(name)
422 if not role_def:
423 continue
424 # ``inherits`` may be an attribute or a dict key depending on the role type
425 inherits = (
426 getattr(role_def, "inherits", None)
427 if hasattr(role_def, "inherits")
428 else None
429 )
430 if inherits is None:
431 inherits = (
432 role_def.get("inherits", []) if isinstance(role_def, dict) else []
433 )
434 for parent in inherits:
435 if parent not in result:
436 result.add(parent)
437 to_process.append(parent)
439 self._role_flatten_cache[cache_key] = (now + self._permission_cache_ttl, result)
440 return result
443__all__ = ["_AuthCheckMixin"]