The CSRF token is invalid. Can't figure out what is wrong #11528
-
First Check
Commit to Help
Example Codeimport os
import sys
import aiohttp
import asyncio
import logging
from django.core.asgi import get_asgi_application
from fastapi import FastAPI, Request, Form, UploadFile, Depends, Response
from fastapi.responses import HTMLResponse, JSONResponse
from fastapi.templating import Jinja2Templates
from fastapi.staticfiles import StaticFiles
from starlette.applications import Starlette
from starlette.routing import Mount, Router
from fastapi_csrf_protect.exceptions import TokenValidationError, InvalidHeaderError
from channels.auth import AuthMiddlewareStack
from channels.routing import ProtocolTypeRouter, URLRouter
from channels.layers import get_channel_layer
from django.urls import path
from fastapi_csrf_protect import CsrfProtect
from django.shortcuts import render
import secrets
from fastapi_csrf_protect import CsrfProtect
from fastapi_csrf_protect.exceptions import CsrfProtectError
from pydantic import BaseModel
django_asgi = get_asgi_application()
fastapi_app = FastAPI(debug=True)
templates_dir = os.path.join(os.path.dirname(__file__), '..', 'web_tool', 'templates', 'web_tool')
templates = Jinja2Templates(directory=templates_dir)
static_path = os.path.join(os.path.dirname(os.path.dirname(__file__)), 'web_tool', 'static')
logging.basicConfig(level=logging.DEBUG)
logger = logging.getLogger(__name__)
class CsrfSettings(BaseModel):
secret_key: str = f"{secrets.token_urlsafe(32)}"
cookie_samesite: str = "lax"
cookie_samesite: str = "none"
cookie_secure: bool = True
@CsrfProtect.load_config
def get_csrf_config():
return CsrfSettings()
def generate_csrf_token() -> str:
return secrets.token_urlsafe(32)
@fastapi_app.get("/csrf-token")
async def get_csrf_token(response: Response):
csrf_token = secrets.token_urlsafe(32)
response.set_cookie(
key="fastapi-csrf-token",
value=csrf_token,
httponly=True,
samesite='lax',
secure=False
)
logger.debug(f"CSRF Token Issued: {csrf_token}")
return JSONResponse({"csrf_token": csrf_token})
@fastapi_app.get("/account_management")
async def get_account_management(request: Request, csrf_protect: CsrfProtect = Depends()):
csrf_token, signed_token = csrf_protect.generate_csrf_tokens()
return templates.TemplateResponse(
'account_management.html',
{"request": request, "csrf_token": csrf_token}
)
csrf_protect.set_csrf_cookie(signed_token, response)
return response
@fastapi_app.post("/account_management", response_class=JSONResponse)
async def post_account_management(request: Request,
csrf_protect: CsrfProtect = Depends(),
account_ids: str = Form(None),
file: UploadFile = None,
acc_operation: str = Form(None),
acc_action: str = Form(None)
):
try:
try:
logger.debug(f"Request Headers: {request.headers}")
logger.debug(f"Request Cookies: {request.cookies}")
logger.debug(f"Received CSRF Token from Headers: {request.headers.get('X-CSRF-Token')}")
logger.debug(f"Received CSRF Token from Cookies: {request.cookies.get('fastapi-csrf-token')}")
await csrf_protect.validate_csrf(request)
logger.debug("CSRF token validation passed")
except (CsrfProtectError, InvalidHeaderError, TokenValidationError) as e:
logger.error(f"CSRF Token error: {e}")
return JSONResponse(status_code=e.status_code, content={"detail": e.message})
return JSONResponse(status_code=e.status_code, content={"detail": e.message})
except (CsrfProtectError, InvalidHeaderError, TokenValidationError, Exception) as e:
logger.error(f"CSRF Token Validation Error: {str(e)}")
return JSONResponse(status_code=e.status_code, content={"detail": e.message})
@fastapi_app.exception_handler(CsrfProtectError)
def csrf_protect_exception_handler(request: Request, exc: CsrfProtectError):
return JSONResponse(status_code=exc.status_code, content={"detail": exc.message})
# Define WebSocket routes
websocket_routes = [
path('ws/acc_management_progress/<str:task_id>/', AccountStatusConsumer.as_asgi()),
]
class CustomASGIApp:
def __init__(self):
self.django_app = django_asgi
self.fastapi_app = Mount("/fastapi", app=fastapi_app)
async def __call__(self, scope, receive, send):
if scope["type"] == "http" and scope["path"].startswith("/fastapi"):
await self.fastapi_app(scope, receive, send)
else:
await self.django_app(scope, receive, send)
application = ProtocolTypeRouter({
"http": CustomASGIApp(), # Custom app that directs requests correctly
"websocket": AuthMiddlewareStack(URLRouter(websocket_routes)),
}) DescriptionHello team.
Operating SystemWindows Operating System DetailsWindows 11 FastAPI Version0.110.3 Pydantic Version2.7.1 Python VersionPython 3.10.6 Additional ContextNo response |
Beta Was this translation helpful? Give feedback.
Replies: 2 comments
-
As I see you have two csrf tokens:
So, I think you should send |
Beta Was this translation helpful? Give feedback.
-
Yes, it helped. Thank you. |
Beta Was this translation helpful? Give feedback.
As I see you have two csrf tokens:
fastapi-csrf-token
, generated bysecrets.token_urlsafe(32)
csrftoken
, generated bycsrf_protect.generate_csrf_tokens()
csrf_protect.validate_csrf(request)
expects that you passcsrftoken
in thex-csrf-token
, but in your request you putfastapi-csrf-token
into this header parameter.So, I think you should send
csrftoken
instead offastapi-csrf-token
in thex-csrf-token
header