feat(auth): deny-by-default gate middleware + require_role

Adds operator_gate Starlette HTTP middleware that gates every route
except an explicit allow-list. Flag defaults OFF so all existing
behaviour and tests are unchanged. wire_operator_auth helper in
conftest lets tests monkeypatch the module-global SessionLocal and
flag, keeping the gate's own DB session pointed at the test engine.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
2026-06-17 19:22:15 +00:00
parent e8fe4845aa
commit 2879abb355
4 changed files with 156 additions and 0 deletions
+69
View File
@@ -9,10 +9,15 @@ import os
import time
import uuid
from datetime import datetime, timedelta
from urllib.parse import quote
from fastapi import Request, HTTPException
from fastapi.responses import JSONResponse, RedirectResponse
from backend.models import OperatorUser
from backend.auth_passwords import hash_password, verify_password, generate_password
from backend.auth_cookies import sign, read, COOKIE_SECURE
from backend.database import SessionLocal
# Feature flag — OFF by default. When off, the gate and require_role both pass
# everything through and the app behaves exactly as it does today.
@@ -156,3 +161,67 @@ def set_operator_role(db, user, role: str):
user.role = role
db.commit()
return user
# Routes reachable with no login. A new route added next year is gated by default.
_EXEMPT_EXACT = {
"/login", "/logout", "/health",
"/manifest.json", "/sw.js", "/favicon.ico", "/offline-db.js",
"/portal", # portal home (its own auth)
# machine endpoints — LAN-only, automated, no human (watchers/heartbeats):
"/emitters/report", "/api/series3/heartbeat", "/api/series4/heartbeat",
}
_EXEMPT_PREFIX = ("/static/", "/portal/")
def _is_exempt(path: str) -> bool:
return path in _EXEMPT_EXACT or path.startswith(_EXEMPT_PREFIX)
async def operator_gate(request: Request, call_next):
"""Deny-by-default gate. Flag off → pass through (app as today). Flag on →
exempt paths pass; otherwise require a valid operator session, stash it on
request.state.operator, and force a password change when pending."""
if not OPERATOR_AUTH_ENABLED:
return await call_next(request)
path = request.url.path
if _is_exempt(path):
return await call_next(request)
db = SessionLocal()
try:
user = current_operator(request, db)
if user is not None:
db.expunge(user) # detach a fully-loaded row so we can close now
finally:
db.close()
if user is None:
if path.startswith("/api/"):
return JSONResponse({"detail": "Not authenticated"}, status_code=401)
return RedirectResponse(f"/login?next={quote(path)}", status_code=303)
if user.must_change_password and path not in ("/change-password", "/logout"):
if path.startswith("/api/"):
return JSONResponse({"detail": "Password change required"}, status_code=403)
return RedirectResponse("/change-password", status_code=303)
request.state.operator = user
return await call_next(request)
def require_role(minimum: str):
"""Dependency factory: require a logged-in operator ranked >= `minimum`.
Respects the flag (off → pass through). When on, the middleware has already
set request.state.operator before this runs."""
def _dep(request: Request):
if not OPERATOR_AUTH_ENABLED:
return None
user = getattr(request.state, "operator", None)
if user is None:
raise HTTPException(status_code=401, detail="Not authenticated")
if not role_at_least(user.role, minimum):
raise HTTPException(status_code=403, detail="Insufficient permissions")
return user
return _dep