fix(users): never cache a session_version read that a revocation raced; single-flight fills

CHANGELOG.md

@@ -24,6 +24,7 @@ All notable changes to this project are documented in this file. The format is b
 - Scaffold new modules with `core/` and `web/` areas and mirrored nested Python tests; existing modules migrate individually. See `docs/module-layout.md`.
 
 ### Security
+- The users session-revocation cache no longer stores a `session_version` read that a revocation raced: a read in flight when "sign out everywhere", a password change or a logout landed could put the old counter back and re-admit the revoked sessions in that worker until the cache TTL expired. Fills now go through `simple_module_core.singleflight.fill` (epoch-checked), and concurrent misses for one account share a single read.
 - **Behaviour change:** local browser auth mutations now refuse `Sec-Fetch-Site: cross-site`
   and require the session-bound CSRF token (no Origin/Referer equality check, so TLS proxies work), extending the shared token support (#504).
   Native API clients without browser headers retain their existing workflow.

modules/users/tests/accounts/session_version/test_users_session_version_fill.py

@@ -0,0 +1,99 @@
+"""Cache fills: one read per account, and never a read that a revocation raced.
+
+A read that started before "sign out everywhere" can finish after the forget.
+Storing it would put the old counter back and re-admit every revoked session
+in this worker until the TTL ran out.
+"""
+
+from __future__ import annotations
+
+import asyncio
+
+import pytest
+from users.accounts.session_version_cache import (
+    SESSION_VERSION_TTL_SECONDS,
+    clear_session_version_cache,
+    configure_session_version_cache,
+    fill_session_version,
+    forget_session_version,
+    peek_session_version,
+)
+
+_USER = "dana"
+
+
+@pytest.fixture(autouse=True)
+def _cold_cache():
+    clear_session_version_cache()
+    yield
+    configure_session_version_cache(SESSION_VERSION_TTL_SECONDS)
+    clear_session_version_cache()
+
+
+class TestARacedReadIsNotCached:
+    @pytest.mark.anyio
+    async def test_a_forget_during_the_read_discards_it_and_rereads(self):
+        reads = []
+
+        async def read():
+            reads.append(len(reads))
+            if len(reads) == 1:
+                # The revocation commits and forgets while this read is in flight.
+                forget_session_version(_USER)
+                return 1, frozenset()
+            return 2, frozenset()
+
+        assert await fill_session_version(_USER, read) == (2, frozenset())
+        assert len(reads) == 2
+        assert peek_session_version(_USER) == 2
+
+    @pytest.mark.anyio
+    async def test_a_cache_clear_during_the_read_also_discards_it(self):
+        calls = 0
+
+        async def read():
+            nonlocal calls
+            calls += 1
+            clear_session_version_cache()
+            return calls, frozenset()
+
+        # Every read is raced, so the last one answers this request uncached.
+        assert await fill_session_version(_USER, read) == (3, frozenset())
+        assert peek_session_version(_USER) is None
+
+
+class TestConcurrentMissesShareOneRead:
+    @pytest.mark.anyio
+    async def test_five_waiters_one_read(self):
+        reads = 0
+        release = asyncio.Event()
+
+        async def read():
+            nonlocal reads
+            reads += 1
+            await release.wait()
+            return 4, frozenset({"retired"})
+
+        waiters = [asyncio.create_task(fill_session_version(_USER, read)) for _ in range(5)]
+        await asyncio.sleep(0)
+        release.set()
+        results = await asyncio.gather(*waiters)
+
+        assert reads == 1
+        assert results == [(4, frozenset({"retired"}))] * 5
+
+
+class TestAFailedReadIsNotCached:
+    @pytest.mark.anyio
+    async def test_the_error_propagates_and_the_next_fill_reads_again(self):
+        async def broken():
+            raise ConnectionError("database unavailable")
+
+        with pytest.raises(ConnectionError):
+            await fill_session_version(_USER, broken)
+        assert peek_session_version(_USER) is None
+
+        async def read():
+            return 5, frozenset()
+
+        assert await fill_session_version(_USER, read) == (5, frozenset())

modules/users/users/accounts/provider.py

@@ -32,6 +32,7 @@
 from users.accounts.session_version_cache import (  # noqa: F401
     SESSION_VERSION_TTL_SECONDS,
     clear_session_version_cache,
+    fill_session_version,
     forget_session_version,
     peek_session_version,
     read_session_version,
@@ -195,12 +196,13 @@ async def _version_still_current(self, scope, user_id: uuid_mod.UUID, session) -
         hit, cached = read_session_version(user_id, session_id)
         if hit:
             return cached is not None and int(cached) == _stamped_version(session)
-        try:
-            from sqlalchemy import and_, select
+        from sqlalchemy import and_, select
 
-            from users.models import RevokedSession, User
+        from users.models import RevokedSession, User
 
-            session_factory = scope["app"].state.sm.db.session_factory
+        session_factory = scope["app"].state.sm.db.session_factory
+
+        async def read() -> tuple[int | None, frozenset[str]]:
             async with session_factory() as db_session:
                 rows = (
                     await db_session.execute(
@@ -215,16 +217,18 @@ async def _version_still_current(self, scope, user_id: uuid_mod.UUID, session) -
                         .where(User.id == user_id)
                     )
                 ).all()
-                stored = rows[0][0] if rows else None
-                revoked = frozenset(row[1] for row in rows if row[1] is not None)
+            stored = rows[0][0] if rows else None
+            return stored, frozenset(row[1] for row in rows if row[1] is not None)
+
+        try:
+            stored, revoked = await fill_session_version(user_id, read)
         except Exception:
             # A database that cannot be reached is not evidence of a
             # revocation. Refusing here would log every signed-in user out of
             # an app that is merely having a bad minute. Not cached — the next
             # request should retry rather than inherit the outage.
             logger.exception("Session version check failed for %s; keeping the session", user_id)
             return True
-        store_session_version(user_id, stored, revoked)
         if stored is None:
             return False
         return int(stored) == _stamped_version(session) and session_id not in revoked

modules/users/users/accounts/session_version_cache.py

@@ -9,12 +9,16 @@
 
 from __future__ import annotations
 
+from collections.abc import Awaitable, Callable
+
 from cachetools import TTLCache
+from simple_module_core.singleflight import MISS, SingleFlight, fill
 
 __all__ = [
     "SESSION_VERSION_TTL_SECONDS",
     "clear_session_version_cache",
     "configure_session_version_cache",
+    "fill_session_version",
     "forget_session_version",
     "peek_session_version",
     "read_session_version",
@@ -73,8 +77,13 @@
 """
 
 
-_MISS = object()
-"""Distinguishes "not cached" from a cached ``None`` in a single lookup."""
+_FLIGHT = SingleFlight()
+# Bumped by every forget so a read that started before a revocation is never
+# cached after it: that would re-admit the revoked sessions for a whole TTL.
+_epoch = 0
+
+type SessionVersionEntry = tuple[int | None, frozenset[str]]
+"""``(session_version, retired session ids)``; ``None`` for a missing account."""
 
 
 def forget_session_version(user_id) -> None:
@@ -83,6 +92,8 @@ def forget_session_version(user_id) -> None:
     Called by whatever just changed it — "sign out everywhere" and a password
     change — so this process stops answering from the value it read before.
     """
+    global _epoch
+    _epoch += 1
     _SESSION_VERSIONS.pop(user_id, None)
 
 
@@ -99,6 +110,8 @@ def peek_session_version(user_id):
 
 def clear_session_version_cache() -> None:
     """Empty the cache — used by tests that need a cold read."""
+    global _epoch
+    _epoch += 1
     _SESSION_VERSIONS.clear()
 
 
@@ -116,8 +129,8 @@ def read_session_version(user_id, session_id: str | None = None) -> tuple[bool,
     caller has no ``except KeyError``, so landing in it turned a revocation
     check into a 500 on an authenticated request.
     """
-    cached = _SESSION_VERSIONS.get(user_id, _MISS)
-    if cached is _MISS:
+    cached = _SESSION_VERSIONS.get(user_id, MISS)
+    if cached is MISS:
         return False, None
     version, retired = cached
     if session_id in retired:
@@ -132,6 +145,24 @@ def store_session_version(
     _SESSION_VERSIONS[user_id] = (None if stored is None else int(stored), revoked)
 
 
+async def fill_session_version(
+    user_id, read: Callable[[], Awaitable[SessionVersionEntry]]
+) -> SessionVersionEntry:
+    """Read through *read* once per account, caching only reads no forget raced.
+
+    Concurrent misses for one account share a single read. *read*'s exceptions
+    propagate uncached, so the caller decides how an outage is answered.
+    """
+    async with _FLIGHT.hold(user_id):
+        stored, revoked = await fill(
+            read,
+            epoch=lambda: _epoch,
+            store=lambda entry: store_session_version(user_id, *entry),
+            cached=lambda: _SESSION_VERSIONS.get(user_id, MISS),
+        )
+    return stored, revoked
+
+
 def session_version_ttl() -> float:
     """The window currently in effect, in seconds."""
     return _SESSION_VERSIONS.ttl
@@ -153,8 +184,9 @@ def configure_session_version_cache(ttl_seconds: int) -> None:
     round trip and can keep the cache. This TTL is then the bound on a *dropped*
     message rather than on every cross-worker revocation.
     """
-    global _SESSION_VERSIONS
+    global _SESSION_VERSIONS, _epoch
     ttl = max(0, int(ttl_seconds))
     if ttl == _SESSION_VERSIONS.ttl:
         return
+    _epoch += 1
     _SESSION_VERSIONS = TTLCache(maxsize=_CACHE_MAXSIZE, ttl=ttl)