feat: bind operator sessions to upstream identity (Closes #1372)

tests/test_dashboard_auth.py

@@ -10,7 +10,7 @@ import pytest
 from fastapi import Request
 
 from src import main
-from src.session_store import SessionStoreError
+from src.session_store import SessionStatus, SessionStoreError
 from src.views import FRONTEND_BUILD
 
 
@@ -29,6 +29,83 @@ def access_control(monkeypatch, tmp_path):
     monkeypatch.setenv("STACKCHAIN_LOGIN_ATTEMPT_DB", str(tmp_path / "login-attempts.sqlite3"))
     monkeypatch.setenv("STACKCHAIN_LOGIN_MAX_FAILURES", "3")
     monkeypatch.setenv("STACKCHAIN_LOGIN_WINDOW_SECONDS", "60")
+    async def upstream_user():
+        return {"id": 42, "login": "timmy"}
+
+    monkeypatch.setattr(main, "current_user", upstream_user)
+
+
+def test_signed_session_carries_bound_upstream_identity(access_control):
+    signed, _ = main.dashboard_auth.issue_session(
+        now=1_000,
+        principal_id=42,
+        principal_login="timmy",
+    )
+
+    verification = main.dashboard_auth.verify_session_with_reason(signed, now=1_001)
+
+    assert verification.session is not None
+    assert verification.session.principal_id == 42
+    assert verification.session.principal_login == "timmy"
+
+
+@pytest.mark.anyio
+async def test_session_is_blocked_when_upstream_account_changes(
+    access_control, monkeypatch
+):
+    identity = {"id": 42, "login": "timmy"}
+
+    async def upstream_user():
+        return dict(identity)
+
+    monkeypatch.setattr(main, "current_user", upstream_user)
+    transport = httpx.ASGITransport(app=main.app)
+    async with httpx.AsyncClient(transport=transport, base_url="https://test") as client:
+        signed_in = await client.post(
+            "/api/v1/session",
+            json={"access_token": "correct horse battery staple"},
+        )
+        identity.update(login="renamed-timmy")
+        same_principal = await client.get("/api/v1/security-events")
+        identity.update(id=84, login="other-operator")
+        protected = await client.get("/api/v1/security-events")
+
+    assert signed_in.status_code == 200
+    assert same_principal.status_code == 200
+    assert protected.status_code == 401
+    assert protected.json() == {
+        "detail": "Connected Gitea account changed; sign in again",
+        "code": "session_idle",
+        "reason": "upstream_identity_changed",
+        "previous_login": "timmy",
+        "current_login": "other-operator",
+    }
+
+
+@pytest.mark.anyio
+async def test_session_identity_lookup_outage_is_retryable(access_control, monkeypatch):
+    available = True
+
+    async def upstream_user():
+        if not available:
+            raise RuntimeError("credential lookup details")
+        return {"id": 42, "login": "timmy"}
+
+    monkeypatch.setattr(main, "current_user", upstream_user)
+    transport = httpx.ASGITransport(app=main.app)
+    async with httpx.AsyncClient(transport=transport, base_url="https://test") as client:
+        signed_in = await client.post(
+            "/api/v1/session",
+            json={"access_token": "correct horse battery staple"},
+        )
+        available = False
+        protected = await client.get("/api/v1/security-events")
+
+    assert signed_in.status_code == 200
+    assert protected.status_code == 503
+    assert protected.json() == {"detail": "Gitea identity is temporarily unavailable"}
+    assert protected.headers["retry-after"] == "5"
+    assert "credential lookup" not in protected.text
 
 
 async def fresh_grant(client, action: str, target: str) -> str:
@@ -1904,7 +1981,7 @@ async def test_logout_revokes_a_captured_cookie_before_gitea(access_control, mon
     assert before.status_code == 200
     assert after.status_code == 401
     assert after.headers["cache-control"] == "no-store"
-    assert calls == 1
+    assert calls == 4
 
 
 @pytest.mark.anyio
@@ -1918,7 +1995,9 @@ async def test_session_registry_latency_does_not_block_the_event_loop(access_con
         class SlowStore:
             def status(self, session_id, expires_at, *, idle_timeout_seconds):
                 time.sleep(0.15)
-                return "active"
+                return SessionStatus(
+                    "active", principal_id=42, principal_login="timmy"
+                )
 
         monkeypatch.setattr(main.dashboard_auth, "_session_store", lambda now=None: SlowStore())
         private_request = asyncio.create_task(client.get("/api/v1/session"))
@@ -2084,7 +2163,9 @@ async def test_single_session_revocation_does_not_block_the_event_loop(
         class SlowStore:
             def status(self, session_id, expires_at, *, idle_timeout_seconds):
                 time.sleep(0.02)
-                return "active"
+                return SessionStatus(
+                    "active", principal_id=42, principal_login="timmy"
+                )
 
             def revoke(self, session_id):
                 time.sleep(0.15)
@@ -2126,6 +2207,7 @@ async def test_session_registry_read_failure_fails_closed_before_gitea(
         await client.post(
             "/api/v1/session", json={"access_token": "correct horse battery staple"}
         )
+        called = False
 
         class BrokenStore:
             def status(self, session_id, expires_at, *, idle_timeout_seconds):
@@ -2174,7 +2256,9 @@ async def test_logout_registry_failure_does_not_claim_revocation(access_control,
 
         class BrokenStore:
             def status(self, session_id, expires_at, *, idle_timeout_seconds):
-                return "active"
+                return SessionStatus(
+                    "active", principal_id=42, principal_login="timmy"
+                )
 
             def revoke(self, session_id):
                 raise SessionStoreError("database path and secret details")

tests/test_dashboard_session_frontend.py

@@ -558,6 +558,28 @@ process.stdout.write(JSON.stringify(state));
     assert result["replacedAfterDeletion"] is True
 
 
+def test_upstream_identity_change_preserves_private_work_and_requires_sign_in():
+    result = run_session_scenario(
+        """
+state.responseStatus = 401;
+state.responsePayload = {
+  detail:'Connected Gitea account changed; sign in again',
+  code:'session_idle', reason:'upstream_identity_changed',
+  previous_login:'timmy', current_login:'other-operator',
+};
+await boundary.fetch('/dashboard/api/v1/live');
+state.remaining = Array.from(storage.values.keys());
+process.stdout.write(JSON.stringify(state));
+"""
+    )
+
+    assert result["remaining"] == ["stackchain.private", "gitea.preference"]
+    assert result["deletedDatabases"] == []
+    assert result["deletedCaches"] == []
+    assert result["workerMessages"] == []
+    assert result["replaced"] == ["/dashboard/login?reason=session-idle"]
+
+
 def test_idle_response_locks_without_clearing_private_queued_work():
     result = run_session_scenario(
         """

tests/test_login_frontend.py

@@ -54,7 +54,7 @@ process.stdout.write(JSON.stringify({{ status: status.textContent }}));
     )
 
     assert json.loads(result.stdout)["status"] == (
-        "Stackchain locked after inactivity. Your drafts and queued work are still "
+        "Stackchain locked. Your drafts and queued work are still "
         "on this device. Sign in to resume."
     )
 

tests/test_security_activity.py

@@ -18,6 +18,11 @@ def security_access(monkeypatch, tmp_path):
     monkeypatch.setenv("STACKCHAIN_SESSION_DB", str(tmp_path / "sessions.sqlite3"))
     monkeypatch.setenv("STACKCHAIN_LOGIN_ATTEMPT_DB", str(tmp_path / "attempts.sqlite3"))
     monkeypatch.setenv("STACKCHAIN_SECURITY_EVENT_DB", str(tmp_path / "security.sqlite3"))
+
+    async def upstream_user():
+        return {"id": 42, "login": "timmy"}
+
+    monkeypatch.setattr(main, "current_user", upstream_user)
     return tmp_path
 
 

tests/test_session_store.py

@@ -88,6 +88,23 @@ def test_session_status_reports_idle_without_background_validation_extending_act
     assert store.status("phone-session", 3_000, idle_timeout_seconds=900) == "idle"
 
 
+def test_active_session_status_carries_bound_upstream_identity(tmp_path):
+    store = SessionStore(tmp_path / "sessions.sqlite3", clock=lambda: 1_000.0)
+
+    store.activate(
+        "phone-session",
+        3_000,
+        principal_id=42,
+        principal_login="timmy",
+    )
+
+    status = store.status("phone-session", 3_000, idle_timeout_seconds=900)
+
+    assert status == "active"
+    assert status.principal_id == 42
+    assert status.principal_login == "timmy"
+
+
 def test_managed_session_status_enforces_absolute_and_idle_expiry(tmp_path):
     now = [1_000.0]
     store = SessionStore(tmp_path / "sessions.sqlite3", clock=lambda: now[0])
@@ -230,6 +247,30 @@ def test_idle_status_migrates_existing_registry_and_starts_legacy_idle_clock_now
     assert last_active_at == 1_000
 
 
+def test_identity_status_migrates_registry_that_already_has_idle_tracking(tmp_path):
+    database = tmp_path / "sessions.sqlite3"
+    digest = SessionStore._digest("existing-session")
+    with sqlite3.connect(database) as connection:
+        connection.execute(
+            "CREATE TABLE active_sessions ("
+            "session_hash TEXT PRIMARY KEY, expires_at INTEGER NOT NULL, "
+            "management_id TEXT, device_label TEXT, created_at INTEGER, last_active_at INTEGER"
+            ")"
+        )
+        connection.execute(
+            "INSERT INTO active_sessions VALUES (?, ?, ?, ?, ?, ?)",
+            (digest, 2_000, "legacy-device", "Existing device", 900, 1_000),
+        )
+
+    status = SessionStore(database, clock=lambda: 1_000.0).status(
+        "existing-session", 2_000, idle_timeout_seconds=900
+    )
+
+    assert status == "active"
+    assert status.principal_id is None
+    assert status.principal_login is None
+
+
 def test_existing_session_can_mint_first_step_up_grant_during_schema_upgrade(tmp_path):
     database = tmp_path / "sessions.sqlite3"
     digest = SessionStore._digest("existing-session")