Enforce server-side session idle expiry with sliding TTL

packages/kaya-session/tests/test_session.py

@@ -1,3 +1,4 @@
+import asyncio
 import unittest
 from typing import Any
 
@@ -9,6 +10,17 @@ from kaya.session import InMemorySessionStore, Session, SessionMiddleware, Sessi
 from kaya.session._cookie import format_set_cookie, parse_cookie_value
 
 
+class FakeClock:
+    def __init__(self, start: float = 0.0) -> None:
+        self._now = start
+
+    def __call__(self) -> float:
+        return self._now
+
+    def advance(self, seconds: float) -> None:
+        self._now += seconds
+
+
 class SessionTest(unittest.TestCase):
     app: KayaApp
     store: InMemorySessionStore
@@ -123,7 +135,7 @@ class SessionTest(unittest.TestCase):
             cookies_before = {c.name: c.value for c in client.cookies.jar}
             old_id = cookies_before.get('session_id')
             self.assertIsNotNone(old_id)
-            self.assertIn(old_id, self.store._sessions)
+            self.assertIn(old_id, self.store._data)
 
             r = await client.get('/rotate')
             self.assertEqual(200, r.status_code)
@@ -133,8 +145,8 @@ class SessionTest(unittest.TestCase):
             new_id = cookies_after.get('session_id')
             self.assertIsNotNone(new_id)
             self.assertNotEqual(old_id, new_id)
-            self.assertNotIn(old_id, self.store._sessions)
-            self.assertIn(new_id, self.store._sessions)
+            self.assertNotIn(old_id, self.store._data)
+            self.assertIn(new_id, self.store._data)
 
             r = await client.get('/read')
             self.assertEqual('visits: 1', r.text)
@@ -148,6 +160,37 @@ class SessionTest(unittest.TestCase):
             self.assertEqual('visits: 1', r.text)
             self.assertIn('Set-Cookie', r.headers)
 
+    @async_test
+    async def test_stale_cookie_cannot_access_old_data(self) -> None:
+        clock = FakeClock()
+        store = InMemorySessionStore(clock=clock)
+        app = KayaApp()
+        session_app = SessionMiddleware(app, store, max_age=60)
+
+        @session_app.GET('/')
+        async def home(ctx: HttpContext) -> None:
+            ctx.session['secret'] = 'super-sensitive'
+            await ctx.send_str(200, 'ok')
+
+        @session_app.GET('/read')
+        async def read(ctx: HttpContext) -> None:
+            await ctx.send_str(200, ctx.session.get('secret', 'none'))
+
+        transport = httpx.ASGITransport(app=session_app)
+        async with httpx.AsyncClient(transport=transport, base_url='http://127.0.0.1:80') as client:
+            r = await client.get('/')
+            self.assertEqual('ok', r.text)
+            old_id = client.cookies['session_id']
+            self.assertIn(old_id, store._data)
+            self.assertIn(old_id, store._expires)
+
+            clock.advance(61)
+
+            r = await client.get('/read')
+            self.assertEqual('none', r.text)
+            self.assertNotIn(old_id, store._data)
+            self.assertNotIn(old_id, store._expires)
+
 
 class SessionUnitTest(unittest.TestCase):
 
@@ -202,9 +245,8 @@ class SessionUnitTest(unittest.TestCase):
         store = InMemorySessionStore()
         session_id = store.new_session_id()
         session = Session(session_id, {'a': 1})
-        self.assertIsNone(store._sessions.get(session_id))
+        self.assertIsNone(store._data.get(session_id))
 
-        import asyncio
         asyncio.run(store.save(session_id, session))
 
         loaded = asyncio.run(store.load(session_id))
@@ -215,6 +257,53 @@ class SessionUnitTest(unittest.TestCase):
         asyncio.run(store.delete(session_id))
         self.assertIsNone(asyncio.run(store.load(session_id)))
 
+    def test_in_memory_store_expires_after_max_age(self) -> None:
+        clock = FakeClock()
+        store = InMemorySessionStore(clock=clock)
+        session_id = store.new_session_id()
+        asyncio.run(store.save(session_id, Session(session_id, {'a': 1}), max_age=60))
+        clock.advance(61)
+        self.assertIsNone(asyncio.run(store.load(session_id, max_age=60)))
+        self.assertIsNone(store._data.get(session_id))
+        self.assertIsNone(store._expires.get(session_id))
+
+    def test_in_memory_store_slides_expiry_on_load(self) -> None:
+        clock = FakeClock()
+        store = InMemorySessionStore(clock=clock)
+        session_id = store.new_session_id()
+        asyncio.run(store.save(session_id, Session(session_id, {'a': 1}), max_age=60))
+        clock.advance(30)
+        loaded = asyncio.run(store.load(session_id, max_age=60))
+        self.assertIsNotNone(loaded)
+        assert loaded is not None
+        self.assertEqual(1, loaded['a'])
+        # Without sliding, the session would expire at t=60. With sliding it is now valid until t=90.
+        clock.advance(35)
+        loaded = asyncio.run(store.load(session_id, max_age=60))
+        self.assertIsNotNone(loaded)
+        assert loaded is not None
+        self.assertEqual(1, loaded['a'])
+
+    def test_in_memory_store_no_expiry_without_max_age(self) -> None:
+        clock = FakeClock()
+        store = InMemorySessionStore(clock=clock)
+        session_id = store.new_session_id()
+        asyncio.run(store.save(session_id, Session(session_id, {'a': 1})))
+        clock.advance(1000000)
+        loaded = asyncio.run(store.load(session_id))
+        self.assertIsNotNone(loaded)
+        assert loaded is not None
+        self.assertEqual(1, loaded['a'])
+
+    def test_in_memory_store_invalidate_removes_expiry(self) -> None:
+        clock = FakeClock()
+        store = InMemorySessionStore(clock=clock)
+        session_id = store.new_session_id()
+        asyncio.run(store.save(session_id, Session(session_id, {'a': 1}), max_age=60))
+        asyncio.run(store.delete(session_id))
+        self.assertIsNone(store._data.get(session_id))
+        self.assertIsNone(store._expires.get(session_id))
+
 
 class CookieUtilTest(unittest.TestCase):