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