"""FastAPI dependency for CSRF verification on admin mutating endpoints. Mount on any route that performs a state change (POST / PUT / DELETE) and sits inside the admin router. GET admin routes do NOT need this — they only need ``require_admin`` to gate access. Two flavors, same underlying check: - :func:`require_csrf_form` — reads ``csrf_token`` from the form body. Preferred for classic HTML forms. - :func:`require_csrf_header` — reads ``X-CSRF-Token`` from the request headers. Used by the live-preview fetch and the drag-drop upload endpoint where sending an extra form field is awkward. Both raise HTTP 403 on mismatch, which surfaces as the generic FastAPI error page. No information leaks about which side (cookie or submitted token) failed — fail-closed is uniform. """ from __future__ import annotations from typing import Optional from fastapi import Form, HTTPException, Request from app.services.csrf import CSRF_COOKIE_NAME, CSRFService def _get_csrf_service(request: Request) -> CSRFService: """Pull the app-scoped :class:`CSRFService` off ``request.app.state``. Private helper: route handlers depend on :func:`require_csrf_form` or :func:`require_csrf_header` directly, not on this lookup. """ return request.app.state.csrf_service def require_csrf_form( request: Request, csrf_token: str = Form(default=""), ) -> None: """Verify a form-submitted CSRF token matches the signed cookie. Raises :class:`fastapi.HTTPException` 403 on any mismatch. On success returns ``None`` — the dependency has no payload. """ service: CSRFService = _get_csrf_service(request) cookie_value: Optional[str] = request.cookies.get(CSRF_COOKIE_NAME) if not service.verify(cookie_value=cookie_value, submitted=csrf_token): raise HTTPException(status_code=403, detail="CSRF verification failed") def require_csrf_header( request: Request, ) -> None: """Verify an ``X-CSRF-Token`` header matches the signed cookie. Used by the JS-driven preview + media upload endpoints. The header name is case-insensitive — Starlette canonicalizes on read. """ service: CSRFService = _get_csrf_service(request) cookie_value: Optional[str] = request.cookies.get(CSRF_COOKIE_NAME) submitted: Optional[str] = request.headers.get("x-csrf-token") if not service.verify(cookie_value=cookie_value, submitted=submitted): raise HTTPException(status_code=403, detail="CSRF verification failed")