在构建 FastAPI 应用时，我们常常面临一个核心挑战：如何设计安全可靠的用户认证系统？今天我们就来聊聊基于 OAuth2 密码模式、JWT 令牌和密码哈希的完整解决方案，这套方案不仅能满足生产环境的安全需求，还能与其他框架（如 Django）实现用户数据共享。话不多说，直接上干货！

一、JWT 基础：安全认证的核心载体

JWT（JSON Web Token）是我们这套方案的核心组件，它将用户信息编码为紧凑的字符串，格式通常像这样：eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiIxMjM0NTY3ODkwIiwibmFtZSI6IkpvaG4gRG9lIiwiaWF0IjoxNTE2MjM5MDIyfQ.SflKxwRJSMeKKF2QT4fwpMeJf36POk6yJV_adQssw5c。

关键特性解析：

• 非加密但签名：JWT 内容可解码查看，但通过密钥签名保证完整性。如果有人篡改令牌（比如修改过期时间），签名验证会失败
• 过期机制：我们可以设置令牌有效期（如 30 分钟），过期后自动失效，强制用户重新认证
• 无状态认证：服务器无需存储令牌状态，每次请求携带令牌即可完成身份验证

实战建议：想直观了解 JWT 结构？推荐使用https://jwt.io在线工具，输入令牌即可解析 payload、签名等细节。

二、环境准备：安装必要的依赖库

1. 安装 PyJWT 处理 JWT 令牌

bash

# 创建并激活虚拟环境后执行
pip install pyjwt
# 如需使用RSA等加密算法，安装扩展依赖
pip install pyjwt[crypto]

2. 安装 PassLib 实现密码哈希

bash

# 推荐使用Bcrypt算法，安全性更高
pip install "passlib[bcrypt]"

PassLib 的隐藏技能：它支持与 Django、Flask 等框架的密码哈希兼容。比如你可以用 FastAPI 读取 Django 生成的密码哈希，实现平滑迁移，这在多系统集成场景中非常实用！

三、核心实现：从密码哈希到 JWT 令牌生成

1. 数据模型设计

我们需要定义以下 Pydantic 模型：

python

from pydantic import BaseModel

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

class TokenData(BaseModel):
    username: str | None = None

class User(BaseModel):
    username: str
    email: str | None = None
    full_name: str | None = None
    disabled: bool | None = None

class UserInDB(User):
    hashed_password: str

2. 密码哈希处理

python

from passlib.context import CryptContext

# 初始化密码哈希上下文，使用bcrypt算法
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

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

def get_password_hash(password):
    """生成密码的哈希值"""
    return pwd_context.hash(password)

3. JWT 令牌生成与验证

python

from datetime import datetime, timedelta, timezone
import jwt
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer

# 生成安全密钥（生产环境务必使用独立生成的密钥）
# 生成命令：openssl rand -hex 32
SECRET_KEY = "09d25e094faa6ca2556c818166b7a9563b93f7099f6f0f4caa6cf63b88e8d3e7"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

def create_access_token(data: dict, expires_delta: timedelta | None = None):
    """创建JWT访问令牌"""
    to_encode = data.copy()
    expire = datetime.now(timezone.utc) + (expires_delta or timedelta(minutes=15))
    to_encode.update({"exp": expire})
    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
    return encoded_jwt

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

async def get_current_user(token: str = Depends(oauth2_scheme)):
    """获取当前用户信息"""
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="无法验证凭据",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        # 解码JWT令牌
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username = payload.get("sub")
        if username is None:
            raise credentials_exception
    except Exception:
        raise credentials_exception
    # 从数据库获取用户信息（这里使用模拟数据）
    user = get_user(fake_users_db, username=username)
    if user is None:
        raise credentials_exception
    return user

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

4. 认证流程整合

python

fake_users_db = {
    "johndoe": {
        "username": "johndoe",
        "full_name": "John Doe",
        "email": "johndoe@example.com",
        "hashed_password": "$2b$12$EixZaYVK1fsbw1ZfbX3OXePaWxn96p36WQoeG6Lruj3vjPGga31lW",
        "disabled": False,
    }
}

def get_user(db, username: str):
    """从数据库获取用户"""
    if username in db:
        user_dict = db[username]
        return UserInDB(**user_dict)

def authenticate_user(db, username: str, password: str):
    """验证用户凭据"""
    user = get_user(db, username)
    if not user:
        return False
    if not verify_password(password, user.hashed_password):
        return False
    return user

@app.post("/token")
async def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends()):
    """用户登录获取访问令牌"""
    user = authenticate_user(fake_users_db, form_data.username, form_data.password)
    if not user:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            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 Token(access_token=access_token, token_type="bearer")

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

@app.get("/users/me/items/")
async def read_own_items(current_user: User = Depends(get_current_active_user)):
    """获取当前用户的项目"""
    return [{"item_id": "Foo", "owner": current_user.username}]

四、实战测试与安全细节

1. 启动服务并测试

运行 FastAPI 服务后，访问http://127.0.0.1:8000/docs打开交互式文档，使用以下凭据测试：

• 用户名：johndoe
• 密码：secret（注意：代码中只存储了哈希值，明文密码仅在首次认证时使用）

2. 关键安全细节

• SECRET_KEY 的重要性：这个密钥是 JWT 签名的关键，必须严格保密！生产环境中应通过环境变量或安全配置加载，绝不能硬编码在代码中
• JWT 的 subject 字段：我们使用sub字段存储用户标识，这是 JWT 规范推荐的做法。在复杂场景中，可以通过前缀避免 ID 冲突（如username:johndoe）
• 令牌传输安全：务必在 HTTPS 环境下使用 JWT，防止令牌被中间人截取
• 密码哈希强度：Bcrypt 算法中的12表示工作因子，数值越高安全性越高，但会增加计算开销，可根据服务器性能调整

五、进阶拓展：基于 Scopes 的权限控制

OAuth2 的 Scopes 机制可以让我们为令牌添加细粒度的权限控制。比如：

• scope:read：仅允许读取数据
• scope:write：允许写入数据
• scope:admin：管理员权限

在 FastAPI 中实现 Scopes 控制并不复杂，后续我们可以深入探讨如何将 Scopes 与 JWT 结合，实现更灵活的权限管理，这也是各大平台（如 Google、GitHub）常用的认证模式。

总结：为什么选择这套方案？

• 标准化：遵循 OAuth2 和 JWT 国际标准，兼容性强
• 安全性：密码哈希 + JWT 签名双重保障，数据库泄露也不会导致明文密码泄露
• 灵活性：不绑定特定数据库或框架，可与 Django 等系统共享用户数据
• 无状态：服务器无需维护会话状态，便于横向扩展

如果本文对你有帮助，别忘了点赞收藏，关注我，一起探索更高效的开发方式～