feat: initial shared library extraction
Contains shared infrastructure for all coop services: - shared/ (factory, urls, user_loader, context, internal_api, jinja_setup) - models/ (User, Order, Calendar, Ticket, Product, Ghost CMS) - db/ (SQLAlchemy async session, base) - suma_browser/app/ (csrf, middleware, errors, authz, redis_cacher, payments, filters, utils) - suma_browser/templates/ (shared base layouts, macros, error pages) - static/ (CSS, JS, fonts, images) - alembic/ (database migrations) - config/ (app-config.yaml) - editor/ (Lexical editor Node.js build) - requirements.txt Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
99
suma_browser/app/csrf.py
Normal file
99
suma_browser/app/csrf.py
Normal file
@@ -0,0 +1,99 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import secrets
|
||||
from typing import Callable, Awaitable, Optional
|
||||
|
||||
from quart import (
|
||||
abort,
|
||||
current_app,
|
||||
request,
|
||||
session as qsession,
|
||||
)
|
||||
|
||||
SAFE_METHODS = {"GET", "HEAD", "OPTIONS", "TRACE"}
|
||||
|
||||
|
||||
def generate_csrf_token() -> str:
|
||||
"""
|
||||
Per-session CSRF token.
|
||||
|
||||
In Jinja:
|
||||
<input type="hidden" name="csrf_token" value="{{ csrf_token() }}">
|
||||
"""
|
||||
token = qsession.get("csrf_token")
|
||||
if not token:
|
||||
token = secrets.token_urlsafe(32)
|
||||
qsession["csrf_token"] = token
|
||||
return token
|
||||
|
||||
|
||||
def _is_exempt_endpoint() -> bool:
|
||||
endpoint = request.endpoint
|
||||
if not endpoint:
|
||||
return False
|
||||
view = current_app.view_functions.get(endpoint)
|
||||
|
||||
# Walk decorator stack (__wrapped__) to find csrf_exempt
|
||||
while view is not None:
|
||||
if getattr(view, "_csrf_exempt", False):
|
||||
return True
|
||||
view = getattr(view, "__wrapped__", None)
|
||||
|
||||
return False
|
||||
|
||||
|
||||
async def protect() -> None:
|
||||
"""
|
||||
Enforce CSRF on unsafe methods.
|
||||
|
||||
Supports:
|
||||
* Forms: hidden input "csrf_token"
|
||||
* JSON: "csrf_token" or "csrfToken" field
|
||||
* HTMX/AJAX: "X-CSRFToken" or "X-CSRF-Token" header
|
||||
"""
|
||||
if request.method in SAFE_METHODS:
|
||||
return
|
||||
|
||||
if _is_exempt_endpoint():
|
||||
return
|
||||
|
||||
session_token = qsession.get("csrf_token")
|
||||
if not session_token:
|
||||
abort(400, "Missing CSRF session token")
|
||||
|
||||
supplied_token: Optional[str] = None
|
||||
|
||||
# JSON body
|
||||
if request.mimetype == "application/json":
|
||||
data = await request.get_json(silent=True) or {}
|
||||
supplied_token = data.get("csrf_token") or data.get("csrfToken")
|
||||
|
||||
# Form body
|
||||
if not supplied_token and request.mimetype != "application/json":
|
||||
form = await request.form
|
||||
supplied_token = form.get("csrf_token")
|
||||
|
||||
# Headers (HTMX / fetch)
|
||||
if not supplied_token:
|
||||
supplied_token = (
|
||||
request.headers.get("X-CSRFToken")
|
||||
or request.headers.get("X-CSRF-Token")
|
||||
)
|
||||
|
||||
if not supplied_token or supplied_token != session_token:
|
||||
abort(400, "Invalid CSRF token")
|
||||
|
||||
|
||||
def csrf_exempt(view: Callable[..., Awaitable]) -> Callable[..., Awaitable]:
|
||||
"""
|
||||
Mark a view as CSRF-exempt.
|
||||
|
||||
from suma_browser.app.csrf import csrf_exempt
|
||||
|
||||
@csrf_exempt
|
||||
@blueprint.post("/hook")
|
||||
async def webhook():
|
||||
...
|
||||
"""
|
||||
setattr(view, "_csrf_exempt", True)
|
||||
return view
|
||||
Reference in New Issue
Block a user