FastApi中无需登录密码表单的OAuth2授权实现方法

实现核心逻辑 #

无需传统账号密码表单，采集用户指定特征数据（如手机号、身份证片段、第三方平台绑定标识等）后，转发至第三方服务校验匹配状态，匹配通过则为用户下发访问凭证，完成授权流程。

前置依赖安装 #

执行以下命令安装所需依赖包：

pip install fastapi uvicorn httpx python-jose[cryptography]

• httpx 用于异步请求第三方校验服务
• python-jose 用于生成和校验JWT访问凭证

完整实现代码 #

from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import OAuth2Bearer
from pydantic import BaseModel
import httpx
from jose import JWTError, jwt
from datetime import datetime, timedelta

# 配置项，生产环境请存入环境变量避免泄露
JWT_SECRET_KEY = "自定义的高强度密钥"
JWT_ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30
THIRD_PARTY_VALIDATE_URL = "第三方校验接口地址"

app = FastAPI()
oauth2_scheme = OAuth2Bearer(tokenUrl="auth")

# 定义需要采集的用户数据结构，可根据实际需求调整字段
class UserAuthFeature(BaseModel):
    phone: str
    id_card_last6: str
    real_name: str

# 生成JWT访问凭证
def create_access_token(data: dict, expires_delta: timedelta | None = None):
    to_encode = data.copy()
    expire = datetime.utcnow() + (expires_delta if expires_delta else timedelta(minutes=15))
    to_encode.update({"exp": expire})
    return jwt.encode(to_encode, JWT_SECRET_KEY, algorithm=JWT_ALGORITHM)

# 请求第三方服务校验用户数据是否匹配
async def third_party_validate(user_features: UserAuthFeature) -> bool:
    try:
        async with httpx.AsyncClient(timeout=10) as client:
            resp = await client.post(
                THIRD_PARTY_VALIDATE_URL,
                json=user_features.dict()
            )
            resp.raise_for_status()
            # 按第三方接口返回的匹配字段调整判断逻辑
            return resp.json().get("matched", False)
    except:
        raise HTTPException(
            status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
            detail="第三方校验服务不可用，请稍后重试"
        )

# 凭证校验依赖，需要授权的接口直接引入即可
async def get_current_user(token: str = Depends(oauth2_scheme)):
    auth_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="无效访问凭证",
        headers={"WWW-Authenticate": "Bearer"}
    )
    try:
        payload = jwt.decode(token, JWT_SECRET_KEY, algorithms=[JWT_ALGORITHM])
        user_id = payload.get("user_id")
        if not user_id:
            raise auth_exception
        return user_id
    except JWTError:
        raise auth_exception

# 授权接口，无需输入密码
@app.post("/auth", summary="无密码授权接口")
async def auth(user_features: UserAuthFeature):
    is_matched = await third_party_validate(user_features)
    if not is_matched:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="用户数据匹配失败，无法完成授权"
        )
    # 校验通过，下发有效期30分钟的访问凭证
    access_token = create_access_token(
        data={"user_id": user_features.phone},
        expires_delta=timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    )
    return {"access_token": access_token, "token_type": "bearer"}

# 测试受保护的业务接口
@app.get("/business/data", summary="受保护的业务接口")
async def get_business_data(current_user: str = Depends(get_current_user)):
    return {"current_user": current_user, "data": "仅授权用户可访问的业务数据"}

运行方式 #

执行以下命令启动服务：

uvicorn main:app --reload

启动后访问http://127.0.0.1:8000/docs即可打开接口调试页面，先调用/auth接口传入用户特征数据获取凭证，再用凭证访问受保护接口即可。

注意事项 #

• 传输用户敏感数据必须使用HTTPS协议，避免数据泄露
• 第三方请求需配置合理的超时和重试逻辑，避免第三方服务故障影响自身服务可用性
• 对用户提交的特征数据需先做格式校验，不符合规则的请求直接拦截，无需转发到第三方
• 访问凭证需设置合理有效期，禁止使用永久有效的凭证
• 如果业务场景不需要JWT，也可替换为Session存储凭证，核心校验逻辑保持一致即可

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