FastAPI OAuth2 + JWT 认证详解

认证是绝大多数 Web API 的核心需求。FastAPI 内置了对 OAuth2 的支持，结合 JWT（JSON Web Token）可以实现一套简洁、安全的无状态认证方案。本文将从头搭建一个完整的 JWT 认证流程，包括密码哈希、令牌签发、令牌验证和获取当前用户。

为什么选择 JWT

JWT 是一种紧凑、自包含的令牌格式，由三部分组成：Header（头部）、Payload（负载）、Signature（签名），用 . 分隔。典型结构如下：

1	eyJhbGciOiJIUzI1NiJ9.eyJ1c2VyX2lkIjoxfQ.xxx-signature

核心优势：

无状态：服务端不需要存储会话信息，令牌本身携带用户标识
可扩展：Payload 中可以放入自定义字段（如角色、权限）
防篡改：签名保证了令牌内容不可被修改

FastAPI 通过 fastapi.security 模块提供了 OAuth2 的完整支持，配合 python-jose（JWT 库）和 passlib（密码哈希库）即可快速实现。

安装依赖

1	pip install python-jose[cryptography] passlib[bcrypt] python-multipart

python-jose：生成和验证 JWT
passlib + bcrypt：密码哈希，不存储明文密码
python-multipart：解析 OAuth2 登录表单数据

核心配置与工具函数

首先创建配置常量和工具函数，集中管理 JWT 相关参数：

# auth.py
from datetime import datetime, timedelta
from typing import Optional

from jose import JWTError, jwt
from passlib.context import CryptContext
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer

# ---------- 配置 ----------
SECRET_KEY = "your-secret-key-keep-it-safe"  # 生产环境使用环境变量
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30  # 令牌有效期

# ---------- 密码哈希 ----------
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

def hash_password(password: str) -> str:
    """将明文密码哈希化"""
    return pwd_context.hash(password)

def verify_password(plain_password: str, hashed_password: str) -> bool:
    """验证明文密码与哈希值是否匹配"""
    return pwd_context.verify(plain_password, hashed_password)

# ---------- JWT 令牌 ----------
def create_access_token(data: dict, expires_delta: Optional[timedelta] = None) -> str:
    """生成 JWT 访问令牌"""
    to_encode = data.copy()
    expire = datetime.utcnow() + (expires_delta or timedelta(minutes=15))
    to_encode.update({"exp": expire})
    return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)

def decode_access_token(token: str) -> Optional[dict]:
    """解析 JWT 令牌，返回 payload；无效则返回 None"""
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        return payload
    except JWTError:
        return None

关键点解释：

CryptContext 自动处理加盐和哈希，verify() 方法不需要你手动提取盐值
JWT 的 exp 字段是标准过期时间声明，jose 库在 decode 时会自动校验
ACCESS_TOKEN_EXPIRE_MINUTES 一般设为 15–30 分钟，配合 refresh token 使用可获得更好的安全体验

OAuth2PasswordBearer —— 自动提取 Token

OAuth2PasswordBearer 是 FastAPI 提供的安全工具，它会从请求头 Authorization: Bearer <token> 中自动提取令牌：

1	oauth2_scheme = OAuth2PasswordBearer(tokenUrl="login")

tokenUrl 指向你的登录接口路径（相对路径），FastAPI 会自动在 Swagger 文档中添加认证按钮
当请求没有携带 token 时，自动返回 401 状态码

伪造用户数据库与模型

为了演示完整流程，我们用一个内存字典模拟数据库，并定义 Pydantic 模型：

# models.py
from pydantic import BaseModel

class User(BaseModel):
    username: str
    email: Optional[str] = None
    disabled: bool = False

class UserInDB(User):
    hashed_password: str

class Token(BaseModel):
    access_token: str
    token_type: str

class LoginForm(BaseModel):
    username: str
    password: str

# 模拟数据库（实际应用中替换为 ORM 查询）
fake_users_db = {
    "alice": {
        "username": "alice",
        "email": "alice@example.com",
        "hashed_password": "",  # 启动时或注册时填充
        "disabled": False,
    }
}

登录接口：验证凭据并签发 Token

# main.py
from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordRequestForm

app = FastAPI()

@app.post("/login", response_model=Token)
async def login(form_data: OAuth2PasswordRequestForm = Depends()):
    """
    OAuth2 标准登录接口。
    使用表单格式：username + password（不是 JSON）
    """
    # 1. 查用户
    user_dict = fake_users_db.get(form_data.username)
    if not user_dict:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="用户名或密码错误",
        )

    # 2. 验密码
    user = UserInDB(**user_dict)
    if not verify_password(form_data.password, user.hashed_password):
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="用户名或密码错误",
        )

    # 3. 签发 JWT，payload 中放 user_id（或其他唯一标识）
    access_token = create_access_token(
        data={"sub": user.username},
        expires_delta=timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES),
    )

    return {"access_token": access_token, "token_type": "bearer"}

注意三个要点：

OAuth2PasswordRequestForm 是 FastAPI 内置的表单解析器，自动从 application/x-www-form-urlencoded 中提取 username 和 password（还可选 scope、grant_type）
密码验证失败返回统一的错误信息，不要区分”用户名不存在”和”密码错误”，防止用户枚举攻击
JWT 的 sub（subject）字段按惯例存放用户唯一标识（如用户名、用户 ID）

获取当前用户：核心依赖注入

这是整个认证体系中最关键的依赖函数，后续所有需要登录的接口都依赖它：

from fastapi import Depends, HTTPException, status

async def get_current_user(token: str = Depends(oauth2_scheme)):
    """从 JWT 中解析当前用户 —— 核心依赖注入函数"""
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="无法验证凭据",
        headers={"WWW-Authenticate": "Bearer"},
    )

    # 1. 解码 token
    payload = decode_access_token(token)
    if payload is None:
        raise credentials_exception

    # 2. 提取用户名
    username: str = payload.get("sub")
    if username is None:
        raise credentials_exception

    # 3. 从数据库获取用户
    user_dict = fake_users_db.get(username)
    if user_dict is None:
        raise credentials_exception

    return UserInDB(**user_dict)

这个函数被设计为 FastAPI 依赖，使用时只需在路由函数参数中声明 Depends(get_current_user)：

@app.get("/users/me", response_model=User)
async def read_current_user(current_user: User = Depends(get_current_user)):
    """获取当前登录用户的信息"""
    return current_user

调用 /users/me 时，get_current_user 会自动执行：提取 token → 解码 → 查库 → 返回用户对象。整个链路透明且可测试。

更细粒度的认证：检查用户状态

有时不仅需要认证，还需检查用户是否被禁用：

async def get_current_active_user(
    current_user: User = Depends(get_current_user)
):
    """获取当前用户，且要求未被禁用"""
    if current_user.disabled:
        raise HTTPException(status_code=400, detail="该用户已被禁用")
    return current_user

@app.get("/users/me/items")
async def read_own_items(
    current_user: User = Depends(get_current_active_user)
):
    return {"owner": current_user.username, "items": ["item1", "item2"]}

这种依赖链式调用的设计是 FastAPI 依赖注入的强大之处：get_current_active_user 复用 get_current_user，逻辑清晰，无重复代码。

完整集成：初始化用户密码

将上述代码整合后，需要在启动时对测试用户做密码哈希：

@app.on_event("startup")
async def startup():
    """服务启动时初始化测试用户的密码哈希"""
    from auth import hash_password
    fake_users_db["alice"]["hashed_password"] = hash_password("secret123")

Swagger 文档中的认证体验

完成上述代码后，打开 http://localhost:8000/docs 会看到：

右上角多了一个 Authorize 🔒 按钮
点击后在弹出的表单中输入 username: alice、password: secret123
授权成功后，所有需要认证的接口自动携带 Authorization: Bearer <token> 请求头
右上角的 🔒 图标会变为已锁定状态

这就是 OAuth2PasswordBearer(tokenUrl="login") 带来的文档集成效果 —— 不需要额外配置。

安全性注意事项

在实际项目中，还需要关注以下要点：

# ✅ 生产环境的 SECRET_KEY 从环境变量读取
import os
SECRET_KEY = os.getenv("JWT_SECRET_KEY", "fallback-for-dev-only")

# ✅ 敏感接口强制 HTTPS（配合反向代理处理）
# ✅ 密码存储必须使用 bcrypt（已完成）

# ✅ 可以为 JWT 增加 token_type 校验，防止混淆攻击
def create_access_token(data: dict, ...):
    to_encode = data.copy()
    to_encode.update({"type": "access"})
    ...

# ✅ refresh token 建议使用更长有效期 + 独立的 token 类型
def create_refresh_token(data: dict):
    to_encode = data.copy()
    to_encode.update({"type": "refresh", "exp": ...})
    return jwt.encode(to_encode, ...)

完整流程总结

客户端                           服务端
  │                                │
  │  POST /login                   │
  │  username=alice&password=...   │
  │ ─────────────────────────────> │
  │                                │  校验密码
  │                                │  签发 JWT (exp=30min)
  │  {"access_token":"eyJ...",     │
  │   "token_type":"bearer"}       │
  │ <───────────────────────────── │
  │                                │
  │  GET /users/me                 │
  │  Authorization: Bearer eyJ...  │
  │ ─────────────────────────────> │
  │                                │  decode JWT
  │                                │  查用户 → 返回
  │  {"username":"alice", ...}     │
  │ <───────────────────────────── │

小结

组件	作用	关键类/函数
密码哈希	安全存储密码	`passlib.CryptContext`
JWT 签发	生成访问令牌	`jose.jwt.encode()`
JWT 校验	解析并验证令牌	`jose.jwt.decode()`
Token 提取	从请求头拿 token	`OAuth2PasswordBearer`
表单解析	解析登录表单	`OAuth2PasswordRequestForm`
依赖注入	获取当前用户	`Depends(get_current_user)`

FastAPI 的 OAuth2 + JWT 方案本质上就是这六个组件的配合。一旦搭建完成，后续只需在需要认证的接口参数中加上 Depends(get_current_user) 即可 —— 简洁、安全、可维护。

文章目录