From 73302cd72b0aa966e5554f0691e05dde3d1bb826 Mon Sep 17 00:00:00 2001 From: deacon Date: Sun, 15 Mar 2026 13:17:39 -0400 Subject: [PATCH 1/3] fix: add server-side session store and invalidation to auth_svc Add in-memory session registry with register, validate, invalidate lifecycle. --- app/service/auth_svc.py | 37 +++++++++++++++ tests/security/test_session_store.py | 67 ++++++++++++++++++++++++++++ 2 files changed, 104 insertions(+) create mode 100644 tests/security/test_session_store.py diff --git a/app/service/auth_svc.py b/app/service/auth_svc.py index 17b2c1aea..68c87cc6c 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,14 @@ 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 + self._active_sessions = {} # {session_token: {username, created_at, expires_at}} @property def default_login_handler(self): @@ -83,6 +88,38 @@ 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.""" + token = str(uuid.uuid4()) + lifetime = (self.get_config('session_lifetime_hours') or self.SESSION_LIFETIME_HOURS) * 3600 + now = time.time() + self._active_sessions[token] = { + 'username': username, + 'created_at': now, + 'expires_at': now + lifetime, + } + return 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..2bb540953 --- /dev/null +++ b/tests/security/test_session_store.py @@ -0,0 +1,67 @@ +import time +import unittest +from unittest.mock import patch, MagicMock + +# We need to mock the services before import +import sys +sys.modules.setdefault('app.service.interfaces.i_auth_svc', MagicMock()) + + +class TestSessionStore(unittest.TestCase): + def _make_auth_svc(self): + from app.service.auth_svc import AuthService + with patch.object(AuthService, 'add_service', return_value=MagicMock()): + svc = AuthService.__new__(AuthService) + svc._active_sessions = {} + svc.log = MagicMock() + svc.user_map = {} + svc._login_handler = None + svc._default_login_handler = None + return svc + + def test_register_session(self): + svc = self._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 = self._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 = self._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 = self._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 = self._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)) + + +if __name__ == '__main__': + unittest.main() From 4b8046b05fe809d3b6f54e6001a7a28673188aa0 Mon Sep 17 00:00:00 2001 From: deacon Date: Mon, 16 Mar 2026 00:43:39 -0400 Subject: [PATCH 2/3] fix: address Copilot review feedback on server-side session store - Fix register_session to cast get_config() result to float before multiplying, handling string config values correctly; use is None check so 0 hours is respected rather than falling back to default - Add purge_expired_sessions() to prevent unbounded _active_sessions growth when expired tokens are never revalidated - Fix test isolation: use patch.dict(sys.modules) instead of bare sys.modules assignment to prevent cross-test contamination - Add tests: string config, zero config, None config default fallback, and purge_expired_sessions behaviour --- app/service/auth_svc.py | 16 +++++- tests/security/test_session_store.py | 77 +++++++++++++++++++++++----- 2 files changed, 79 insertions(+), 14 deletions(-) diff --git a/app/service/auth_svc.py b/app/service/auth_svc.py index 68c87cc6c..afef344cf 100644 --- a/app/service/auth_svc.py +++ b/app/service/auth_svc.py @@ -91,7 +91,9 @@ async def create_user(self, username, password, group): def register_session(self, username): """Create a server-side session and return the token.""" token = str(uuid.uuid4()) - lifetime = (self.get_config('session_lifetime_hours') or self.SESSION_LIFETIME_HOURS) * 3600 + _cfg = self.get_config('session_lifetime_hours') + lifetime_hours = float(_cfg) if _cfg is not None else self.SESSION_LIFETIME_HOURS + lifetime = lifetime_hours * 3600 now = time.time() self._active_sessions[token] = { 'username': username, @@ -100,6 +102,18 @@ def register_session(self, username): } 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) diff --git a/tests/security/test_session_store.py b/tests/security/test_session_store.py index 2bb540953..1bf1bee73 100644 --- a/tests/security/test_session_store.py +++ b/tests/security/test_session_store.py @@ -2,25 +2,34 @@ import unittest from unittest.mock import patch, MagicMock -# We need to mock the services before import -import sys -sys.modules.setdefault('app.service.interfaces.i_auth_svc', MagicMock()) +def _make_auth_svc(): + """Create an AuthService instance with dependencies mocked. -class TestSessionStore(unittest.TestCase): - def _make_auth_svc(self): - from app.service.auth_svc import AuthService - with patch.object(AuthService, 'add_service', return_value=MagicMock()): + Uses patch.dict to scope sys.modules injection to the import call, + avoiding global test-process contamination. Calls __init__ so that + _active_sessions and other attributes are properly initialised. + """ + 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 + return svc + +class TestSessionStore(unittest.TestCase): def test_register_session(self): - svc = self._make_auth_svc() + svc = _make_auth_svc() with patch.object(type(svc), 'get_config', return_value=8): token = svc.register_session('admin') self.assertIsNotNone(token) @@ -28,21 +37,21 @@ def test_register_session(self): self.assertEqual(svc._active_sessions[token]['username'], 'admin') def test_validate_session(self): - svc = self._make_auth_svc() + 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 = self._make_auth_svc() + 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 = self._make_auth_svc() + svc = _make_auth_svc() with patch.object(type(svc), 'get_config', return_value=8): t1 = svc.register_session('admin') t2 = svc.register_session('admin') @@ -53,7 +62,7 @@ def test_invalidate_all_for_user(self): self.assertTrue(svc.is_session_valid(t3)) def test_expired_session(self): - svc = self._make_auth_svc() + svc = _make_auth_svc() token = 'test-token' svc._active_sessions[token] = { 'username': 'admin', @@ -62,6 +71,48 @@ def test_expired_session(self): } 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.""" + from app.service.auth_svc import AuthService + svc = _make_auth_svc() + with patch.object(type(svc), 'get_config', return_value=None): + token = svc.register_session('admin') + session = svc._active_sessions[token] + expected_lifetime = AuthService.SESSION_LIFETIME_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) + if __name__ == '__main__': unittest.main() From a72e684b84eaeea26d5a23ed462e26b80955f047 Mon Sep 17 00:00:00 2001 From: deacon Date: Mon, 16 Mar 2026 10:02:55 -0400 Subject: [PATCH 3/3] Address Copilot review: single-process docs, config error handling, tests - Document single-process limitation of in-memory session store - Add try/except for non-numeric config values in register_session - Call purge_expired_sessions on each registration to bound memory - Fix test helper docstring to accurately describe __new__ usage - Add test for invalid string config ('8h') falling back to default - Fix test_none_config to access class constant via type(svc) - Remove __main__ block from test file --- app/service/auth_svc.py | 18 ++++++++++++++++-- tests/security/test_session_store.py | 27 ++++++++++++++++++--------- 2 files changed, 34 insertions(+), 11 deletions(-) diff --git a/app/service/auth_svc.py b/app/service/auth_svc.py index afef344cf..27a6d843d 100644 --- a/app/service/auth_svc.py +++ b/app/service/auth_svc.py @@ -65,6 +65,11 @@ def __init__(self): 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 @@ -89,10 +94,19 @@ 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.""" + """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') - lifetime_hours = float(_cfg) if _cfg is not None else self.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] = { diff --git a/tests/security/test_session_store.py b/tests/security/test_session_store.py index 1bf1bee73..2791fe092 100644 --- a/tests/security/test_session_store.py +++ b/tests/security/test_session_store.py @@ -6,9 +6,10 @@ 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. Calls __init__ so that - _active_sessions and other attributes are properly initialised. + 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 @@ -92,12 +93,24 @@ def test_zero_config_lifetime(self): def test_none_config_uses_default_lifetime(self): """get_config=None should fall back to SESSION_LIFETIME_HOURS.""" - from app.service.auth_svc import AuthService 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 = AuthService.SESSION_LIFETIME_HOURS * 3600 + 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) @@ -112,7 +125,3 @@ def test_purge_expired_sessions(self): self.assertNotIn('expired1', svc._active_sessions) self.assertNotIn('expired2', svc._active_sessions) self.assertIn('valid1', svc._active_sessions) - - -if __name__ == '__main__': - unittest.main()