Coverage for src/lexigram/admin/auth/store/direct_sql.py: 69%
281 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-21 14:56 +0800
1"""
2Direct SQL admin user store implementation.
3"""
5from __future__ import annotations
7from typing import Any
8import uuid
10from lexigram.admin.auth.errors import SetupAlreadyCompletedError
11from lexigram.admin.sql_dialect import is_postgres
12from lexigram.auth import PasswordHasher
13from lexigram.contracts.auth import PasswordHasherProtocol
14from lexigram.contracts.data import DatabaseProviderProtocol
15from lexigram.di.decorators import inject
16from lexigram.logging import get_logger
17from lexigram.result import Err, Ok, Result
18from lexigram.serialization import dumps_str
19from lexigram.serialization import loads as json_loads
21logger = get_logger(__name__)
24def _parse_list(value: Any) -> list[str]:
25 """Normalize a roles/permissions column value into a string list.
27 Handles real lists (Postgres JSONB arrays), Postgres array literals
28 (``"{admin,editor}"``), and JSON text (SQLite TEXT columns).
30 Args:
31 value: Raw column value.
33 Returns:
34 List of string entries; empty when the value is ``None`` or empty.
35 """
36 if value is None:
37 return []
38 if isinstance(value, list):
39 return [str(v) for v in value]
40 text = str(value).strip()
41 if text.startswith("{") and text.endswith("}"):
42 inner = text[1:-1].strip()
43 return [part.strip() for part in inner.split(",")] if inner else []
44 if text.startswith("[") and text.endswith("]"):
45 try:
46 parsed = json_loads(text)
47 except ValueError:
48 return []
49 return [str(v) for v in parsed] if isinstance(parsed, list) else []
50 return []
53def _row_to_user(row: dict[str, Any]) -> Any:
54 """Build a mutable user record object from an ``admin_users`` row."""
56 class _UserObj:
57 def __init__(self, row: dict[str, Any]) -> None:
58 self.user_id = str(row.get("id") or row.get("user_id"))
59 self.name = row.get("name")
60 self.email = row.get("email")
61 self.roles = _parse_list(row.get("roles"))
62 self.permissions = _parse_list(row.get("permissions"))
63 self.hashed_password = row.get("hashed_password")
64 self.is_active = row.get("is_active")
66 def record_login(self) -> Any:
67 """Record login - no-op for admin users"""
69 return _UserObj(row)
72class _CreatedUser:
73 """Lightweight created-user record returned by claim_first_admin."""
75 def __init__(self, uid: str, name: str, email: str) -> None:
76 self.user_id = uid
77 self.name = name
78 self.username = name
79 self.email = email
82@inject
83class DirectSQLAdminUserStore:
84 """Simple store implementation for admin_users table using direct SQL.
86 This class provides the subset of operations required by AdminAuthAdapter
87 (create_user, get_user_by_username, update_user, delete_user).
88 Renamed from AdminUserStore to avoid conflict with config-backed store.
89 """
91 def __init__(
92 self,
93 db_provider: DatabaseProviderProtocol,
94 password_hasher: PasswordHasherProtocol | None = None,
95 ) -> None:
96 self.db_provider = db_provider
97 self._password_hasher = password_hasher or PasswordHasher()
98 self._initialized = False
100 async def ensure_schema(self) -> None:
101 """Ensure admin_users table exists (create if needed)."""
102 if self._initialized:
103 return
105 try:
106 # Table check (supports both Postgres and SQLite)
107 db_type = getattr(self.db_provider, "database_type", "") or ""
108 exists = False
110 if db_type.lower() in ("postgres", "postgresql"):
111 check_sql = "SELECT EXISTS (SELECT FROM information_schema.tables WHERE table_schema = 'public' AND table_name = 'admin_users')"
112 result = await self.db_provider.execute_query(check_sql, [])
113 if hasattr(result, "rows") and result.rows:
114 exists = result.rows[0].get("exists", False)
115 elif isinstance(result, list) and result:
116 exists = result[0].get("exists", False)
117 else:
118 # SQLite fallback
119 check_sql = "SELECT name FROM sqlite_master WHERE type='table' AND name='admin_users'"
120 result = await self.db_provider.execute_query(check_sql, [])
121 # Direct check if result has any rows
122 if hasattr(result, "rows"):
123 exists = len(result.rows) > 0
124 elif isinstance(result, list):
125 exists = len(result) > 0
126 else:
127 exists = bool(result)
129 logger.debug("admin_users exists=%s (db_type=%s)", exists, db_type)
131 if not exists:
132 logger.info("Table 'admin_users' not found; creating it...")
133 # Note: UUID in SQLite is TEXT, JSONB is TEXT
134 sql = """
135 CREATE TABLE admin_users (
136 id VARCHAR(255) PRIMARY KEY,
137 name VARCHAR(255) NOT NULL,
138 email VARCHAR(255) UNIQUE NOT NULL,
139 hashed_password TEXT,
140 roles TEXT,
141 permissions TEXT,
142 is_active BOOLEAN DEFAULT true,
143 created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
144 updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
145 )
146 """
147 if db_type.lower() in ("postgres", "postgresql"):
148 sql = """
149 CREATE TABLE admin_users (
150 id VARCHAR(255) PRIMARY KEY,
151 name VARCHAR(255) NOT NULL,
152 email VARCHAR(255) UNIQUE NOT NULL,
153 hashed_password TEXT,
154 roles JSONB,
155 permissions JSONB,
156 is_active BOOLEAN DEFAULT true,
157 created_at TIMESTAMPTZ DEFAULT NOW(),
158 updated_at TIMESTAMPTZ DEFAULT NOW()
159 )
160 """
162 await self.db_provider.execute(sql, [])
163 logger.info("✅ admin_users table created successfully")
164 else:
165 logger.debug("Table 'admin_users' already exists")
167 self._initialized = True
168 except Exception as _schema_err: # noqa: BLE001 — schema setup may fail with DB-specific errors; log and propagate
169 logger.exception("Failed to ensure admin_users table exists")
170 raise
172 async def list_users(self) -> list[Any]:
173 """Return all admin users ordered by creation time.
175 Returns:
176 Mutable user record objects (same type as
177 ``get_user_by_email``) — one per row, oldest first.
178 """
179 result = await self.db_provider.execute_query(
180 "SELECT * FROM admin_users ORDER BY created_at",
181 [],
182 )
183 rows = []
184 if hasattr(result, "rows") and result.rows:
185 rows = list(result.rows)
186 elif isinstance(result, list):
187 rows = result
188 elif isinstance(result, dict):
189 rows = [result]
190 return [_row_to_user(row) for row in rows]
192 async def get_admin_count(self) -> int:
193 """Count total admin users."""
194 await self.ensure_schema()
195 sql = "SELECT COUNT(*) as count FROM admin_users"
196 result = await self.db_provider.execute_query(sql, [])
198 if hasattr(result, "rows") and result.rows:
199 return result.rows[0].get("count", 0)
200 if isinstance(result, list) and result:
201 return result[0].get("count", 0)
202 if isinstance(result, dict):
203 return result.get("count", 0)
205 return 0
207 async def create_user(
208 self,
209 name: str,
210 email: str,
211 hashed_password: str,
212 roles: list[str] | None = None,
213 permissions: list[str] | None = None,
214 **kwargs, # Accept and ignore extra parameters like 'profile'
215 ) -> Any:
216 await self.ensure_schema()
217 admin_id = str(uuid.uuid4())
218 # SQLite cannot bind Python lists — store roles/permissions as JSON text.
219 serialize_lists = not is_postgres(self.db_provider)
220 payload = {
221 "id": admin_id,
222 "name": name,
223 "email": email,
224 "hashed_password": hashed_password,
225 "roles": dumps_str(roles or []) if serialize_lists else roles or [],
226 "permissions": (
227 dumps_str(permissions or []) if serialize_lists else permissions or []
228 ),
229 "is_active": True,
230 }
232 # Attempt an atomic upsert when running against Postgres to avoid races
233 db_type = getattr(self.db_provider, "database_type", "") or ""
234 if db_type.lower() in ("postgres", "postgresql"):
235 # Use RETURNING to fetch created/updated row atomically.
236 # Explicit ::jsonb casts are required because asyncpg cannot
237 # infer the column type for unparameterised JSONB columns
238 # (DataError: expected str, got list).
239 sql = (
240 "INSERT INTO admin_users (id, name, email, hashed_password, roles, permissions, is_active) "
241 "VALUES (?, ?, ?, ?, ?, ?, ?) "
242 "ON CONFLICT (email) DO UPDATE SET "
243 "name = EXCLUDED.name, hashed_password = EXCLUDED.hashed_password, "
244 "roles = EXCLUDED.roles, permissions = EXCLUDED.permissions, is_active = EXCLUDED.is_active, updated_at = NOW() "
245 "RETURNING id, name, email"
246 )
247 params = [
248 admin_id,
249 name,
250 email,
251 hashed_password,
252 roles or [],
253 permissions or [],
254 True,
255 ]
256 try:
257 result = await self.db_provider.execute(sql, params)
258 # db_provider.execute() returns a QueryResult object
259 if hasattr(result, "success") and not result.success:
260 raise RuntimeError(
261 f"UPSERT failed: {getattr(result, 'error_message', 'unknown error')}"
262 )
263 # Extract row from QueryResult.rows
264 row = None
265 if hasattr(result, "rows") and result.rows:
266 row = result.rows[0]
267 elif isinstance(result, list) and result:
268 row = result[0]
269 elif isinstance(result, dict):
270 row = result
272 if row:
274 class _CreatedUser:
275 def __init__(self, uid: str, name: str, email: str) -> None:
276 self.user_id = uid
277 self.name = name
278 self.username = name
279 self.email = email
281 logger.info("Created or updated admin user %s via upsert", name)
282 return _CreatedUser(
283 str(row.get("id")),
284 str(row.get("name") or ""),
285 str(row.get("email") or ""),
286 )
287 except Exception as _upsert_err: # noqa: BLE001 — pragma: no cover - fallback path exercised by tests; DB upsert may raise DB-specific errors
288 # Fall through to non-upsert approach below
289 logger.exception(
290 "Postgres upsert for admin_users failed, falling back to insert"
291 )
293 # Fallback (DB-agnostic): try a manual insert with JSONB casts for Postgres,
294 # falling back to execute_insert for other databases.
295 try:
296 if db_type.lower() in ("postgres", "postgresql"):
297 # Use explicit ::jsonb casts for Postgres
298 fallback_sql = (
299 "INSERT INTO admin_users (id, name, email, hashed_password, roles, permissions, is_active) "
300 "VALUES (?, ?, ?, ?, ?, ?, ?)"
301 )
302 fallback_params = [
303 admin_id,
304 name,
305 email,
306 hashed_password,
307 roles or [],
308 permissions or [],
309 True,
310 ]
311 fallback_result = await self.db_provider.execute(
312 fallback_sql, fallback_params
313 )
314 if hasattr(fallback_result, "success") and not fallback_result.success:
315 raise RuntimeError(
316 f"Fallback INSERT failed: {getattr(fallback_result, 'error_message', 'unknown error')}"
317 )
318 else:
319 await self.db_provider.execute_insert("admin_users", payload)
321 # Return lightweight created object similar shape used by adapter
322 class _CreatedUser: # type: ignore[no-redef]
323 def __init__(self, uid: str, name: str, email: str) -> None:
324 self.user_id = uid
325 self.name = name
326 self.username = name
327 self.email = email
329 logger.info("Created admin user %s via AdminUserStore", name)
330 return _CreatedUser(admin_id, name, email)
331 except Exception as e: # noqa: BLE001 — duplicate-key detection requires inspecting the exception type from any DB driver
332 # Handle duplicate-key errors by resolving existing user and updating if needed
333 err_str = str(e).lower()
334 from lexigram.contracts.exceptions import DuplicateKeyError
336 if (
337 isinstance(e, DuplicateKeyError)
338 or "duplicate key" in err_str
339 or "unique constraint" in err_str
340 ):
341 logger.info(
342 "AdminUserStore.create_user detected existing user for %s; resolving existing user",
343 name,
344 )
345 existing = None
346 try:
347 existing = await self.get_user_by_email(email)
348 except (RuntimeError, ValueError, OSError):
349 existing = None
351 if existing:
352 # Optionally update roles/permissions/hashed_password
353 try:
354 # Populate missing attrs if provided
355 existing_roles = getattr(existing, "roles", []) or []
356 if existing_roles != (roles or []):
357 existing.roles = roles or []
358 existing_perms = getattr(existing, "permissions", []) or []
359 if existing_perms != (permissions or []):
360 existing.permissions = permissions or []
362 existing_hash = getattr(existing, "hashed_password", None)
363 if hashed_password and existing_hash != hashed_password:
364 existing.hashed_password = hashed_password
365 await self.update_user(existing)
366 logger.info(
367 "Updated existing admin user %s after duplicate create",
368 name,
369 )
370 except BaseException:
371 logger.exception(
372 "Failed to update existing admin user %s after duplicate create",
373 name,
374 )
376 # Return lightweight object
377 class _ExistingUser:
378 def __init__(self, uid: str, name: str, email: str) -> None:
379 self.user_id = uid
380 self.name = name
381 self.email = email
383 return _ExistingUser(
384 str(
385 getattr(
386 existing, "user_id", getattr(existing, "id", None) or ""
387 )
388 ),
389 str(existing.name or ""),
390 str(existing.email or ""),
391 )
393 # Re-raise if it's some other error
394 logger.exception(
395 "DirectSQLAdminUserStore.create_user failed for %s",
396 name,
397 )
398 raise
400 async def claim_first_admin(
401 self,
402 name: str,
403 email: str,
404 hashed_password: str,
405 roles: list[str],
406 ) -> Result[Any, SetupAlreadyCompletedError]:
407 """Atomically insert the first admin account only if none exists.
409 Runs a single ``INSERT ... SELECT ... WHERE NOT EXISTS`` statement so
410 that concurrent first-run submissions cannot both insert.
412 Args:
413 name: Display name.
414 email: Unique email address — used as the login identifier.
415 hashed_password: Pre-hashed credential.
416 roles: Role strings for the new account.
418 Returns:
419 Ok(_CreatedUser) when this call inserted the first admin account;
420 ``Err(SetupAlreadyCompletedError)`` when the table already holds
421 an admin account and nothing was inserted.
422 """
423 await self.ensure_schema()
424 admin_id = str(uuid.uuid4())
425 serialize_lists = not is_postgres(self.db_provider)
427 if serialize_lists:
428 sql = (
429 "INSERT INTO admin_users "
430 "(id, name, email, hashed_password, roles, permissions, is_active) "
431 "SELECT ?, ?, ?, ?, ?, ?, ? "
432 "WHERE NOT EXISTS (SELECT 1 FROM admin_users)"
433 )
434 params: list[Any] = [
435 admin_id,
436 name,
437 email,
438 hashed_password,
439 dumps_str(roles),
440 "[]",
441 True,
442 ]
443 else:
444 sql = (
445 "INSERT INTO admin_users "
446 "(id, name, email, hashed_password, roles, permissions, is_active) "
447 "SELECT ?, ?, ?, ?, roles::jsonb, permissions::jsonb, ? "
448 "WHERE NOT EXISTS (SELECT 1 FROM admin_users)"
449 )
450 params = [admin_id, name, email, hashed_password, roles, [], True]
452 result = await self.db_provider.execute(sql, params)
453 if hasattr(result, "success") and not result.success:
454 raise RuntimeError(
455 "claim_first_admin failed: "
456 f"{getattr(result, 'error_message', 'unknown error')}"
457 )
459 # 1 inserted row → Ok; 0 rows → Err. Postgres reports the insert via
460 # RETURNING-style rows, SQLite via row_count on the QueryResult.
461 row = None
462 if hasattr(result, "rows") and result.rows:
463 row = result.rows[0]
464 elif isinstance(result, list) and result:
465 row = result[0]
466 elif isinstance(result, dict):
467 row = result
469 inserted = bool(row) or getattr(result, "row_count", 0) > 0
470 if not inserted:
471 return Err(SetupAlreadyCompletedError())
473 if row:
474 return Ok(
475 _CreatedUser(
476 str(row.get("id")),
477 str(row.get("name") or ""),
478 str(row.get("email") or ""),
479 )
480 )
481 return Ok(_CreatedUser(admin_id, name, email))
483 # Also need helper for email lookup if we use it in logic
484 async def get_user_by_email(self, email: str) -> Any | None:
485 await self.ensure_schema()
486 from lexigram.logging import get_logger
488 logger = get_logger(__name__)
490 sql = "SELECT * FROM admin_users WHERE email = ?"
491 result = await self.db_provider.execute_query(sql, [email])
492 logger.debug(
493 "get_user_by_email result type: %s, result: %s",
494 type(result),
495 result,
496 )
498 row = None
499 if hasattr(result, "rows") and result.rows:
500 row = result.rows[0]
501 elif hasattr(result, "fetchone"):
502 row = result.fetchone()
503 elif isinstance(result, dict):
504 row = result
505 elif isinstance(result, list) and result:
506 row = result[0] if result else None
508 logger.debug("Parsed row: %s", row)
510 if not row:
511 logger.debug("No user found with email: %s", email)
512 return None
514 logger.debug(
515 "Found user by email: %s, is_active: %s, has_password: %s",
516 email,
517 row.get("is_active"),
518 bool(row.get("hashed_password")),
519 )
521 return _row_to_user(row)
523 async def get_user_by_id(self, user_id: str) -> Any | None:
524 await self.ensure_schema()
525 from lexigram.logging import get_logger
527 logger = get_logger(__name__)
529 sql = "SELECT * FROM admin_users WHERE id = ?"
530 result = await self.db_provider.execute_query(sql, [user_id])
531 row = None
532 if hasattr(result, "rows") and result.rows:
533 row = result.rows[0]
534 elif hasattr(result, "fetchone"):
535 row = result.fetchone()
536 elif isinstance(result, dict):
537 row = result
538 elif isinstance(result, list) and result:
539 row = result[0]
541 if not row:
542 logger.debug("No user found with id: %s", user_id)
543 return None
545 logger.debug(
546 "Found user by id: %s, is_active: %s",
547 user_id,
548 row.get("is_active"),
549 )
551 return _row_to_user(row)
553 async def update_user(self, user: Any) -> None:
554 await self.ensure_schema()
555 # SQLite cannot bind Python lists — store roles/permissions as JSON text.
556 serialize_lists = not is_postgres(self.db_provider)
557 payload = {
558 "name": user.name,
559 "email": user.email,
560 "hashed_password": getattr(user, "hashed_password", None),
561 "roles": (
562 dumps_str(getattr(user, "roles", []))
563 if serialize_lists
564 else getattr(user, "roles", [])
565 ),
566 "permissions": (
567 dumps_str(getattr(user, "permissions", []))
568 if serialize_lists
569 else getattr(user, "permissions", [])
570 ),
571 "is_active": getattr(user, "is_active", True),
572 }
573 await self.db_provider.execute_update(
574 "admin_users",
575 payload,
576 "id = ?",
577 [user.user_id],
578 )
580 async def authenticate(self, email: str, password: str) -> Any | None:
581 """Authenticate an admin user by email and bcrypt-hashed password.
583 Args:
584 email: Email address to look up.
585 password: Plain-text password to verify against the stored hash.
587 Returns:
588 User object when credentials are valid and account is active,
589 ``None`` otherwise.
590 """
591 user = await self.get_user_by_email(email)
592 if not user:
593 return None
594 if not getattr(user, "is_active", True):
595 return None
596 hashed = getattr(user, "hashed_password", None)
597 if not hashed:
598 return None
599 try:
600 hashed_str = hashed.decode("utf-8") if isinstance(hashed, bytes) else hashed
601 if await self._password_hasher.verify(password, hashed_str):
602 return user
603 except (ValueError, TypeError) as exc:
604 logger.warning(
605 "authenticate.password_check_failed", email=email, error=str(exc)
606 )
607 return None
609 async def delete_user(self, user_id: str) -> None:
610 await self.ensure_schema()
611 await self.db_provider.execute_delete("admin_users", "id = ?", [user_id])
613 async def get_by_id(self, admin_id: str) -> Any | None:
614 """Alias for get_user_by_id — used by AdminAuthMiddleware."""
615 return await self.get_user_by_id(admin_id)