第 12 章:认证与授权(OAuth2 + JWT)
第 11 章的接口谁都能调。本章补上访问控制:密码怎么存、JWT 怎么签发与校验、
get_current_user依赖怎么写、401 与 403 怎么分、角色权限怎么收口。
学习目标
- 分清认证(Authentication)与授权(Authorization),并知道各自该返回哪个状态码
- 能用
pwdlib+ Argon2 完成密码哈希与校验,理解为什么不能存明文 - 理解 JWT 的三段结构,说清「签名」与「加密」的区别
- 能独立实现
/token端点与get_current_user依赖 - 会用
Annotated定义CurrentUser,并用工厂函数实现角色权限依赖 - 能把
SECRET_KEY等敏感配置从代码里赶到环境变量,并写出.env.example
12.1 认证与授权
两个词经常被混用,但它们在代码里对应完全不同的两件事:
| 认证(Authentication) | 授权(Authorization) | |
|---|---|---|
| 回答的问题 | 你是谁 | 你能做什么 |
| 输入 | 凭据(用户名 + 密码、token) | 已确认的身份 + 目标资源 |
| 失败状态码 | 401 Unauthorized | 403 Forbidden |
| 响应头要求 | 必须带 WWW-Authenticate: Bearer | 无特殊要求 |
| 典型实现 | /token 端点、get_current_user 依赖 | require_role("admin")、资源归属校验 |
12.2 依赖安装
uv add "pwdlib[argon2]" pyjwt python-multipart| 包 | 用途 | 不装会怎样 |
|---|---|---|
pwdlib[argon2] | 密码哈希(Argon2 算法) | 导入 PasswordHash 失败 |
pyjwt | JWT 的签发与校验 | 导入 jwt 失败 |
python-multipart | 解析 application/x-www-form-urlencoded | /token 报 python-multipart must be installed |
⚠️ FastAPI 官方文档已经从
passlib迁移到pwdlib。passlib长期缺乏维护,且从 Python 3.13 起无法正常工作。新项目直接用pwdlib,它支持 Argon2 与 Bcrypt,但不提供 MD5、SHA1 这类已经不该出现的算法。
12.3 密码哈希
密码必须单向哈希后存储:数据库泄露时攻击者拿到的是散列值,无法直接反推出密码,也无法拿去撞其他站点(很多人到处用同一个密码)。
# app/core/security.py
from pwdlib import PasswordHash
# 推荐配置:Argon2id,参数由 pwdlib 按当前硬件给出
password_hash = PasswordHash.recommended()
# 用户名不存在时用来比对的一次性哈希,见 12.5 的防时序攻击说明
DUMMY_HASH = password_hash.hash("dummy-password-for-timing")
def hash_password(plain: str) -> str:
"""哈希明文密码,返回可直接入库的字符串(含算法与盐)。"""
return password_hash.hash(plain)
def verify_password(plain: str, hashed: str) -> bool:
"""校验明文与散列是否匹配;散列格式损坏时返回 False,不抛异常。"""
try:
return password_hash.verify(plain, hashed)
except Exception:
return False❌ 不推荐:
import hashlib
def hash_password(plain: str) -> str:
# MD5/SHA1/SHA256 是快速哈希,GPU 每秒能试几十亿次
return hashlib.md5(plain.encode()).hexdigest()
def verify_password(plain: str, hashed: str) -> bool:
return hash_password(plain) == hashed # 比较也不是恒定时间✅ 推荐:
def hash_password(plain: str) -> str:
# 慢哈希 + 随机盐:同样的密码每次哈希结果都不同
return password_hash.hash(plain)
def verify_password(plain: str, hashed: str) -> bool:
return password_hash.verify(plain, hashed) # 内部使用恒定时间比较Argon2 散列的形态是这样的,算法、版本、内存/迭代参数与盐都编码在字符串里,所以同一个密码两次哈希的结果不同,校验时必须用 verify 而不是直接比较:
$argon2id$v=19$m=65536,t=3,p=4$wagCPXjifgvUFBzq4hqe3w$CYaIb8sB+wtD+Vu/P4uod1+Qof8h+1g7bbDlBID48Rc💡 参数调整与算法迁移时,用
PasswordHash((primary, [secondary...]))组合多个哈希器:新密码用主算法,旧散列也能继续校验,用户下次登录时再静默升级。
12.4 JWT 结构与签发
JWT(JSON Web Token)是一串用 . 分成三段的字符串:
eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiI0MiIsImV4cCI6MTc2NzIyNTYwMH0.7m0X2Fq...
└──────────── Header ────────────┘ └──────────── Payload ───────────┘ └─ Signature ─┘| 段 | 内容 | 是否可被客户端读取 |
|---|---|---|
| Header | {"alg": "HS256", "typ": "JWT"} | ✅ Base64URL,明文可读 |
| Payload | 声明(claims),如 sub / exp / iat / jti | ✅ Base64URL,明文可读 |
| Signature | 用密钥对前两段做的 HMAC | ✅ 可读,但无法伪造 |
JWT 是签名,不是加密。 任何人都能 Base64 解码出 payload,所以绝对不能把密码、身份证号、内部权限清单塞进去。签名保证的是「这段内容没被篡改、且确实是我签发的」。
常用声明(claims):
| 声明 | 全称 | 语义 | 注意事项 |
|---|---|---|---|
sub | subject | 主体标识 | 必须是字符串,见下方警告 |
exp | expiration time | 过期时间(UTC 时间戳) | 必须设置,否则 token 永不失效 |
iat | issued at | 签发时间 | 用于判断 token 年龄 |
jti | JWT ID | token 唯一标识 | 做黑名单/一次性 token 时需要 |
nbf | not before | 生效时间 | 提前签发场景用 |
# app/core/security.py(续)
from datetime import datetime, timedelta, timezone
import jwt
from app.core.config import settings
def create_access_token(subject: str | int, expires_delta: timedelta | None = None) -> str:
"""签发访问令牌。sub 会强制转成字符串——PyJWT 校验时对类型敏感。"""
now = datetime.now(timezone.utc)
expire = now + (expires_delta or timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES))
payload = {
"sub": str(subject), # 关键:字符串!传 int 会在解码时抛 InvalidSubjectError
"iat": now,
"exp": expire, # 必须设置
}
return jwt.encode(payload, settings.SECRET_KEY, algorithm=settings.ALGORITHM)关于时间的三条硬规则:
- 一律用
datetime.now(timezone.utc)。用datetime.now()(本地时间)或utcnow()(无时区 naive 对象)在容器时区变化或跨时区部署时会算出偏移的exp,表现为「刚登录就过期」或「token 活了 8 小时」。 exp不是可选项。省略它等于签发了一张永久通行证,一旦泄露无法挽回。sub必须是字符串。PyJWT 2.10 起对sub做类型校验,jwt.decode遇到非字符串的sub会抛InvalidSubjectError。数据库主键是int,所以encode时str(subject)、decode后int(...)转回来。
12.5 校验与异常分支
# app/core/security.py(续)
import jwt
from jwt.exceptions import ExpiredSignatureError, InvalidTokenError
def decode_access_token(token: str) -> dict:
"""解码并验签。失败一律抛 InvalidTokenError 的子类,由调用方翻译成 401。"""
try:
return jwt.decode(token, settings.SECRET_KEY, algorithms=[settings.ALGORITHM])
except ExpiredSignatureError as exc: # exp 已过
raise InvalidTokenError("token 已过期") from exc
except InvalidTokenError: # 签名不匹配、格式损坏、sub 类型错
raisealgorithms 参数必须显式传白名单列表。这不是可选的礼貌写法:如果把客户端 header 里的 alg 直接拿来用,攻击者可以改成 none 或换成非对称算法来绕过验签。
| 异常 | 触发条件 | 给客户端的状态码 |
|---|---|---|
ExpiredSignatureError | exp 早于当前时间 | 401 |
InvalidSignatureError | 签名与密钥不匹配(被篡改或换了密钥) | 401 |
DecodeError | 不是合法的 JWT 三段结构 | 401 |
InvalidSubjectError | sub 不是字符串 | 401(同时说明签发端有 bug) |
InvalidAlgorithmError | alg 不在允许列表内 | 401 |
ExpiredSignatureError 是 InvalidTokenError 的子类,所以 except InvalidTokenError 一个分支就能兜住全部情况;单列过期分支只是为了给用户更友好的提示文案。
12.6 /token 端点
# app/services/auth.py
from app.core.security import DUMMY_HASH, create_access_token, verify_password
from app.models.user import User
from app.repositories.user import UserRepository
class AuthService:
"""认证相关业务:校验凭据、签发 token。"""
def __init__(self, repo: UserRepository) -> None:
self.repo = repo
async def authenticate(self, email: str, password: str) -> User | None:
user = await self.repo.get_by_email(email)
if user is None:
# 用户不存在时也做一次哈希校验,抹平响应时间差,防止用户名枚举
verify_password(password, DUMMY_HASH)
return None
if not verify_password(password, user.hashed_password):
return None
return user
async def issue_token(self, user: User) -> str:
return create_access_token(subject=user.id)# app/api/v1/auth.py
from typing import Annotated
from fastapi import APIRouter, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordRequestForm
from pydantic import BaseModel
from app.api.deps import AuthServiceDep
from app.schemas.user import UserOut
router = APIRouter(prefix="/auth", tags=["auth"])
class Token(BaseModel):
access_token: str
token_type: str = "bearer"
@router.post("/token", response_model=Token, summary="登录换取访问令牌")
async def login(
form: Annotated[OAuth2PasswordRequestForm, Depends()],
service: AuthServiceDep,
) -> Token:
user = await service.authenticate(form.username, form.password)
if user is None:
# 认证失败:401 + WWW-Authenticate
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="邮箱或密码错误",
headers={"WWW-Authenticate": "Bearer"},
)
return Token(access_token=await service.issue_token(user))三个容易踩的细节:
OAuth2PasswordRequestForm的字段名固定是username/password,即使你的业务用邮箱登录也不能改名。想支持邮箱,就让用户在username里填邮箱,或者自己写一个Depends读取表单。- 请求体是
application/x-www-form-urlencoded,不是 JSON。前端用 JSON 提交会得到 422,Swagger 的 Authorize 按钮则是对的。 - 错误信息要统一为「邮箱或密码错误」。分别提示「用户不存在」和「密码错误」等于免费提供了用户名枚举接口。
12.7 get_current_user 依赖
# app/api/deps.py(衔接第 11 章)
from typing import Annotated
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from jwt.exceptions import InvalidTokenError
from sqlalchemy.ext.asyncio import AsyncSession
from app.core.security import decode_access_token
from app.models.user import User
from app.repositories.user import UserRepository
from app.services.auth import AuthService
# tokenUrl 是「相对根路径」的地址,Swagger 的 Authorize 按钮靠它拿 token
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/api/v1/auth/token")
async def get_db() -> AsyncIterator[AsyncSession]:
... # 第 10 章的会话依赖,此处省略
DbSession = Annotated[AsyncSession, Depends(get_db)]
async def get_current_user(
token: Annotated[str, Depends(oauth2_scheme)],
session: DbSession,
) -> User:
credentials_error = HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="凭证无效或已过期",
headers={"WWW-Authenticate": "Bearer"}, # 401 必须带这个头
)
try:
payload = decode_access_token(token)
user_id = int(payload["sub"])
except (InvalidTokenError, KeyError, ValueError) as exc:
raise credentials_error from exc
user = await UserRepository(session).get(user_id)
if user is None:
# token 有效但用户已被删除:仍然是认证失败
raise credentials_error
return user
CurrentUser = Annotated[User, Depends(get_current_user)]认证服务照第 11 章的方式组装(app/api/deps.py 续),/token 端点用的就是这个别名:
from app.services.auth import AuthService
def get_auth_service(repo: Annotated[UserRepository, Depends(get_user_repository)]) -> AuthService:
return AuthService(repo)
AuthServiceDep = Annotated[AuthService, Depends(get_auth_service)]几个关键点:
oauth2_scheme负责抽取Authorization: Bearer <token>,缺失或格式不对时它自己就会抛 401,你不需要手写解析。tokenUrl指的是完整路径。写错(例如漏掉/api/v1前缀)不会影响运行时校验,但会让 Swagger 的 Authorize 按钮拿着 404 的响应去解析 token,表现为「授权成功但请求还是 401」。- 必须查库,不能只信 token。用户可能在 token 有效期内被删除、被禁用或改了角色,只有数据库是当前事实。
int(payload["sub"])包在同一个 try 里:sub缺失抛KeyError,被篡改成非数字抛ValueError,都该翻译成 401。
12.8 401 与 403 的正确使用
这两个状态码的误用比错误本身更麻烦——它会让前端的「自动跳登录页」逻辑失效。
| 场景 | 状态码 | 是否带 WWW-Authenticate: Bearer |
|---|---|---|
完全没带 Authorization 头 | 401 | 是 |
| token 过期、签名不对、格式损坏 | 401 | 是 |
| token 有效,但用户已被删除 | 401 | 是 |
用户名或密码错误(/token) | 401 | 是 |
| token 有效,但用户被禁用 | 403 | 否 |
| token 有效,但角色/权限不足 | 403 | 否 |
| token 有效,但访问的不是自己的资源 | 403(或 404,避免资源存在性泄露) | 否 |
判断口诀:「你的身份还没被确认」→ 401;「身份确认了,但这事你不能干」→ 403。
# app/api/deps.py(续)
async def get_current_active_user(current_user: CurrentUser) -> User:
"""在 get_current_user 之上叠加一层状态检查。"""
if not current_user.is_active:
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="用户已被禁用")
return current_user
ActiveUser = Annotated[User, Depends(get_current_active_user)]
async def get_current_superuser(current_user: ActiveUser) -> User:
if not current_user.is_superuser:
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="需要超级管理员权限")
return current_user
SuperUser = Annotated[User, Depends(get_current_superuser)]这种「依赖套依赖」的叠加写法是 FastAPI 权限模型的骨架:每一层只负责一个判断,组合出任意精细的规则,而且每一层都能被 dependency_overrides 单独替换。
12.9 基于角色的权限依赖
角色检查比 is_superuser 布尔值更通用,也更容易扩展。模型上先加一列:
# app/models/user.py(在第 10 章的基础上补一行)
from sqlalchemy import JSON
roles: Mapped[list[str]] = mapped_column(JSON, default=list)然后写一个依赖工厂——它本身不是依赖,而是「返回依赖的函数」:
# app/api/deps.py(续)
from collections.abc import Callable
from app.core.exceptions import PermissionDeniedError
def require_role(*allowed: str) -> Callable[[ActiveUser], User]:
"""返回一个依赖:要求当前用户的角色命中 allowed 中的任意一个。"""
async def _checker(current_user: ActiveUser) -> User:
if not set(current_user.roles) & set(allowed):
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail=f"需要以下角色之一:{', '.join(allowed)}",
)
return current_user
return _checker
AdminUser = Annotated[User, Depends(require_role("admin"))]
EditorUser = Annotated[User, Depends(require_role("admin", "editor"))]用法:
# app/api/v1/admin.py
from fastapi import APIRouter
from app.api.deps import AdminUser, EditorUser
router = APIRouter(prefix="/admin", tags=["admin"])
@router.get("/stats", summary="仅管理员")
async def stats(current_user: AdminUser) -> dict[str, str]:
return {"operator": current_user.email}
@router.post("/posts/{post_id}/publish", summary="管理员或编辑")
async def publish(post_id: int, current_user: EditorUser) -> dict[str, object]:
return {"post_id": post_id, "published_by": current_user.id}Depends(require_role("admin")) 每次调用都会新建一个闭包,FastAPI 以函数对象为缓存键,因此不同角色的检查不会互相串用。
💡 角色适合粗粒度收口(管理员、编辑、普通用户);细粒度(「只能删自己的文章」)属于资源归属校验,应该放在 Service 层里用
post.author_id != current_user.id判断,而不是堆更多角色。
12.10 配置与密钥管理
密钥硬编码进代码 = 密钥进了 Git 历史 = 全网可读。正确做法是用 pydantic-settings 从环境变量读(第 17 章会完整展开配置体系):
# app/core/config.py
from pydantic import Field
from pydantic_settings import BaseSettings, SettingsConfigDict
class Settings(BaseSettings):
model_config = SettingsConfigDict(env_file=".env", env_file_encoding="utf-8", extra="ignore")
PROJECT_NAME: str = "Blog API"
ENVIRONMENT: str = "dev"
DATABASE_URL: str = "sqlite+aiosqlite:///./app.db"
# 生产环境必须通过环境变量覆盖;开发期给一个占位值,方便本地直接启动
SECRET_KEY: str = "dev-only-insecure-key-change-me"
ALGORITHM: str = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES: int = Field(default=30, gt=0)
settings = Settings()# .env(加入 .gitignore,绝不提交)
SECRET_KEY=09d25e094faa6ca2556c818166b7a9563b93f7099f6f0f4caa6cf63b88e8d3e7
ALGORITHM=HS256
ACCESS_TOKEN_EXPIRE_MINUTES=30
DATABASE_URL=postgresql+asyncpg://postgres:secret@127.0.0.1:5432/blog# .env.example(提交进仓库,只放键名与占位值)
SECRET_KEY=please-generate-with-openssl-rand-hex-32
ALGORITHM=HS256
ACCESS_TOKEN_EXPIRE_MINUTES=30
DATABASE_URL=postgresql+asyncpg://user:password@localhost:5432/dbname# .gitignore 里的关键几行
.env
.env.local
*.pem生成一个足够强的密钥:
openssl rand -hex 32
# 或
python -c "import secrets; print(secrets.token_hex(32))"❌ 不推荐:
SECRET_KEY = "supersecret" # 进了仓库,任何能读到代码的人都能签发合法 token
ALGORITHM = "HS256"✅ 推荐:
class Settings(BaseSettings):
model_config = SettingsConfigDict(env_file=".env")
SECRET_KEY: str # 无默认值:环境变量缺失时启动即报错
ALGORITHM: str = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES: int = 30生产环境刻意不给 SECRET_KEY 默认值,让配置缺失在启动阶段就暴露,而不是带着一个公开的弱密钥上线。
12.11 端到端验证
# 1. 登录,拿 token
curl -X POST http://127.0.0.1:8000/api/v1/auth/token \
-H "Content-Type: application/x-www-form-urlencoded" \
-d "username=alice@example.com&password=correct-horse"
# {"access_token":"eyJhbGciOiJIUzI1NiJ9...","token_type":"bearer"}
# 2. 带上 token 访问受保护端点
TOKEN="eyJhbGciOiJIUzI1NiJ9..."
curl http://127.0.0.1:8000/api/v1/users/me -H "Authorization: Bearer $TOKEN"
# 3. 不带 token → 401 + WWW-Authenticate
curl -i http://127.0.0.1:8000/api/v1/users/me对应的受保护端点:
# app/api/v1/users.py(在第 11 章的基础上补两个端点)
from fastapi import APIRouter
from app.api.deps import ActiveUser, CurrentUser, SuperUser, UserServiceDep
from app.schemas.user import UserOut
router = APIRouter(prefix="/users", tags=["users"])
@router.get("/me", response_model=UserOut, summary="当前登录用户")
async def read_me(current_user: CurrentUser) -> UserOut:
return UserOut.model_validate(current_user)
@router.get("/me/profile", response_model=UserOut, summary="需要账号处于激活状态")
async def read_profile(current_user: ActiveUser) -> UserOut:
return UserOut.model_validate(current_user)
@router.get("/admin/all", response_model=list[UserOut], summary="仅超级管理员")
async def list_all(current_user: SuperUser, service: UserServiceDep) -> list[UserOut]:
return await service.list_all(limit=100)⚠️ 路由注册顺序很重要:
/users/me必须写在/users/{user_id}之前,否则me会被当成路径参数去转int,得到 422。这个坑第 3 章讲过,在加认证端点时最容易再次踩到。
12.12 上线前的安全清单
| 项 | 做法 | 不做的后果 |
|---|---|---|
| 传输加密 | 全站 HTTPS,HTTP 301 跳转 | token 在链路上被直接截获 |
| token 有效期 | 访问令牌 15 ~ 30 分钟 | 泄露后长期可用 |
| 刷新机制 | 长效 refresh token 只换发 access token,可单独吊销 | 要么频繁登录,要么发一个长期令牌 |
| 令牌传递 | 只放 Authorization 头 | 放 URL query 会进 Nginx 日志、浏览器历史与 Referer |
| 注销 | 服务端维护 jti 黑名单(Redis,带 TTL)或缩短有效期 | 「退出登录」只是前端删了本地存储,token 依然有效 |
| 密码策略 | 最少 8 位,配 Argon2 | 弱密码可被离线爆破 |
| 日志 | 绝不打印 Authorization 头与密码字段 | 凭据落进日志系统 |
❌ 不推荐:
GET /api/v1/users/me?token=eyJhbGciOiJIUzI1NiJ9... # 进日志、进历史、进 Referer✅ 推荐:
GET /api/v1/users/me
Authorization: Bearer eyJhbGciOiJIUzI1NiJ9...常见坑与排查
| 现象 | 原因 | 解决 |
|---|---|---|
jwt.exceptions.InvalidSubjectError: Subject must be a string | 签发时 sub 传了 int(如 user.id) | 编码时 str(user.id),解码后 int(payload["sub"]);PyJWT 2.10+ 对类型敏感 |
| 刚登录就提示 token 过期,或有效期比配置长好几小时 | exp 用了本地时间或 naive utcnow(),与 UTC 校验基准有偏移 | 一律 datetime.now(timezone.utc);不要用 datetime.utcnow() |
RuntimeError: python-multipart must be installed | /token 的 OAuth2PasswordRequestForm 需要表单解析库 | uv add python-multipart(或直接装 fastapi[standard]) |
| Swagger 里 Authorize 成功,但请求仍返回 401 | OAuth2PasswordBearer(tokenUrl=...) 路径写错,Swagger 拿不到 token | tokenUrl 写完整路径(含 /api/v1 前缀),与路由注册保持一致 |
| 前端收不到 401 的自动跳转 | 401 响应缺少 WWW-Authenticate: Bearer 头 | 每个 401 的 HTTPException 都传 headers={"WWW-Authenticate": "Bearer"} |
接口返回里出现了 hashed_password | response_model 用了 ORM 模型或含密码字段的 schema | 出参 schema 只声明允许暴露的字段;UserOut 不含密码,必要时 model_config = ConfigDict(from_attributes=True) |
| token 泄露后长期有效,无法止损 | 有效期设得过长(几天甚至不过期),且没有吊销机制 | access token 15 ~ 30 分钟;用 refresh token 换发;注销时按 jti 拉黑 |
| 密钥泄漏、任何人都能签发合法 token | SECRET_KEY 硬编码在代码里并提交进了仓库 | 移到环境变量;泄漏后立即轮换(旧 token 全部失效);历史记录里也要清理 |
| 用户不存在与密码错误返回不同提示 | 分别返回「用户不存在」「密码错误」,等于提供枚举接口 | 统一为「邮箱或密码错误」,并对不存在的用户也跑一次哈希校验抹平耗时 |
| 用户被禁用后仍能继续调接口 | 只校验了 token,没查库或没查 is_active | get_current_user 必须查库;用 get_current_active_user 叠加状态检查 |
| 401 与 403 混用,前端逻辑混乱 | 把「权限不足」也返回 401 | 认证失败用 401 + WWW-Authenticate,已认证但无权用 403 |
本章小结
| 要点 | 说明 |
|---|---|
| 认证 vs 授权 | 认证答「你是谁」(401),授权答「你能做什么」(403) |
| 密码存储 | pwdlib + Argon2,PasswordHash.recommended() 的 hash() / verify();绝不用 MD5/SHA1 |
| 依赖 | uv add "pwdlib[argon2]" pyjwt python-multipart(官方已从 passlib 迁移到 pwdlib) |
| JWT 本质 | 三段 Header.Payload.Signature,是签名不是加密,payload 可被任何人解码 |
| 签发 | jwt.encode(payload, settings.SECRET_KEY, algorithm="HS256"),sub 转字符串,exp 必设且用 UTC |
| 校验 | jwt.decode(token, key, algorithms=["HS256"]),algorithms 必须显式白名单 |
| 登录端点 | OAuth2PasswordBearer(tokenUrl="/api/v1/auth/token") + OAuth2PasswordRequestForm,返回 access_token / token_type |
| 当前用户 | get_current_user 解码 → 查库 → 返回 User;用 Annotated 定义 CurrentUser |
| 状态码 | 未认证/凭证无效 → 401 带 WWW-Authenticate;已认证但权限不足 → 403 |
| 权限依赖 | require_role("admin") 工厂返回依赖;细粒度归属校验放 Service 层 |
| 密钥管理 | SECRET_KEY 走环境变量(pydantic-settings),生产不设默认值,.env 进 .gitignore,同时提供 .env.example |
| 上线加固 | HTTPS、短期 access token + refresh token、token 只走 Authorization 头、注销用 jti 黑名单 |
练习题
给
User模型加上roles: Mapped[list[str]],实现require_role依赖工厂,并写一个「仅 admin 可访问」的DELETE /api/v1/users/{user_id}端点。实现 refresh token 流程:
/token同时返回短期access_token与长期refresh_token,新增/token/refresh用 refresh token 换新的 access token。说明如何区分两类 token(提示:加typ声明)以及为什么不能互相混用。用
dependency_overrides写一个测试,覆盖get_current_user返回一个固定的User,验证GET /api/v1/users/me在「未带 token」「token 已过期」「token 合法但用户被禁用」三种情况下的状态码分别是 401、401、403。下面这段签发代码有两个会直接导致运行时错误的问题,请指出并修正:
payload = {"sub": user.id, "exp": datetime.utcnow() + timedelta(minutes=60)}
token = jwt.encode(payload, "my-secret-key", algorithm="HS256")
data = jwt.decode(token, "my-secret-key", algorithms=["HS256"])
user_id = data["sub"]- 设计一个
jti黑名单方案:注销时把 token 的jti写入 Redis 并设置 TTL,get_current_user校验时查一次黑名单。说明 TTL 应该设为多少,以及这个方案给每个请求增加了什么开销。
下一章预告
认证依赖里已经出现了
async def与await,但你可能还没弄清「什么时候该用async def、什么时候用def」。下一章把异步模型讲透:事件循环、阻塞调用的陷阱、线程池的边界,以及用httpx调用外部服务。