第 15 章 · SimpleJWT、限流与 OAuth2
本章目标:使用 djangorestframework-simplejwt 实现 access/refresh token 与自定义 claims;配置 DRF Throttling(及 django-ratelimit 简介)返回 429;理解 OAuth2 授权码流程(虚构 auth.example.com)与 django-oauth-toolkit 定位;掌握 Token 刷新、黑名单概念与 API 安全清单;在 shop-demo 为 user-demo SPA 提供 JWT 鉴权。
学时建议:5~6 小时(含 2 小时 JWT 联调 + 1 小时 OAuth2 概念阅读)
前置:本模块 ch01~ch14;重点复习 ch03 Session 认证、ch07 DRF 权限、ch14 Swagger BearerAuth 配置。
15.1 JWT 与 Session 选型
| 对比项 | Session + Cookie(ch03 管理页) | JWT Bearer(本章 API) |
|---|---|---|
| 状态 | 服务端 session 存储 | 默认无状态(黑名单辅助注销) |
| 跨域 | 需 credentials: include | Authorization: Bearer 头 |
| 注销 | logout 清 session | 短 access + refresh 轮换 / 黑名单 |
| 适用 | 同源 Django 模板后台 | user-demo SPA / 移动端 |
| 风险 | CSRF | XSS 窃取 Token、日志泄露 |
shop-demo 建议:运营后台继续 Session;/api/ 对外 JWT,供 user-demo 调用 https://api.example.com。
15.2 安装 SimpleJWT
cd ~/python-learn/shop-demo
pip install "djangorestframework-simplejwt>=5.3,<6"
pip install django-redis # 黑名单 / 限流计数(可选)
shopdemo/settings.py:
from datetime import timedelta
INSTALLED_APPS += [
"rest_framework_simplejwt",
"rest_framework_simplejwt.token_blacklist", # 可选:注销黑名单
]
REST_FRAMEWORK = {
# ...
"DEFAULT_AUTHENTICATION_CLASSES": [
"rest_framework_simplejwt.authentication.JWTAuthentication",
"rest_framework.authentication.SessionAuthentication", # 保留 Browsable API
],
"DEFAULT_THROTTLE_CLASSES": [
"rest_framework.throttling.AnonRateThrottle",
"rest_framework.throttling.UserRateThrottle",
],
"DEFAULT_THROTTLE_RATES": {
"anon": "200/hour",
"user": "1000/hour",
"login": "10/minute",
"burst": "60/minute",
},
}
SIMPLE_JWT = {
"ACCESS_TOKEN_LIFETIME": timedelta(minutes=15),
"REFRESH_TOKEN_LIFETIME": timedelta(days=7),
"ROTATE_REFRESH_TOKENS": True,
"BLACKLIST_AFTER_ROTATION": True,
"UPDATE_LAST_LOGIN": True,
"ALGORITHM": "HS256",
"SIGNING_KEY": os.environ.get("JWT_SIGNING_KEY", "dev-only-change-me-in-production"),
"AUTH_HEADER_TYPES": ("Bearer",),
"USER_ID_FIELD": "id",
"USER_ID_CLAIM": "user_id",
"TOKEN_OBTAIN_SERIALIZER": "accounts.serializers.ShopTokenObtainPairSerializer",
}
.env.example:
JWT_SIGNING_KEY=请替换为随机长字符串至少32字符
REDIS_URL=redis://127.0.0.1:6379/1
严禁将真实 JWT_SIGNING_KEY 提交 Git 或写入答辩材料。
15.3 自定义 Token claims
accounts/serializers.py:
from rest_framework_simplejwt.serializers import TokenObtainPairSerializer
class ShopTokenObtainPairSerializer(TokenObtainPairSerializer):
@classmethod
def get_token(cls, user):
token = super().get_token(user)
token["username"] = user.username
token["role"] = "staff" if user.is_staff else "customer"
return token
注意:不要覆盖 SimpleJWT 的保留 claim(token_type、exp、iat、jti、user_id)。此前示例给token_type赋值会污染 refresh token 的类型判定,属错误写法;自定义信息请用非保留名(如role、app_scope)。
| Claim | 含义 |
|---|---|
user_id | 用户主键(SimpleJWT 默认) |
username | 用户名(自定义) |
role | staff / customer(自定义,非敏感) |
token_type / exp / iat / jti | SimpleJWT 保留字段,勿覆盖 |
不要在 JWT 中存放:密码、手机号明文、完整权限列表(体积与泄露风险)。
15.4 URL 与登录端点
# shopdemo/urls.py
from django.urls import path
from rest_framework_simplejwt.views import TokenRefreshView, TokenBlacklistView
from accounts.views import me, ThrottledLoginView
urlpatterns = [
path("api/auth/login/", ThrottledLoginView.as_view()),
path("api/auth/refresh/", TokenRefreshView.as_view()),
path("api/auth/logout/", TokenBlacklistView.as_view()),
path("api/me/", me),
# api/、ch14 文档路由 ...
]
| 端点 | 方法 | 说明 |
|---|---|---|
/api/auth/login/ | POST | {"username","password"} → access + refresh |
/api/auth/refresh/ | POST | {"refresh"} → 新 access(可轮换 refresh) |
/api/auth/logout/ | POST | Header Bearer + body refresh → 黑名单 |
curl -X POST http://127.0.0.1:8000/api/auth/login/ \
-H "Content-Type: application/json" \
-d '{"username": "user-demo", "password": "demo-pass"}'
15.5 保护 API 与 /me 端点
accounts/throttles.py(登录专用限流,配合 15.6 的 scope 配置):
from rest_framework.throttling import ScopedRateThrottle
class LoginRateThrottle(ScopedRateThrottle):
scope = "login"
accounts/views.py:
from drf_spectacular.utils import extend_schema
from rest_framework.decorators import api_view, permission_classes
from rest_framework.permissions import IsAuthenticated
from rest_framework.response import Response
from rest_framework_simplejwt.views import TokenObtainPairView
from .throttles import LoginRateThrottle
class ThrottledLoginView(TokenObtainPairView):
"""登录端点叠加限流,防爆破"""
throttle_classes = [LoginRateThrottle]
@extend_schema(tags=["auth"])
@api_view(["GET"])
@permission_classes([IsAuthenticated])
def me(request):
u = request.user
return Response({"id": u.id, "username": u.username, "email": u.email, "is_staff": u.is_staff})
settings.py 中声明 scope 速率:
REST_FRAMEWORK = {
# ...
"DEFAULT_THROTTLE_RATES": {"login": "10/minute", "burst": "60/minute"},
}
catalog/views.py 中写操作视图继续使用 IsAuthenticated + IsAdminUser 权限组合(见 ch07 §7.6)。