"""Centralized CSRF protection utilities.
This module is the single source of truth for CSRF token generation and
validation. All state-changing API routes (POST/PUT/DELETE/PATCH) must be
decorated with ``require_csrf`` unless they are explicitly exempt
(public forms like login/signup/demo-request).
Startup Audit
-------------
On ``app.factory`` completion (called from ``app/__init__.py``),
``audit_exempt_routes()`` scans the Flask URL map and warns if any
state-changing route lacks ``@require_csrf``. Log lines go to
``app.logger`` — they do **not** abort startup but surface in logs
for security review.
"""
from __future__ import annotations
import secrets
import logging
import uuid
from functools import wraps
from flask import current_app, request, jsonify, session as flask_session, session
from flask_login import current_user
logger = logging.getLogger(__name__)
def generate_csrf_token() -> str:
"""Return a cryptographically secure CSRF token string."""
return secrets.token_hex(32)
# ─── Exempt routes ──────────────────────────────────────────────────────────
# Routes that legitimately skip CSRF: public auth endpoints, webhooks, and
# health checks. Each entry is a (endpoint_name, methods) tuple.
# ``endpoint_name`` is the Flask blueprint function name (e.g. ``auth_api.login``).
CSRF_EXEMPT_ENDPOINTS: set[str] = {
# Auth (public)
"auth_api.login",
"auth_api.signup",
"auth_api.logout",
"auth_api.forgot_password",
"auth_api.reset_password",
# Webhooks — external services (no user session)
"billing.stripe_webhook",
"billing.stripe_portal",
"sms_webhooks.ngrok_webhook",
"sms_webhooks.twilio_webhook",
"lead_gen.google_ads_webhook",
"lead_gen.facebook_ads_webhook",
# Health / misc
"health_check",
"auth_api.csrf_token",
# OAuth callbacks
"connectors.oauth_callback",
"connectors.oauth_callback_url",
}
def require_csrf(f):
"""Decorator: verify X-CSRF-Token header matches the session token.
Skip unauthenticated requests (no session token to match) and Flask
test mode (``app.testing = True``). Browsers with SameSite=Lax cookies
are partially protected already; this is defense-in-depth.
"""
@wraps(f)
def decorated(*args, **kwargs):
# Skip in test mode
if current_app.testing:
return f(*args, **kwargs)
# Only enforce for authenticated sessions
if current_user.is_authenticated:
session_token = flask_session.get("csrf_token")
request_token = (
request.headers.get("X-CSRF-Token")
or request.headers.get("X-Csrf-Token")
)
if not session_token or not request_token:
return jsonify({"error": "CSRF token missing"}), 403
if not secrets.compare_digest(session_token, request_token):
return jsonify({"error": "CSRF token invalid"}), 403
return f(*args, **kwargs)
return decorated
def audit_exempt_routes(app):
"""Scan the Flask URL map and warn about unprotected state-changing routes.
Called from ``create_app()`` after blueprint registration. Logs a
WARNING for every route that accepts POST/PUT/DELETE/PATCH but is
not decorated with ``require_csrf`` and is not in the exempt list.
"""
unprotected = []
for rule in app.url_map.iter_rules():
if rule.methods is None:
continue
state_methods = rule.methods & {"POST", "PUT", "DELETE", "PATCH"}
if not state_methods:
continue
endpoint = rule.endpoint
# Skip static, SPA fallback, and exempt endpoints
if endpoint in ("static", "send_static"):
continue
if endpoint in CSRF_EXEMPT_ENDPOINTS:
continue
# Check if the view function has the require_csrf wrapper
view_func = app.view_functions.get(endpoint)
if view_func is None:
continue
# The decorator preserves the original function name via @wraps,
# so we check the __wrapped__ chain for 'require_csrf'
has_csrf = False
_check = view_func
while hasattr(_check, "__wrapped__"):
if getattr(_check, "__name__", None) == "require_csrf":
has_csrf = True
break
_check = _check.__wrapped__
if not has_csrf:
unprotected.append(f"{rule.rule} [{', '.join(sorted(state_methods))}] → {endpoint}")
if unprotected:
logger.warning(
"CSRF audit: %d unprotected state-changing route(s):\n %s",
len(unprotected),
"\n ".join(unprotected),
)
else:
logger.info("CSRF audit: all state-changing routes are protected.")