bvandeusenandClaude Opus 5.5
96126b1c83
#5031 "Forged-request protection: SameSite cookie, per-session CSRF token on every non-GET, HTMX header". A page on another site could make a logged-in browser POST to Steward with the session cookie attached. - steward/auth/csrf.py: a per-session token, checked in one before_request for every POST/PUT/PATCH/DELETE (header X-CSRF-Token or form field csrf_token, constant-time compare). The bearer-authenticated host-agent ingest is exempt; /mcp never reaches Quart. - Every POST form renders {{ csrf_field() }}; base.html carries the token in a meta tag and adds it to every HTMX request; the dashboard's fetch() calls send it. - The session cookie is SameSite=Lax; login issues a fresh token. - PUBLIC_BASE_URL is now loaded from Settings at startup (it was only set when Settings → General was saved). - tests/test_csrf.py: refusal, acceptance, rotation, the ingest exemption, and a guard that every POST form in the templates carries the field. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
99 lines
3.6 KiB
Python
99 lines
3.6 KiB
Python
from __future__ import annotations
|
|
import functools
|
|
from urllib.parse import quote, urlsplit
|
|
from quart import session, redirect, url_for, abort, g, request, Response
|
|
from steward.models.users import UserRole
|
|
|
|
_ROLE_ORDER = [UserRole.viewer, UserRole.operator, UserRole.admin]
|
|
|
|
# Paths we never bounce back to after login (would loop or make no sense).
|
|
_NO_RETURN = {"/login", "/logout", "/setup"}
|
|
|
|
|
|
def role_meets(user_role: UserRole, minimum_role: UserRole) -> bool:
|
|
"""True when user_role is at least minimum_role in the viewer<operator<admin order."""
|
|
return _ROLE_ORDER.index(user_role) >= _ROLE_ORDER.index(minimum_role)
|
|
|
|
|
|
def safe_next_url(raw: str | None) -> str | None:
|
|
"""Return a safe same-site relative path+query from `raw`, else None.
|
|
|
|
Accepts a relative path or a full URL (e.g. HTMX's HX-Current-URL) and
|
|
reduces it to path[?query]. Rejects off-site/protocol-relative targets and
|
|
the auth pages themselves, so it can't be abused as an open redirect.
|
|
"""
|
|
if not raw:
|
|
return None
|
|
parts = urlsplit(raw)
|
|
# Protocol-relative (//host/…) has a netloc but no scheme — reject it; a full
|
|
# same-site URL (HX-Current-URL) has a scheme and we keep only its path.
|
|
if parts.netloc and not parts.scheme:
|
|
return None
|
|
path = parts.path
|
|
if parts.query:
|
|
path += "?" + parts.query
|
|
if not path.startswith("/") or path.startswith("//") or path.startswith("/\\"):
|
|
return None
|
|
if path.split("?", 1)[0] in _NO_RETURN:
|
|
return None
|
|
return path
|
|
|
|
|
|
def _login_redirect():
|
|
"""Redirect an unauthenticated request to /login, preserving where to return.
|
|
|
|
For HTMX requests the destination is the page the user is on (HX-Current-URL)
|
|
and we use HX-Redirect so the whole browser navigates (not a fragment swap);
|
|
for normal requests it's the requested path.
|
|
"""
|
|
is_htmx = request.headers.get("HX-Request") == "true"
|
|
nxt = safe_next_url(request.headers.get("HX-Current-URL")) if is_htmx else None
|
|
if nxt is None:
|
|
nxt = safe_next_url(request.full_path)
|
|
login_url = url_for("auth.login")
|
|
if nxt:
|
|
login_url = f"{login_url}?next={quote(nxt, safe='/')}"
|
|
if is_htmx:
|
|
resp = Response("", status=401)
|
|
resp.headers["HX-Redirect"] = login_url
|
|
return resp
|
|
return redirect(login_url)
|
|
|
|
|
|
def require_role(minimum_role: UserRole):
|
|
"""Decorator: requires authenticated user with at least minimum_role.
|
|
Also allows access for validated share-token requests (viewer level only).
|
|
"""
|
|
def decorator(f):
|
|
@functools.wraps(f)
|
|
async def wrapper(*args, **kwargs):
|
|
# Share-token requests bypass session auth for viewer-level endpoints
|
|
if getattr(g, "share_viewer", False) and minimum_role == UserRole.viewer:
|
|
return await f(*args, **kwargs)
|
|
user_id = session.get("user_id")
|
|
if not user_id:
|
|
return _login_redirect()
|
|
user_role = UserRole(session.get("user_role", "viewer"))
|
|
if _ROLE_ORDER.index(user_role) < _ROLE_ORDER.index(minimum_role):
|
|
abort(403)
|
|
return await f(*args, **kwargs)
|
|
return wrapper
|
|
return decorator
|
|
|
|
|
|
def login_user(user) -> None:
|
|
"""Write user identity into the session, with a fresh CSRF token."""
|
|
from steward.auth.csrf import rotate
|
|
rotate()
|
|
session["user_id"] = user.id
|
|
session["user_role"] = user.role.value
|
|
session["username"] = user.username
|
|
|
|
|
|
def logout_user() -> None:
|
|
session.clear()
|
|
|
|
|
|
def current_user_id() -> str | None:
|
|
return session.get("user_id")
|