第 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。
uv add pyjwt "pwdlib[argon2]"14.3 OAuth2 密码模式全流程
整体流程四步:
- 客户端向
/token提交表单格式的用户名+密码; - 服务端校验通过后签发 JWT(带过期时间);
- 客户端后续请求在
Authorization: Bearer <token>头里携带它; - 依赖从请求中提取并解码 token,取出当前用户。
先搭骨架(依赖注入部分):
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_userOAuth2PasswordBearer(tokenUrl="token") 本身不校验任何东西——它只是一个"提取器"依赖:从 Authorization: Bearer xxx 头中取出 token 字符串;没有该头则抛 401。真正的校验发生在你的 get_current_user 里。
14.4 密码哈希与 /token 端点
pwdlib 用法与 Argon2 哈希:
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 登录端点:
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)提交username和password,不是 JSON——这是 OAuth2 规范决定的,所以需要uv add python-multipart;- JWT 是签名而非加密:任何人都能解码看到内容,因此绝不能往里放密码等敏感信息;
exp声明由 PyJWT 自动校验,过期即抛异常,被我们统一转成 401。sub必须是字符串,且在整个应用内唯一;若同一 id 会出现在多种实体上,官方建议加前缀如"username:johndoe"。
14.5 防时序攻击的小心思
官方示例里有个精妙细节:用户名不存在时,仍然用一个假哈希去跑一次 verify_password:
# 用户不存在也执行一次校验,保证两种情况响应耗时接近,
# 攻击者无法通过响应时间差枚举有效用户名
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接收表单格式凭据,签发含exp和sub的 JWT;- JWT 只签名不加密,别放敏感数据;密钥用环境变量管理。
🧪 随堂测验
点击你认为正确的选项。答错时会展示正确答案与原因解析。
1. 按当前 FastAPI 官方教程,密码哈希和 JWT 分别推荐用什么库?
2. OAuth2PasswordBearer(tokenUrl="token") 这个实例作为依赖被调用时做了什么?
3. 关于 JWT,下列说法正确的是?
4. /token 端点用 OAuth2PasswordRequestForm 接收参数时,客户端该如何提交?
🛠️ 动手实践
- 把本章代码补全成可运行版本(内存假数据库 + 两个测试用户),在 Swagger UI 里点 Authorize 完成登录并访问
/users/me/,观察 401/200 两种响应头差异。 - 给 token 增加
scopes字段并在get_current_user中打印出来,为下一章做准备;故意用过期 token 请求,确认 PyJWT 抛出的具体异常类型。 - 用
openssl rand -hex 32生成密钥并改造成从环境变量读取(结合 pydantic-settings),写一个单元测试断言"篡改签名后的 token 返回 401"。
单用户登录已经不够用了?请进入下一章:Scopes 细粒度权限与 API Key。