Coverage for src/lexigram/auth/authz/service.py: 36%
116 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-26 00:58 +0800
1"""Unified Authorization Service for Lexigram.
3This service consolidates RBAC (Role-Based Access Control) logic, combining
4the best features of permission management, role hierarchies, and ABAC
5(Attribute-Based Access Control) policies.
7It serves as the single source of truth for all authorization decisions
8in a Lexigram application.
10Example:
11 Checking authorization::
13 from lexigram.auth.authz import AuthorizationService
15 auth_service = AuthorizationService()
17 # Define roles with inheritance
18 auth_service.set_roles({
19 "admin": {"permissions": ["*"]},
20 "editor": {"inherits": ["viewer"], "permissions": ["articles.write"]},
21 "viewer": {"permissions": ["articles.read"]},
22 })
24 # Authorize a user action
25 result = await auth_service.authorize(user, "articles", "read")
26 if result.is_ok():
27 # User can read articles
28 pass
30 Using with dependency injection::
32 from lexigram.di import inject
34 class ArticleController(Controller):
35 @inject
36 def __init__(self, auth: AuthorizationService):
37 self.auth = auth
39 async def check_permission(self, user, resource: str, action: str) -> bool:
40 result = await self.auth.authorize(user, resource, action)
41 return result.is_ok()
43See Also:
44 - :class:`lexigram.auth.policies.engine.PolicyEngine`: ABAC policy engine.
45 - :class:`lexigram.auth.types.RoleDefinition`: Role definition model.
46"""
48from __future__ import annotations
50import asyncio
51from typing import TYPE_CHECKING, Any, Protocol, runtime_checkable
53from lexigram.auth.authz._check_mixin import _AuthCheckMixin
54from lexigram.auth.authz._parsers import (
55 ListValueParser,
56 NoneValueParser,
57 StringValueParser,
58 ValueParser,
59 ValueParserRegistry,
60)
61from lexigram.auth.policies.engine import PolicyEngine
62from lexigram.auth.types import RoleDefinition
63from lexigram.contracts.exceptions import DependencyError, UnresolvableDependencyError
64from lexigram.logging import get_logger
66if TYPE_CHECKING:
67 from lexigram.contracts.audit import AuditLoggerProtocol
70# Use local types if available, otherwise define minimal protocols
71@runtime_checkable
72class UserProtocol(Protocol):
73 roles: list[str]
74 permissions: list[str]
77logger = get_logger(__name__)
80class AuthorizationService(_AuthCheckMixin):
81 """Central service for all authorization checks."""
83 def __init__(
84 self,
85 permission_cache_ttl: float = 300.0,
86 *,
87 max_cache_entries: int = 10000,
88 audit_logger: AuditLoggerProtocol | None = None,
89 ) -> None:
90 """Initialize a new authorization service instance.
92 Args:
93 permission_cache_ttl: TTL in seconds for the per-user permission
94 cache. Defaults to 300 seconds (5 minutes).
95 max_cache_entries: Maximum number of entries in the permission cache.
96 When the cache reaches this size the oldest entry (by insertion
97 order) is evicted before adding a new one. Defaults to 10 000.
98 audit_logger: Optional :class:`~lexigram.contracts.audit.AuditLoggerProtocol`
99 used to record authorization decisions (granted / denied).
100 No audit entries are written when *None*.
101 """
102 self._lock = asyncio.Lock()
103 self._roles: dict[str, Any] = {}
104 self._schemas: dict[str, Any] = {}
105 self._policy_engine: Any | None = None
106 self._delegation_manager: Any | None = None
107 self._value_parsers = ValueParserRegistry()
108 self._role_flatten_cache: dict[frozenset[str], tuple[float, set[str]]] = {}
109 self._permission_cache: dict[str, tuple[float, set[str]]] = {}
110 self._permission_cache_ttl = permission_cache_ttl
111 self._max_cache_entries = max_cache_entries
112 self._audit_logger: AuditLoggerProtocol | None = audit_logger
114 def set_policies(self, policies: list[Any]) -> None:
115 """Set the ABAC policies and initialize the engine."""
116 self._policy_engine = PolicyEngine(policies)
117 logger.info("✓ ABAC Policy Engine configured with %d policies", len(policies))
119 def set_roles(self, roles: dict[str, RoleDefinition | dict[str, Any]]) -> None:
120 """Set the global role definitions (usually from config/seed)."""
121 for name, role in roles.items():
122 self.register_role(name, role)
123 logger.info("✓ RBAC Roles configured with %d definitions", len(self._roles))
125 def __repr__(self) -> str:
126 """Return developer-friendly string representation."""
127 return f"AuthorizationService(roles={len(self._roles)}, schemas={len(self._schemas)})"
129 def register_role(self, name: str, role: RoleDefinition | dict[str, Any]) -> None:
130 """Register a role definition."""
131 # OPT-AUTH-1: Invalidate cache when roles change
132 self._role_flatten_cache.clear()
134 if isinstance(role, dict):
135 role_def = RoleDefinition(
136 name=name,
137 description=role.get("description", ""),
138 permissions=self._parse_list(role.get("permissions", [])),
139 inherits=self._parse_list(role.get("inherits", [])),
140 )
141 else:
142 role_def = role
144 self._roles[name] = role_def
145 # Invalidate caches
146 self._role_flatten_cache.clear()
147 self._permission_cache.clear()
148 logger.debug("Registered role: %s (inherits: %s)", name, role_def.inherits)
150 def get_role(self, name: str) -> Any | None:
151 return self._roles.get(name)
153 async def sync_from_db(self, container: Any) -> None:
154 """Load roles from database and merge with existing (YAML) roles."""
155 # Dynamic import to avoid circular deps
156 from lexigram.contracts.data import DatabaseProviderProtocol
158 # GenericRepository is a concrete class from lexigram-sql.
159 # We resolve it dynamically to avoid top-level sibling dependencies.
161 # We assume a standard Role entity exists or use a generic dict approach
162 # For now, strict typing is relaxed to allow flexible adoption
164 db = await container.resolve(DatabaseProviderProtocol)
165 if db:
166 # Try to query a standard roles table if it exists
167 try:
168 # 1. Try 'roles' first (modern standard)
169 try:
170 generic_repository = await container.resolve(
171 "GenericRepository",
172 )
173 except (
174 DependencyError,
175 UnresolvableDependencyError,
176 KeyError,
177 AttributeError,
178 ):
179 # GenericRepository not registered — skip DB role sync
180 logger.debug(
181 "GenericRepository not found in container; skipping DB role sync",
182 )
183 return
185 repo = generic_repository(db, "roles", dict)
186 db_roles = await repo.find_many()
187 except (RuntimeError, OSError, LookupError):
188 # 2. Fallback to 'admin_roles' (legacy)
189 try:
190 repo = generic_repository(db, "admin_roles", dict)
191 db_roles = await repo.find_many()
192 except (RuntimeError, OSError, LookupError) as e:
193 logger.debug(
194 "Table 'admin_roles' not found or inaccessible: %s",
195 e,
196 )
197 db_roles = []
199 for role in db_roles:
200 # DB takes priority over YAML if same name
201 if isinstance(role, dict):
202 name = role.get("name")
203 if name:
204 self.register_role(name, role)
205 else:
206 name = getattr(role, "name", None)
207 if name:
208 self.register_role(name, role)
210 if db_roles:
211 logger.info(
212 "✓ RBAC Roles synced from database (%d roles)",
213 len(db_roles),
214 )
216 def create_role(
217 self,
218 name: str,
219 permissions: list[str] | None = None,
220 inherits: list[str] | None = None,
221 ) -> None:
222 """Create or update a role definition."""
224 self.register_role(
225 name,
226 RoleDefinition(
227 name=name,
228 permissions=permissions or [],
229 inherits=inherits or [],
230 ),
231 )
233 def add_role_permission(self, role_name: str, permission: str) -> None:
234 """Add a permission to an existing role."""
235 role = self._roles.get(role_name)
236 if not role:
237 self.create_role(role_name, [permission])
238 return
240 # Handle dict or object
241 if isinstance(role, dict):
242 perms = role.get("permissions")
243 if isinstance(perms, list):
244 perms = set(perms)
245 role["permissions"] = perms
246 elif isinstance(perms, set):
247 pass
248 else:
249 role["permissions"] = set()
250 role["permissions"].add(permission)
251 elif hasattr(role, "permissions"):
252 # If object has mutable permissions set/list
253 if isinstance(role.permissions, list):
254 if permission not in role.permissions:
255 role.permissions.append(permission)
256 elif isinstance(role.permissions, set):
257 role.permissions.add(permission)
259 # Invalidate caches
260 self._role_flatten_cache.clear()
261 self._permission_cache.clear()
263 def get_role_permissions(self, role: str) -> set[str]:
264 """Get all permissions for a role, including inherited ones."""
265 effective_roles = self._flatten_roles({role})
266 return self._get_user_permissions(effective_roles)
268 def remove_role(self, name: str) -> None:
269 """Remove a role definition and invalidate dependent caches.
271 Missing roles are a no-op. In-memory merges mean a removed role
272 can reappear if DB-synced again; callers persist the deletion.
274 Args:
275 name: Role name to remove.
276 """
277 self._roles.pop(name, None)
278 self._role_flatten_cache.clear()
279 self._permission_cache.clear()
280 logger.debug("Removed role: %s", name)
282 def invalidate_user(self, user_id: str) -> None:
283 """Invalidate the permission cache for a specific user."""
284 self._permission_cache.pop(user_id, None)
285 logger.debug("Invalidated permission cache for user: %s", user_id)
288def __getattr__(name: str) -> Any:
289 if name == "authorization_service":
290 raise AttributeError(
291 "authorization_service is now async. Use the container to resolve AuthorizationService.",
292 )
293 raise AttributeError(f"module {__name__} has no attribute {name}")
296__all__ = [
297 "AuthorizationService",
298 "ListValueParser",
299 "NoneValueParser",
300 "StringValueParser",
301 "UserProtocol",
302 "ValueParser",
303 "ValueParserRegistry",
304 "logger",
305]