关于CSRF的后端问题（FastAPI、AuthX、SQLAlchemy、Pydantic等）

人工智能 2026-07-11

我在尝试创建一个带有登录系统和角色的简单API。我的代码有个问题：

from fastapi import FastAPI, Response, HTTPException, Request, Depends
from fastapi.responses import JSONResponse
from pydantic import BaseModel
from sqlalchemy import select
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from typing import Annotated
from authx import AuthX, AuthXConfig

app = FastAPI()

config = AuthXConfig()
config.JWT_SECRET_KEY = "secret_key"
config.JWT_ACCESS_COOKIE_NAME = "cookie"
config.JWT_TOKEN_LOCATION = ["cookies"]
config.JWT_COOKIE_CSRF_PROTECT = True
security = AuthX(config=config)

engine = create_async_engine("sqlite+aiosqlite:///basedata.db")
Session = async_sessionmaker(engine, expire_on_commit=False)

async def get_session():
    async with Session() as session:
        yield session

SessionDep = Annotated[AsyncSession, Depends(get_session)]

class Base(DeclarativeBase):
    pass

class LogModel(Base):
    __tablename__ = "Log_Model"
    Login: Mapped[str] = mapped_column(primary_key=True)
    Pass: Mapped[str]
    Role: Mapped[str] = mapped_column(default="user")

class LogAddSchema(BaseModel):
    Login: str
    Pass: str

@app.post("/run_engine")
async def run_engine():
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    return {"Data": "Runned"}

@app.post("/add_new_user")
async def add_user(user: LogAddSchema, session: SessionDep):
    stmt = select(LogModel).where(LogModel.Login == user.Login)
    result = await session.execute(stmt)
    if result.scalar_one_or_none():
        return {"error": "Користувач уже існує"}
    new_user = LogModel(Login=user.Login, Pass=user.Pass)
    session.add(new_user)
    await session.commit()
    return {"message": "Користувач створений"}

@app.post("/add_admin")
async def add_admin(user: LogAddSchema, session: SessionDep):
    stmt = select(LogModel).where(LogModel.Login == user.Login)
    result = await session.execute(stmt)
    if result.scalar_one_or_none():
        return {"error": "Користувач уже існує"}
    new_user = LogModel(Login=user.Login, Pass=user.Pass, Role="admin")
    session.add(new_user)
    await session.commit()
    return {"message": "Адміністратор створений"}

@app.post("/user_login")
async def user_login(user: LogAddSchema, session: SessionDep, response: Response):
    stmt = select(LogModel).where(LogModel.Login == user.Login)
    result = await session.execute(stmt)
    db_user = result.scalar_one_or_none()
    if db_user is None or db_user.Pass != user.Pass:
        raise HTTPException(status_code=400, detail="Incorrect login or password")
    token = security.create_access_token(uid=db_user.Login, data={"role": db_user.Role})
    response = JSONResponse(content={"message": "Вхід успішний"})
    security.set_access_cookies(response, token)

    return response

@app.get("/profile")
async def profile(request: Request):
    try:
        payload = await security.access_token_required(request)
        return {"user": payload.sub, "role": getattr(payload, "role", "user")}
    except HTTPException:
        return {"Please": "Login"}

@app.get("/admin")
async def admin_panel(request: Request):
    try:
        payload = await security.access_token_required(request)
        role = getattr(payload, "role", "user")
        if role != "admin":
            raise HTTPException(status_code=403, detail="Access denied")
        return {"message": "Admin panel"}
    except HTTPException:
        return {"Please": "Login"}

@app.delete("/delete_user/{login}")
async def delete_user(login: str, request: Request, session: SessionDep):
    payload = await security.access_token_required(request)
    role = getattr(payload, "role", "user")
    if role != "admin":
        raise HTTPException(status_code=403, detail="Access denied")
    stmt = select(LogModel).where(LogModel.Login == login)
    result = await session.execute(stmt)
    user = result.scalar_one_or_none()
    if not user:
        raise HTTPException(status_code=404, detail="User not found")
    await session.delete(user)
    await session.commit()
    return {"message": f"User '{login}' deleted successfully"}

当我想使用user_login端点时，出现如下错误：

Traceback (most recent call last):
  File "D:\Codes\Python\env\Lib\site-packages\uvicorn\protocols\http\httptools_impl.py", line 416, in run_asgi
    result = await app(  # type: ignore[func-returns-value]
             ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
        self.scope, self.receive, self.send
        ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
    )
    ^
  File "D:\Codes\Python\env\Lib\site-packages\uvicorn\middleware\proxy_headers.py", line 60, in __call__
    return await self.app(scope, receive, send)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "D:\Codes\Python\env\Lib\site-packages\fastapi\applications.py", line 1134, in __call__
    await super().__call__(scope, receive, send)
  File "D:\Codes\Python\env\Lib\site-packages\starlette\applications.py", line 107, in __call__
    await self.middleware_stack(scope, receive, send)
  File "D:\Codes\Python\env\Lib\site-packages\starlette\middleware\errors.py", line 186, in __call__
    raise exc
  File "D:\Codes\Python\env\Lib\site-packages\starlette\middleware\errors.py", line 164, in __call__
    await self.app(scope, receive, _send)
  File "D:\Codes\Python\env\Lib\site-packages\starlette\middleware\exceptions.py", line 63, in __call__
    await wrap_app_handling_exceptions(self.app, conn)(scope, receive, send)
  File "D:\Codes\Python\env\Lib\site-packages\starlette\_exception_handler.py", line 53, in wrapped_app
    raise exc
  File "D:\Codes\Python\env\Lib\site-packages\starlette\_exception_handler.py", line 42, in wrapped_app
    await app(scope, receive, sender)
  File "D:\Codes\Python\env\Lib\site-packages\fastapi\middleware\asyncexitstack.py", line 18, in __call__
    await self.app(scope, receive, send)
  File "D:\Codes\Python\env\Lib\site-packages\starlette\routing.py", line 716, in __call__
    await self.middleware_stack(scope, receive, send)
  File "D:\Codes\Python\env\Lib\site-packages\starlette\routing.py", line 736, in app
    await route.handle(scope, receive, send)
  File "D:\Codes\Python\env\Lib\site-packages\starlette\routing.py", line 290, in handle
    await self.app(scope, receive, send)
  File "D:\Codes\Python\env\Lib\site-packages\fastapi\routing.py", line 119, in app
    await wrap_app_handling_exceptions(app, request)(scope, receive, send)
  File "D:\Codes\Python\env\Lib\site-packages\starlette\_exception_handler.py", line 53, in wrapped_app
    raise exc
  File "D:\Codes\Python\env\Lib\site-packages\starlette\_exception_handler.py", line 42, in wrapped_app
    await app(scope, receive, sender)
  File "D:\Codes\Python\env\Lib\site-packages\fastapi\routing.py", line 105, in app
    response = await f(request)
               ^^^^^^^^^^^^^^^^
  File "D:\Codes\Python\env\Lib\site-packages\fastapi\routing.py", line 424, in app
    raw_response = await run_endpoint_function(
                   ^^^^^^^^^^^^^^^^^^^^^^^^^^^^
    ...<3 lines>...
    )
    ^
  File "D:\Codes\Python\env\Lib\site-packages\fastapi\routing.py", line 312, in run_endpoint_function
    return await dependant.call(**values)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "D:\Codes\Python\backend.py", line 78, in user_login
    security.set_access_cookies(response, token)
    ~~~~~~~~~~~~~~~~~~~~~~~~~~~^^^^^^^^^^^^^^^^^
  File "D:\Codes\Python\env\Lib\site-packages\authx\main.py", line 482, in set_access_cookies
    self._set_cookies(token=token, type="access", response=response, max_age=max_age)
    ~~~~~~~~~~~~~~~~~^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "D:\Codes\Python\env\Lib\site-packages\authx\main.py", line 194, in _set_cookies
    response.set_cookie(
    ^^^^^^^^^^^^^^^^^^^
AttributeError: 'str' object has no attribute 'set_cookie'
ERROR:uvicorn.error:Exception in ASGI application
Traceback (most recent call last):
  File "D:\Codes\Python\env\Lib\site-packages\uvicorn\protocols\http\httptools_impl.py", line 416, in run_asgi
    result = await app(  # type: ignore[func-returns-value]
             ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
        self.scope, self.receive, self.send
        ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
    )
    ^
  File "D:\Codes\Python\env\Lib\site-packages\uvicorn\middleware\proxy_headers.py", line 60, in __call__
    return await self.app(scope, receive, send)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "D:\Codes\Python\env\Lib\site-packages\fastapi\applications.py", line 1134, in __call__
    await super().__call__(scope, receive, send)
  File "D:\Codes\Python\env\Lib\site-packages\starlette\applications.py", line 107, in __call__
    await self.middleware_stack(scope, receive, send)
  File "D:\Codes\Python\env\Lib\site-packages\starlette\middleware\errors.py", line 186, in __call__
    raise exc
  File "D:\Codes\Python\env\Lib\site-packages\starlette\middleware\errors.py", line 164, in __call__
    await self.app(scope, receive, _send)
  File "D:\Codes\Python\env\Lib\site-packages\starlette\middleware\exceptions.py", line 63, in __call__
    await wrap_app_handling_exceptions(self.app, conn)(scope, receive, send)
  File "D:\Codes\Python\env\Lib\site-packages\starlette\_exception_handler.py", line 53, in wrapped_app
    raise exc
  File "D:\Codes\Python\env\Lib\site-packages\starlette\_exception_handler.py", line 42, in wrapped_app
    await app(scope, receive, sender)
  File "D:\Codes\Python\env\Lib\site-packages\fastapi\middleware\asyncexitstack.py", line 18, in __call__
    await self.app(scope, receive, send)
  File "D:\Codes\Python\env\Lib\site-packages\starlette\routing.py", line 716, in __call__
    await self.middleware_stack(scope, receive, send)
  File "D:\Codes\Python\env\Lib\site-packages\starlette\routing.py", line 736, in app
    await route.handle(scope, receive, send)
  File "D:\Codes\Python\env\Lib\site-packages\starlette\routing.py", line 290, in handle
    await self.app(scope, receive, send)
  File "D:\Codes\Python\env\Lib\site-packages\fastapi\routing.py", line 119, in app
    await wrap_app_handling_exceptions(app, request)(scope, receive, send)
  File "D:\Codes\Python\env\Lib\site-packages\starlette\_exception_handler.py", line 53, in wrapped_app
    raise exc
  File "D:\Codes\Python\env\Lib\site-packages\starlette\_exception_handler.py", line 42, in wrapped_app
    await app(scope, receive, sender)
  File "D:\Codes\Python\env\Lib\site-packages\fastapi\routing.py", line 105, in app
    response = await f(request)
               ^^^^^^^^^^^^^^^^
  File "D:\Codes\Python\env\Lib\site-packages\fastapi\routing.py", line 424, in app
    raw_response = await run_endpoint_function(
                   ^^^^^^^^^^^^^^^^^^^^^^^^^^^^
    ...<3 lines>...
    )
    ^
  File "D:\Codes\Python\env\Lib\site-packages\fastapi\routing.py", line 312, in run_endpoint_function
    return await dependant.call(**values)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "D:\Codes\Python\backend.py", line 78, in user_login
    security.set_access_cookies(response, token)
    ~~~~~~~~~~~~~~~~~~~~~~~~~~~^^^^^^^^^^^^^^^^^
  File "D:\Codes\Python\env\Lib\site-packages\authx\main.py", line 482, in set_access_cookies
    self._set_cookies(token=token, type="access", response=response, max_age=max_age)
    ~~~~~~~~~~~~~~~~~^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "D:\Codes\Python\env\Lib\site-packages\authx\main.py", line 194, in _set_cookies
    response.set_cookie(
    ^^^^^^^^^^^^^^^^^^^
AttributeError: 'str' object has no attribute 'set_cookie'

有些AI说这是AuthX和 CSRF的问题，但我不确定。

解决方案

这并非主要的CSRF问题。导致直接崩溃的原因是set_access_cookies() 的参数顺序颠倒了。AuthX的文档指出该方法的签名是set_access_cookies(token, response, max_age=None)，因此token必须在前，Response对象在后。在你的代码中你调用security.set_access_cookies(response, token)，这会让AuthX将 JWT字符串当作响应对象。随后它尝试在该字符串上调用set_cookie()，从而引发AttributeError: 'str' 对象没有属性 'set_cookie'。

所以这行 security.set_access_cookies(response, token) 应该改为 security.set_access_cookies(token, response)。

你的user_login端点应该像这样：

@app.post("/user_login")
async def user_login(user: LogAddSchema, session: SessionDep):
    stmt = select(LogModel).where(LogModel.Login == user.Login)
    result = await session.execute(stmt)
    db_user = result.scalar_one_or_none()

    if db_user is None or db_user.Pass != user.Pass:
        raise HTTPException(status_code=400, detail="Incorrect login or password")

    token = security.create_access_token(
        uid=db_user.Login,
        data={"role": db_user.Role}
    )

    response = JSONResponse(content={"message": "Вхід успішний"})
    security.set_access_cookies(token, response)
    return response

在 /profile和 /admin这两个端点中也存在一个逻辑问题。它会捕获所有HTTPExceptions，包括你自己的403 Access denied，并把它们变成 "Please": "Login"。因此经过身份验证但非管理员的用户可能会看到错误的提示。

except HTTPException:
    return {"Please": "Login"}

可以像下面这样写一个更精炼的版本：

@app.get("/profile")
async def profile(request: Request):
    payload = await security.access_token_required(request)
    return {"user": payload.sub, "role": getattr(payload, "role", "user")}

@app.get("/admin")
async def admin_panel(request: Request):
    payload = await security.access_token_required(request)
    role = getattr(payload, "role", "user")
    if role != "admin":
        raise HTTPException(status_code=403, detail="Access denied")
    return {"message": "Admin panel"}

另外，为了安全起见，请不要以明文存储密码。在保存用户信息之前，使用密码哈希（如passlib/bcrypt）进行存储。

站内所有文章版权归属LeftHeroAI导航站，无授权禁止任何主体转载、抄袭、复制内容，亦不得私自架设镜像站点。一经侵权，本站将通过法律途径追责。

相关文章

C#使用libcurl查找已协商的TLS版本
3
在ASP.NET Core中启用并使用双因素认证
1
在带有文件上传端点的ASP.NET Core 9中，Swagger UI无法加载，并显示“Internal Server Error /swagger/v1/swagger.json”
22
为什么GitLab CI/CD无法从另一个仓库访问LFS对象？
1
在DigitalOcean应用平台上将FastAPI挂载到Flask的问题
0
如何为Nuxt UI的 UChip设置自定义颜色？
1
汇集全球数千款 AI 工具，每日持续更新，配套 AI 开发学习资源，轻松用 AI 提升工作效率。快捷键 Ctrl+D/⌘+D 一键收藏本站。

超人AI导航关于我们免责声明

Copyright © 超人AI导航  辽ICP备2024029316号-4