如何为FastAPI的后端连接场景创建OAuth唯一访问令牌?
实现方案
完全可以实现,不需要定义任何用户登录模型。你需要的是OAuth2的客户端凭证(Client Credentials)流,专门用于服务到服务的身份认证场景,全程不涉及普通用户的登录、授权逻辑。
核心思路
- 不需要用到OAuth2授权码、密码模式等涉及用户的复杂流程,仅保留两个核心逻辑:令牌签发、令牌校验
- 提前给每个调用方分配唯一的客户端ID(client_id)和客户端密钥(client_secret),调用方先用这对凭证兑换访问令牌,后续业务请求携带令牌即可完成身份校验
- 令牌推荐用JWT格式,自带过期时间属性,服务端仅需验签即可完成校验,不需要额外依赖数据库存储令牌
具体实现步骤
1. 安装依赖
pip install fastapi uvicorn python-jose[cryptography] passlib[bcrypt]
2. 完整代码示例
from fastapi import FastAPI, Depends, HTTPException, status from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm from jose import JWTError, jwt from datetime import datetime, timedelta from typing import Optional # 配置项,生产环境请放到环境变量/配置中心 SECRET_KEY = "生产环境替换为随机生成的32位以上强密钥" ALGORITHM = "HS256" ACCESS_TOKEN_EXPIRE_MINUTES = 30 # 合法调用方凭证,生产环境建议存在数据库,client_secret需哈希存储不要明文 VALID_CLIENTS = { "service_order": "hash_after_bcrypt_123456", "service_pay": "hash_after_bcrypt_654321" } app = FastAPI() oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token") # 生成JWT访问令牌 def create_access_token(data: dict, expires_delta: Optional[timedelta] = 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, SECRET_KEY, algorithm=ALGORITHM) # 校验令牌合法性 async def verify_token(token: str = Depends(oauth2_scheme)): auth_exception = HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid credentials", headers={"WWW-Authenticate": "Bearer"}, ) try: payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM]) client_id = payload.get("sub") if not client_id or client_id not in VALID_CLIENTS: raise auth_exception except JWTError: raise auth_exception return client_id # 令牌兑换接口,调用方用client_id和client_secret换令牌 @app.post("/token") async def get_token(form_data: OAuth2PasswordRequestForm = Depends()): # 生产环境如果client_secret是哈希存储,此处需要做哈希比对 if form_data.username not in VALID_CLIENTS or form_data.password != VALID_CLIENTS[form_data.username]: raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Incorrect client id or secret", headers={"WWW-Authenticate": "Bearer"}, ) access_token = create_access_token( data={"sub": form_data.username}, expires_delta=timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES) ) return {"access_token": access_token, "token_type": "bearer"} # 受保护的业务接口示例,加依赖即可自动校验令牌 @app.get("/your-business-api") async def business_route(current_client: str = Depends(verify_token)): return {"code": 0, "msg": "请求成功", "caller": current_client}
3. 调用方使用流程
- 第一步请求
/token接口,form参数中username填分配的client_id,password填分配的client_secret,获取返回的访问令牌 - 第二步请求业务接口时,请求头携带
Authorization: Bearer <获取到的令牌>即可正常请求
生产环境优化建议
- 所有敏感配置不要硬编码,统一存入环境变量或配置中心
- client_secret不要明文存储,用bcrypt哈希后存储,校验时先对传入的密钥做哈希再比对
- 如果需要支持令牌提前撤销,可以增加Redis黑名单机制,校验令牌时先排查是否在黑名单内
内容的提问来源于stack exchange,提问作者cYbersYn
相关产品推荐
相关产品推荐

