如何在FastAPI中正确验证以HttpOnly Cookie发送的JWT?
问题核心是你使用的OAuth2PasswordBearer默认仅从Authorization: Bearer <token>请求头提取令牌,完全不会处理Cookie中的内容。要解决这个问题,你需要自定义一个依赖函数,让它从Cookie中读取令牌,替代默认的oauth2_scheme。
步骤1:自定义令牌提取依赖
修改你的OAuth2认证代码,替换默认的OAuth2PasswordBearer:
from fastapi import Depends, HTTPException, status, Request from jose import JWTError, jwt from sqlalchemy.orm import Session import database, schemas, models from settings import settings # 自定义令牌提取逻辑:优先从Cookie拿,可回退到Authorization头 def oauth2_scheme(request: Request): # 从Cookie中获取令牌(假设Cookie键名为access_token,存储纯token,不带Bearer前缀) token = request.cookies.get("access_token") if not token: # 可选:Cookie无令牌时,尝试从Authorization头读取(兼容原有方式) auth_header = request.headers.get("Authorization") if auth_header and auth_header.startswith("Bearer "): token = auth_header.split(" ")[1] else: raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="未提供有效令牌", headers={"WWW-Authenticate": "Bearer"} ) return token SECRET_KEY = settings.SECRET_KEY ALGORITHM = settings.ALGORITHM ACCESS_TOKEN_EXPIRE_MINUTES = settings.ACCESS_TOKEN_EXPIRE_MINUTES def verify_access_token(token: str, credentials_exception): try: payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM]) id: str = payload.get("user_id") if id is None: raise credentials_exception token_data = schemas.TokenData(id=id) except JWTError: raise credentials_exception return token_data def get_current_user(token: str = Depends(oauth2_scheme), db: Session = Depends(database.get_db)): credentials_exception = HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="无法验证凭据", headers={"WWW-Authenticate": "Bearer"} ) token_data = verify_access_token(token, credentials_exception) user = db.query(models.User).filter(models.User.id == token_data.id).first() return user
步骤2:修改登录接口,设置HttpOnly Cookie
确保登录成功后,把JWT令牌存入HttpOnly Cookie,而非返回给前端存储:
from fastapi import APIRouter, Depends, HTTPException, status, Response from fastapi.security import OAuth2PasswordRequestForm from jose import jwt from datetime import datetime, timedelta import database, schemas, models, utils from settings import settings router = APIRouter(tags=["认证"]) def create_access_token(data: dict): to_encode = data.copy() expire = datetime.utcnow() + timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES) to_encode.update({"exp": expire}) encoded_jwt = jwt.encode(to_encode, settings.SECRET_KEY, algorithm=settings.ALGORITHM) return encoded_jwt @router.post("/login") def login(response: Response, form_data: OAuth2PasswordRequestForm = Depends(), db: Session = Depends(database.get_db)): # 验证用户 user = db.query(models.User).filter(models.User.email == form_data.username).first() if not user or not utils.verify(form_data.password, user.password): raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail="用户名或密码错误" ) # 生成令牌 access_token = create_access_token(data={"user_id": user.id}) # 设置HttpOnly Cookie response.set_cookie( key="access_token", value=access_token, # 存储纯token,不带Bearer前缀 httponly=True, secure=True, # 生产环境必须开启,仅HTTPS下有效 samesite="Lax", # 防范CSRF攻击 max_age=settings.ACCESS_TOKEN_EXPIRE_MINUTES * 60 ) # 可选:返回用户基础信息,无需返回令牌 return {"user_id": user.id, "email": user.email}
步骤3:前端代码保持不变
你的前端getPosts函数无需修改,只要保留credentials: "include",浏览器会自动带上HttpOnly Cookie:
async function getPosts() { const url = "http://localhost:8000/posts"; const fetchConfig = { credentials: "include", }; const response = await fetch(url, fetchConfig); const posts = await response.json(); console.log(posts); }
关键注意事项
- Cookie属性:生产环境必须开启
secure=True,确保Cookie仅在HTTPS连接下传递;samesite="Lax"或"Strict"可有效防范CSRF攻击。 - 令牌格式:示例中Cookie存储纯JWT令牌,若你的Cookie存的是
Bearer <token>,需在自定义oauth2_scheme中添加token = token.replace("Bearer ", "")处理前缀。 - 兼容性:自定义
oauth2_scheme保留了从Authorization头读取令牌的逻辑,原有通过请求头传递令牌的方式依然可用。
内容的提问来源于stack exchange,提问作者snkennedy21
相关产品推荐
相关产品推荐

