Skip to content

第 14 章 · 安全基础:OAuth2 与 JWT

本章目标:分清认证与授权,跑通"登录换 token → 带 token 访问 → 校验取用户"的完整 OAuth2 密码模式,使用官方当前推荐的 PyJWT 与 pwdlib。

14.1 认证 vs 授权

  • 认证(Authentication):你是谁?——验证用户名/密码、token 等;
  • 授权(Authorization):你能干什么?——验证你是否有权限访问某个资源。

本章实现认证 + 最基础的授权(未登录返回 401),下一章的 scopes 再做细粒度授权。FastAPI 在 fastapi.security 里提供了若干安全工具类,它们不仅帮你解析凭据,还会把对应的安全方案写进 OpenAPI 文档(Swagger UI 自动出现 "Authorize" 按钮)。

14.2 官方推荐的库

版本演进

早期官方教程使用 passlib[bcrypt] 做密码哈希、python-jose 处理 JWT。当前官方教程已改为 pwdlib(推荐 Argon2 算法)和 PyJWT。python-jose 已停止维护;passlib 也多年未更新且与 bcrypt 4.x 有兼容问题。新项目请直接用 pwdlib + PyJWT。

bash
uv add pyjwt "pwdlib[argon2]"

14.3 OAuth2 密码模式全流程

整体流程四步:

  1. 客户端向 /token 提交表单格式的用户名+密码;
  2. 服务端校验通过后签发 JWT(带过期时间);
  3. 客户端后续请求在 Authorization: Bearer <token> 头里携带它;
  4. 依赖从请求中提取并解码 token,取出当前用户。

先搭骨架(依赖注入部分):

python
from datetime import datetime, timedelta, timezone
from typing import Annotated

import jwt
from fastapi import Depends, FastAPI, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from pydantic import BaseModel

app = FastAPI()

SECRET_KEY = "09d25e094faa..."   # 生产用 openssl rand -hex 32 生成,放环境变量!
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

# tokenUrl 指向获取 token 的端点,供 Swagger UI 的 Authorize 表单使用
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")


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


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


class UserInDB(User):
    hashed_password: str


async def get_current_user(token: Annotated[str, Depends(oauth2_scheme)]) -> User:
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},  # 规范要求 401 带此头
    )
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username: str = payload.get("sub")
        if username is None:
            raise credentials_exception
    except jwt.InvalidTokenError:   # 覆盖过期、签名错误等所有无效情况
        raise credentials_exception
    user = get_user_from_db(username)
    if user is None:
        raise credentials_exception
    return user


async def get_current_active_user(
    current_user: Annotated[User, Depends(get_current_user)],
) -> User:
    if current_user.disabled:
        raise HTTPException(status_code=400, detail="Inactive user")
    return current_user


@app.get("/users/me/")
async def read_me(current_user: Annotated[User, Depends(get_current_active_user)]):
    return current_user

OAuth2PasswordBearer(tokenUrl="token") 本身不校验任何东西——它只是一个"提取器"依赖:从 Authorization: Bearer xxx 头中取出 token 字符串;没有该头则抛 401。真正的校验发生在你的 get_current_user 里。

14.4 密码哈希与 /token 端点

pwdlib 用法与 Argon2 哈希:

python
from pwdlib import PasswordHash

password_hash = PasswordHash.recommended()  # 当前推荐配置即 Argon2

def hash_password(password: str) -> str:
    return password_hash.hash(password)

def verify_password(plain: str, hashed: str) -> bool:
    return password_hash.verify(plain, hashed)

# 数据库里存的是哈希,例如:
# $argon2id$v=19$m=65536,t=3,p=4$... (永远不要存明文)

签发 token 并完成 /token 登录端点:

python
def create_access_token(data: dict, expires_delta: timedelta | None = None) -> str:
    to_encode = data.copy()
    expire = datetime.now(timezone.utc) + (
        expires_delta or timedelta(minutes=15)
    )
    to_encode.update({"exp": expire})
    return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)


@app.post("/token")
async def login_for_access_token(
    form_data: Annotated[OAuth2PasswordRequestForm, Depends()],
) -> Token:
    user = authenticate_user(form_data.username, form_data.password)
    if not user:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Incorrect username or password",
            headers={"WWW-Authenticate": "Bearer"},
        )
    access_token = create_access_token(
        data={"sub": user.username},   # sub = subject,JWT 标准的"主体"字段
        expires_delta=timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES),
    )
    return Token(access_token=access_token, token_type="bearer")

三个必须理解的细节:

  • OAuth2PasswordRequestForm 要求客户端以 表单编码application/x-www-form-urlencoded)提交 usernamepassword,不是 JSON——这是 OAuth2 规范决定的,所以需要 uv add python-multipart
  • JWT 是签名而非加密:任何人都能解码看到内容,因此绝不能往里放密码等敏感信息;
  • exp 声明由 PyJWT 自动校验,过期即抛异常,被我们统一转成 401。
  • sub 必须是字符串,且在整个应用内唯一;若同一 id 会出现在多种实体上,官方建议加前缀如 "username:johndoe"

14.5 防时序攻击的小心思

官方示例里有个精妙细节:用户名不存在时,仍然用一个假哈希去跑一次 verify_password

python
# 用户不存在也执行一次校验,保证两种情况响应耗时接近,
# 攻击者无法通过响应时间差枚举有效用户名
def authenticate_user(username: str, password: str) -> UserInDB | None:
    user = get_user_from_db(username)
    if not user:
        verify_password(password, DUMMY_HASH)
        return None
    if not verify_password(password, user.hashed_password):
        return None
    return user

这类"恒定时间"防御在认证代码里是值得养成的习惯。

14.6 本章小结

  • 认证回答"你是谁",授权回答"你能干什么";401 表示没认证成功,403 表示认证了但权限不足;
  • 当前官方栈:PyJWT(签发/校验)+ pwdlib[argon2](哈希),替代旧的 python-jose/passlib;
  • OAuth2PasswordBearer 只负责提取 Bearer token 并写入 OpenAPI 文档;
  • /token 接收表单格式凭据,签发含 expsub 的 JWT;
  • JWT 只签名不加密,别放敏感数据;密钥用环境变量管理。

🧪 随堂测验

点击你认为正确的选项。答错时会展示正确答案与原因解析。

1. 按当前 FastAPI 官方教程,密码哈希和 JWT 分别推荐用什么库?

2. OAuth2PasswordBearer(tokenUrl="token") 这个实例作为依赖被调用时做了什么?

3. 关于 JWT,下列说法正确的是?

4. /token 端点用 OAuth2PasswordRequestForm 接收参数时,客户端该如何提交?

🛠️ 动手实践

  1. 把本章代码补全成可运行版本(内存假数据库 + 两个测试用户),在 Swagger UI 里点 Authorize 完成登录并访问 /users/me/,观察 401/200 两种响应头差异。
  2. 给 token 增加 scopes 字段并在 get_current_user 中打印出来,为下一章做准备;故意用过期 token 请求,确认 PyJWT 抛出的具体异常类型。
  3. openssl rand -hex 32 生成密钥并改造成从环境变量读取(结合 pydantic-settings),写一个单元测试断言"篡改签名后的 token 返回 401"。

单用户登录已经不够用了?请进入下一章:Scopes 细粒度权限与 API Key