"""CSRF protection middleware for AgentForms. Extracted from app/app.py to keep the application factory clean. Provides token generation, validation, and the before_request hook. Security model: - Form submissions: validated against per-session CSRF token (Synchronizer Token Pattern) - JSON API from authenticated sessions: origin-checked instead of trusting Content-Type - API key / token auth: bypasses CSRF (token-based auth is inherently CSRF-safe) - Exempt paths: webhooks, public endpoints, login/registration entry points """ import secrets from flask import abort, current_app, jsonify, request, session # ─── CSRF-exempt paths ──────────────────────────────────────────────────────── # Routes that legitimately don't need CSRF protection: # - Webhooks (external services posting to us) # - Login/registration entry points (no session to protect yet) # - Public API endpoints (token/auth-key auth, no session cookie) # - Tracking/beacon (anonymous, no session) CSRF_EXEMPT_PATHS = { "/billing/stripe/webhook", "/auth/magic", # Login entry point — no session to protect "/auth/login", # JSON login — no session yet "/auth/register", # Registration entry point — no session yet "/documents/generate", # JSON API — authenticated sessions skip CSRF (origin-checked) "/documents/history", # JSON API "/documents/gallery", # JSON API "/api/waitlist", # Public waitlist — anonymous users, no session "/api/beacon", # Public beacon — anonymous SDK impression tracking "/api/sessions", # Public sessions — anonymous SDK analytics "/resend/webhook", # Resend webhook — external events "/tracking/bounce", # Bounce webhook — external SMTP relay "/api/v2/templates", # Email templates API — API key auth, no session "/api/v2/email/settings", # Email settings API — API key auth "/api/v2/email/domain/verify", # Domain verification API — API key auth "/api/v2/email/domain/check", # Domain check API — API key auth "/api/submit", # Public form submission — token auth, no session "/api/v2/agent/deploy", # Agent I/O — API key auth "/api/v2/agent/document", # Agent I/O — API key auth "/api/v2/agent/submissions", # Agent I/O — API key auth "/api/v2/agent/forms", # Agent I/O — API key auth } def ensure_csrf_token(): """Ensure a CSRF token exists in the current session, generating one if needed. Call this from context processors, route handlers, or middleware. Returns the token (str). """ if "csrf_token" not in session: session["csrf_token"] = secrets.token_hex(32) return session["csrf_token"] def csrf_protect(): """Flask before_request hook for CSRF protection. Returns None to allow the request, or an abort() response to block it. Protection strategy: 1. Safe HTTP methods (GET/HEAD/OPTIONS/TRACE) are always allowed 2. Testing mode skips CSRF (tests use client sessions without tokens) 3. Exempt paths (webhooks, public APIs, login entry points) are skipped 4. Dynamic form action paths for authenticated users are skipped 5. API key auth (Bearer afk_* / X-API-Key) bypasses CSRF 6. JSON requests from authenticated sessions: origin-checked (NOT content-type trusted) 7. All other requests: validated against session CSRF token """ # Safe methods never need CSRF protection if request.method in ("GET", "HEAD", "OPTIONS", "TRACE"): return None # Skip in test mode — tests use client sessions without CSRF tokens if current_app.config.get("TESTING", False): return None # Exempt paths — webhooks, public APIs, login entry points if request.path in CSRF_EXEMPT_PATHS: return None # Form actions API — dynamic paths, JSON API with session auth path = request.path parts = path.strip("/").split("/") if (len(parts) == 3 and parts[0] == "sites" and parts[2] == "actions") or ( len(parts) >= 2 and parts[0] == "actions" and parts[1].isdigit() ): if "user_id" in session: return None # API key auth (Bearer afk_* or X-API-Key) bypasses CSRF — token-based auth is CSRF-safe auth_header = request.headers.get("Authorization", "") api_key_header = request.headers.get("X-API-Key", "") if auth_header.startswith("Bearer afk_") or api_key_header.startswith("af_"): return None # Ensure a CSRF token exists in session for subsequent checks ensure_csrf_token() # ─── JSON API requests from authenticated sessions ───────────────────── # DO NOT trust request.content_type alone — it is attacker-controlled # (Content-Type header can be spoofed to bypass CSRF). # # Instead, verify the request origin matches our domain. # Cross-origin requests cannot attach SameSite cookies, so legitimate # same-origin requests will have a matching origin. if request.content_type and "application/json" in request.content_type: if "user_id" in session: origin = request.headers.get("Origin", "") host = request.host # If no Origin header, it's a same-origin request (browser omits Origin # for same-origin requests). Allow it. if not origin: return None # If Origin is present, verify the host appears in the origin URL. # This handles both localhost:5060 and https://agentforms.io. if host and host not in origin: return jsonify({"error": "Cross-origin request blocked"}), 403 return None # ─── Form submissions: Synchronizer Token Pattern ────────────────────── # Verify token from hidden form field or X-CSRF-Token header token = request.form.get("csrf_token") or request.headers.get("X-CSRF-Token") if not token or not secrets.compare_digest( token.encode("utf-8"), session["csrf_token"].encode("utf-8"), ): return abort(403, description="CSRF token missing or invalid") return None def csrf_context_processor(): """Flask context_processor that injects csrf_token into all templates. Call via: @app.context_processor\ndef inject_csrf():\n return csrf_context_processor() """ csrf_token = ensure_csrf_token() nonce = getattr(request, "csp_nonce", None) or secrets.token_urlsafe(16) return {"csrf_token": csrf_token, "csp_nonce": nonce}