security: purge data after remote session revocation (#337)

src/dashboard_auth.py

@@ -30,6 +30,12 @@ class Session:
     expires_at: int
 
 
+@dataclass(frozen=True)
+class SessionVerification:
+    session: Session | None
+    reason: str | None = None
+
+
 def access_token() -> str:
     return os.getenv("STACKCHAIN_DASHBOARD_ACCESS_TOKEN", "")
 
@@ -112,13 +118,15 @@ def issue_session(
     return f"{encoded}.{signature}", session
 
 
-def verify_session(value: str | None, now: int | None = None) -> Session | None:
+def verify_session_with_reason(
+    value: str | None, now: int | None = None
+) -> SessionVerification:
     if not value or "." not in value or not enabled():
-        return None
+        return SessionVerification(None)
     encoded, supplied_signature = value.rsplit(".", 1)
     expected = _encode(hmac.new(_secret(), encoded.encode(), hashlib.sha256).digest())
     if not hmac.compare_digest(supplied_signature, expected):
-        return None
+        return SessionVerification(None)
     try:
         payload = json.loads(_decode(encoded))
         session = Session(
@@ -127,7 +135,7 @@ def verify_session(value: str | None, now: int | None = None) -> Session | None:
             expires_at=int(payload["exp"]),
         )
     except (ValueError, TypeError, KeyError, json.JSONDecodeError):
-        return None
+        return SessionVerification(None)
     current = int(time.time() if now is None else now)
     if (
         session.expires_at <= current
@@ -136,8 +144,14 @@ def verify_session(value: str | None, now: int | None = None) -> Session | None:
         or not isinstance(session.csrf, str)
         or not session.csrf
     ):
-        return None
-    return session if _session_store(now).is_active(session.session_id, session.expires_at) else None
+        return SessionVerification(None)
+    if not _session_store(now).is_active(session.session_id, session.expires_at):
+        return SessionVerification(None, "session_revoked")
+    return SessionVerification(session)
+
+
+def verify_session(value: str | None, now: int | None = None) -> Session | None:
+    return verify_session_with_reason(value, now).session
 
 
 def revoke_session(session: Session) -> None:
@@ -160,6 +174,12 @@ async def request_session(request: Request) -> Session | None:
     return await asyncio.to_thread(verify_session, request.cookies.get(SESSION_COOKIE))
 
 
+async def request_session_verification(request: Request) -> SessionVerification:
+    return await asyncio.to_thread(
+        verify_session_with_reason, request.cookies.get(SESSION_COOKIE)
+    )
+
+
 def cookie_path(request: Request) -> str:
     root_path = request.scope.get("root_path", "").rstrip("/")
     return root_path or "/"

src/main.py

@@ -524,9 +524,12 @@ async def require_operator_session(request: Request, call_next):
         or (path == "/api/v1/session" and request.method == "POST")
     )
     session = None
+    session_reason = None
     if not public:
         try:
-            session = await dashboard_auth.request_session(request)
+            verification = await dashboard_auth.request_session_verification(request)
+            session = verification.session
+            session_reason = verification.reason
         except dashboard_auth.SessionStoreError:
             return JSONResponse(
                 {"detail": "Session registry is temporarily unavailable"},
@@ -535,13 +538,21 @@ async def require_operator_session(request: Request, call_next):
             )
     if not public and session is None:
         if path.startswith("/api/"):
+            payload = {"detail": "Authentication required"}
+            if session_reason == "session_revoked":
+                payload["code"] = session_reason
             return JSONResponse(
-                {"detail": "Authentication required"},
+                payload,
                 status_code=401,
                 headers={"Cache-Control": "no-store"},
             )
+        login_redirect = (
+            "login?reason=session-revoked"
+            if session_reason == "session_revoked"
+            else _share_target_login_redirect(request)
+        )
         return RedirectResponse(
-            _share_target_login_redirect(request),
+            login_redirect,
             status_code=303,
             headers={"Cache-Control": "no-store"},
         )

src/views.py

@@ -14,7 +14,7 @@ LOGIN_HTML = """<!doctype html>
 <style>body{margin:0;background:#07111f;color:#eef6ff;font:16px system-ui;display:grid;min-height:100vh;place-items:center}main{box-sizing:border-box;width:min(90vw,24rem);padding:2rem;border:1px solid #29415d;border-radius:1rem;background:#0d1b2b}label,input,button{display:block;width:100%;box-sizing:border-box}input,button{min-height:48px;margin-top:.6rem;border-radius:.6rem;border:1px solid #49647f;padding:.75rem}button{margin-top:1rem;background:#55d6be;color:#06121b;font-weight:700}p{color:#a9bed3}</style></head>
 <body><main><h1>Operator sign in</h1><p>Enter the dashboard access token. It is exchanged for a private, short-lived session and is never stored on this device.</p>
 <form id="sign-in"><label>Device name<input name="device_label" type="text" autocomplete="name" maxlength="64" value="This device" required></label><label>Access token<input name="access_token" type="password" autocomplete="current-password" required></label><button id="submit-sign-in">Sign in</button><p id="status" role="status" aria-live="polite"></p></form></main>
-<script src="static/login.js"></script></body></html>"""
+<script src="static/private-device-data.js"></script><script src="static/login.js"></script></body></html>"""
 
 
 class RevalidatingHTMLResponse(HTMLResponse):