Coverage for src/local_deep_research/web/auth/session_manager.py: 100%
110 statements
« prev ^ index » next coverage.py v7.16.0, created at 2026-09-06 15:42 +0000
1"""
2Session management for encrypted database connections.
3Handles session creation, validation, and cleanup.
4"""
6import datetime
7import secrets
8import threading
9from datetime import UTC
10from typing import Dict, Optional, Set
12from loguru import logger
14from ...security import get_security_default
17class SessionManager:
18 """Manages user sessions and database connection lifecycle.
20 Note: session state is in-memory and per-process (no Redis or other
21 shared store). In multi-worker deployments (e.g. gunicorn with more
22 than one worker) each worker holds its own ``sessions`` dict, and a
23 worker restart drops all sessions it held. ``validate_session`` only
24 ever sees sessions created by *this* process — including for callers
25 like the socket connect-time session gate, which now relies on this
26 lookup. See ``security/account_lockout.py`` and
27 ``web/routers/rag.py`` for the same caveat elsewhere in the codebase.
28 """
30 def __init__(self):
31 self.sessions: Dict[str, dict] = {}
32 # WARNING: never call logger.* while holding _lock. The synchronous
33 # frontend_progress_sink can re-enter emit_to_subscribers ->
34 # touch_session -> _lock on the same thread and self-deadlock (this is
35 # a plain Lock, not an RLock).
36 self._lock = threading.Lock()
37 # Load session timeouts from security settings
38 session_hours = get_security_default(
39 "security.session_timeout_hours", 2
40 )
41 remember_days = get_security_default(
42 "security.session_remember_me_days", 30
43 )
44 self.session_timeout = datetime.timedelta(hours=session_hours)
45 self.remember_me_timeout = datetime.timedelta(days=remember_days)
47 def create_session(self, username: str, remember_me: bool = False) -> str:
48 """Create a new session for a user.
50 Args:
51 username: The username to create a session for.
52 remember_me: If True, use extended session timeout.
54 Returns:
55 The session ID as a URL-safe string.
56 """
57 session_id = secrets.token_urlsafe(32)
59 with self._lock:
60 self.sessions[session_id] = {
61 "username": username,
62 "created_at": datetime.datetime.now(UTC),
63 "last_access": datetime.datetime.now(UTC),
64 "remember_me": remember_me,
65 }
67 logger.debug(f"Created session {session_id[:8]}... for user {username}")
68 return session_id
70 def validate_session(self, session_id: str) -> Optional[str]:
71 """
72 Validate a session and return username if valid.
73 Updates last access time.
74 """
75 expired_username = None
76 with self._lock:
77 if session_id not in self.sessions:
78 return None
80 session_data = self.sessions[session_id]
81 now = datetime.datetime.now(UTC)
83 # Check timeout
84 timeout = (
85 self.remember_me_timeout
86 if session_data["remember_me"]
87 else self.session_timeout
88 )
89 if now - session_data["last_access"] > timeout:
90 # Session expired — remove inline (already under lock).
91 # Defer the log call until after the lock is released: see
92 # the warning in __init__ about not logging while holding
93 # _lock.
94 expired_username = session_data["username"]
95 del self.sessions[session_id]
96 else:
97 # Update last access
98 session_data["last_access"] = now
99 return str(session_data["username"])
101 logger.debug(
102 f"Session {session_id[:8]}... expired for {expired_username}"
103 )
104 return None
106 def touch_session(self, session_id: str) -> None:
107 """Refresh a session's ``last_access`` without enforcing the timeout.
109 Called on socket activity: a user actively watching a long research
110 run makes no HTTP requests, so the request-path idle-timeout refresh
111 (in ``validate_session``) never fires and the session would silently
112 expire mid-run. Unlike ``validate_session`` this never deletes an
113 expired session — it only bumps ``last_access`` for one still present.
114 """
115 if not session_id:
116 return
117 with self._lock:
118 session_data = self.sessions.get(session_id)
119 if session_data is not None:
120 session_data["last_access"] = datetime.datetime.now(UTC)
122 def destroy_session(self, session_id: str):
123 """Destroy a session and clean up."""
124 username = None
125 with self._lock:
126 if session_id in self.sessions:
127 username = self.sessions[session_id]["username"]
128 del self.sessions[session_id]
130 # Logged after releasing the lock: see the warning in __init__
131 # about not logging while holding _lock.
132 if username is not None:
133 logger.debug(
134 f"Destroyed session {session_id[:8]}... for user {username}"
135 )
137 def destroy_all_user_sessions(self, username: str) -> int:
138 """Destroy all sessions for a given user. Returns count destroyed."""
139 with self._lock:
140 to_delete = [
141 sid
142 for sid, data in self.sessions.items()
143 if data["username"] == username
144 ]
145 for sid in to_delete:
146 del self.sessions[sid]
147 if to_delete:
148 logger.debug(
149 f"Destroyed {len(to_delete)} session(s) for user {username}"
150 )
151 return len(to_delete)
153 def cleanup_expired_sessions(self):
154 """Remove all expired sessions and disconnect their live sockets."""
155 now = datetime.datetime.now(UTC)
156 expired = []
158 with self._lock:
159 for session_id, data in self.sessions.items():
160 timeout = (
161 self.remember_me_timeout
162 if data["remember_me"]
163 else self.session_timeout
164 )
165 if now - data["last_access"] > timeout:
166 expired.append(session_id)
168 for session_id in expired:
169 del self.sessions[session_id]
171 if expired:
172 logger.info(f"Cleaned up {len(expired)} expired sessions")
173 # Tear down any sockets still attached to these now-dead sessions.
174 # A socket is authorised once at handshake and then frozen, so
175 # without this an expired session's tab keeps receiving the user's
176 # events. Done AFTER releasing the lock (disconnect reaches into
177 # the socket server) and best-effort — socket teardown must never
178 # break session cleanup.
179 self._disconnect_expired_sockets(expired)
181 @staticmethod
182 def _disconnect_expired_sockets(session_ids: list[str]) -> None:
183 """Best-effort disconnect of the sockets for expired session ids.
185 Lazy-imports the socket layer to avoid an import cycle and to
186 tolerate its absence in non-web contexts (CLI, tests). Each session is
187 torn down independently so one failure can't strand the rest.
189 Retargeted from the deleted Flask ``SocketIOService`` onto the ASGI
190 layer. The old import raised ``ModuleNotFoundError`` into the bare
191 ``except Exception`` below, so this path silently disconnected
192 nothing: an idle-expired session's sockets stayed live and kept
193 receiving that user's events -- including ``settings_changed``, which
194 carries plaintext secrets. It was masked whenever the idle-DB-close
195 sweep happened to tear the user down anyway, but not when they had
196 another live session or an in-flight research run.
197 """
198 try:
199 from ..services.socketio_asgi import disconnect_session
200 except Exception:
201 logger.opt(exception=True).warning(
202 "Failed to resolve socket layer for expired-session teardown"
203 )
204 return
205 for session_id in session_ids:
206 try:
207 disconnect_session(session_id)
208 except Exception:
209 logger.opt(exception=True).warning(
210 "Failed to disconnect sockets for an expired session"
211 )
213 def get_active_sessions_count(self) -> int:
214 """Get count of active sessions."""
215 self.cleanup_expired_sessions()
216 with self._lock:
217 return len(self.sessions)
219 def get_user_sessions(self, username: str) -> list:
220 """Get all active sessions for a user."""
221 user_sessions = []
222 with self._lock:
223 for session_id, data in self.sessions.items():
224 if data["username"] == username:
225 user_sessions.append(
226 {
227 "session_id": session_id[:8] + "...",
228 "created_at": data["created_at"],
229 "last_access": data["last_access"],
230 "remember_me": data["remember_me"],
231 }
232 )
233 return user_sessions
235 def get_active_usernames(self) -> Set[str]:
236 """Return set of usernames with at least one non-expired session."""
237 now = datetime.datetime.now(UTC)
238 with self._lock:
239 return {
240 data["username"]
241 for data in self.sessions.values()
242 if now - data["last_access"]
243 <= (
244 self.remember_me_timeout
245 if data["remember_me"]
246 else self.session_timeout
247 )
248 }
250 def has_active_sessions_for(self, username: str) -> bool:
251 """Check if a user has any non-expired sessions."""
252 now = datetime.datetime.now(UTC)
253 with self._lock:
254 for data in self.sessions.values():
255 if data["username"] != username:
256 continue
257 timeout = (
258 self.remember_me_timeout
259 if data["remember_me"]
260 else self.session_timeout
261 )
262 if now - data["last_access"] <= timeout:
263 return True
264 return False
267# Module-level singleton
268session_manager = SessionManager()