diff --git a/app/service/auth_svc.py b/app/service/auth_svc.py index 17b2c1aea..27a6d843d 100644 --- a/app/service/auth_svc.py +++ b/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()) diff --git a/tests/security/test_session_store.py b/tests/security/test_session_store.py new file mode 100644 index 000000000..2791fe092 --- /dev/null +++ b/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)