Coverage for src/lexigram/admin/auth/services/session_service.py: 70%
97 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-21 14:56 +0800
1"""Admin session lifecycle management service."""
3from __future__ import annotations
5from datetime import UTC, datetime, timedelta
6import hashlib
7import hmac
8import secrets
9from typing import Any
11from lexigram.admin.auth.protocols import AdminSessionServiceProtocol
12from lexigram.contracts.auth.repositories import SessionRepositoryProtocol
13from lexigram.di.decorators import inject
14from lexigram.logging import get_logger
15from lexigram.serialization import dumps
16from lexigram.serialization.backends.json import loads as _json_loads
18logger = get_logger(__name__)
21@inject
22class AdminSessionService:
23 """Admin session lifecycle management service.
25 Wraps ``SessionRepositoryProtocol`` to handle creation, idle-timeout
26 enforcement, absolute-expiry enforcement, activity tracking, and
27 single/bulk revocation.
29 The ``fingerprint`` column in the underlying ``admin_sessions`` table is
30 used to carry ``email`` and ``roles`` as a JSON-serialisable dict.
31 PostgreSQL stores this as JSONB (auto-parsed on read); SQLite stores it
32 as TEXT, which this service parses back to a dict transparently.
34 The ``fingerprint_sig`` column stores an HMAC-SHA256 signature of the
35 fingerprint JSON to detect tampering (AUTH-05).
36 """
38 def __init__(
39 self,
40 session_repo: SessionRepositoryProtocol,
41 session_lifetime: int = 86400,
42 idle_timeout: int = 3600,
43 fingerprint_secret: str = "",
44 ) -> None:
45 """Initialize with repository and lifetime configuration.
47 Args:
48 session_repo: Repository that satisfies
49 ``SessionRepositoryProtocol`` from ``lexigram-contracts``.
50 session_lifetime: Absolute session lifetime in seconds
51 (default 86 400 = 24 h).
52 idle_timeout: Idle inactivity timeout in seconds
53 (default 3 600 = 1 h). A session that has not been
54 touched within this window is treated as expired.
55 fingerprint_secret: HMAC key for fingerprint signing.
56 When empty, signing is skipped (backward compatibility).
57 """
58 self._repo = session_repo
59 self._session_lifetime = session_lifetime
60 self._idle_timeout = idle_timeout
61 self._fingerprint_key: bytes = (
62 hashlib.sha256(fingerprint_secret.encode()).digest()
63 if fingerprint_secret
64 else b""
65 )
66 self._signing_enabled = bool(fingerprint_secret)
68 # ------------------------------------------------------------------
69 # AdminSessionServiceProtocol
70 # ------------------------------------------------------------------
72 async def create_session(
73 self,
74 user_id: str,
75 email: str,
76 roles: list[str],
77 ip_address: str,
78 user_agent: str,
79 ) -> str:
80 """Create a new session and return the session ID.
82 Generates a cryptographically secure session identifier, persists
83 the session with absolute expiry and initial last-active timestamp,
84 and stores ``email`` and ``roles`` in the ``fingerprint`` column so
85 that they can be retrieved without a secondary user-table lookup.
87 When ``fingerprint_secret`` is configured, the fingerprint JSON is
88 HMAC-SHA256 signed and the signature stored in ``fingerprint_sig``.
90 Args:
91 user_id: Admin user UUID.
92 email: Admin user email.
93 roles: User's role names.
94 ip_address: Client IP address.
95 user_agent: Client user agent string.
97 Returns:
98 New session identifier (``secrets.token_urlsafe(32)``).
99 """
100 session_id = secrets.token_urlsafe(32)
101 now = datetime.now(UTC)
102 expires_at = now + timedelta(seconds=self._session_lifetime)
104 fingerprint: dict[str, Any] = {"email": email, "roles": roles}
106 payload: dict[str, Any] = {
107 "session_id": session_id,
108 "admin_id": user_id,
109 "ip_address": ip_address,
110 "user_agent": user_agent,
111 "fingerprint": fingerprint,
112 "expires_at": expires_at,
113 "created_at": now,
114 "last_active_at": now,
115 }
117 if self._signing_enabled:
118 payload["fingerprint_sig"] = self._sign_fingerprint(fingerprint)
120 await self._repo.insert(payload)
121 logger.info(
122 "session.created",
123 session_id=session_id,
124 user_id=user_id,
125 expires_at=expires_at.isoformat(),
126 )
127 return session_id
129 async def get_session(self, session_id: str) -> dict[str, Any] | None:
130 """Retrieve session data if valid (not expired, not revoked).
132 Checks absolute expiry first, then idle timeout. Either condition
133 triggers an immediate revocation of the session record so that
134 subsequent requests cannot use it even if the ``is_active`` flag
135 would otherwise still be TRUE.
137 When ``fingerprint_secret`` is configured, the fingerprint signature
138 is verified on read. A mismatch causes immediate revocation.
140 Args:
141 session_id: Session to retrieve.
143 Returns:
144 Session data dict (including parsed ``fingerprint``) or ``None``
145 if the session does not exist, is revoked, or has expired.
146 """
147 row = await self._repo.find_active(session_id)
148 if row is None:
149 return None
151 now = datetime.now(UTC)
153 # --- Absolute expiry check ---
154 expires_at = _parse_dt(row.get("expires_at"))
155 if expires_at is not None and expires_at <= now:
156 await self._repo.revoke(session_id)
157 logger.debug("session.expired_absolute", session_id=session_id)
158 return None
160 # --- Idle timeout check ---
161 last_active_at = _parse_dt(row.get("last_active_at"))
162 if last_active_at is not None:
163 idle_deadline = last_active_at + timedelta(seconds=self._idle_timeout)
164 if idle_deadline <= now:
165 await self._repo.revoke(session_id)
166 logger.debug("session.expired_idle", session_id=session_id)
167 return None
169 # Return a copy with fingerprint normalised to a dict regardless of
170 # whether the underlying store returned it as JSONB (already dict) or
171 # TEXT (requires JSON parsing — SQLite path).
172 result = dict(row)
173 fingerprint = result.get("fingerprint")
174 if isinstance(fingerprint, str):
175 try:
176 result["fingerprint"] = _json_loads(fingerprint)
177 except (ValueError, TypeError):
178 result["fingerprint"] = {}
180 # --- Fingerprint HMAC verification (AUTH-05) ---
181 if self._signing_enabled:
182 stored_sig: str | None = result.get("fingerprint_sig")
183 if not stored_sig or not self._verify_fingerprint(
184 result["fingerprint"], stored_sig
185 ):
186 logger.warning(
187 "session.fingerprint_tampered",
188 session_id=session_id,
189 )
190 await self._repo.revoke(session_id)
191 return None
193 return result
195 # ------------------------------------------------------------------
196 # Fingerprint HMAC helpers (AUTH-05)
197 # ------------------------------------------------------------------
199 def _sign_fingerprint(self, fingerprint: dict[str, Any]) -> str:
200 """HMAC-SHA256 sign the fingerprint dict.
202 Args:
203 fingerprint: Fingerprint dict (email, roles).
205 Returns:
206 Hex-encoded HMAC-SHA256 signature.
207 """
208 raw = dumps(fingerprint, sort_keys=True)
209 return hmac.new(self._fingerprint_key, raw, hashlib.sha256).hexdigest()
211 def _verify_fingerprint(self, fingerprint: dict[str, Any], signature: str) -> bool:
212 """Verify the HMAC-SHA256 signature of a fingerprint.
214 Args:
215 fingerprint: Fingerprint dict to verify.
216 signature: Previously stored hex-encoded signature.
218 Returns:
219 True if the signature matches, False otherwise.
220 """
221 expected = self._sign_fingerprint(fingerprint)
222 return hmac.compare_digest(expected, signature)
224 # ------------------------------------------------------------------
225 # Touch / revoke
226 # ------------------------------------------------------------------
228 async def touch_session(self, session_id: str) -> None:
229 """Update session last-active timestamp to now (UTC).
231 Args:
232 session_id: Session to touch.
233 """
234 now = datetime.now(UTC)
235 await self._repo.update_activity(session_id, now)
236 logger.debug("session.touched", session_id=session_id)
238 async def revoke_session(self, session_id: str) -> None:
239 """Revoke a single session (logout).
241 Args:
242 session_id: Session to revoke.
243 """
244 await self._repo.revoke(session_id)
245 logger.info("session.revoked", session_id=session_id)
247 async def revoke_all_user_sessions(self, user_id: str) -> None:
248 """Revoke all active sessions for a user.
250 Delegates to the repository's ``revoke_all`` which issues a single
251 bulk UPDATE rather than fetching and revoking sessions individually.
253 Args:
254 user_id: Admin user UUID whose sessions are to be revoked.
255 """
256 await self._repo.revoke_all(user_id)
257 logger.info("session.revoked_all", user_id=user_id)
260# ---------------------------------------------------------------------------
261# Module-level helpers
262# ---------------------------------------------------------------------------
265def _parse_dt(value: Any) -> datetime | None:
266 """Parse a datetime from a DB row value.
268 Handles both timezone-aware ``datetime`` objects returned by async
269 PostgreSQL drivers (e.g. asyncpg) and ISO-8601 strings returned by
270 SQLite drivers. Naïve datetimes are assumed to be UTC.
272 Args:
273 value: Raw column value from the database row.
275 Returns:
276 Timezone-aware ``datetime`` in UTC, or ``None`` if ``value`` is
277 ``None`` or cannot be parsed.
278 """
279 if value is None:
280 return None
281 if isinstance(value, datetime):
282 if value.tzinfo is None:
283 return value.replace(tzinfo=UTC)
284 return value
285 if isinstance(value, str):
286 try:
287 dt = datetime.fromisoformat(value)
288 if dt.tzinfo is None:
289 dt = dt.replace(tzinfo=UTC)
290 return dt
291 except ValueError:
292 return None
293 return None
296# Verify structural subtyping at import time: AdminSessionService must
297# satisfy AdminSessionServiceProtocol without explicit inheritance.
298_: AdminSessionServiceProtocol = AdminSessionService.__new__(AdminSessionService)
300__all__ = ["AdminSessionService"]