Coverage for src / lexigram / admin / auth / adapter.py: 0%
183 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-08-11 02:25 +0800
1"""Adapter to align lexigram-admin's YAML/config-driven admin users and roles
2with the canonical `lexigram-auth` provider and role manager.
4This keeps `AdminProvider` UX but delegates core authn/authz to `lexigram-auth`.
5"""
7from __future__ import annotations
9from typing import TYPE_CHECKING, Any
11from lexigram.contracts import (
12 AuthorizerProtocol,
13 AuthProviderProtocol,
14 DatabaseProviderProtocol,
15)
16from lexigram.contracts.auth import PasswordHasherProtocol
17from lexigram.contracts.exceptions import UnresolvableDependencyError
18from lexigram.di.decorators import inject
19from lexigram.logging import get_logger
21if TYPE_CHECKING:
22 from lexigram.contracts.core.di import ContainerProtocol, ContainerResolverProtocol
24try:
25 from lexigram.admin.auth.session_manager import AdminSessionManager
26 from lexigram.admin.auth.store import (
27 AdminSessionSqlRepository,
28 DirectSQLAdminUserStore,
29 )
30except ImportError:
31 AdminSessionManager = None # type: ignore[assignment, misc]
32 AdminSessionSqlRepository = None # type: ignore[assignment,misc]
33 DirectSQLAdminUserStore = None # type: ignore[assignment, misc]
35logger = get_logger(__name__)
38@inject
39class AdminAuthAdapter:
40 """Sync admin config (users + roles) into ``lexigram-auth`` components.
42 Usage::
44 adapter = AdminAuthAdapter(admin_auth_config)
45 await adapter.register(container)
46 """
48 def __init__(
49 self,
50 auth_config: Any,
51 authorization_service: AuthorizerProtocol,
52 ) -> None:
53 self.auth_config = auth_config
54 self.authorization_service = authorization_service
56 async def sync(self, container: ContainerProtocol) -> None:
57 """Sync admin config (roles + users) into the core Auth system.
59 This method should be called during the boot phase. It should NOT
60 call container.singleton() as the container is likely frozen.
61 """
62 if not getattr(self.auth_config, "enabled", True):
63 logger.info("AdminAuthAdapter: disabled via config; skipping sync")
64 return
66 # Resolve or create AuthProvider
67 auth_provider: AuthProviderProtocol | None = None
68 try:
69 auth_provider = await container.resolve("AdminAuthProvider")
70 logger.info("Found dedicated AdminAuthProvider in container; using it")
71 except UnresolvableDependencyError:
72 try:
73 auth_provider = await container.resolve(AuthProviderProtocol)
74 logger.info("Found existing global AuthProvider in container; using it")
75 except UnresolvableDependencyError:
76 raise RuntimeError(
77 "AuthProvider not found in container; please register one using the Provider pattern",
78 ) from None
80 # Sync roles into authorization_service
81 logger.debug(
82 "Starting role sync with config: %s",
83 getattr(self.auth_config, "roles", {}),
84 )
85 if getattr(self.auth_config, "roles", None):
86 for name, role_def in self.auth_config.roles.items():
87 logger.debug("Syncing role: %s", name)
88 if isinstance(role_def, dict):
89 permissions = role_def.get("permissions", [])
90 else:
91 # RoleDefinition-like object
92 permissions = getattr(role_def, "permissions", []) or []
93 self.authorization_service.create_role(name, list(permissions)) # type: ignore[attr-defined]
94 for perm in permissions:
95 self.authorization_service.add_role_permission(name, perm) # type: ignore[attr-defined]
96 logger.debug(
97 "Registered admin role '%s' with permissions %s",
98 name,
99 permissions,
100 )
102 # Prefer an AbstractAdminUserStore backed by admin_users table to avoid
103 # writing admin accounts into the application `users` table.
104 admin_store = None
105 try:
106 # Try to resolve an explicit admin store if registered
107 admin_store = await container.resolve(
108 "lexigram.admin.auth.AbstractAdminUserStore",
109 )
110 except UnresolvableDependencyError:
111 admin_store = None
113 # If no explicit admin store was registered, try to construct one from DatabaseProvider
114 if admin_store is None:
115 try:
116 db_provider = await container.resolve(DatabaseProviderProtocol)
117 logger.debug("Resolved DatabaseProvider: %s", db_provider)
118 if DirectSQLAdminUserStore: # type: ignore[truthy-function]
119 admin_store = DirectSQLAdminUserStore(db_provider)
120 logger.info(
121 "Constructed AbstractAdminUserStore from DatabaseProvider for admin sync",
122 )
123 else:
124 logger.warning(
125 "DirectSQLAdminUserStore not available (ImportError?); skipping construction",
126 )
127 except UnresolvableDependencyError:
128 logger.exception(
129 "Failed to construct AbstractAdminUserStore from DatabaseProvider",
130 )
131 admin_store = None
133 # Choose the store to use for admin user operations. Prefer AbstractAdminUserStore if available.
134 if admin_store is not None:
135 # If we are using a dedicated AdminAuthProvider or even the global one,
136 # we need to ensure its user_store is set to this admin_store if we want separation.
137 if auth_provider is not None:
138 auth_provider.user_store = admin_store
139 logger.info(
140 "Updated AuthProvider user_store to AbstractAdminUserStore for separation",
141 )
143 # Attach AdminSessionManager using admin_sessions table (FK to admin_users)
144 try:
145 if hasattr(admin_store, "db_provider") and admin_store.db_provider:
146 if AdminSessionManager and AdminSessionSqlRepository: # type: ignore[truthy-function]
147 auth_provider.session_manager = AdminSessionManager(
148 AdminSessionSqlRepository(admin_store.db_provider),
149 )
150 logger.info(
151 "Attached AdminSessionManager to AuthProvider for admin user store",
152 )
153 else:
154 logger.warning(
155 "AdminSessionManager not available; skipping attachment",
156 )
157 except (AttributeError, TypeError):
158 logger.exception(
159 "Failed to attach AdminSessionManager to AuthProvider for admin user store",
160 )
162 # Auto-seeding of admin users from config has been removed in favor
163 # of the one-time super admin registration flow (SetupModule).
164 logger.info("AdminAuthAdapter: sync complete")
166 def _is_duplicate_key_error(self, e: Exception) -> bool:
167 """Check if exception is a duplicate key error."""
168 err_str = str(e).lower()
169 return "duplicate key" in err_str or "unique constraint" in err_str
171 async def _find_user(
172 self,
173 store: Any,
174 email: str,
175 ) -> Any | None:
176 """Find user by email."""
177 try:
178 return await store.get_user_by_email(email)
179 except (
180 ConnectionError,
181 RuntimeError,
182 ValueError,
183 TypeError,
184 AttributeError,
185 ) as e:
186 logger.debug("Store lookup failed for %s: %s", email, e)
187 return None
189 async def _log_audit_event(
190 self,
191 container: ContainerResolverProtocol,
192 table: str,
193 entity_id: str,
194 action: str,
195 values: dict | None = None,
196 ) -> None:
197 """Log an audit event if an audit logger is available."""
198 try:
199 audit_logger = await container.resolve("AuditLogger")
200 if audit_logger:
201 resource = f"{table}:{entity_id}"
202 details = {
203 "action": action,
204 "table": table,
205 "entity_id": entity_id,
206 "values": values or {},
207 }
208 await audit_logger.log_change(
209 user_id="system",
210 resource=resource,
211 action=action,
212 details=details,
213 )
214 except UnresolvableDependencyError:
215 pass
216 except (OSError, ValueError, TypeError) as e:
217 logger.debug("Optional audit logging failed: %s", e)
219 async def create_user(
220 self,
221 container: ContainerResolverProtocol,
222 username: str,
223 email: str,
224 password: str | None = None,
225 hashed_password: str | None = None,
226 roles: list | None = None,
227 permissions: list | None = None,
228 ) -> Any:
229 """Create a new admin user via the configured AuthProvider/UI store.
231 This is intentionally minimal — it delegates to the AuthProvider.create_user
232 when available, otherwise falls back to user_store.create_user.
233 """
234 roles = roles or []
235 permissions = permissions or []
237 # Resolve or create AuthProvider
238 auth_provider: AuthProviderProtocol | None = None
239 try:
240 auth_provider = await container.resolve(AuthProviderProtocol)
241 except (ImportError, AttributeError, RuntimeError):
242 auth_provider = None
244 # Prefer user_store.create_user if available (auth providers don't own user creation)
245 if auth_provider and getattr(auth_provider, "user_store", None):
246 user_store = auth_provider.user_store
247 try:
248 _hasher = await container.resolve(PasswordHasherProtocol)
249 _hashed = hashed_password or await _hasher.hash(password or "")
250 created = await user_store.create_user( # type: ignore[union-attr]
251 name=username,
252 email=email,
253 hashed_password=_hashed,
254 roles=roles,
255 permissions=permissions,
256 )
257 await self._log_audit_event(
258 container,
259 "admin_users",
260 getattr(created, "user_id", username),
261 "INSERT",
262 {"username": username, "email": email},
263 )
264 return created
265 except Exception as e: # noqa: BLE001 — duplicate-key detection inspects exception type from any DB driver
266 # Check if it's a duplicate key error before re-raising
267 if not isinstance(
268 e,
269 (
270 ConnectionError,
271 RuntimeError,
272 ValueError,
273 TypeError,
274 AttributeError,
275 ),
276 ):
277 # Only check for duplicate if not one of the known types
278 if not self._is_duplicate_key_error(e):
279 raise
281 # Handle duplicate key error
282 if self._is_duplicate_key_error(e):
283 logger.info(
284 "user_store.create_user detected existing user for %s; loading existing user",
285 username,
286 )
287 existing = await self._find_user(user_store, email)
288 if existing:
289 if roles and set(getattr(existing, "roles", [])) != set(roles):
290 existing.roles = roles
291 try:
292 await user_store.update_user(existing) # type: ignore[union-attr]
293 except (
294 ConnectionError,
295 RuntimeError,
296 ValueError,
297 TypeError,
298 AttributeError,
299 ):
300 logger.exception(
301 "Failed to update roles for existing user %s",
302 username,
303 )
304 return existing
305 logger.exception("user_store.create_user failed for %s", username)
306 raise
308 raise RuntimeError(
309 "No AuthProvider or user_store available to create user",
310 ) from None
312 async def delete_user(
313 self, container: ContainerResolverProtocol, user_id: str
314 ) -> None:
315 """Remove a user via AuthProvider.delete_user or user_store.delete_user."""
316 try:
317 auth_provider = await container.resolve(AuthProviderProtocol)
318 except (ImportError, AttributeError, RuntimeError):
319 auth_provider = None
321 if auth_provider and hasattr(auth_provider, "delete_user"):
322 try:
323 await auth_provider.delete_user(user_id)
324 except (
325 ConnectionError,
326 RuntimeError,
327 ValueError,
328 TypeError,
329 AttributeError,
330 ):
331 logger.exception("AuthProvider.delete_user failed for %s", user_id)
332 raise
334 # Fallback to user_store
335 if auth_provider and getattr(auth_provider, "user_store", None):
336 try:
337 await auth_provider.user_store.delete_user(user_id)
338 logger.info("Deleted user %s via user_store", user_id)
339 await self._log_audit_event(container, "admin_users", user_id, "DELETE")
340 except (
341 ConnectionError,
342 RuntimeError,
343 ValueError,
344 TypeError,
345 AttributeError,
346 ):
347 logger.exception("user_store.delete_user failed for %s", user_id)
348 raise
349 else:
350 return
352 raise RuntimeError(
353 "No AuthProvider or user_store available to delete user",
354 ) from None
357@inject
358class AdminAuthServiceAdapter:
359 """Bridges admin auth operations to ``lexigram-auth``'s ``AuthProviderProtocol``.
361 Usage::
363 adapter = AdminAuthServiceAdapter(auth_provider=real_provider)
364 user = await adapter.verify_token(token)
365 """
367 def __init__(
368 self,
369 auth_provider: AuthProviderProtocol | None = None,
370 ) -> None:
371 self._provider = auth_provider or _NoOpAuthProvider()
373 async def verify_token(self, token: str) -> dict[str, Any] | None:
374 """Verify a JWT or session token via the auth provider.
376 Args:
377 token: The token string to verify.
379 Returns:
380 Decoded token claims dict, or ``None`` if verification fails.
381 """
382 try:
383 result = await self._provider.verify_token(token)
384 if hasattr(result, "is_ok") and result.is_ok():
385 return result.unwrap() if hasattr(result, "unwrap") else result
386 return None
387 except Exception: # noqa: BLE001
388 logger.warning("admin.auth.token_verification_failed")
389 return None
391 async def get_user(self, user_id: str) -> Any | None:
392 """Fetch a user by ID via the auth provider.
394 Args:
395 user_id: The unique identifier of the user.
397 Returns:
398 The user object, or ``None`` if not found.
399 """
400 try:
401 return await self._provider.get_user(user_id)
402 except Exception: # noqa: BLE001
403 return None
405 async def validate_session(self, token: str) -> Any | None:
406 """Validate a session and return user info.
408 Args:
409 token: The session token to validate.
411 Returns:
412 User info dict, or ``None`` if invalid.
413 """
414 try:
415 result = await self._provider.validate_session(token) # type: ignore[union-attr]
416 if hasattr(result, "is_ok") and result.is_ok():
417 return result.unwrap() if hasattr(result, "unwrap") else result
418 return result
419 except Exception: # noqa: BLE001
420 return None
423class _NoOpAuthProvider:
424 """Fallback when lexigram-auth is not installed."""
426 async def verify_token(self, token: str) -> Any:
427 return None
429 async def get_user(self, user_id: str) -> Any | None:
430 return None
432 async def validate_session(self, token: str) -> Any:
433 return None
436__all__ = ["AdminAuthAdapter", "AdminAuthServiceAdapter"]