deerflow-code/offline-backend-20260512/backend/app/gateway/csrf_middleware.py

"""CSRF protection middleware for FastAPI.

Per RFC-001:
State-changing operations require CSRF protection.
"""

import secrets
from collections.abc import Callable

from app.gateway.auth.disabled_mode import is_auth_disabled
from app.gateway.proxy_path import app_relative_path
from fastapi import Request, Response
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.responses import JSONResponse
from starlette.types import ASGIApp

CSRF_COOKIE_NAME = "csrf_token"
CSRF_HEADER_NAME = "X-CSRF-Token"
CSRF_TOKEN_LENGTH = 64  # bytes


def is_secure_request(request: Request) -> bool:
    """Detect whether the original client request was made over HTTPS."""
    return request.headers.get("x-forwarded-proto", request.url.scheme) == "https"


def generate_csrf_token() -> str:
    """Generate a secure random CSRF token."""
    return secrets.token_urlsafe(CSRF_TOKEN_LENGTH)


def should_check_csrf(request: Request) -> bool:
    """Determine if a request needs CSRF validation.

    CSRF is checked for state-changing methods (POST, PUT, DELETE, PATCH).
    GET, HEAD, OPTIONS, and TRACE are exempt per RFC 7231.
    """
    if request.method not in ("POST", "PUT", "DELETE", "PATCH"):
        return False

    # Match the application-relative path so a reverse-proxy prefix
    # (e.g. /xxx/api/v1/auth/me) does not defeat these whitelists.
    path = app_relative_path(request)
    stripped = path.rstrip("/")
    # Exempt /api/v1/auth/me endpoint
    if stripped == "/api/v1/auth/me":
        return False
    # Anonymous public endpoints (mirror auth_middleware._PUBLIC_PATH_PREFIXES).
    # These intentionally have no cookie session, so the Double-Submit Cookie
    # pattern is both impossible (no token to compare) and unnecessary (no
    # session for an attacker to ride). Without this exemption every public
    # PUT/POST returns 403 even though auth was bypassed earlier in the chain.
    if path.startswith("/api/public/"):
        return False
    return True


_AUTH_EXEMPT_PATHS: frozenset[str] = frozenset(
    {
        "/api/v1/auth/login/local",
        "/api/v1/auth/login/username",
        # Token-exchange login is sessionless on first call (no CSRF cookie yet),
        # exactly like the other login endpoints — mirror auth_middleware's
        # _PUBLIC_EXACT_PATHS, which already whitelists it.
        "/api/v1/auth/login/token",
        "/api/v1/auth/logout",
        "/api/v1/auth/register",
        "/api/v1/auth/initialize",
        # 并行多智能体面板专用取 token 端点（公共 + CSRF 豁免）。
        "/api/parallel-agents/auth/token",
    }
)


def is_auth_endpoint(request: Request) -> bool:
    """Check if the request is to an auth endpoint.

    Auth endpoints don't need CSRF validation on first call (no token).
    Matches the application-relative path so a reverse-proxy prefix does not
    defeat the whitelist.
    """
    return app_relative_path(request).rstrip("/") in _AUTH_EXEMPT_PATHS


def _has_bearer_auth(request: Request) -> bool:
    """Return True when Authorization header carries a Bearer token."""
    authorization = request.headers.get("authorization", "").strip()
    if not authorization:
        return False
    scheme, _, token = authorization.partition(" ")
    return scheme.lower() == "bearer" and bool(token.strip())


def _has_cookie_session(request: Request) -> bool:
    """Return True when the request carries a cookie-based session.

    Double-Submit-Cookie CSRF only protects requests the browser authenticates
    **automatically via cookies** — that is the exact surface a cross-site
    forgery can ride. A request with no ``access_token`` session cookie has no
    session for an attacker to exploit, so enforcing CSRF on it is both
    pointless and harmful: a freshly-loaded browser (e.g. one that just cleared
    its cache/cookies) hitting the login endpoint would otherwise be rejected
    with "CSRF token missing" before it can even sign in. Such an
    unauthenticated request, if forged, is rejected by the auth layer (401)
    anyway.
    """
    return bool(request.cookies.get("access_token"))


class CSRFMiddleware(BaseHTTPMiddleware):
    """Middleware that implements CSRF protection using Double Submit Cookie pattern."""

    def __init__(self, app: ASGIApp) -> None:
        super().__init__(app)

    async def dispatch(self, request: Request, call_next: Callable) -> Response:
        _is_auth = is_auth_endpoint(request)

        if is_auth_disabled():
            return await call_next(request)

        # Double-submit-cookie CSRF applies to cookie-based browser sessions.
        # API clients using explicit Bearer tokens are not CSRF-prone, and a
        # request with no session cookie at all (e.g. a fresh browser signing
        # in) has no session to forge — only enforce when an ``access_token``
        # cookie is actually present.
        if should_check_csrf(request) and not _is_auth and not _has_bearer_auth(request) and _has_cookie_session(request):
            cookie_token = request.cookies.get(CSRF_COOKIE_NAME)
            header_token = request.headers.get(CSRF_HEADER_NAME)

            if not cookie_token or not header_token:
                return JSONResponse(
                    status_code=403,
                    content={"detail": "CSRF token missing. Include X-CSRF-Token header."},
                )

            if not secrets.compare_digest(cookie_token, header_token):
                return JSONResponse(
                    status_code=403,
                    content={"detail": "CSRF token mismatch."},
                )

        response = await call_next(request)

        # For auth endpoints that set up session, also set CSRF cookie
        if _is_auth and request.method == "POST":
            # Generate a new CSRF token for the session
            csrf_token = generate_csrf_token()
            is_https = is_secure_request(request)
            response.set_cookie(
                key=CSRF_COOKIE_NAME,
                value=csrf_token,
                httponly=False,  # Must be JS-readable for Double Submit Cookie pattern
                secure=is_https,
                samesite="strict",
            )

        return response


def get_csrf_token(request: Request) -> str | None:
    """Get the CSRF token from the current request's cookies.

    This is useful for server-side rendering where you need to embed
    token in forms or headers.
    """
    return request.cookies.get(CSRF_COOKIE_NAME)