feat: add server-side session store with expiry and invalidation to AuthService

app/service/auth_svc.py

@@ -1,4 +1,6 @@
 import base64
+import uuid
+import time
 from collections import namedtuple
 from hmac import compare_digest
 from importlib import import_module
@@ -56,11 +58,19 @@ async def helper(*args, **params):
 class AuthService(AuthServiceInterface, BaseService):
     User = namedtuple('User', ['username', 'password', 'permissions'])
 
+    SESSION_LIFETIME_HOURS = 8
+
     def __init__(self):
         self.user_map = dict()
         self.log = self.add_service('auth_svc', self)
         self._login_handler = None
         self._default_login_handler = None
+        # In-memory session store.  This only tracks sessions within a single
+        # process; in multi-worker or multi-process deployments sessions will
+        # not be shared across workers and forced logout will not propagate.
+        # For production multi-worker setups a shared backing store (e.g.
+        # Redis) behind this same API would be needed.
+        self._active_sessions = {}  # {session_token: {username, created_at, expires_at}}
 
     @property
     def default_login_handler(self):
@@ -83,6 +93,61 @@ async def apply(self, app, users):
     async def create_user(self, username, password, group):
         self.user_map[username] = self.User(username, password, (group, 'app'), )
 
+    def register_session(self, username):
+        """Create a server-side session and return the token.
+
+        Calls :meth:`purge_expired_sessions` on each registration to keep the
+        in-memory store bounded.
+        """
+        self.purge_expired_sessions()
+        token = str(uuid.uuid4())
+        _cfg = self.get_config('session_lifetime_hours')
+        try:
+            lifetime_hours = float(_cfg) if _cfg is not None else self.SESSION_LIFETIME_HOURS
+        except (ValueError, TypeError):
+            self.log.warning('Invalid session_lifetime_hours config value %r; using default', _cfg)
+            lifetime_hours = self.SESSION_LIFETIME_HOURS
+        lifetime = lifetime_hours * 3600
+        now = time.time()
+        self._active_sessions[token] = {
+            'username': username,
+            'created_at': now,
+            'expires_at': now + lifetime,
+        }
+        return token
+
+    def purge_expired_sessions(self):
+        """Remove all expired sessions from the in-memory store.
+
+        This prevents _active_sessions from growing without bound when sessions
+        are never revalidated after expiry (e.g. tokens issued but never used
+        again).  Call this periodically (e.g. during server maintenance loops).
+        """
+        now = time.time()
+        expired = [t for t, s in self._active_sessions.items() if now > s['expires_at']]
+        for token in expired:
+            del self._active_sessions[token]
+
+    def invalidate_session(self, token):
+        """Remove a session from the server-side store."""
+        self._active_sessions.pop(token, None)
+
+    def is_session_valid(self, token):
+        """Check if a session exists and is not expired."""
+        session = self._active_sessions.get(token)
+        if not session:
+            return False
+        if time.time() > session['expires_at']:
+            self._active_sessions.pop(token, None)
+            return False
+        return True
+
+    def invalidate_all_sessions_for_user(self, username):
+        """Remove all sessions for a given user."""
+        to_remove = [t for t, s in self._active_sessions.items() if s['username'] == username]
+        for token in to_remove:
+            del self._active_sessions[token]
+
     @staticmethod
     async def logout_user(request):
         await forget(request, web.Response())

tests/security/test_session_store.py

@@ -0,0 +1,127 @@
+import time
+import unittest
+from unittest.mock import patch, MagicMock
+
+
+def _make_auth_svc():
+    """Create an AuthService instance with dependencies mocked.
+
+    Uses ``patch.dict`` to scope ``sys.modules`` injection to the import
+    call, avoiding global test-process contamination.  Uses ``__new__``
+    to bypass ``__init__`` (which calls ``add_service``), then manually
+    initialises the attributes that the session methods rely on.
+    """
+    mock_interface_module = MagicMock()
+    mock_interface_module.AuthServiceInterface = object  # plain base class
+
+    with patch.dict('sys.modules', {
+        'app.service.interfaces.i_auth_svc': mock_interface_module,
+    }):
+        with patch('app.service.auth_svc.BaseService.add_service', return_value=MagicMock()):
+            from app.service.auth_svc import AuthService
+            svc = AuthService.__new__(AuthService)
+            svc._active_sessions = {}
+            svc.log = MagicMock()
+            svc.user_map = {}
+            svc._login_handler = None
+            svc._default_login_handler = None
+    return svc
+
+
+class TestSessionStore(unittest.TestCase):
+    def test_register_session(self):
+        svc = _make_auth_svc()
+        with patch.object(type(svc), 'get_config', return_value=8):
+            token = svc.register_session('admin')
+        self.assertIsNotNone(token)
+        self.assertIn(token, svc._active_sessions)
+        self.assertEqual(svc._active_sessions[token]['username'], 'admin')
+
+    def test_validate_session(self):
+        svc = _make_auth_svc()
+        with patch.object(type(svc), 'get_config', return_value=8):
+            token = svc.register_session('admin')
+        self.assertTrue(svc.is_session_valid(token))
+        self.assertFalse(svc.is_session_valid('invalid-token'))
+
+    def test_invalidate_session(self):
+        svc = _make_auth_svc()
+        with patch.object(type(svc), 'get_config', return_value=8):
+            token = svc.register_session('admin')
+        svc.invalidate_session(token)
+        self.assertFalse(svc.is_session_valid(token))
+
+    def test_invalidate_all_for_user(self):
+        svc = _make_auth_svc()
+        with patch.object(type(svc), 'get_config', return_value=8):
+            t1 = svc.register_session('admin')
+            t2 = svc.register_session('admin')
+            t3 = svc.register_session('other')
+        svc.invalidate_all_sessions_for_user('admin')
+        self.assertFalse(svc.is_session_valid(t1))
+        self.assertFalse(svc.is_session_valid(t2))
+        self.assertTrue(svc.is_session_valid(t3))
+
+    def test_expired_session(self):
+        svc = _make_auth_svc()
+        token = 'test-token'
+        svc._active_sessions[token] = {
+            'username': 'admin',
+            'created_at': time.time() - 100000,
+            'expires_at': time.time() - 1,
+        }
+        self.assertFalse(svc.is_session_valid(token))
+
+    def test_string_config_lifetime(self):
+        """get_config may return a string; register_session must handle it correctly."""
+        svc = _make_auth_svc()
+        with patch.object(type(svc), 'get_config', return_value='2'):
+            token = svc.register_session('admin')
+        session = svc._active_sessions[token]
+        expected_lifetime = 2 * 3600
+        actual_lifetime = session['expires_at'] - session['created_at']
+        self.assertAlmostEqual(actual_lifetime, expected_lifetime, delta=5)
+
+    def test_zero_config_lifetime(self):
+        """get_config=0 should use 0-hour lifetime, not fall back to default."""
+        svc = _make_auth_svc()
+        with patch.object(type(svc), 'get_config', return_value=0):
+            token = svc.register_session('admin')
+        session = svc._active_sessions[token]
+        # 0 hours -> expires_at == created_at
+        self.assertAlmostEqual(session['expires_at'], session['created_at'], delta=1)
+
+    def test_none_config_uses_default_lifetime(self):
+        """get_config=None should fall back to SESSION_LIFETIME_HOURS."""
+        svc = _make_auth_svc()
+        # Access the class constant via type(svc) to avoid import-order issues
+        default_hours = type(svc).SESSION_LIFETIME_HOURS
+        with patch.object(type(svc), 'get_config', return_value=None):
+            token = svc.register_session('admin')
+        session = svc._active_sessions[token]
+        expected_lifetime = default_hours * 3600
+        actual_lifetime = session['expires_at'] - session['created_at']
+        self.assertAlmostEqual(actual_lifetime, expected_lifetime, delta=5)
+
+    def test_invalid_string_config_falls_back_to_default(self):
+        """Non-numeric config string (e.g. '8h') should fall back to default."""
+        svc = _make_auth_svc()
+        default_hours = type(svc).SESSION_LIFETIME_HOURS
+        with patch.object(type(svc), 'get_config', return_value='8h'):
+            token = svc.register_session('admin')
+        session = svc._active_sessions[token]
+        expected_lifetime = default_hours * 3600
+        actual_lifetime = session['expires_at'] - session['created_at']
+        self.assertAlmostEqual(actual_lifetime, expected_lifetime, delta=5)
+
+    def test_purge_expired_sessions(self):
+        """purge_expired_sessions should remove expired entries without touching valid ones."""
+        svc = _make_auth_svc()
+        now = time.time()
+        svc._active_sessions['expired1'] = {'username': 'a', 'created_at': now - 200, 'expires_at': now - 100}
+        svc._active_sessions['expired2'] = {'username': 'b', 'created_at': now - 200, 'expires_at': now - 50}
+        svc._active_sessions['valid1'] = {'username': 'c', 'created_at': now, 'expires_at': now + 3600}
+        svc.purge_expired_sessions()
+        self.assertNotIn('expired1', svc._active_sessions)
+        self.assertNotIn('expired2', svc._active_sessions)
+        self.assertIn('valid1', svc._active_sessions)