JWT 全链路、限流与 OAuth2

第 15 章 · JWT 全链路、限流与 OAuth2

本章目标：使用 PyJWT 实现 access/refresh token 签发与校验，编写 @jwt_required 装饰器；配置 Flask-Limiter 按 IP/用户限流并返回 429；理解 OAuth2 授权码流程（虚构 auth.example.com）及与 Session 的对比；实现 Token 黑名单/注销与刷新端点；掌握 API 安全清单。

学时建议：5～6 小时（含 2 小时 JWT 联调 + 1 小时 OAuth2 概念阅读）

前置：本模块 ch01～ch14；重点复习 ch06 Flask-Login Session、ch07 401 错误码、ch12 认证方案对比、ch14 Swagger Bearer 配置。

15.1 JWT 与 Session 选型

api-demo 建议：管理页继续 Session；/api/v1 对外提供 JWT，供 user-demo SPA 调用。

15.2 安装与配置

pip install PyJWT Flask-Limiter redis

# api_demo/config.py
import os
from datetime import timedelta

class Config:
    JWT_SECRET_KEY = os.environ.get("JWT_SECRET_KEY", "dev-only-change-me")
    JWT_ALGORITHM = "HS256"
    JWT_ACCESS_EXPIRES = timedelta(minutes=15)
    JWT_REFRESH_EXPIRES = timedelta(days=7)
    RATELIMIT_STORAGE_URI = os.environ.get("REDIS_URL", "memory://")
    RATELIMIT_DEFAULT = "200 per hour"

.env.example：JWT_SECRET_KEY=请替换为随机长字符串。严禁将真实密钥提交 Git。

15.3 Token 签发与载荷

api_demo/auth/jwt_utils.py：

from datetime import datetime, timezone
import jwt
from flask import current_app


def _utcnow():
    return datetime.now(timezone.utc)


def create_access_token(user_id: int, username: str) -> str:
    now = _utcnow()
    payload = {
        "sub": str(user_id), "username": username, "type": "access",
        "iat": now, "exp": now + current_app.config["JWT_ACCESS_EXPIRES"],
    }
    return jwt.encode(payload, current_app.config["JWT_SECRET_KEY"],
                      algorithm=current_app.config["JWT_ALGORITHM"])


def create_refresh_token(user_id: int) -> str:
    now = _utcnow()
    payload = {
        "sub": str(user_id), "type": "refresh",
        "iat": now, "exp": now + current_app.config["JWT_REFRESH_EXPIRES"],
    }
    return jwt.encode(payload, current_app.config["JWT_SECRET_KEY"],
                      algorithm=current_app.config["JWT_ALGORITHM"])


def decode_token(token: str) -> dict:
    return jwt.decode(token, current_app.config["JWT_SECRET_KEY"],
                      algorithms=[current_app.config["JWT_ALGORITHM"]])

15.4 登录端点

# api_demo/api/auth_jwt.py
from flask import request, current_app
from werkzeug.security import check_password_hash
from api_demo.extensions import limiter
from api_demo.models.user import User
from api_demo.auth.jwt_utils import create_access_token, create_refresh_token
from . import api_bp
from .utils import ok, fail


@api_bp.post("/auth/login")
@limiter.limit("10 per minute")
def login():
    body = request.get_json(silent=True) or {}
    username = (body.get("username") or "").strip()
    password = body.get("password") or ""
    user = User.query.filter_by(username=username).first()
    if not user or not check_password_hash(user.password_hash, password):
        return fail(40101, "用户名或密码错误", status=401)
    return ok(data={
        "access_token": create_access_token(user.id, user.username),
        "refresh_token": create_refresh_token(user.id),
        "token_type": "Bearer",
        "expires_in": int(current_app.config["JWT_ACCESS_EXPIRES"].total_seconds()),
    })

15.5 @jwt_required 装饰器

api_demo/auth/decorators.py 核心逻辑：

def jwt_required(token_type="access"):
    def decorator(fn):
        @wraps(fn)
        def wrapper(*args, **kwargs):
            token = _extract_bearer()          # 解析 Authorization: Bearer
            if not token:
                return fail(40101, "缺少 Bearer Token", status=401)
            if is_token_revoked(token):
                return fail(40102, "Token 已注销", status=401)
            try:
                payload = decode_token(token)
            except jwt.ExpiredSignatureError:
                return fail(40103, "Token 已过期", status=401)
            except jwt.InvalidTokenError:
                return fail(40101, "Token 无效", status=401)
            if payload.get("type") != token_type:
                return fail(40101, "Token 类型错误", status=401)
            g.current_user_id = int(payload["sub"])
            g.jwt_payload = payload
            return fn(*args, **kwargs)
        return wrapper
    return decorator

GET /api/v1/me 加 @jwt_required()，从 g.current_user_id 查用户返回。

15.6 Refresh 与注销

POST /api/v1/auth/refresh：校验 refresh_token 的 type=refresh 且未过期 → 签发新 access。

流程：登录 → access(15min) + refresh(7d) → access 过期 → refresh → refresh 过期 → 重新登录。

黑名单（Redis，api_demo/auth/token_blacklist.py）：

def revoke_token(token: str, exp_timestamp: int) -> None:
    key = f"jwt:bl:{hashlib.sha256(token.encode()).hexdigest()}"
    ttl = max(exp_timestamp - int(datetime.now(timezone.utc).timestamp()), 1)
    redis_client.setex(key, ttl, "1")

POST /auth/logout：@jwt_required() 后将当前 access 写入黑名单；body 可附带 refresh_token 一并注销。