Skip to content

第 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 Unauthorized403 Forbidden
响应头要求必须带 WWW-Authenticate: Bearer无特殊要求
典型实现/token 端点、get_current_user 依赖require_role("admin")、资源归属校验

12.2 依赖安装 ​

bash
uv add "pwdlib[argon2]" pyjwt python-multipart
包用途不装会怎样
pwdlib[argon2]密码哈希(Argon2 算法)导入 PasswordHash 失败
pyjwtJWT 的签发与校验导入 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 密码哈希 ​

密码必须单向哈希后存储:数据库泄露时攻击者拿到的是散列值,无法直接反推出密码,也无法拿去撞其他站点(很多人到处用同一个密码)。

python
# 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

❌ 不推荐:

python
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   # 比较也不是恒定时间

✅ 推荐:

python
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 而不是直接比较:

text
$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)是一串用 . 分成三段的字符串:

text
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):

声明全称语义注意事项
subsubject主体标识必须是字符串,见下方警告
expexpiration time过期时间(UTC 时间戳)必须设置,否则 token 永不失效
iatissued at签发时间用于判断 token 年龄
jtiJWT IDtoken 唯一标识做黑名单/一次性 token 时需要
nbfnot before生效时间提前签发场景用
python
# 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)

关于时间的三条硬规则:

  1. 一律用 datetime.now(timezone.utc)。用 datetime.now()(本地时间)或 utcnow()(无时区 naive 对象)在容器时区变化或跨时区部署时会算出偏移的 exp,表现为「刚登录就过期」或「token 活了 8 小时」。
  2. exp 不是可选项。省略它等于签发了一张永久通行证,一旦泄露无法挽回。
  3. sub 必须是字符串。PyJWT 2.10 起对 sub 做类型校验,jwt.decode 遇到非字符串的 sub 会抛 InvalidSubjectError。数据库主键是 int,所以 encode 时 str(subject)、decode 后 int(...) 转回来。

12.5 校验与异常分支 ​

python
# 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 类型错
        raise

algorithms 参数必须显式传白名单列表。这不是可选的礼貌写法:如果把客户端 header 里的 alg 直接拿来用,攻击者可以改成 none 或换成非对称算法来绕过验签。

异常触发条件给客户端的状态码
ExpiredSignatureErrorexp 早于当前时间401
InvalidSignatureError签名与密钥不匹配(被篡改或换了密钥)401
DecodeError不是合法的 JWT 三段结构401
InvalidSubjectErrorsub 不是字符串401(同时说明签发端有 bug)
InvalidAlgorithmErroralg 不在允许列表内401

ExpiredSignatureError 是 InvalidTokenError 的子类,所以 except InvalidTokenError 一个分支就能兜住全部情况;单列过期分支只是为了给用户更友好的提示文案。


12.6 /token 端点 ​

python
# 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)
python
# 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 依赖 ​

python
# 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 端点用的就是这个别名:

python
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。

python
# 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 布尔值更通用,也更容易扩展。模型上先加一列:

python
# app/models/user.py(在第 10 章的基础上补一行)
from sqlalchemy import JSON

roles: Mapped[list[str]] = mapped_column(JSON, default=list)

然后写一个依赖工厂——它本身不是依赖,而是「返回依赖的函数」:

python
# 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"))]

用法:

python
# 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 章会完整展开配置体系):

python
# 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()
ini
# .env(加入 .gitignore,绝不提交)
SECRET_KEY=09d25e094faa6ca2556c818166b7a9563b93f7099f6f0f4caa6cf63b88e8d3e7
ALGORITHM=HS256
ACCESS_TOKEN_EXPIRE_MINUTES=30
DATABASE_URL=postgresql+asyncpg://postgres:secret@127.0.0.1:5432/blog
ini
# .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
text
# .gitignore 里的关键几行
.env
.env.local
*.pem

生成一个足够强的密钥:

bash
openssl rand -hex 32
# 或
python -c "import secrets; print(secrets.token_hex(32))"

❌ 不推荐:

python
SECRET_KEY = "supersecret"        # 进了仓库,任何能读到代码的人都能签发合法 token
ALGORITHM = "HS256"

✅ 推荐:

python
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 端到端验证 ​

bash
# 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

对应的受保护端点:

python
# 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 头与密码字段凭据落进日志系统

❌ 不推荐:

text
GET /api/v1/users/me?token=eyJhbGciOiJIUzI1NiJ9...    # 进日志、进历史、进 Referer

✅ 推荐:

text
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 成功,但请求仍返回 401OAuth2PasswordBearer(tokenUrl=...) 路径写错,Swagger 拿不到 tokentokenUrl 写完整路径(含 /api/v1 前缀),与路由注册保持一致
前端收不到 401 的自动跳转401 响应缺少 WWW-Authenticate: Bearer 头每个 401 的 HTTPException 都传 headers={"WWW-Authenticate": "Bearer"}
接口返回里出现了 hashed_passwordresponse_model 用了 ORM 模型或含密码字段的 schema出参 schema 只声明允许暴露的字段;UserOut 不含密码,必要时 model_config = ConfigDict(from_attributes=True)
token 泄露后长期有效,无法止损有效期设得过长(几天甚至不过期),且没有吊销机制access token 15 ~ 30 分钟;用 refresh token 换发;注销时按 jti 拉黑
密钥泄漏、任何人都能签发合法 tokenSECRET_KEY 硬编码在代码里并提交进了仓库移到环境变量;泄漏后立即轮换(旧 token 全部失效);历史记录里也要清理
用户不存在与密码错误返回不同提示分别返回「用户不存在」「密码错误」,等于提供枚举接口统一为「邮箱或密码错误」,并对不存在的用户也跑一次哈希校验抹平耗时
用户被禁用后仍能继续调接口只校验了 token,没查库或没查 is_activeget_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 黑名单

练习题 ​

  1. 给 User 模型加上 roles: Mapped[list[str]],实现 require_role 依赖工厂,并写一个「仅 admin 可访问」的 DELETE /api/v1/users/{user_id} 端点。

  2. 实现 refresh token 流程:/token 同时返回短期 access_token 与长期 refresh_token,新增 /token/refresh 用 refresh token 换新的 access token。说明如何区分两类 token(提示:加 typ 声明)以及为什么不能互相混用。

  3. 用 dependency_overrides 写一个测试,覆盖 get_current_user 返回一个固定的 User,验证 GET /api/v1/users/me 在「未带 token」「token 已过期」「token 合法但用户被禁用」三种情况下的状态码分别是 401、401、403。

  4. 下面这段签发代码有两个会直接导致运行时错误的问题,请指出并修正:

python
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"]
  1. 设计一个 jti 黑名单方案:注销时把 token 的 jti 写入 Redis 并设置 TTL,get_current_user 校验时查一次黑名单。说明 TTL 应该设为多少,以及这个方案给每个请求增加了什么开销。

下一章预告 ​

认证依赖里已经出现了 async def 与 await,但你可能还没弄清「什么时候该用 async def、什么时候用 def」。下一章把异步模型讲透:事件循环、阻塞调用的陷阱、线程池的边界,以及用 httpx 调用外部服务。

👉 第 13 章:异步编程与并发模型

📖本文阅读--次|📊全站访问--次|👥访客--人