"""Pluggable session cache abstraction. This module defines a cache interface for authenticated sessions and a default process-local in-memory implementation. The backend can swap the cache implementation without changing the authentication dependency logic. ⚠️ PROCESS-LOCAL CONSTRAINT (InMemorySessionCache) ==================================================== InMemorySessionCache stores validated sessions in a process-local dict. This means: - Each worker process has its own independent session cache. - A session invalidated (logout) by worker A is still valid in worker B. - Changes to the cache in one process are NOT visible to other processes. IMPACT IN MULTI-WORKER DEPLOYMENTS: - User logs out (worker A clears session from its cache). - User makes a new request → routed to worker B. - Worker B still has the stale session in its cache → request is accepted. - User appears still logged in (from their perspective). This is a security issue: logout does not work reliably across workers. MULTI-WORKER SOLUTION: To deploy BanGUI with multiple workers (e.g., via gunicorn -w 4), replace InMemorySessionCache with a shared backend such as: - RedisSessionCache — backed by Redis (recommended for production). - DatabaseSessionCache — backed by SQLite or PostgreSQL. - SharedMemorySessionCache — backed by IPC (for local multi-process). The SessionCache Protocol is already designed for pluggable backends: class SessionCache(Protocol): def get(token: str) -> Session | None: ... def set(token: str, session: Session, ttl_seconds: float) -> None: ... def invalidate(token: str) -> None: ... def clear() -> None: ... To add Redis support: 1. Create RedisSessionCache in this module (implements SessionCache). 2. Update runtime_state.set_runtime_settings() to instantiate RedisSessionCache when REDIS_URL is configured. 3. See Backend-Development.md § "Session Cache Pluggability" for details. SINGLE-WORKER ENFORCEMENT: See TASK-002 in Docs/Tasks.md for deployment configuration that enforces single-worker mode, preventing this issue entirely. For now, BanGUI is deployed as single-worker only — this constraint is acceptable and keeps the implementation simple. """ from __future__ import annotations import time from typing import TYPE_CHECKING, Protocol if TYPE_CHECKING: # pragma: no cover from app.models.auth import Session class SessionCache(Protocol): """Interface for session token validation cache backends.""" def get(self, token: str) -> Session | None: """Return the cached session for *token*, or ``None`` if missing.""" def set(self, token: str, session: Session, ttl_seconds: float) -> None: """Cache the validated *session* for *token* for *ttl_seconds*.""" def invalidate(self, token: str) -> None: """Remove *token* from the cache if it exists.""" def clear(self) -> None: """Remove all entries from the cache.""" class InMemorySessionCache: """A process-local session cache implementation.""" def __init__(self) -> None: self._entries: dict[str, tuple[Session, float]] = {} def get(self, token: str) -> Session | None: entry = self._entries.get(token) if entry is None: return None session, expires_at = entry if time.monotonic() >= expires_at: self._entries.pop(token, None) return None return session def set(self, token: str, session: Session, ttl_seconds: float) -> None: self._entries[token] = (session, time.monotonic() + ttl_seconds) def invalidate(self, token: str) -> None: self._entries.pop(token, None) def clear(self) -> None: self._entries.clear() class NoOpSessionCache: """A no-op session cache used when caching is disabled.""" def get(self, token: str) -> Session | None: return None def set(self, token: str, session: Session, ttl_seconds: float) -> None: return None def invalidate(self, token: str) -> None: return None def clear(self) -> None: return None