feat: revoke operator sessions on logout (#268)

tests/test_dashboard_auth.py

@@ -2,13 +2,15 @@ import httpx
 import pytest
 
 from src import main
+from src.session_store import SessionStoreError
 
 
 @pytest.fixture
-def access_control(monkeypatch):
+def access_control(monkeypatch, tmp_path):
     monkeypatch.setenv("STACKCHAIN_DASHBOARD_AUTH_MODE", "operator")
     monkeypatch.setenv("STACKCHAIN_DASHBOARD_ACCESS_TOKEN", "correct horse battery staple")
     monkeypatch.setenv("STACKCHAIN_DASHBOARD_SESSION_SECRET", "a-separate-session-signing-secret-with-enough-entropy")
+    monkeypatch.setenv("STACKCHAIN_SESSION_DB", str(tmp_path / "sessions.sqlite3"))
 
 
 @pytest.mark.anyio
@@ -207,6 +209,132 @@ async def test_logout_clears_session_and_blocks_private_routes(access_control):
     assert all("Max-Age=0" in value for value in response.headers.get_list("set-cookie"))
 
 
+@pytest.mark.anyio
+async def test_logout_revokes_a_captured_cookie_before_gitea(access_control, monkeypatch):
+    calls = 0
+
+    async def user():
+        nonlocal calls
+        calls += 1
+        return {"id": 1, "login": "timmy"}
+
+    monkeypatch.setattr(main, "current_user", user)
+    transport = httpx.ASGITransport(app=main.app)
+    async with httpx.AsyncClient(transport=transport, base_url="https://test") as client:
+        await client.post(
+            "/api/v1/session", json={"access_token": "correct horse battery staple"}
+        )
+        captured = client.cookies["stackchain_session"]
+        csrf = client.cookies["stackchain_csrf"]
+
+        async with httpx.AsyncClient(
+            transport=transport,
+            base_url="https://test",
+            headers={"Cookie": f"stackchain_session={captured}"},
+        ) as replay:
+            before = await replay.get("/api/v1/background-identity")
+
+        await client.delete(
+            "/api/v1/session",
+            headers={"Origin": "https://test", "X-CSRF-Token": csrf},
+        )
+
+        async with httpx.AsyncClient(
+            transport=transport,
+            base_url="https://test",
+            headers={"Cookie": f"stackchain_session={captured}"},
+        ) as replay:
+            after = await replay.get("/api/v1/background-identity")
+
+    assert before.status_code == 200
+    assert after.status_code == 401
+    assert after.headers["cache-control"] == "no-store"
+    assert calls == 1
+
+
+@pytest.mark.anyio
+async def test_session_registry_read_failure_fails_closed_before_gitea(
+    access_control, monkeypatch
+):
+    called = False
+
+    async def user():
+        nonlocal called
+        called = True
+        return {"id": 1, "login": "timmy"}
+
+    monkeypatch.setattr(main, "current_user", user)
+    transport = httpx.ASGITransport(app=main.app)
+    async with httpx.AsyncClient(transport=transport, base_url="https://test") as client:
+        await client.post(
+            "/api/v1/session", json={"access_token": "correct horse battery staple"}
+        )
+
+        class BrokenStore:
+            def is_active(self, session_id, expires_at):
+                raise SessionStoreError("database path and secret details")
+
+        monkeypatch.setattr(main.dashboard_auth, "_session_store", lambda now=None: BrokenStore())
+        response = await client.get("/api/v1/background-identity")
+        health = await client.get("/healthz")
+
+    assert response.status_code == 503
+    assert response.json() == {"detail": "Session registry is temporarily unavailable"}
+    assert response.headers["cache-control"] == "no-store"
+    assert "database path" not in response.text
+    assert called is False
+    assert health.status_code == 200
+
+
+@pytest.mark.anyio
+async def test_sign_in_registry_write_failure_issues_no_cookie(access_control, monkeypatch):
+    class BrokenStore:
+        def activate(self, session_id, expires_at):
+            raise SessionStoreError("database path and secret details")
+
+    monkeypatch.setattr(main.dashboard_auth, "_session_store", lambda now=None: BrokenStore())
+    transport = httpx.ASGITransport(app=main.app)
+    async with httpx.AsyncClient(transport=transport, base_url="https://test") as client:
+        response = await client.post(
+            "/api/v1/session", json={"access_token": "correct horse battery staple"}
+        )
+
+    assert response.status_code == 503
+    assert response.json() == {"detail": "Session registry is temporarily unavailable"}
+    assert response.headers["cache-control"] == "no-store"
+    assert response.headers.get_list("set-cookie") == []
+    assert "database path" not in response.text
+
+
+@pytest.mark.anyio
+async def test_logout_registry_failure_does_not_claim_revocation(access_control, monkeypatch):
+    transport = httpx.ASGITransport(app=main.app)
+    async with httpx.AsyncClient(transport=transport, base_url="https://test") as client:
+        await client.post(
+            "/api/v1/session", json={"access_token": "correct horse battery staple"}
+        )
+        csrf = client.cookies["stackchain_csrf"]
+
+        class BrokenStore:
+            def is_active(self, session_id, expires_at):
+                return True
+
+            def revoke(self, session_id):
+                raise SessionStoreError("database path and secret details")
+
+        monkeypatch.setattr(main.dashboard_auth, "_session_store", lambda now=None: BrokenStore())
+        response = await client.delete(
+            "/api/v1/session",
+            headers={"Origin": "https://test", "X-CSRF-Token": csrf},
+        )
+
+    assert response.status_code == 503
+    assert response.json() == {"detail": "Session registry is temporarily unavailable"}
+    assert response.headers["cache-control"] == "no-store"
+    assert response.headers.get_list("set-cookie") == []
+    assert "database path" not in response.text
+
+
 @pytest.mark.anyio
 async def test_public_routes_remain_available_and_readiness_hides_identity(access_control, monkeypatch):
     async def user():

tests/test_session_store.py

@@ -0,0 +1,31 @@
+import sqlite3
+
+from src.session_store import SessionStore
+
+
+def test_revocation_is_durable_and_scoped_to_one_session(tmp_path):
+    now = [1_000.0]
+    database = tmp_path / "sessions.sqlite3"
+    first = SessionStore(database, clock=lambda: now[0])
+    first.activate("first-session-secret", 2_000)
+    first.activate("second-session-secret", 2_000)
+
+    reconstructed = SessionStore(database, clock=lambda: now[0])
+    reconstructed.revoke("first-session-secret")
+
+    assert reconstructed.is_active("first-session-secret", 2_000) is False
+    assert reconstructed.is_active("second-session-secret", 2_000) is True
+    assert b"first-session-secret" not in database.read_bytes()
+    assert b"second-session-secret" not in database.read_bytes()
+
+
+def test_expired_sessions_are_rejected_and_pruned(tmp_path):
+    now = [1_000.0]
+    store = SessionStore(tmp_path / "sessions.sqlite3", clock=lambda: now[0])
+    store.activate("expiring-session", 1_001)
+
+    now[0] = 1_001.0
+
+    assert store.is_active("expiring-session", 1_001) is False
+    with sqlite3.connect(store.path) as connection:
+        assert connection.execute("SELECT COUNT(*) FROM active_sessions").fetchone() == (0,)