The "Missing CSRF token" error #720
Description
I did the authorization according to the guide using the JWT token in cookies, everything works well until I check if the user has rights, here is my code:
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from sqlalchemy import select
from authx import AuthX, AuthXConfig
from fastapi import FastAPI, Depends, HTTPException, Response
from pydantic import BaseModel, Field
from typing import Annotated
app = FastAPI()
config = AuthXConfig()
config.JWT_SECRET_KEY = "<--Bef,eT$qme~^yS|gH(c4{IbU$/?AwD~[F5"
config.JWT_ACCESS_COOKIE_NAME = "access_cookie"
config.JWT_TOKEN_LOCATION = ["cookies"]
security = AuthX(config=config)
engine = create_async_engine('sqlite+aiosqlite:///books.db')
new_session = async_sessionmaker(engine, expire_on_commit=False)
async def get_session():
async with new_session() as session:
yield session
SessionDep = Annotated[AsyncSession, Depends(get_session)]
class Base(DeclarativeBase):
pass
class BookModel(Base):
__tablename__ = "books"
id: Mapped[int] = mapped_column(primary_key=True)
title: Mapped[str]
author: Mapped[str]
class BookPostSchema(BaseModel):
title: str = Field(max_length= 30)
author: str = Field(max_length= 20)
class BookSchema(BookPostSchema):
id: int
class LoginUserSchema(BaseModel):
login: str
password: str = Field(min_length=8)
@app.post("/setup_db", summary="Creates a new database", tags=["The database"], dependencies=[Depends(security.access_token_required)])
async def setup_db():
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.drop_all)
await conn.run_sync(Base.metadata.create_all)
return {"ok": True}
@app.post("/books", summary="Adds a new book to the database", tags=["Books"], dependencies=[Depends(security.access_token_required)])
async def add_book(data: BookPostSchema, session: SessionDep):
new_book = BookModel(
title = data.title,
author = data.author,
)
session.add(new_book)
await session.commit()
@app.get("/books", summary="Outputs all books that are in the database", tags=["Books"])
async def get_books(session: SessionDep) -> list[BookSchema]:
query = select(BookModel)
result = await session.execute(query)
return result.scalars().all()
@app.post("/login", summary="Authorizes the user", tags=["User"])
def login(creds: LoginUserSchema, response: Response):
if creds.login == "admin" and creds.password == "admin1234":
token = security.create_access_token(uid="3422342")
response.set_cookie(config.JWT_ACCESS_COOKIE_NAME, token)
return {"access_token": token}
raise HTTPException(status_code=401, detail="Incorrect login or password")Please help, I don't know how to solve this problem.