feat: require operator sessions for privileged access (#258)

src/dashboard_auth.py

@@ -0,0 +1,102 @@
+"""Signed, short-lived single-operator sessions for the dashboard boundary."""
+
+import base64
+import hashlib
+import hmac
+import json
+import os
+import secrets
+import time
+from dataclasses import dataclass
+
+from fastapi import Request
+
+SESSION_COOKIE = "stackchain_session"
+CSRF_COOKIE = "stackchain_csrf"
+DEFAULT_TTL_SECONDS = 8 * 60 * 60
+
+
+@dataclass(frozen=True)
+class Session:
+    csrf: str
+    expires_at: int
+
+
+def access_token() -> str:
+    return os.getenv("STACKCHAIN_DASHBOARD_ACCESS_TOKEN", "")
+
+
+def enabled() -> bool:
+    return bool(access_token())
+
+
+def _secret() -> bytes:
+    configured = os.getenv("STACKCHAIN_DASHBOARD_SESSION_SECRET", "")
+    if configured:
+        return configured.encode()
+    return hmac.new(
+        access_token().encode(), b"stackchain-dashboard-session-signing", hashlib.sha256
+    ).digest()
+
+
+def _encode(raw: bytes) -> str:
+    return base64.urlsafe_b64encode(raw).rstrip(b"=").decode()
+
+
+def _decode(value: str) -> bytes:
+    return base64.urlsafe_b64decode(value + "=" * (-len(value) % 4))
+
+
+def issue_session(now: int | None = None) -> tuple[str, Session]:
+    issued_at = int(time.time() if now is None else now)
+    ttl = int(os.getenv("STACKCHAIN_DASHBOARD_SESSION_TTL_SECONDS", str(DEFAULT_TTL_SECONDS)))
+    session = Session(csrf=secrets.token_urlsafe(24), expires_at=issued_at + max(1, ttl))
+    payload = json.dumps(
+        {"csrf": session.csrf, "exp": session.expires_at},
+        separators=(",", ":"),
+        sort_keys=True,
+    ).encode()
+    encoded = _encode(payload)
+    signature = _encode(hmac.new(_secret(), encoded.encode(), hashlib.sha256).digest())
+    return f"{encoded}.{signature}", session
+
+
+def verify_session(value: str | None, now: int | None = None) -> Session | None:
+    if not value or "." not in value or not enabled():
+        return 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
+    try:
+        payload = json.loads(_decode(encoded))
+        session = Session(csrf=payload["csrf"], expires_at=int(payload["exp"]))
+    except (ValueError, TypeError, KeyError, json.JSONDecodeError):
+        return None
+    current = int(time.time() if now is None else now)
+    if session.expires_at <= current or not isinstance(session.csrf, str) or not session.csrf:
+        return None
+    return session
+
+
+def request_session(request: Request) -> Session | None:
+    return verify_session(request.cookies.get(SESSION_COOKIE))
+
+
+def cookie_path(request: Request) -> str:
+    root_path = request.scope.get("root_path", "").rstrip("/")
+    return root_path or "/"
+
+
+def application_path(request: Request) -> str:
+    path = request.url.path
+    root_path = request.scope.get("root_path", "").rstrip("/")
+    if root_path and (path == root_path or path.startswith(root_path + "/")):
+        return path[len(root_path):] or "/"
+    return path
+
+
+def same_origin(request: Request) -> bool:
+    origin = request.headers.get("origin", "")
+    expected = f"{request.url.scheme}://{request.url.netloc}"
+    return bool(origin) and hmac.compare_digest(origin.rstrip("/"), expected)

src/main.py

@@ -1,4 +1,5 @@
 import asyncio
+import hmac
 import math
 import os
 import time
@@ -8,12 +9,12 @@ from datetime import datetime
 from pathlib import Path
 from typing import Any, Literal
 
-from fastapi import FastAPI, Header, HTTPException, Path as PathParam, Query
-from fastapi.responses import JSONResponse
+from fastapi import FastAPI, Header, HTTPException, Path as PathParam, Query, Request, Response
+from fastapi.responses import JSONResponse, RedirectResponse
 from fastapi.staticfiles import StaticFiles
 from pydantic import BaseModel, Field, PositiveInt, field_validator, model_validator
 
-from src import gitea_proxy
+from src import dashboard_auth, gitea_proxy
 from src.gitea_proxy import (
     activity_events,
     current_user,
@@ -124,6 +125,10 @@ class ReadinessPayloadError(ValueError):
     """Raised when Gitea returns a structurally invalid readiness payload."""
 
 
+class DashboardSignIn(BaseModel):
+    access_token: str = Field(min_length=1, max_length=1_024)
+
+
 class NotificationReadBatch(BaseModel):
     ids: list[PositiveInt] = Field(min_length=1, max_length=50)
 
@@ -423,21 +428,63 @@ app.mount("/static", StaticFiles(directory=FRONTEND_DIR), name="static")
 app.include_router(frontend_router)
 
 
+@app.middleware("http")
+async def require_operator_session(request: Request, call_next):
+    if not dashboard_auth.enabled():
+        return await call_next(request)
+
+    path = dashboard_auth.application_path(request)
+    public = (
+        path in {"/healthz", "/readyz", "/login", "/manifest.webmanifest"}
+        or path.startswith("/static/")
+        or (path == "/api/v1/session" and request.method == "POST")
+    )
+    session = dashboard_auth.request_session(request)
+    if not public and session is None:
+        if path.startswith("/api/"):
+            return JSONResponse(
+                {"detail": "Authentication required"},
+                status_code=401,
+                headers={"Cache-Control": "no-store"},
+            )
+        return RedirectResponse("login", status_code=303, headers={"Cache-Control": "no-store"})
+
+    if not public and request.method not in {"GET", "HEAD", "OPTIONS"}:
+        supplied_csrf = request.headers.get("x-csrf-token", "")
+        if (
+            session is None
+            or not dashboard_auth.same_origin(request)
+            or not supplied_csrf
+            or not hmac.compare_digest(supplied_csrf, session.csrf)
+        ):
+            return JSONResponse(
+                {"detail": "Valid same-origin CSRF proof required"},
+                status_code=403,
+                headers={"Cache-Control": "no-store"},
+            )
+    request.state.dashboard_session = session
+    response = await call_next(request)
+    if path in {"/login", "/api/v1/session"}:
+        response.headers["Cache-Control"] = "no-store"
+    return response
+
+
 @app.middleware("http")
 async def prevent_live_api_caching(request, call_next):
     response = await call_next(request)
-    if request.url.path in {"/api/v1/context", "/api/v1/background-identity", "/api/v1/events", "/api/v1/live", "/api/v1/available-issues", "/api/v1/search", "/api/v1/work-route"} or request.url.path.startswith("/api/v1/work/") or (
-        request.url.path.startswith("/api/v1/repos/")
-        and request.url.path.endswith("/review")
-    ) or request.url.path.startswith("/api/v1/notifications") or (
-        request.url.path.startswith("/api/v1/repos/")
-        and ("/issues/" in request.url.path or "/pulls/" in request.url.path)
+    path = dashboard_auth.application_path(request)
+    if path in {"/api/v1/context", "/api/v1/background-identity", "/api/v1/events", "/api/v1/live", "/api/v1/available-issues", "/api/v1/search", "/api/v1/work-route"} or path.startswith("/api/v1/work/") or (
+        path.startswith("/api/v1/repos/")
+        and path.endswith("/review")
+    ) or path.startswith("/api/v1/notifications") or (
+        path.startswith("/api/v1/repos/")
+        and ("/issues/" in path or "/pulls/" in path)
     ) or (
-        request.url.path.startswith("/api/v1/repos/")
-        and request.url.path.endswith("/issues")
+        path.startswith("/api/v1/repos/")
+        and path.endswith("/issues")
     ) or (
-        request.url.path.startswith("/api/v1/repos/")
-        and request.url.path.endswith(("/labels", "/milestones"))
+        path.startswith("/api/v1/repos/")
+        and path.endswith(("/labels", "/milestones"))
     ):
         response.headers["Cache-Control"] = "no-store"
     return response
@@ -449,6 +496,68 @@ def health() -> dict[str, str]:
     return {"status": "ok", "service": "stackchain-dashboard"}
 
 
+@app.post("/api/v1/session")
+async def sign_in(payload: DashboardSignIn, request: Request, response: Response):
+    configured_token = dashboard_auth.access_token()
+    if not configured_token or not hmac.compare_digest(
+        payload.access_token, configured_token
+    ):
+        raise HTTPException(status_code=401, detail="Invalid access token")
+    signed, session = dashboard_auth.issue_session()
+    path = dashboard_auth.cookie_path(request)
+    max_age = max(1, session.expires_at - int(time.time()))
+    response.set_cookie(
+        dashboard_auth.SESSION_COOKIE,
+        signed,
+        max_age=max_age,
+        path=path,
+        secure=True,
+        httponly=True,
+        samesite="strict",
+    )
+    response.set_cookie(
+        dashboard_auth.CSRF_COOKIE,
+        session.csrf,
+        max_age=max_age,
+        path=path,
+        secure=True,
+        httponly=False,
+        samesite="strict",
+    )
+    response.headers["Cache-Control"] = "no-store"
+    return {"authenticated": True}
+
+
+@app.get("/api/v1/session")
+async def session_status(request: Request):
+    session = dashboard_auth.request_session(request)
+    return {
+        "authenticated": session is not None,
+        "csrf_token": session.csrf if session is not None else "",
+    }
+
+
+@app.delete("/api/v1/session")
+async def sign_out(request: Request, response: Response):
+    path = dashboard_auth.cookie_path(request)
+    response.delete_cookie(
+        dashboard_auth.SESSION_COOKIE,
+        path=path,
+        secure=True,
+        httponly=True,
+        samesite="strict",
+    )
+    response.delete_cookie(
+        dashboard_auth.CSRF_COOKIE,
+        path=path,
+        secure=True,
+        httponly=False,
+        samesite="strict",
+    )
+    response.headers["Cache-Control"] = "no-store"
+    return {"authenticated": False, "clear_private_device_data": True}
+
+
 @app.get("/readyz")
 async def readiness():
     """Return readiness after verifying the configured Gitea connection."""
@@ -482,11 +591,13 @@ async def readiness():
                 "Retry-After": str(max(1, math.ceil(READINESS_TIMEOUT_SECONDS)))
             } if timed_out else None,
         )
-    return {
+    payload = {
         "status": "ready",
         "service": "stackchain-dashboard",
-        "gitea_user": user["login"],
     }
+    if not dashboard_auth.enabled():
+        payload["gitea_user"] = user["login"]
+    return payload
 
 
 @app.get("/api/v1/context")

src/views.py

@@ -8,6 +8,14 @@ DASHBOARD_FILE = Path(__file__).resolve().parent.parent / "frontend" / "index.ht
 MANIFEST_FILE = DASHBOARD_FILE.parent / "manifest.webmanifest"
 SERVICE_WORKER_FILE = DASHBOARD_FILE.parent / "service-worker.js"
 
+LOGIN_HTML = """<!doctype html>
+<html lang="en"><head><meta charset="utf-8"><meta name="viewport" content="width=device-width,initial-scale=1">
+<title>Sign in · Stackchain Dashboard</title>
+<style>body{margin:0;background:#07111f;color:#eef6ff;font:16px system-ui;display:grid;min-height:100vh;place-items:center}main{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>Access token<input name="access_token" type="password" autocomplete="current-password" required></label><button>Sign in</button><p id="status" role="status"></p></form></main>
+<script>document.getElementById('sign-in').addEventListener('submit',async(event)=>{event.preventDefault();const status=document.getElementById('status');status.textContent='Signing in…';const access_token=new FormData(event.currentTarget).get('access_token');const response=await fetch('api/v1/session',{method:'POST',headers:{'Content-Type':'application/json'},body:JSON.stringify({access_token})});if(response.ok){location.replace('./');return}status.textContent='Sign-in failed. Check the token and try again.';event.currentTarget.reset()});</script></body></html>"""
+
 
 class RevalidatingHTMLResponse(HTMLResponse):
     def __init__(self, content, status_code=200, headers=None, media_type=None, background=None):
@@ -21,6 +29,11 @@ async def dashboard() -> str:
     return DASHBOARD_FILE.read_text()
 
 
+@router.get("/login", response_class=RevalidatingHTMLResponse)
+async def login() -> str:
+    return LOGIN_HTML
+
+
 @router.get("/manifest.webmanifest", response_class=FileResponse)
 async def web_app_manifest() -> FileResponse:
     return FileResponse(MANIFEST_FILE, media_type="application/manifest+json")