Django RESTful 架构与 DRF 进阶全链路深度剖析
1. RESTful 架构在 Django/DRF 中的本质与落地
1.1 RESTful 架构核心原则
REST（Representational State Transfer）是一种架构风格，强调资源（Resource）为中心，使用统一接口（Uniform Interface），无状态（Stateless），可缓存（Cacheable），分层系统（Layered System），按需代码（Code on Demand，可选）。
1.1.1 资源的抽象与建模
在企业级系统中，资源不仅仅是数据库表，更是业务实体的抽象。例如，订单、用户、商品、库存、日志、审批流等都可以作为资源。
# models.py
class Order(models.Model):
"""订单模型，代表一个业务资源"""
user = models.ForeignKey(User, on_delete=models.CASCADE, related_name='orders') # 订单所属用户
status = models.CharField(max_length=20, choices=ORDER_STATUS_CHOICES) # 订单状态
total_price = models.DecimalField(max_digits=10, decimal_places=2) # 订单总价
created_at = models.DateTimeField(auto_now_add=True) # 创建时间
updated_at = models.DateTimeField(auto_now=True) # 更新时间
# 订单的商品项通过OrderItem模型关联
解释： 这里的Order模型就是一个RESTful资源的典型例子。
1.1.2 统一接口与HTTP动词的语义映射
GET /orders/：获取订单列表
POST /orders/：创建新订单
GET /orders/{id}/：获取订单详情
PUT /orders/{id}/：全量更新订单
PATCH /orders/{id}/：部分更新订单
DELETE /orders/{id}/：删除订单
# urls.py
from rest_framework.routers import DefaultRouter
from .views import OrderViewSet
router = DefaultRouter()
router.register(r'orders', OrderViewSet, basename='order')
urlpatterns = router.urls
解释： 使用DRF的路由器自动将资源与HTTP动词映射。
1.1.3 资源的表现层与序列化
表现层（Representation）是资源的不同视图，常见为JSON、XML、YAML等。DRF通过序列化器（Serializer）实现表现层转换。
# serializers.py
class OrderSerializer(serializers.ModelSerializer):
"""订单序列化器，将Order模型转换为JSON表现层"""
class Meta:
model = Order
fields = '__all__'
解释： 该序列化器负责将Order对象转为API输出的JSON格式。
1.1.4 无状态性与幂等性
RESTful要求服务端不保存客户端状态，每个请求都应自包含所有认证和上下文信息。PUT/DELETE等操作应幂等。
# views.py
class OrderViewSet(viewsets.ModelViewSet):
"""订单视图集，自动实现RESTful接口"""
queryset = Order.objects.all()
serializer_class = OrderSerializer
permission_classes = [IsAuthenticated]
# 认证信息每次请求都需携带（如Token/JWT），服务端不保存会话状态
解释： 这里每个请求都需认证，服务端不保存会话。
1.1.5 可缓存性
RESTful架构允许对GET请求结果进行缓存，提升性能。
# views.py
from django.views.decorators.cache import cache_page
from django.utils.decorators import method_decorator
@method_decorator(cache_page(60*5), name='dispatch') # 缓存5分钟
class ProductListView(APIView):
"""商品列表接口，支持缓存"""
def get(self, request):
products = Product.objects.all()
serializer = ProductSerializer(products, many=True)
return Response(serializer.data)
解释： 通过Django缓存装饰器对API结果进行缓存。
1.1.6 分层系统与微服务
RESTful架构天然支持分层系统，前端、API网关、业务服务、数据服务可分层部署。DRF可作为微服务架构中的业务API层。
1.2 RESTful API 设计规范与企业级最佳实践
1.2.1 资源命名规范
资源名用复数（orders、users、products）
子资源用嵌套（/orders/{order_id}/items/）
操作用动词（/orders/{id}/cancel/）
1.2.2 状态码与错误处理
200 OK：成功
201 Created：创建成功
204 No Content：删除成功
400 Bad Request：参数错误
401 Unauthorized：未认证
403 Forbidden：无权限
404 Not Found：资源不存在
409 Conflict：资源冲突
422 Unprocessable Entity：语义错误
500 Internal Server Error：服务器错误
# views.py
def create(self, request, *args, **kwargs):
serializer = self.get_serializer(data=request.data)
if not serializer.is_valid():
return Response({
'code': 422, 'msg': '参数校验失败', 'errors': serializer.errors}, status=422)
self.perform_create(serializer)
return Response({
'code': 201, 'msg': '创建成功', 'data': serializer.data}, status=201)
解释： 统一API响应结构，明确状态码和错误信息。
1.2.3 版本管理
企业API需支持多版本共存，避免前端/客户端升级时服务中断。
# urls.py
urlpatterns = [
path('api/v1/', include('app.api.v1.urls')),
path('api/v2/', include('app.api.v2.urls')),
]
解释： 通过URL前缀实现API版本隔离。
1.2.4 分页、过滤、排序
企业级API通常数据量大，需支持分页、过滤、排序。
# settings.py
REST_FRAMEWORK = {
'DEFAULT_PAGINATION_CLASS': 'rest_framework.pagination.PageNumberPagination',
'PAGE_SIZE': 20,
'DEFAULT_FILTER_BACKENDS': [
'django_filters.rest_framework.DjangoFilterBackend',
'rest_framework.filters.OrderingFilter',
'rest_framework.filters.SearchFilter',
],
}
# views.py
class ProductViewSet(viewsets.ModelViewSet):
queryset = Product.objects.all()
serializer_class = ProductSerializer
filterset_fields = ['category', 'status'] # 支持字段过滤
search_fields = ['name', 'description'] # 支持模糊搜索
ordering_fields = ['price', 'created_at'] # 支持排序
解释： 配置分页、过滤、排序，提升API灵活性。
1.2.5 HATEOAS（超媒体作为应用状态的引擎）
RESTful推荐在响应中包含相关资源的链接，便于客户端发现和导航。
# serializers.py
from rest_framework.reverse import reverse
class OrderSerializer(serializers.ModelSerializer):
links = serializers.SerializerMethodField()
def get_links(self, obj):
request = self.context.get('request')
return {
'self': reverse('order-detail', args=[obj.pk], request=request),
'items': reverse('orderitem-list', args=[obj.pk], request=request),
}
class Meta:
model = Order
fields = '__all__'
解释： 在序列化输出中动态生成相关资源的URL。
2. DRF进阶：底层机制与高级应用
2.1 认证与权限深度定制
2.1.1 多重认证机制
企业级API常需支持多种认证方式（Session、Token、JWT、第三方SSO）。
# settings.py
REST_FRAMEWORK = {
'DEFAULT_AUTHENTICATION_CLASSES': [
'rest_framework.authentication.SessionAuthentication', # 浏览器会话
'rest_framework.authentication.TokenAuthentication', # Token认证
'rest_framework_simplejwt.authentication.JWTAuthentication', # JWT认证
'app.auth.EnterpriseSSOAuth', # 企业自定义SSO
]
}
解释： 支持多种认证方式，兼容不同客户端。
2.1.2 自定义权限类
企业常见复杂权限需求，如RBAC、ABAC、数据行级权限。
# permissions.py
from rest_framework.permissions import BasePermission
class IsOrderOwnerOrAdmin(BasePermission):
"""只有订单所有者或管理员可操作订单"""
def has_object_permission(self, request, view, obj):
return obj.user == request.user or request.user.is_staff
# views.py
class OrderViewSet(viewsets.ModelViewSet):
permission_classes = [IsOrderOwnerOrAdmin]
解释： 通过自定义权限类实现细粒度控制。
2.1.3 动态权限与策略引擎
企业级系统常需根据业务规则动态授权，可集成策略引擎如Casbin。
# permissions.py
from casbin import Enforcer
class CasbinPermission(BasePermission):
"""基于Casbin的动态权限控制"""
def has_permission(self, request, view):
enforcer = Enforcer('path/to/model.conf', 'path/to/policy.csv')
sub = request.user.username
obj = request.path
act = request.method.lower()
return enforcer.enforce(sub, obj, act)
解释： 集成第三方策略引擎实现动态权限。
2.2 复杂序列化与反序列化
2.2.1 嵌套写入与事务
企业业务常需在一个API中同时创建主对象和子对象，需保证原子性。
# serializers.py
class OrderItemSerializer(serializers.ModelSerializer):
class Meta:
model = OrderItem
fields = ['product', 'quantity', 'price']
class OrderCreateSerializer(serializers.ModelSerializer):
items = OrderItemSerializer(many=True)
class Meta:
model = Order
fields = ['user', 'status', 'total_price', 'items']
def create(self, validated_data):
items_data = validated_data.pop('items')
with transaction.atomic(): # 保证原子性
order = Order.objects.create(**validated_data)
for item_data in items_data:
OrderItem.objects.create(order=order, **item_data)
return order
解释： 嵌套序列化器+事务，保证主子表数据一致性。
2.2.2 动态字段与多态序列化
企业API常需根据用户角色、请求参数返回不同字段，或对多态模型序列化。
# serializers.py
class DynamicFieldsModelSerializer(serializers.ModelSerializer):
"""根据context动态返回字段"""
def __init__(self, *args, **kwargs):
fields = kwargs.pop('fields', None)
super().__init__(*args, **kwargs)
if fields:
allowed = set(fields)
existing = set(self.fields)
for field_name in existing - allowed:
self.fields.pop(field_name)
# 多态序列化
class PaymentSerializer(serializers.Serializer):
def to_representation(self, instance):
if isinstance(instance, CreditCardPayment):
return CreditCardPaymentSerializer(instance).data
elif isinstance(instance, PaypalPayment):
return PaypalPaymentSerializer(instance).data
return super().to_representation(instance)
解释： 支持动态字段和多态对象的序列化。
2.3 API性能优化与大规模数据处理
2.3.1 查询优化与N+1问题
企业API常因ORM不当导致N+1查询，需用select_related/prefetch_related优化。
# views.py
class OrderViewSet(viewsets.ModelViewSet):
def get_queryset(self):
return Order.objects.select_related('user').prefetch_related('items__product')
解释： 预加载关联对象，避免N+1查询。
2.3.2 批量操作与高性能API
企业级API常需支持批量创建、更新、删除。
# views.py
from rest_framework import status
from rest_framework.decorators import action
from rest_framework.response import Response
class ProductViewSet(viewsets.ModelViewSet):
@action(detail=False, methods=['post'])
def bulk_create(self, request):
serializer = ProductSerializer(data=request.data, many=True)
serializer.is_valid(raise_exception=True)
Product.objects.bulk_create([Product(**item) for item in serializer.validated_data])
return Response({
'msg': '批量创建成功'}, status=status.HTTP_201_CREATED)
解释： 支持批量创建，提升数据导入效率。
2.3.3 缓存与分布式缓存
高并发场景下，API需结合Redis等缓存提升性能。
# views.py
from django.core.cache import cache
class ProductListView(APIView):
def get(self, request):
cache_key = 'product_list'
data = cache.get(cache_key)
if not data:
products = Product.objects.all()
serializer = ProductSerializer(products, many=True)
data = serializer.data
cache.set(cache_key, data, 60*10) # 缓存10分钟
return Response(data)
解释： 使用Django缓存API响应，减轻数据库压力。
2.4 API安全与防护
2.4.1 防止SQL注入与XSS
DRF序列化器自动防止SQL注入，前端需对输出做HTML转义防XSS。
2.4.2 防止CSRF与请求伪造
Session认证需开启CSRF防护，Token/JWT认证可关闭。
# settings.py
REST_FRAMEWORK = {
'DEFAULT_AUTHENTICATION_CLASSES': [
'rest_framework.authentication.TokenAuthentication',
],
'DEFAULT_RENDERER_CLASSES': [
'rest_framework.renderers.JSONRenderer',
],
}
解释： 关闭Session认证时可关闭CSRF。
2.4.3 限流与防刷
企业API需防止恶意刷接口，可用DRF自带限流或接入第三方。
# settings.py
REST_FRAMEWORK = {
'DEFAULT_THROTTLE_CLASSES': [
'rest_framework.throttling.UserRateThrottle',
'rest_framework.throttling.AnonRateThrottle',
],
'DEFAULT_THROTTLE_RATES': {
'user': '1000/day',
'anon': '100/day',
}
}
解释： 配置用户和匿名用户的访问频率限制。
2.4.4 日志与审计
企业API需记录操作日志，便于审计和追踪。
# middleware.py
import logging
logger = logging.getLogger('api.audit')
class APILogMiddleware:
"""API操作日志中间件"""
def __init__(self, get_response):
self.get_response = get_response
def __call__(self, request):
response = self.get_response(request)
if request.path.startswith('/api/'):
logger.info(f'User:{
request.user} Path:{
request.path} Method:{
request.method} Status:{
response.status_code}')
return response
解释： 记录每次API请求的用户、路径、方法和状态码。
2.5 自动化文档与API可发现性
2.5.1 Swagger/OpenAPI集成
企业API需自动生成文档，便于前后端协作。
# settings.py
INSTALLED_APPS += ['drf_yasg']
# urls.py
from drf_yasg.views import get_schema_view
from drf_yasg import openapi
schema_view = get_schema_view(
openapi.Info(
title="企业API文档",
default_version='v1',
description="企业级API接口文档",
),
public=True,
)
urlpatterns += [
path('swagger/', schema_view.with_ui('swagger', cache_timeout=0), name='schema-swagger-ui'),
]
解释： 集成Swagger自动生成API文档。
2.5.2 Redoc/Apiary等多种文档风格
urlpatterns += [
path('redoc/', schema_view.with_ui('redoc', cache_timeout=0), name='schema-redoc'),
]
解释： 支持多种API文档风格，满足不同团队需求。
2.6 复杂业务场景与企业级案例
2.6.1 审批流API设计
企业常见审批流（如请假、报销、采购）需支持多级审批、状态流转、操作日志。
# models.py
class Approval(models.Model):
"""审批流主表"""
applicant = models.ForeignKey(User, on_delete=models.CASCADE)
status = models.CharField(max_length=20, choices=APPROVAL_STATUS_CHOICES)
current_step = models.IntegerField(default=1)
created_at = models.DateTimeField(auto_now_add=True)
class ApprovalStep(models.Model):
"""审批流步骤表"""
approval = models.ForeignKey(Approval, on_delete=models.CASCADE, related_name='steps')
step = models.IntegerField()
approver = models.ForeignKey(User, on_delete=models.CASCADE)
status = models.CharField(max_length=20, choices=STEP_STATUS_CHOICES)
comment = models.TextField(blank=True)
operated_at = models.DateTimeField(auto_now=True)
# serializers.py
class ApprovalStepSerializer(serializers.ModelSerializer):
class Meta:
model = ApprovalStep
fields = '__all__'
class ApprovalSerializer(serializers.ModelSerializer):
steps = ApprovalStepSerializer(many=True, read_only=True)
class Meta:
model = Approval
fields = '__all__'
# views.py
class ApprovalViewSet(viewsets.ModelViewSet):
queryset = Approval.objects.all()
serializer_class = ApprovalSerializer
@action(detail=True, methods=['post'])
def approve(self, request, pk=None):
"""审批通过操作"""
approval = self.get_object()
# 业务逻辑：找到当前步骤，设置为通过，流转到下一个步骤
# 记录操作日志
return Response({
'msg': '审批通过'})
解释： 审批流API设计，支持多级审批和操作日志。
2.6.2 多租户SaaS系统API设计
企业SaaS系统需支持多租户隔离，API需自动过滤租户数据。
# models.py
class Tenant(models.Model):
"""租户模型"""
name = models.CharField(max_length=100)
domain = models.CharField(max_length=100, unique=True)
class TenantAwareModel(models.Model):
"""所有租户相关模型继承此基类"""
tenant = models.ForeignKey(Tenant, on_delete=models.CASCADE)
class Meta:
abstract = True
class Customer(TenantAwareModel):
"""租户下的客户"""
name = models.CharField(max_length=100)
# middleware.py
class TenantMiddleware:
"""根据域名自动识别租户"""
def __init__(self, get_response):
self.get_response = get_response
def __call__(self, request):
domain = request.get_host().split(':')[0]
request.tenant = Tenant.objects.get(domain=domain)
return self.get_response(request)
# views.py
class CustomerViewSet(viewsets.ModelViewSet):
def get_queryset(self):
return Customer.objects.filter(tenant=self.request.tenant)
解释： 多租户API自动隔离数据，防止数据串租。
2.7 运维、监控与高可用
2.7.1 健康检查与自动化运维
企业API需支持健康检查接口，便于K8s等平台自动探活。
# views.py
from django.http import JsonResponse
def health_check(request):
"""健康检查接口"""
return JsonResponse({
'status': 'ok'})
解释： 提供/healthz接口供运维平台探测。
2.7.2 Prometheus监控与指标采集
集成Prometheus采集API请求量、响应时间、错误率等指标。
# middleware.py
from prometheus_client import Counter, Histogram
REQUEST_COUNT = Counter('api_requests_total', 'API请求总数', ['method', 'endpoint', 'status'])
REQUEST_LATENCY = Histogram('api_request_latency_seconds', 'API请求延迟', ['endpoint'])
class PrometheusMiddleware:
"""Prometheus监控中间件"""
def __init__(self, get_response):
self.get_response = get_response
def __call__(self, request):
import time
start = time.time()
response = self.get_response(request)
latency = time.time() - start
endpoint = request.path
REQUEST_COUNT.labels(request.method, endpoint, response.status_code).inc()
REQUEST_LATENCY.labels(endpoint).observe(latency)
return response
解释： 采集API调用指标，便于监控和告警。
2.7.3 灰度发布与流量分流
企业API需支持灰度发布、A/B测试，可结合Nginx、Istio等实现流量分流。
2.8 自动化测试与CI/CD
2.8.1 单元测试与集成测试
企业API需覆盖单元测试、集成测试，保障质量。
# tests.py
from rest_framework.test import APITestCase
class OrderAPITestCase(APITestCase):
def test_create_order(self):
"""测试订单创建接口"""
url = '/api/v1/orders/'
data = {
'user': 1, 'status': 'pending', 'total_price': 100}
response = self.client.post(url, data, format='json')
self.assertEqual(response.status_code, 201)
self.assertEqual(response.data['status'], 'pending')
解释： 使用DRF测试工具自动化API测试。
2.8.2 持续集成与自动部署
结合Jenkins/GitLab CI/GitHub Actions实现自动化测试、构建、部署。
# .github/workflows/ci.yml
name: Django CI
on: [push, pull_request]
jobs:
build:
runs-on: ubuntu-latest
services:
postgres:
image: postgres:12
ports: [5432:5432]
env:
POSTGRES_DB: testdb
POSTGRES_USER: testuser
POSTGRES_PASSWORD: testpass
steps:
- uses: actions/checkout@v2
- name: Set up Python
uses: actions/setup-python@v2
with:
python-version: 3.9
- name: Install dependencies
run: |
python -m pip install --upgrade pip
pip install -r requirements.txt
- name: Run tests
env:
DATABASE_URL: postgres://testuser:testpass@localhost:5432/testdb
run: |
python manage.py migrate
python manage.py test
解释： 自动化测试和部署保障API质量。
2.9 微服务、异步与分布式
2.9.1 Celery异步任务与API解耦
企业API常需处理异步任务（如邮件、短信、报表），可用Celery解耦。
# tasks.py
from celery import shared_task
@shared_task
def send_order_email(order_id):
"""异步发送订单邮件"""
order = Order.objects.get(pk=order_id)
# 发送邮件逻辑
return True
# views.py
class OrderViewSet(viewsets.ModelViewSet):
@action(detail=True, methods=['post'])
def send_email(self, request, pk=None):
send_order_email.delay(pk) # 异步调用
return Response({
'msg': '邮件发送任务已提交'})
解释： API与耗时任务解耦，提升响应速度。
2.9.2 分布式事务与幂等性
分布式场景下需保证API幂等性和事务一致性。
# models.py
class IdempotencyKey(models.Model):
"""幂等性Key表"""
key = models.CharField(max_length=64, unique=True)
created_at = models.DateTimeField(auto_now_add=True)
# views.py
class PaymentView(APIView):
def post(self, request):
idempotency_key = request.headers.get('Idempotency-Key')
if IdempotencyKey.objects.filter(key=idempotency_key).exists():
return Response({
'msg': '重复请求'}, status=409)
with transaction.atomic():
# 业务逻辑
IdempotencyKey.objects.create(key=idempotency_key)
return Response({
'msg': '支付成功'})
解释： 通过幂等性Key防止重复请求。
2.10 国际化、多语言与多时区
2.10.1 API国际化
企业API需支持多语言响应。
# settings.py
LANGUAGES = [
('en', 'English'),
('zh-hans', '简体中文'),
('ja', '日本語'),
]
USE_I18N = True
# views.py
from django.utils.translation import gettext as _
class HelloWorldView(APIView):
def get(self, request):
return Response({
'msg': _('Hello, world!')})
解释： API响应内容支持多语言。
2.10.2 多时区处理
API需支持多时区，时间统一用UTC存储，前端按需转换。
# settings.py
USE_TZ = True
TIME_ZONE = 'UTC'
解释： 统一用UTC存储时间，避免时区混乱。
2.11 数据治理与合规
2.11.1 数据脱敏与隐私保护
企业API需对敏感数据脱敏，保护用户隐私。
# serializers.py
class UserSerializer(serializers.ModelSerializer):
phone = serializers.SerializerMethodField()
def get_phone(self, obj):
# 手机号脱敏
return obj.phone[:3] + '****' + obj.phone[-4:]
class Meta:
model = User
fields = ['id', 'username', 'phone']
解释： 手机号等敏感信息脱敏输出。
2.11.2 数据访问审计与合规
API需记录数据访问日志，满足合规要求。
# middleware.py
class DataAccessAuditMiddleware:
def __init__(self, get_response):
self.get_response = get_response
def __call__(self, request):
response = self.get_response(request)
if request.path.startswith('/api/'):
# 记录访问日志到合规系统
pass
return response
解释： 记录API访问日志，满足合规要求。
2.12 复杂API网关与服务治理
2.12.1 API网关限流、认证、聚合
企业API常部署在API网关后，统一做限流、认证、聚合。
2.12.2 服务注册与发现
微服务架构下，API服务需支持注册与发现（如Consul、Eureka）。
2.13 团队协作与API治理
2.13.1 API Mock与前后端协作
集成API Mock工具（如YApi、Postman）提升前后端协作效率。
2.13.2 API变更管理与灰度
API变更需有变更管理机制，支持灰度发布。
2.14 领域驱动设计(DDD)与API分层
2.14.1 领域模型与应用服务分层
企业API应分为表示层、应用层、领域层、基础设施层。
2.15 复杂场景下的DRF定制
2.15.1 自定义渲染器与解析器
支持多种数据格式（如CSV、Excel、Protobuf）。
# renderers.py
from rest_framework.renderers import BaseRenderer
class CSVRenderer(BaseRenderer):
media_type = 'text/csv'
format = 'csv'
def render(self, data, media_type=None, renderer_context=None):
# 实现CSV渲染逻辑
pass
# views.py
class ExportView(APIView):
renderer_classes = [CSVRenderer]
def get(self, request):
# 返回CSV格式数据
pass
解释： 支持API导出为CSV等格式。
2.16 事件驱动与API异步推送
2.16.1 Webhook与事件订阅
企业API需支持Webhook，向第三方推送事件。
# models.py
class Webhook(models.Model):
url = models.URLField()
event = models.CharField(max_length=50)
is_active = models.BooleanField(default=True)
# tasks.py
@shared_task
def send_webhook(event, data):
for webhook in Webhook.objects.filter(event=event, is_active=True):
requests.post(webhook.url, json=data)
解释： 事件发生时异步推送Webhook。
2.17 API网关与多协议支持
2.17.1 GraphQL、gRPC等多协议
企业API可同时支持REST、GraphQL、gRPC等多协议。
2.18 复杂数据结构与大数据API
2.18.1 分页优化与游标分页
大数据量场景下，游标分页比传统分页性能更优。
# settings.py
REST_FRAMEWORK = {
'DEFAULT_PAGINATION_CLASS': 'rest_framework.pagination.CursorPagination',
'PAGE_SIZE': 100,
}
解释： 使用游标分页提升大数据量API性能。
2.19 复杂业务规则与校验
2.19.1 复杂校验逻辑
企业API常需跨字段、跨表校验。
# serializers.py
class OrderCreateSerializer(serializers.ModelSerializer):
def validate(self, data):
# 校验库存
for item in data['items']:
if item['quantity'] > item['product'].stock:
raise serializers.ValidationError(f"{
item['product'].name}库存不足")
return data
解释： 跨字段、跨表业务校验。
Django RESTful 架构与 DRF 进阶全链路深度剖析
3. DRF 认证与权限的更深层次探索 (Continued from Advanced Authentication and Permissions)
在之前的章节中，我们已经讨论了 DRF 内建的认证类，如 BasicAuthentication、SessionAuthentication 和 TokenAuthentication，并初步探讨了 JWT。现在，我们将更深入地研究 JWT 的细节、OAuth 2.0 在 DRF 中的应用，以及更复杂的权限控制策略。
3.1 JSON Web Tokens (JWT) 深度解析与企业级实践
JWT (RFC 7519) 是一种紧凑且自包含的方式，用于在各方之间安全地传输信息（声明）。由于其无状态、易于跨域、适合移动端等特性，在现代 Web API 认证中被广泛采用。
3.1.1 JWT 结构详解
一个 JWT 由三部分组成，用点 (.) 分隔：
Header (头部):
通常包含两部分：令牌的类型（即 JWT）和所使用的签名算法（如 HMAC SHA256 或 RSA）。
示例: {"alg": "HS256", "typ": "JWT"}
这部分会进行 Base64Url 编码形成 JWT 的第一部分。
Payload (载荷):
包含声明 (Claims)。声明是关于实体（通常是用户）和附加数据的语句。有三种类型的声明：
Registered Claims (注册声明): 这些是一组预定义的声明，非强制但建议使用，以提供一组有用的、可互操作的声明。例如：
iss (Issuer): 签发者
sub (Subject): 主题（通常是用户ID）
aud (Audience): 受众
exp (Expiration Time): 过期时间戳 (NumericDate)
nbf (Not Before Time): 生效时间戳 (NumericDate)
iat (Issued At Time): 签发时间戳 (NumericDate)
jti (JWT ID): JWT 的唯一标识符
Public Claims (公共声明): 这些可以由使用 JWT 的人随意定义。为避免冲突，应在 IANA JSON Web Token Registry 中定义它们，或将其定义为包含抗冲突命名空间的 URI。
Private Claims (私有声明): 这些是为在同意使用它们的各方之间共享信息而创建的自定义声明，既非注册声明也非公共声明。
示例: {"sub": "1234567890", "name": "John Doe", "admin": true, "iat": 1516239022, "exp": 1516242622}
这部分也会进行 Base64Url 编码形成 JWT 的第二部分。
重要提示: 载荷是 Base64Url 编码的，不是加密的。任何拥有 JWT 的人都可以解码并读取其内容。因此，永远不要在 JWT 载荷中存放敏感信息，除非它也被加密 (JWE – JSON Web Encryption，是另一种规范)。JWT 的安全性在于其签名，用于验证发送者身份和信息未被篡改。
Signature (签名):
要创建签名部分，你必须获取编码后的头部、编码后的载荷、一个密钥（secret），使用头部中指定的算法进行签名，然后对结果进行 Base64Url 编码。
例如，如果使用 HMAC SHA256 算法，签名将通过以下方式创建：
HMACSHA256(base64UrlEncode(header) + "." + base64UrlEncode(payload), secret)
签名用于验证消息在传递过程中没有被更改，并且对于使用私钥签名的令牌，它还可以验证 JWT 的发送者确实是它所声称的身份。
完整的 JWT 示例 (三部分由 . 连接):
eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiIxMjM0NTY3ODkwIiwibmFtZSI6IkpvaG4gRG9lIiwiYWRtaW4iOnRydWUsImlhdCI6MTUxNjIzOTAyMiwiZXhwIjoxNTE2MjM5MDUyfQ.SflKxwRJSMeKKF2QT4fwpMeJf36POk6yJV_adQssw5c
3.1.2 JWT 在 DRF 中的常用库: djangorestframework-simplejwt
djangorestframework-simplejwt 是一个流行的、功能齐全的库，用于在 DRF 中实现 JWT 认证。
安装:
pip install djangorestframework-simplejwt
配置 (settings.py):
# settings.py
INSTALLED_APPS = [
# ...
'rest_framework',
'rest_framework_simplejwt', # 添加 simplejwt
# ...
]
REST_FRAMEWORK = {
'DEFAULT_AUTHENTICATION_CLASSES': [
'rest_framework_simplejwt.authentication.JWTAuthentication', # 设置为默认认证类
# 'rest_framework.authentication.SessionAuthentication', # 可以与其他认证方式并存
],
# ...
}
from datetime import timedelta
SIMPLE_JWT = {
# simplejwt 的详细配置
'ACCESS_TOKEN_LIFETIME': timedelta(minutes=60), # Access Token 的有效期，例如 60 分钟
'REFRESH_TOKEN_LIFETIME': timedelta(days=1), # Refresh Token 的有效期，例如 1 天
'ROTATE_REFRESH_TOKENS': True, # 如果为 True, 每次使用 Refresh Token 获取新的 Access Token 时，也会返回一个新的 Refresh Token (旧的失效或进入黑名单)
'BLACKLIST_AFTER_ROTATION': True, # 如果 ROTATE_REFRESH_TOKENS 为 True, 则将旧的 Refresh Token 加入黑名单
'UPDATE_LAST_LOGIN': True, # 如果为 True, 登录时会更新用户的 last_login 字段
'ALGORITHM': 'HS256', # 签名算法
'SIGNING_KEY': SECRET_KEY, # 用于签名的密钥，通常使用 Django 的 SECRET_KEY
'VERIFYING_KEY': None, # 用于验证签名的公钥 (当使用非对称算法如 RS256 时需要)
'AUDIENCE': None, # 受众
'ISSUER': None, # 签发者
'JWK_URL': None, # JSON Web Key Set URL (用于动态获取公钥)
'LEEWAY': 0, # 允许的时间误差 (单位秒)，用于处理时钟不同步
'AUTH_HEADER_TYPES': ('Bearer',), # 授权头部的类型，例如 'Authorization: Bearer <token>'
'AUTH_HEADER_NAME': 'HTTP_AUTHORIZATION', # Django HttpRequest.META 中授权头部的键名
'USER_ID_FIELD': 'id', # 用户模型中用作 JWT 'sub' 声明的字段
'USER_ID_CLAIM': 'user_id', # JWT 载荷中代表用户 ID 的声明名称 (simplejwt 默认使用 USER_ID_FIELD 的值)
'USER_AUTHENTICATION_RULE': 'rest_framework_simplejwt.authentication.default_user_authentication_rule', # 验证用户是否有效的规则
'AUTH_TOKEN_CLASSES': ('rest_framework_simplejwt.tokens.AccessToken',), # Access Token 的类
'TOKEN_TYPE_CLAIM': 'token_type', # JWT 载荷中代表令牌类型的声明名称 (例如 'access' 或 'refresh')
'TOKEN_USER_CLASS': 'rest_framework_simplejwt.models.TokenUser', # 用于从 Token 中创建的虚拟用户类
'JTI_CLAIM': 'jti', # JWT ID 声明的名称
'SLIDING_TOKEN_REFRESH_EXP_CLAIM': 'refresh_exp', # 滑动刷新 Token 的过期时间声明
'SLIDING_TOKEN_LIFETIME': timedelta(minutes=5), # 原始 Access Token 的有效期 (当 SLIDING_TOKEN_REFRESH_LIFETIME 激活时)
'SLIDING_TOKEN_REFRESH_LIFETIME': timedelta(days=1), # 如果 Access Token 在此有效期内被刷新，则会获得一个新的 Access Token
# SLIDING_TOKEN 通常用于希望用户保持活跃登录状态的场景，但如果一段时间不活跃则需要重新登录
# 自定义 Payload 声明
# 'CLAIMS_NAMESPACE': 'custom_claims', # 可以将自定义声明放到一个命名空间下
}
URL 配置 (urls.py):
simplejwt 提供了几个内置视图用于获取和刷新令牌。
# myproject/urls.py
from django.contrib import admin
from django.urls import path, include
from rest_framework_simplejwt.views import ( # 导入 simplejwt 提供的视图
TokenObtainPairView, # 获取 Access Token 和 Refresh Token 对
TokenRefreshView, # 使用 Refresh Token 获取新的 Access Token
TokenVerifyView, # 验证 Access Token 是否有效
TokenBlacklistView, # 将 Refresh Token 加入黑名单 (用于登出)
)
urlpatterns = [
path('admin/', admin.site.urls),
path('api/auth/token/', TokenObtainPairView.as_view(), name='token_obtain_pair'), # 登录并获取token对
path('api/auth/token/refresh/', TokenRefreshView.as_view(), name='token_refresh'), # 刷新access token
path('api/auth/token/verify/', TokenVerifyView.as_view(), name='token_verify'), # 验证token
path('api/auth/token/blacklist/', TokenBlacklistView.as_view(), name='token_blacklist'), # 将refresh token加入黑名单 (登出)
path('api/v1/', include('myapp.urls')), # 包含你的应用 API URLs
]
TokenObtainPairView:
接收 POST 请求，包含用户名和密码 (通常是用户模型中 USERNAME_FIELD 和 password 字段)。
如果凭证有效，返回一个 JSON 响应，包含 access 和 refresh 两个令牌。
{
"refresh": "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJ0b2t...",
"access": "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJ0b2t..."
}
TokenRefreshView:
接收 POST 请求，包含一个有效的 refresh 令牌。
返回一个新的 access 令牌。
{
"access": "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJ0b2t..."
// 如果 SIMPLE_JWT['ROTATE_REFRESH_TOKENS'] = True，可能还会返回一个新的 refresh token
}
TokenVerifyView:
接收 POST 请求，包含一个 token (通常是 Access Token)。
如果令牌有效，返回 HTTP 200 OK (空响应体)。
如果无效 (例如过期、签名错误)，返回 HTTP 401 Unauthorized。
TokenBlacklistView:
需要 rest_framework_simplejwt.token_blacklist 应用在 INSTALLED_APPS 中，并运行迁移。
# settings.py
INSTALLED_APPS = [
# ...
'rest_framework_simplejwt.token_blacklist', # 添加黑名单应用
# ...
]
# 然后 python manage.py migrate
接收 POST 请求，包含一个要加入黑名单的 refresh 令牌。
成功后返回 HTTP 200 OK。
这使得该 Refresh Token 无法再用于获取新的 Access Token，是实现“登出”的一种方式。
3.1.3 自定义 JWT 载荷 (Customizing JWT Claims)
企业应用中，经常需要在 JWT 载荷中包含额外的用户信息，如用户角色、权限、部门ID等，以便API后端可以快速获取这些信息，而无需每次都查询数据库。
方法1: 重写 TokenObtainPairSerializer
# myapp/serializers.py
from rest_framework_simplejwt.serializers import TokenObtainPairSerializer
from rest_framework_simplejwt.tokens import RefreshToken # 导入 RefreshToken 类
class MyTokenObtainPairSerializer(TokenObtainPairSerializer): # 继承默认的 Serializer
@classmethod
def get_token(cls, user): # 重写 get_token 方法
token = super().get_token(user) # 调用父类方法获取基础 token
# 添加自定义声明
token['username'] = user.username # 添加用户名
token['email'] = user.email # 添加邮箱
# 假设用户模型有 'roles' ManyToManyField 或类似属性
# token['roles'] = list(user.roles.values_list('name', flat=True)) if hasattr(user, 'roles') else []
# 假设用户模型有 'department_id'
# token['department_id'] = user.department_id if hasattr(user, 'department_id') else None
# 示例：添加用户是否为超级用户的信息
token['is_superuser'] = user.is_superuser
# 示例：添加用户所属的用户组
groups = user.groups.all().values_list('name', flat=True)
token['groups'] = list(groups)
# 你可以添加任何你需要的、且不敏感的信息
# 避免在 Token 中存放过多数据，以保持其紧凑性
return token
# 可选: 如果你还想修改返回给客户端的 Token 响应体结构 (除了 access 和 refresh 之外还想加东西)
# def validate(self, attrs):
# data = super().validate(attrs) # 调用父类的 validate 获取原始的 access 和 refresh token
# # refresh = self.get_token(self.user) # 这会生成带有自定义声明的 Refresh Token
# # data['refresh'] = str(refresh)
# # data['access'] = str(refresh.access_token) # 从 Refresh Token 生成 Access Token
# # 添加额外信息到响应中 (与 Token 载荷不同，这是在获取 Token 接口的响应体中)
# data['user_id'] = self.user.id
# data['username'] = self.user.username
# # data['full_name'] = self.user.get_full_name()
# return data
# myapp/views.py (或直接在项目 urls.py 中配置)
from rest_framework_simplejwt.views import TokenObtainPairView
# from .serializers import MyTokenObtainPairSerializer # 导入自定义的 Serializer
class MyTokenObtainPairView(TokenObtainPairView): # 创建一个视图类，使用自定义的 Serializer
serializer_class = MyTokenObtainPairSerializer
# myproject/urls.py (使用自定义的视图)
# from myapp.views import MyTokenObtainPairView
# path('api/auth/token/', MyTokenObtainPairView.as_view(), name='token_obtain_pair_custom'),
现在，当用户通过 MyTokenObtainPairView 获取令牌时，生成的 Access Token 和 Refresh Token (如果也通过 get_token 生成) 的载荷中会包含这些自定义声明。
方法2: 使用 USER_ID_CLAIM 和 TOKEN_TYPE_CLAIM (简单场景)
如果只是想改变默认的 user_id 和 token_type 声明名称，可以在 SIMPLE_JWT 配置中修改。
方法3: 使用信号 (不推荐用于修改载荷，更适合 Token 创建/吊销时的额外操作)
虽然可以监听 token_obtained 信号，但不适合直接修改已生成的令牌载荷。
访问自定义声明:
在受保护的视图中，当 JWTAuthentication 成功认证后，解码后的 Token 载荷可以通过 request.auth.payload (如果 request.auth 是 Token 对象) 或直接通过解析 request.META['HTTP_AUTHORIZATION'] 中的 Token (不推荐手动解析) 来访问。
更常见的方式是，JWTAuthentication 会将 user_id (或 USER_ID_CLAIM 指定的字段) 从载荷中取出，并用它来获取 User 对象，赋值给 request.user。如果你需要在 request 对象上方便地访问其他自定义载荷信息，可以考虑创建一个自定义的认证后端或中间件来附加它们。
# myapp/auth_backends.py
from rest_framework_simplejwt.authentication import JWTAuthentication
from rest_framework_simplejwt.tokens import UntypedToken # 用于解码未验证类型的 token
from rest_framework_simplejwt.exceptions import InvalidToken, TokenError
from django.conf import settings
from django.contrib.auth import get_user_model
User = get_user_model()
class CustomJWTAuthentication(JWTAuthentication): # 继承 JWTAuthentication
def get_user(self, validated_token):
"""
尝试从给定的已验证 token 中获取用户对象。
并可以将 token payload 中的自定义声明附加到 request 对象上。
"""
try:
user_id = validated_token[settings.SIMPLE_JWT['USER_ID_CLAIM']] # 从 token 中获取用户 ID
except KeyError:
raise InvalidToken('Token contained no recognizable user identification')
try:
user = User.objects.get(**{
settings.SIMPLE_JWT['USER_ID_FIELD']: user_id})
except User.DoesNotExist:
raise AuthenticationFailed('User not found', code='user_not_found')
if not user.is_active:
raise AuthenticationFailed('User is inactive', code='user_inactive')
# --- 附加自定义声明到 request (需要 request 对象) ---
# 注意: get_user 方法本身不直接访问 request，但 authenticate 方法可以
# 这里我们仅演示如何从 token 中获取，实际附加 request 的操作应在 authenticate 中进行
# 假设我们有一个 request 对象 (在实际的 authenticate 方法中是可用的)
# request.custom_claims = {}
# for claim_name, claim_value in validated_token.payload.items():
# if claim_name not in [settings.SIMPLE_JWT['USER_ID_CLAIM'],
# settings.SIMPLE_JWT['TOKEN_TYPE_CLAIM'],
# 'exp', 'iat', 'jti']: # 排除标准声明
# request.custom_claims[claim_name] = claim_value
return user
def authenticate(self, request): # 重写 authenticate 方法
header = self.get_header(request) # 获取 Authorization 头部
if header is None:
return None # 如果没有头部，不处理
raw_token = self.get_raw_token(header) # 从头部提取原始 token 字符串
if raw_token is None:
return None # 如果没有 token，不处理
validated_token = self.get_validated_token(raw_token) # 验证 token (签名、过期等)
# 验证通过后，获取用户
user = self.get_user(validated_token)
# --- 在这里附加自定义声明到 request ---
if user is not None:
request.custom_jwt_claims = {
} # 创建一个属性来存放自定义声明
# 例如，从 validated_token (它是 UntypedToken 的实例，可以访问 payload)
# 或者直接从解码后的 payload (validated_token 就是解码后的)
for claim_name, claim_value in validated_token.payload.items():
# 避免重复存储标准声明或敏感信息，按需选择
if claim_name not in ['user_id', 'token_type', 'exp', 'iat', 'jti', 'password']:
request.custom_jwt_claims[claim_name] = claim_value
# 将解码后的 token (payload) 赋值给 request.auth
# simplejwt 默认的 JWTAuthentication 返回的是 (user, validated_token)
# validated_token 就是一个包含 payload 的 Token 对象
return (user, validated_token)
return None
# settings.py (使用自定义的认证类)
# REST_FRAMEWORK = {
# 'DEFAULT_AUTHENTICATION_CLASSES': [
# 'myapp.auth_backends.CustomJWTAuthentication',
# ],
# }
# 在视图中访问:
# class MyProtectedView(APIView):
# authentication_classes = [CustomJWTAuthentication]
# permission_classes = [IsAuthenticated]
#
# def get(self, request):
# user_roles = getattr(request, 'custom_jwt_claims', {}).get('roles', [])
# department_id = getattr(request, 'custom_jwt_claims', {}).get('department_id')
# # ... 使用 user_roles 和 department_id ...
# return Response({'message': f'Hello {request.user.username}, your roles: {user_roles}'})
注意: CustomJWTAuthentication 的实现需要小心处理 validated_token 的类型和 payload 的访问。simplejwt 的 JWTAuthentication 在成功验证后返回的 auth (即 request.auth) 就是一个 Token 类的实例 (如 AccessToken 或 RefreshToken)，它有一个 payload 属性可以直接访问解码后的声明。所以通常不需要像上面那样手动在 authenticate 中再次解析并附加到 request，可以直接在视图中通过 request.auth.payload.get('custom_claim_name') 来访问。上面的示例更多是展示如果需要深度定制时的思路。
更简洁的访问方式是在视图中直接从 request.auth 获取：
# myapp/views.py
from rest_framework.views import APIView
from rest_framework.response import Response
from rest_framework.permissions import IsAuthenticated
# 假设 DEFAULT_AUTHENTICATION_CLASSES 已设置为 JWTAuthentication 或其子类
class UserProfileView(APIView):
permission_classes = [IsAuthenticated] # 确保用户已通过 JWT 认证
def get(self, request):
user = request.user # 当前认证的用户对象
# request.auth 是认证成功后设置的 Token 对象 (例如 AccessToken 实例)
# 它有一个 payload 属性，包含了 JWT 的所有声明
jwt_payload = request.auth.payload if hasattr(request.auth, 'payload') else {
}
# 从 payload 中获取自定义声明
username_from_token = jwt_payload.get('username') # 我们在 MyTokenObtainPairSerializer 中添加的
email_from_token = jwt_payload.get('email')
is_superuser_from_token = jwt_payload.get('is_superuser')
groups_from_token = jwt_payload.get('groups', [])
# 准备响应数据
data = {
'db_user_id': user.id,
'db_username': user.username,
'token_user_id': jwt_payload.get(settings.SIMPLE_JWT['USER_ID_CLAIM']), # 'user_id'
'token_username': username_from_token,
'token_email': email_from_token,
'token_is_superuser': is_superuser_from_token,
'token_groups': groups_from_token,
'token_issued_at': jwt_payload.get('iat'),
'token_expires_at': jwt_payload.get('exp'),
'token_type': jwt_payload.get(settings.SIMPLE_JWT['TOKEN_TYPE_CLAIM']), # 'token_type'
}
return Response(data)
这个视图展示了如何在用户通过 JWT 认证后，从 request.auth.payload 中读取包括自定义声明在内的所有 JWT 载荷信息。
3.1.4 JWT 的安全性考量
HTTPS: 必须始终通过 HTTPS 传输 JWT。因为载荷是可读的，签名只能防止篡改，不能防止窃听。
密钥安全 (SIGNING_KEY):
对于对称算法 (如 HS256)，密钥必须保密。泄露密钥意味着任何人都可以伪造有效的 JWT。
对于非对称算法 (如 RS256)，私钥必须保密，公钥可以公开。
SECRET_KEY 应足够复杂和随机。
算法选择:
避免使用 alg: "none" (虽然 simplejwt 默认不允许)。
HS256 (HMAC with SHA-256) 是常用的对称算法。
RS256 (RSA Signature with SHA-256) 是常用的非对称算法，更适合需要第三方验证签名的场景 (签发方用私钥，验证方用公钥)。
令牌过期 (exp声明):
Access Token 的有效期应较短 (例如几分钟到几小时)，以减少泄露后的风险。
Refresh Token 的有效期可以较长 (例如几天到几周)，用于获取新的 Access Token。
令牌吊销/黑名单:
JWT 本身是无状态的，一旦签发，在过期前都有效 (除非密钥更改)。
对于需要立即吊销令牌的场景 (例如用户修改密码、账户被盗、用户登出)，需要实现黑名单机制。simplejwt 的 TokenBlacklistView 和 OutstandingToken 模型提供了这种支持。
黑名单会增加状态性，需要在性能和安全性之间权衡。
jti (JWT ID): 可以为每个 JWT 生成唯一 ID，用于防止重放攻击或实现更精细的黑名单。
aud (Audience) 和 iss (Issuer): 用于指定令牌的预期接收者和签发者，增加安全性，防止令牌被误用。
传输方式:
通常通过 HTTP Authorization: Bearer <token> 头部传输。
避免将 JWT 放在 URL 参数中，因为 URL 可能会被记录在服务器日志或浏览器历史中。
如果使用 Cookie 存储 JWT (例如用于 SSR 或某些特定场景)，需要设置 HttpOnly 和 Secure 标志，并考虑 CSRF 防护。但通常 JWT 的优势之一就是避免 Cookie 依赖。
第三方库的安全性: 确保使用的 JWT 库是维护良好且没有已知漏洞的。
3.1.5 JWT 与其他认证方式的比较
| 特性 | Session-Cookie (有状态) | DRF Token (简单令牌) | JWT (无状态令牌) | OAuth 2.0 (授权框架) |
|---|---|---|---|---|
| 状态性 | 有状态 (服务器存储会话) | 半有状态 (DB存Token) | 无状态 (自包含信息) | 通常无状态 (Access Token) |
| 服务器开销 | 较高 (会话存储、查询) | 中等 (Token 查询) | 较低 (仅验证签名) | 较低 (Token 验证) |
| 可扩展性 | 较差 (需共享会话) | 中等 (DB 可扩展) | 好 (无需共享状态) | 好 |
| CSRF防护 | 需要 | 不需要 | 不需要 (若不存Cookie) | 不需要 (若不存Cookie) |
| 跨域/多端 | 复杂 (Cookie 跨域问题) | 较好 | 非常好 | 非常好 |
| 令牌过期 | 会话超时 | 不直接支持 (需自定义) | 支持 (exp 声明) |
支持 |
| 令牌刷新 | – | 不直接支持 (需自定义) | 支持 (Refresh Token) | 支持 (Refresh Token) |
| 标准载荷 | – | – | 是 (Registered Claims) | – (Token内容由AS定义) |
| 细粒度权限 | 依赖后端实现 | 依赖后端实现 | 可在载荷中携带 | 可通过 Scope 实现 |
| 安全性 | 中 (依赖CSRF, Cookie安全) | 中 (Token 泄露风险) | 中高 (依赖密钥, HTTPS) | 高 (复杂但安全) |
| 复杂度 | 低 | 低 | 中等 | 高 |
选择 JWT 的时机:
单页应用 (SPA) 和移动应用 (iOS, Android) 的后端 API。
无状态、可水平扩展的微服务架构。
需要跨域认证的场景。
希望在令牌中携带少量非敏感用户信息的场景。
不希望依赖 Cookie 的场景。
3.2 OAuth 2.0 与 DRF 集成 (概念与方向)
OAuth 2.0 (RFC 6749) 是一个行业标准的授权框架，允许第三方应用代表用户访问其在某个服务提供商上的受保护资源，而无需将用户的凭证（用户名/密码）直接暴露给第三方应用。
核心角色:
Resource Owner (资源所有者): 通常是最终用户，能够授权访问其受保护资源。
Resource Server (资源服务器): 托管受保护资源的服务器 (例如你的 DRF API)。
Client (客户端): 代表资源所有者请求访问受保护资源的第三方应用。
Authorization Server (授权服务器): 负责验证资源所有者的身份，并在获得授权后向客户端颁发访问令牌 (Access Token)。
OAuth 2.0 不是认证协议，而是授权协议。 它通常与 OpenID Connect (OIDC) 结合使用来实现认证。OIDC 构建在 OAuth 2.0 之上，添加了身份层，允许客户端验证最终用户的身份并获取基本的配置文件信息。
常见的 OAuth 2.0 授权流程 (Grant Types):
Authorization Code Grant (授权码模式):
最常用且最安全的流程，适用于传统的 Web 应用 (服务器端应用)。
流程：客户端将用户重定向到授权服务器 -> 用户登录并授权 -> 授权服务器将用户重定向回客户端，并附带一个授权码 (Authorization Code) -> 客户端使用授权码和自己的凭证向授权服务器请求访问令牌。
Implicit Grant (隐式模式): (已不推荐，被 Authorization Code with PKCE 取代)
适用于纯前端应用 (如 SPA)，直接在重定向 URI 中返回访问令牌。
安全性较低，因为访问令牌直接暴露在浏览器中。
Resource Owner Password Credentials Grant (密码模式):
客户端直接使用资源所有者的用户名和密码向授权服务器请求访问令牌。
仅当客户端是高度受信任的 (例如服务提供商自己开发的第一方应用) 时才应使用。
不推荐用于第三方客户端。
Client Credentials Grant (客户端凭证模式):
用于机器对机器 (M2M) 的通信，客户端以自己的名义（而不是代表用户）请求访问受保护资源。
客户端使用其客户端 ID 和客户端密钥向授权服务器请求访问令牌。
Device Authorization Grant (设备授权流程):
用于输入受限的设备 (如智能电视、游戏机)。
Refresh Token Grant (刷新令牌模式):
用于使用长期有效的刷新令牌获取新的、短期有效的访问令牌。
PKCE (Proof Key for Code Exchange):
PKCE (RFC 7636) 是对授权码模式的扩展，旨在防止授权码拦截攻击，特别适用于移动应用和 SPA。
在 DRF 中实现 OAuth 2.0:
DRF 本身不包含完整的 OAuth 2.0 实现。通常需要借助第三方库：
django-oauth-toolkit (DOT):
一个非常流行且功能全面的库，可以帮助你在 Django 应用中同时实现 OAuth 2.0 提供商 (Authorization Server 和 Resource Server) 的功能。
支持多种授权流程、令牌管理、范围 (Scopes)、客户端管理等。
与 DRF 集成良好，可以保护 DRF API 端点。
# settings.py
INSTALLED_APPS = [
# ...
'oauth2_provider', # django-oauth-toolkit
'rest_framework',
# ...
]
REST_FRAMEWORK = {
'DEFAULT_AUTHENTICATION_CLASSES': [
'oauth2_provider.contrib.rest_framework.OAuth2Authentication', # 用于保护资源服务器 API
# 'rest_framework.authentication.SessionAuthentication',
],
'DEFAULT_PERMISSION_CLASSES': [
'rest_framework.permissions.IsAuthenticated',
# 'oauth2_provider.contrib.rest_framework.TokenHasReadWriteScope', # 示例：基于 Scope 的权限
]
}
OAUTH2_PROVIDER = {
# DOT 的配置
# 'SCOPES': {'read': 'Read scope', 'write': 'Write scope', 'groups': 'Access to your groups'},
# ... 其他 DOT 配置 ...
}
# urls.py (项目级)
# urlpatterns += [
# path('o/', include('oauth2_provider.urls', namespace='oauth2_provider')), # 包含 DOT 提供的 URL (例如 /o/authorize/, /o/token/)
# ]
使用 DOT，你的 Django/DRF 应用可以成为一个 OAuth 2.0 授权服务器。客户端（例如你的前端 SPA、移动 App 或第三方应用）将通过 OAuth 流程从你的授权服务器获取访问令牌，然后使用该令牌访问你的 DRF API（资源服务器）。
集成第三方 OAuth 2.0 / OIDC 提供商 (如 Auth0, Okta, Keycloak, Google, Facebook):
如果你的应用需要允许用户通过“使用 Google/Facebook 登录”等方式认证，或者企业内部已有统一的身份认证平台 (IDP)。
这种情况下，你的 DRF API 通常扮演资源服务器的角色。认证由外部 IDP 处理。
你需要配置 DRF 来验证从外部 IDP 获取的访问令牌 (通常是 JWT)。这可能需要自定义认证后端，或者使用特定于提供商的库。
例如，如果外部 IDP 签发 JWT，你可以配置 JWTAuthentication 使用该 IDP 的公钥 (通过 JWK_URL 或直接配置 VERIFYING_KEY) 来验证令牌签名和声明 (如 iss, aud)。
企业级场景:
对外开放 API 平台: 允许第三方开发者构建应用来访问你平台上的用户数据（在用户授权的前提下）。
单点登录 (SSO): 企业内部多个应用可以使用 OAuth 2.0 / OIDC 实现单点登录，用户只需登录一次即可访问所有授权应用。DRF API 可以接受来自中央 IDP 的令牌。
移动应用和 SPA 的安全认证: 使用授权码 + PKCE 流程是最推荐的方式。
OAuth 2.0 是一个复杂但强大的框架。在 DRF 中集成它通常意味着将其配置为 OAuth 2.0 资源服务器，验证由授权服务器（可能是应用本身，也可能是外部的）颁发的访问令牌。
3.3 更高级的权限控制策略
DRF 的权限系统是可插拔的，允许你实现非常精细的访问控制。
3.3.1 对象级权限 (has_object_permission) 的深入应用
我们之前提过 BasePermission 的 has_permission (视图级) 和 has_object_permission (对象实例级) 方法。在企业应用中，对象级权限至关重要。
示例：实现一个项目管理系统的权限
假设有 Project 和 Task 模型。
用户只能查看他们作为成员参与的项目。
用户只能编辑/删除他们自己创建的任务，或者他们是项目管理员才能编辑/删除项目内的任何任务。
# myapp/models.py
# from django.db import models
# from django.conf import settings
#
# class Project(models.Model):
# name = models.CharField(max_length=100)
# owner = models.ForeignKey(settings.AUTH_USER_MODEL, related_name='owned_projects', on_delete=models.CASCADE)
# members = models.ManyToManyField(settings.AUTH_USER_MODEL, related_name='projects_joined', blank=True)
# # ...
#
# class Task(models.Model):
# project = models.ForeignKey(Project, related_name='tasks', on_delete=models.CASCADE)
# title = models.CharField(max_length=200)
# assigned_to = models.ForeignKey(settings.AUTH_USER_MODEL, null=True, blank=True, related_name='assigned_tasks', on_delete=models.SET_NULL)
# created_by = models.ForeignKey(settings.AUTH_USER_MODEL, related_name='created_tasks', on_delete=models.CASCADE)
# # ...
# myapp/permissions.py
from rest_framework import permissions
class IsProjectMember(permissions.BasePermission):
"""
允许访问 (查看) 项目，如果用户是项目的成员或所有者。
"""
def has_object_permission(self, request, view, obj): # obj 是 Project 实例
# 允许 GET, HEAD, OPTIONS 请求 (通常认为是安全方法)
if request.method in permissions.SAFE_METHODS:
return obj.owner == request.user or request.user in obj.members.all()
# 对于其他方法 (PUT, PATCH, DELETE)，可以更严格，例如只允许所有者
return obj.owner == request.user # 示例：只有所有者可以修改项目本身
class IsTaskCreatorOrProjectOwner(permissions.BasePermission):
"""
允许修改/删除任务，如果用户是任务的创建者或任务所属项目的所有者。
查看任务的权限可能由 IsProjectMember 控制 (通过视图的 queryset 过滤)。
"""
def has_object_permission(self, request, view, obj): # obj 是 Task 实例
# 允许所有者或项目管理员进行非安全操作
return obj.created_by == request.user or obj.project.owner == request.user
# 对于查看 (GET)，如果任务列表已按项目成员过滤，则可能不需要额外的对象级权限
# if request.method in permissions.SAFE_METHODS:
# return True # 假设视图 queryset 已处理查看权限
# return obj.created_by == request.user or obj.project.owner == request.user
# myapp/views.py
# from rest_framework import viewsets, permissions
# from .models import Project, Task
# from .serializers import ProjectSerializer, TaskSerializer
# from .permissions import IsProjectMember, IsTaskCreatorOrProjectOwner
# class ProjectViewSet(viewsets.ModelViewSet):
# serializer_class = ProjectSerializer
# permission_classes = [permissions.IsAuthenticated, IsProjectMember] # 先认证，再检查是否项目成员/所有者
# def get_queryset(self):
# # 用户只能看到他们是成员或所有者的项目
# user = self.request.user
# return Project.objects.filter(
# models.Q(owner=user) | models.Q(members=user)
# ).distinct()
# def perform_create(self, serializer):
# serializer.save(owner=self.request.user) # 创建项目时，所有者是当前用户
# class TaskViewSet(viewsets.ModelViewSet):
# serializer_class = TaskSerializer
# # 注意权限顺序和逻辑
# # IsProjectMember 可以在视图级别检查用户是否有权访问该项目下的任务 (一般通过 queryset 实现)
# # IsTaskCreatorOrProjectOwner 在对象级别检查修改/删除权限
# permission_classes = [permissions.IsAuthenticated, IsTaskCreatorOrProjectOwner]
# def get_queryset(self):
# # 用户只能看到他们有权访问的项目的任务
# user = self.request.user
# # 获取用户有权访问的项目 ID 列表
# allowed_project_ids = Project.objects.filter(
# models.Q(owner=user) | models.Q(members=user)
# ).values_list('id', flat=True)
# # 如果是通过嵌套路由 /projects/{project_pk}/tasks/ 访问
# project_pk = self.kwargs.get('project_pk')
# if project_pk:
# # 确保用户有权访问这个特定的 project_pk
# if not Project.objects.filter(
# pk=project_pk,
# owner=user # 或者更宽松的成员检查，取决于需求
# ).exists() and not Project.objects.filter(pk=project_pk, members=user).exists():
# # 如果项目不存在或用户无权访问，返回空 queryset 或抛出 PermissionDenied
# return Task.objects.none()
# return Task.objects.filter(project_id=project_pk)
# else:
# # 如果是顶级 /tasks/ 端点，则返回所有用户有权访问的项目的任务
# return Task.objects.filter(project_id__in=allowed_project_ids)
# def perform_create(self, serializer):
# # 创建任务时，创建者是当前用户
# # 需要确保 project 字段在 validated_data 中，并且用户有权在该项目下创建任务
# project_id = serializer.validated_data.get('project').id # 假设 project 是一个 Project 实例或 ID
# if not Project.objects.filter(
# pk=project_id,
# # 允许在自己拥有或作为成员的项目中创建任务
# (models.Q(owner=self.request.user) | models.Q(members=self.request.user))
# ).exists():
# raise permissions.PermissionDenied("You do not have permission to create tasks in this project.")
# serializer.save(created_by=self.request.user)
这个例子展示了如何结合视图级的 queryset 过滤和对象级的 has_object_permission 来实现复杂的权限逻辑。get_queryset 确保用户在列表和检索时只能看到他们有权访问的对象。has_object_permission 则在进行更新或删除等操作时对单个对象实例进行更细致的检查。
3.3.2 基于角色的访问控制 (RBAC)
RBAC 是企业应用中非常常见的权限模型。用户被分配到一个或多个角色，角色拥有一组权限。
实现思路:
定义角色和权限模型:
# myapp/models.py
# class Permission(models.Model):
# name = models.CharField(max_length=100, unique=True) # e.g., 'view_project', 'edit_task'
# codename = models.CharField(max_length=100, unique=True) # e.g., 'can_view_project'
#
# class Role(models.Model):
# name = models.CharField(max_length=100, unique=True) # e.g., 'Project Manager', 'Developer', 'Viewer'
# permissions = models.ManyToManyField(Permission, blank=True)
#
# # 扩展 User 模型或使用 Profile 模型来关联角色
# class UserProfile(models.Model):
# user = models.OneToOneField(settings.AUTH_USER_MODEL, on_delete=models.CASCADE, related_name='profile')
# roles = models.ManyToManyField(Role, blank=True)
创建自定义权限类 HasRolePermission:
# myapp/permissions.py
# class HasRolePermission(permissions.BasePermission):
# required_permissions = [] # 在视图中定义此权限类所需的权限 codename 列表
#
# def has_permission(self, request, view):
# if not request.user or not request.user.is_authenticated:
# return False
#
# try:
# user_profile = request.user.profile
# except UserProfile.DoesNotExist: # 假设 UserProfile 可能不存在
# return False
#
# user_permissions = set()
# for role in user_profile.roles.all():
# for perm in role.permissions.all():
# user_permissions.add(perm.codename)
#
# # 检查用户是否拥有所有必需的权限
# # 可以是 AND (拥有全部) 或 OR (拥有任一) 逻辑，取决于需求
# # 示例为 AND 逻辑：
# required_perms_set = set(getattr(view, 'required_permissions', [])) # 从视图获取 required_permissions
# if not required_perms_set: # 如果视图没有定义 required_permissions，则默认允许 (或拒绝，取决于策略)
# return True
#
# return required_perms_set.issubset(user_permissions)
#
# # has_object_permission 也可以类似实现，检查角色对特定对象的权限
在视图中使用:
# myapp/views.py
# class SensitiveDataViewSet(viewsets.ViewSet):
# permission_classes = [permissions.IsAuthenticated, HasRolePermission]
# required_permissions = ['view_sensitive_data', 'edit_sensitive_data'] # 定义此视图所需的权限
#
# def list(self, request):
# # 只有拥有 'view_sensitive_data' 权限的用户可以访问
# # (HasRolePermission 会检查 required_permissions 中的第一个，如果只需要一个)
# # 如果需要更细致，可以在 HasRolePermission 中修改逻辑或创建不同的权限类
# return Response(...)
#
# def create(self, request):
# # 假设创建操作也需要 'edit_sensitive_data' (或 'create_sensitive_data')
# # HasRolePermission 已经检查了，如果用户没有所有 required_permissions，则无法进入此方法
# return Response(...)
更复杂的 RBAC 可能需要更细致的权限定义 (例如，对模型字段的读/写权限) 和更灵活的权限检查逻辑。
3.3.3 使用第三方权限库
对于非常复杂的权限需求，可以考虑使用专门的权限库：
django-guardian: 提供对象级权限管理，允许将权限直接授予用户或用户组针对特定的对象实例。它与 Django 内建的权限系统集成。
# pip install django-guardian
# INSTALLED_APPS += ['guardian']
# AUTHENTICATION_BACKENDS = (
# 'django.contrib.auth.backends.ModelBackend', # Django's default
# 'guardian.backends.ObjectPermissionBackend', # Guardian's backend
# )
# # 授予权限
# from guardian.shortcuts import assign_perm
# project_instance = Project.objects.get(pk=1)
# user_instance = User.objects.get(username='developer')
# assign_perm('change_project', user_instance, project_instance) # 'change_project' 是 'app_label.change_modelname' 格式
# # 检查权限
# user_instance.has_perm('change_project', project_instance) # True
# DRF 权限类
# from rest_framework_guardian.permissions import DjangoObjectPermissions
# class ProjectViewSet(viewsets.ModelViewSet):
# permission_classes = [DjangoObjectPermissions] # 使用 guardian 的对象权限
# queryset = Project.objects.all()
# serializer_class = ProjectSerializer
DjangoObjectPermissions 会检查用户是否对特定对象实例拥有 view_model, add_model, change_model, delete_model 等权限。
django-rules: 一个轻量级但功能强大的应用，用于编写可重用的、声明式的权限规则。
# pip install django-rules
# INSTALLED_APPS += ['rules']
# AUTHENTICATION_BACKENDS = (
# 'rules.permissions.ObjectPermissionBackend', # rules backend
# 'django.contrib.auth.backends.ModelBackend',
# )
# # myapp/rules.py
# import rules
#
# @rules.predicate
# def is_project_owner(user, project):
# if not project: return False
# return project.owner == user
#
# @rules.predicate
# def is_project_member(user, project):
# if not project: return False
# return user in project.members.all() or is_project_owner(user, project)
#
# rules.add_perm('myapp.view_project', is_project_member)
# rules.add_perm('myapp.change_project', is_project_owner)
# # DRF 权限类
# from rules.contrib.rest_framework import AutoPermissionViewSetMixin
# class ProjectViewSet(AutoPermissionViewSetMixin, viewsets.ModelViewSet): # Mixin 自动应用权限
# queryset = Project.objects.all()
# serializer_class = ProjectSerializer
# # AutoPermissionViewSetMixin 会根据动作 (list, retrieve, create, update, destroy)
# # 自动检查 'app_label.view_model', 'app_label.add_model' 等权限
Casbin (with django-casbin-adapter): 用于实现更通用的访问控制模型，如 ABAC (Attribute-Based Access Control)。
权限策略存储在外部（通常是文件或数据库）。
模型定义 (PERM model): [request_definition], [policy_definition], [role_definition], [policy_effect], [matchers]
策略规则 (Policy rules): p, subject, object, action (例如 p, alice, /api/data1, read)
# pip install casbin casbin-django-adapter
# settings.py
# CASBIN_MODEL = os.path.join(BASE_DIR, 'casbin_model.conf') # Casbin 模型文件路径
# CASBIN_POLICY_CSV = os.path.join(BASE_DIR, 'casbin_policy.csv') # 策略文件路径 (可选)
# INSTALLED_APPS += ['casbin_adapter'] # (或者你使用的具体 adapter)
# # myapp/permissions.py
# import casbin
# from django.conf import settings
#
# class CasbinPermission(permissions.BasePermission):
# def get_enforcer(self):
# # 获取或初始化 Enforcer 实例
# # 这可能需要从 settings 加载模型和适配器 (用于策略存储)
# # from casbin_adapter.adapter import Adapter # 示例
# # adapter = Adapter()
# # return casbin.Enforcer(settings.CASBIN_MODEL, adapter)
# # 简单的文件适配器示例：
# return casbin.Enforcer(settings.CASBIN_MODEL, settings.CASBIN_POLICY_CSV)
#
# def has_permission(self, request, view):
# enforcer = self.get_enforcer()
# sub = str(request.user) if request.user.is_authenticated else 'anonymous'
# obj = request.path # 或者更细粒度的资源标识
# act = request.method.lower()
#
# return enforcer.enforce(sub, obj, act)
#
# def has_object_permission(self, request, view, obj):
# # 对于对象级权限，obj 可以是模型实例的标识符
# enforcer = self.get_enforcer()
# sub = str(request.user) if request.user.is_authenticated else 'anonymous'
# # obj_identifier = f"{obj._meta.app_label}.{obj._meta.model_name}:{obj.pk}" # 构建唯一对象标识
# obj_identifier = str(obj) # 简单示例
# act = view.action if hasattr(view, 'action') else request.method.lower() # 获取 viewset action 或 HTTP method
#
# return enforcer.enforce(sub, obj_identifier, act)
# # casbin_model.conf (示例 ABAC 模型)
# [request_definition]
# r = sub, obj, act
#
# [policy_definition]
# p = sub, obj, act, eft
#
# [role_definition]
# g = _, _
# g2 = _, _
#
# [policy_effect]
# e = some(where (p.eft == allow)) && !some(where (p.eft == deny))
#
# [matchers]
# m = (g(r.sub, p.sub) || keyMatch(r.sub, p.sub)) &&
# (g2(r.obj, p.obj) || keyMatch(r.obj, p.obj) || globMatch(r.obj, p.obj)) &&
# (keyMatch(r.act, p.act) || regexMatch(r.act, p.act))
# # casbin_policy.csv (示例策略)
# p, role:admin, /*, (GET)|(POST)|(PUT)|(DELETE), allow # 管理员对所有路径所有方法允许
# p, role:editor, /api/articles/*, (GET)|(POST)|(PUT), allow # 编辑者对文章资源有读写权限
# p, anonymous, /api/public/*, GET, allow # 匿名用户对公共资源有读权限
# g, alice, role:admin # alice 是管理员
# g, bob, role:editor # bob 是编辑者
Casbin 提供了极大的灵活性，但学习曲线也较陡峭。
选择哪种权限方案取决于项目的复杂性、团队熟悉度以及对灵活性和性能的要求。对于许多企业级应用，结合 DRF 内建机制和 django-guardian 或 django-rules 可以满足大部分需求。对于需要高度动态和外部化策略管理的系统，Casbin 是一个强大的选项。
暂无评论内容