FastAPI中OAuth2的refresh_token grant_type触发422错误排查

问题描述 #

我是OAuth2和FastAPI的新手，之前使用都正常，但现在在识别refresh_token类型的grant_type时遇到了问题。通过表单能获取到grant_type为password或refresh_token，但传入grant_type=refresh_token时，代码根本进不了对应的条件分支，直接返回422 Unprocessable Entity错误，怀疑是OAuth2PasswordRequestForm不支持该类型导致的。

原代码：

from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm

@app.post("/token", response_model=Token) 
async def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends()):
    user = authenticate_user(fake_users_db, form_data.username, form_data.password)
    
    #for login grant_type = password
    #for refreshToken grant_type = refresh_token
    grant_type_str = str(form_data.grant_type)
    print('167', str(form_data.grant_type))
    try:
        user.username
    except:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail=user['detail'],
            headers={"WWW-Authenticate": "Bearer"},
        )
    try:
        form_data.grant_type
    except:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="must specifiy grant type",
            headers={"WWW-Authenticate": "Bearer"},
        )
    if grant_type_str.startswith('password'):
        access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
        access_token = create_access_token(data={"sub": user.username}, expires_delta=access_token_expires)
        return {"access_token": access_token, "expires_in": ACCESS_TOKEN_EXPIRE_MINUTES,
            "token_type": "bearer", "scope": "read write groups", "grant_type": "password"}
    elif grant_type_str.startswith('refresh_token'):
        access_token_expires_refresh = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES_REFRESH)
        access_token = create_refresh_token(
            data={"sub": user.username}, expires_delta=access_token_expires_refresh
        )
        return {"access_token": access_token, "expires_in": ACCESS_TOKEN_EXPIRE_MINUTES_REFRESH,
            "token_type": "bearer", "scope": "read write groups", "grant_type": "refresh_token"}

问题原因 #

没错，OAuth2PasswordRequestForm是FastAPI专门为**密码模式（password grant）**设计的表单类，它默认只接受grant_type=password，并且强制要求传入username和password参数。当你传入grant_type=refresh_token时，它会因为参数不符合预期（比如缺少username/password，或者grant_type不在允许范围内）直接抛出422错误，根本到不了你自己的逻辑分支。

解决办法 #

自定义一个支持多种grant_type的表单类，替代默认的OAuth2PasswordRequestForm：

步骤1：自定义表单类 #

from fastapi import Form, Depends
from typing import Optional

class OAuth2TokenRequestForm:
    def __init__(
        self,
        grant_type: str = Form(..., regex="^(password|refresh_token)$"),
        username: Optional[str] = Form(None),
        password: Optional[str] = Form(None),
        refresh_token: Optional[str] = Form(None),
        scope: str = Form("")
    ):
        self.grant_type = grant_type
        self.username = username
        self.password = password
        self.refresh_token = refresh_token
        self.scope = scope.split()

这个类通过正则限制grant_type只能是password或refresh_token，同时根据不同类型开放对应的可选参数：

• grant_type=password时，需要传入username和password
• grant_type=refresh_token时，需要传入refresh_token

步骤2：修改接口逻辑 #

替换原表单类，并针对不同grant_type做针对性处理：

@app.post("/token", response_model=Token) 
async def login_for_access_token(form_data: OAuth2TokenRequestForm = Depends()):
    # 处理密码模式
    if form_data.grant_type == "password":
        if not form_data.username or not form_data.password:
            raise HTTPException(
                status_code=status.HTTP_400_BAD_REQUEST,
                detail="username和password为密码模式必填参数",
                headers={"WWW-Authenticate": "Bearer"},
            )
        user = authenticate_user(fake_users_db, form_data.username, form_data.password)
        try:
            user.username
        except:
            raise HTTPException(
                status_code=status.HTTP_401_UNAUTHORIZED,
                detail=user['detail'],
                headers={"WWW-Authenticate": "Bearer"},
            )
        access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
        access_token = create_access_token(data={"sub": user.username}, expires_delta=access_token_expires)
        return {"access_token": access_token, "expires_in": ACCESS_TOKEN_EXPIRE_MINUTES,
                "token_type": "bearer", "scope": "read write groups", "grant_type": "password"}
    
    # 处理刷新令牌模式
    elif form_data.grant_type == "refresh_token":
        if not form_data.refresh_token:
            raise HTTPException(
                status_code=status.HTTP_400_BAD_REQUEST,
                detail="refresh_token为刷新令牌模式必填参数",
                headers={"WWW-Authenticate": "Bearer"},
            )
        # 验证refresh_token有效性并解析用户信息
        payload = decode_refresh_token(form_data.refresh_token)
        username = payload.get("sub")
        if not username:
            raise HTTPException(
                status_code=status.HTTP_401_UNAUTHORIZED,
                detail="无效的refresh token",
                headers={"WWW-Authenticate": "Bearer"},
            )
        user = get_user(fake_users_db, username=username)
        if not user:
            raise HTTPException(
                status_code=status.HTTP_401_UNAUTHORIZED,
                detail="用户不存在",
                headers={"WWW-Authenticate": "Bearer"},
            )
        access_token_expires_refresh = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES_REFRESH)
        access_token = create_refresh_token(
            data={"sub": user.username}, expires_delta=access_token_expires_refresh
        )
        return {"access_token": access_token, "expires_in": ACCESS_TOKEN_EXPIRE_MINUTES_REFRESH,
                "token_type": "bearer", "scope": "read write groups", "grant_type": "refresh_token"}
    
    # 处理不支持的grant_type
    else:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="不支持的grant_type",
            headers={"WWW-Authenticate": "Bearer"},
        )

关键说明 #

1. 自定义表单通过regex限制grant_type可选值，提前拦截无效输入
2. 不同grant_type分支需做对应参数校验，避免非法请求
3. 刷新令牌时需先验证refresh_token有效性并解析用户信息，无需再次调用用户名密码验证逻辑

内容的提问来源于stack exchange，提问作者jKraut