Agent Zero
93 lines
3.7 KiB
Python
93 lines
3.7 KiB
Python
"""CSRF middleware — Origin header + CSRF token validation for state-changing requests."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import logging
|
|
|
|
from fastapi import Request, status
|
|
from starlette.middleware.base import BaseHTTPMiddleware
|
|
from starlette.responses import JSONResponse
|
|
|
|
from app.config import get_settings
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class CSRFMiddleware(BaseHTTPMiddleware):
|
|
"""Validate Origin header and CSRF token on all state-changing requests.
|
|
|
|
SameSite=Strict cookie + Origin validation + double-submit CSRF token.
|
|
The CSRF token is generated at login and stored in the Redis session.
|
|
The client must send it via the X-CSRF-Token header on unsafe methods.
|
|
"""
|
|
|
|
UNSAFE_METHODS = {"POST", "PATCH", "PUT", "DELETE"}
|
|
|
|
async def dispatch(self, request: Request, call_next):
|
|
if request.method in self.UNSAFE_METHODS:
|
|
# 1. Origin header check
|
|
origin = request.headers.get("origin")
|
|
if not origin:
|
|
return JSONResponse(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
content={"detail": "Missing Origin header", "code": "csrf_missing_origin"},
|
|
)
|
|
|
|
settings = get_settings()
|
|
allowed = settings.cors_origin_list
|
|
if origin not in allowed:
|
|
return JSONResponse(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
content={"detail": "Invalid Origin", "code": "csrf_invalid_origin"},
|
|
)
|
|
|
|
# 2. CSRF token validation (double-submit pattern)
|
|
# Skip CSRF token check for auth endpoints (login/password-reset)
|
|
path = request.url.path
|
|
if path.endswith("/auth/login") or path.endswith("/auth/logout") or "/password-reset" in path:
|
|
return await call_next(request)
|
|
|
|
csrf_header = request.headers.get("x-csrf-token")
|
|
if not csrf_header:
|
|
return JSONResponse(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
content={"detail": "Missing X-CSRF-Token header", "code": "csrf_missing_token"},
|
|
)
|
|
|
|
# Get session ID from cookie to look up stored CSRF token
|
|
session_id = request.cookies.get(settings.session_cookie_name)
|
|
if not session_id:
|
|
return JSONResponse(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
content={"detail": "No session for CSRF validation", "code": "csrf_no_session"},
|
|
)
|
|
|
|
# Look up CSRF token from Redis session
|
|
import redis.asyncio as aioredis
|
|
|
|
redis = aioredis.from_url(settings.redis_url, decode_responses=True)
|
|
try:
|
|
raw = await redis.get(f"session:{session_id}")
|
|
if raw is None:
|
|
return JSONResponse(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
content={"detail": "Session expired for CSRF validation", "code": "csrf_session_expired"},
|
|
)
|
|
|
|
session_data = json.loads(raw)
|
|
stored_token = session_data.get("csrf_token")
|
|
|
|
if not stored_token or stored_token != csrf_header:
|
|
return JSONResponse(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
content={"detail": "CSRF token mismatch", "code": "csrf_token_mismatch"},
|
|
)
|
|
|
|
# Sliding session: also extend TTL on CSRF-validated unsafe requests
|
|
await redis.expire(f"session:{session_id}", settings.session_ttl_seconds)
|
|
finally:
|
|
await redis.close()
|
|
|
|
return await call_next(request)
|