Coverage for src/lexigram/auth/session/manager.py: 84%
98 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-26 00:58 +0800
1"""Session Management for Lexigram Auth."""
3from __future__ import annotations
5from collections.abc import Awaitable, Callable
6from datetime import datetime, timedelta
7from typing import Any
9from lexigram.auth.exceptions import (
10 AuthenticationError,
11 SessionNotFoundError,
12 TokenExpiredError,
13)
14from lexigram.auth.models.session import UserSession
15from lexigram.auth.session.fingerprint import generate_device_id
16from lexigram.auth.storage.in_memory_stores import InMemorySessionStore
17from lexigram.auth.storage.session_store import SessionStore
18from lexigram.contracts.audit import AuditEntry, AuditLoggerProtocol
19from lexigram.contracts.core.identity import IdGeneratorProtocol
20from lexigram.di.decorators import inject
21from lexigram.logging import get_logger
22from lexigram.primitives import clock as ambient_clock
23from lexigram.result import Err, Ok, Result
25logger = get_logger(__name__)
27# Minimum seconds between last_active_at DB writes for the same session.
28# Prevents an excessive write on every authenticated request under load.
29_ACTIVITY_DEBOUNCE_SECONDS: float = 60.0
32@inject
33class SessionManagerImpl:
34 """Manages persistent user sessions with device awareness.
36 Uses SessionStore protocol for storage abstraction, allowing different
37 backends (SQL, Redis, in-memory).
38 """
40 DEFAULT_EXPIRY_DAYS = 30
42 def __init__(
43 self,
44 session_store: SessionStore | None = None,
45 ids: IdGeneratorProtocol | None = None,
46 *,
47 audit_logger: AuditLoggerProtocol | None = None,
48 revoke_token: Callable[[str], Awaitable[None]] | None = None,
49 max_sessions_per_user: int | None = None,
50 ) -> None:
51 """Initialise the session manager.
53 Args:
54 session_store: Optional storage backend. Defaults to an
55 :class:`~lexigram.auth.storage.in_memory_stores.InMemorySessionStore`
56 when not provided.
57 audit_logger: Optional :class:`~lexigram.contracts.audit.AuditLoggerProtocol`
58 used to record security-sensitive session events (create /
59 revoke). No audit entries are written when *None*.
60 revoke_token: Optional async callable that accepts a JTI string and
61 adds it to the token blacklist. When provided, revoking a
62 session also invalidates its associated JWT access token.
63 max_sessions_per_user: Maximum concurrent sessions per user before
64 the least-recently-used session is evicted. ``None`` (the
65 default) means no limit is enforced.
66 """
67 self._store: SessionStore = session_store or InMemorySessionStore()
68 self._ids = ids
69 self._audit_logger: AuditLoggerProtocol | None = audit_logger
70 self._revoke_token: Callable[[str], Awaitable[None]] | None = revoke_token
71 self._max_sessions_per_user: int | None = max_sessions_per_user
72 # Tracks the last monotonic timestamp at which last_active_at was written
73 # for each session, so that we can debounce the per-request DB write.
74 self._last_activity_write: dict[str, float] = {}
76 def __repr__(self) -> str:
77 """Return developer-friendly string representation."""
78 return f"SessionManagerImpl(store={self._store!r})"
80 async def create_session(
81 self,
82 user_id: str,
83 fingerprint_data: dict[str, Any],
84 ip_address: str | None = None,
85 user_agent: str | None = None,
86 expires_days: int = DEFAULT_EXPIRY_DAYS,
87 ) -> UserSession:
88 """Create a new session for a user."""
89 device_id = generate_device_id(fingerprint_data)
91 active_sessions = await self.get_active_sessions(user_id)
92 if (
93 self._max_sessions_per_user is not None
94 and len(active_sessions) >= self._max_sessions_per_user
95 ):
96 sorted_sessions = sorted(
97 active_sessions,
98 key=lambda s: s.last_active_at or s.created_at or datetime.min,
99 )
100 await self.revoke_session(sorted_sessions[0].session_id)
101 logger.info("Revoked oldest session for user %s due to limit", user_id)
103 session_id = (
104 self._ids.generate_for("Session")
105 if self._ids
106 else self._generate_fallback_id()
107 )
108 now = ambient_clock.now()
109 expires_at = now + timedelta(days=expires_days)
111 session = UserSession(
112 session_id=session_id,
113 user_id=user_id,
114 device_id=device_id,
115 ip_address=ip_address,
116 user_agent=user_agent,
117 fingerprint=fingerprint_data,
118 expires_at=expires_at,
119 created_at=now,
120 last_active_at=now,
121 )
123 await self._store.create(session)
125 logger.info(
126 "Created session %s for user %s (Device: %s)",
127 session_id,
128 user_id,
129 device_id,
130 )
132 if self._audit_logger is not None:
133 await self._audit_logger.log(
134 AuditEntry(
135 action="session.created",
136 actor_id=user_id,
137 resource_type="Session",
138 resource_id=session_id,
139 outcome="success",
140 metadata={
141 "device_id": device_id,
142 "ip_address": ip_address,
143 "user_agent": user_agent,
144 },
145 )
146 )
148 return session
150 async def validate_session(
151 self, session_id: str
152 ) -> Result[
153 UserSession, AuthenticationError | SessionNotFoundError | TokenExpiredError
154 ]:
155 """Validate a session and update its activity.
157 Args:
158 session_id: The session identifier to validate.
160 Returns:
161 Ok(session) if the session is valid and active.
162 Err(SessionNotFoundError) if the session does not exist.
163 Err(TokenExpiredError) if the session has expired (it is also revoked).
164 Err(AuthenticationError) if the session exists but is inactive.
165 """
166 session = await self._store.get(session_id)
168 if not session:
169 return Err(SessionNotFoundError(session_id))
171 if session.is_expired():
172 await self.revoke_session(session_id)
173 logger.warning("Session %s expired", session_id)
174 return Err(TokenExpiredError("Session has expired"))
176 if not session.is_active:
177 return Err(AuthenticationError("Session is inactive"))
179 now = ambient_clock.monotonic()
180 last_write = self._last_activity_write.get(session_id, 0.0)
181 if now - last_write >= _ACTIVITY_DEBOUNCE_SECONDS:
182 wall_now = ambient_clock.now()
183 await self._store.update(
184 session_id,
185 last_active_at=wall_now,
186 )
187 self._last_activity_write[session_id] = now
188 self._prune_activity_write_cache(now)
190 return Ok(session)
192 def _prune_activity_write_cache(self, now: float) -> None:
193 """Remove stale entries from the activity-write tracker.
195 Entries older than 2x the debounce window can never influence a future
196 debounce decision, so they are safe to discard. Pruning on every write
197 (rather than on a timer) keeps memory bounded without background threads.
198 """
199 cutoff = now - (_ACTIVITY_DEBOUNCE_SECONDS * 2)
200 stale = [sid for sid, ts in self._last_activity_write.items() if ts < cutoff]
201 for sid in stale:
202 del self._last_activity_write[sid]
204 async def revoke_session(self, session_id: str) -> bool:
205 """Revoke a specific session.
207 If a ``revoke_token`` callback was provided at construction time and
208 the session has an associated ``token_jti``, the JWT is added to the
209 token blacklist so it cannot be reused until natural expiry.
210 """
211 # Fetch the session before revoking so we can include user_id in audit log
212 # and blacklist its associated JWT if a token-revocation callback is set.
213 needs_fetch = self._audit_logger is not None or self._revoke_token is not None
214 session = await self._store.get(session_id) if needs_fetch else None
216 await self._store.update(
217 session_id,
218 is_active=False,
219 updated_at=ambient_clock.now(),
220 )
221 logger.info("Revoked session: %s", session_id)
223 # Blacklist the associated JWT access token if we have a JTI and callback
224 if session and session.token_jti and self._revoke_token is not None:
225 try:
226 await self._revoke_token(session.token_jti)
227 logger.debug(
228 "Blacklisted JWT jti=%s on session revocation", session.token_jti
229 )
230 except (RuntimeError, OSError, ConnectionError, ValueError) as e:
231 logger.warning(
232 "Failed to blacklist token jti=%s during session revocation: %s",
233 session.token_jti,
234 e,
235 )
237 if self._audit_logger is not None:
238 actor = session.user_id if session else session_id
239 await self._audit_logger.log(
240 AuditEntry(
241 action="session.revoked",
242 actor_id=actor,
243 resource_type="Session",
244 resource_id=session_id,
245 outcome="success",
246 )
247 )
249 return True
251 async def verify_mfa(self, session_id: str) -> bool:
252 """Mark the session as MFA verified."""
253 now = ambient_clock.now()
254 await self._store.update(
255 session_id,
256 mfa_verified_at=now,
257 updated_at=now,
258 )
259 logger.info("MFA verified for session: %s", session_id)
260 return True
262 async def revoke_all_sessions(self, user_id: str) -> bool:
263 """Revoke all active sessions for a user."""
264 await self._store.delete_all_for_user(user_id)
265 logger.info("Revoked all sessions for user: %s", user_id)
266 return True
268 async def get_active_sessions(self, user_id: str) -> list[UserSession]:
269 """List all active sessions for a user."""
270 return await self._store.list_for_user(user_id)
272 def _generate_fallback_id(self) -> str:
273 """Generate session ID when IdGenerator is not injected."""
274 import uuid as uuid_module
276 return str(uuid_module.uuid4())
279__all__ = [
280 "SessionManagerImpl",
281 "logger",
282]