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>
64 lines
2.3 KiB
Python
64 lines
2.3 KiB
Python
"""Forged-request (CSRF) protection for the browser side of Steward.
|
|
|
|
A page on another site can make a logged-in browser POST to Steward, and the
|
|
browser sends Steward's session cookie with it. Every state-changing request
|
|
from a browser therefore has to carry a token that only Steward's own pages
|
|
know: a random value kept in the session, written into each form by
|
|
`csrf_field()` and sent by HTMX and fetch() as the `X-CSRF-Token` header (set
|
|
once on <body> in base.html).
|
|
|
|
One check in a before_request covers every route, so a new route cannot
|
|
forget it. Callers that authenticate with a bearer token instead of the
|
|
session cookie (the host agent's ingest; /mcp never reaches Quart) are exempt:
|
|
a forged browser request cannot carry their token.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import hmac
|
|
import secrets
|
|
|
|
from markupsafe import Markup, escape
|
|
from quart import Response, request, session
|
|
|
|
SESSION_KEY = "_csrf"
|
|
FORM_FIELD = "csrf_token"
|
|
HEADER = "X-CSRF-Token"
|
|
UNSAFE_METHODS = ("POST", "PUT", "PATCH", "DELETE")
|
|
# Bearer-authenticated endpoints: no session cookie, nothing to forge.
|
|
EXEMPT_PATHS = frozenset({"/plugins/host_agent/ingest"})
|
|
|
|
|
|
def csrf_token() -> str:
|
|
"""This session's token, created on first use."""
|
|
token = session.get(SESSION_KEY)
|
|
if not token:
|
|
token = session[SESSION_KEY] = secrets.token_urlsafe(32)
|
|
return token
|
|
|
|
|
|
def csrf_field() -> Markup:
|
|
"""The hidden input every POST form carries."""
|
|
return Markup(f'<input type="hidden" name="{FORM_FIELD}" value="{escape(csrf_token())}">')
|
|
|
|
|
|
def rotate() -> None:
|
|
"""Drop the token, so a new one is issued — at login, where the session
|
|
changes hands."""
|
|
session.pop(SESSION_KEY, None)
|
|
|
|
|
|
async def check_request() -> Response | None:
|
|
"""before_request hook: refuse an unsafe request without the session's token."""
|
|
if request.method not in UNSAFE_METHODS or request.path in EXEMPT_PATHS:
|
|
return None
|
|
expected = session.get(SESSION_KEY)
|
|
sent = request.headers.get(HEADER, "")
|
|
if not sent:
|
|
form = await request.form
|
|
sent = form.get(FORM_FIELD, "")
|
|
if expected and sent and hmac.compare_digest(expected, sent):
|
|
return None
|
|
return Response(
|
|
"This form has expired or did not come from Steward. Reload the page and try again.",
|
|
status=400, content_type="text/plain; charset=utf-8")
|