Django REST Framework 认证机制与源码深度剖析
前置核心概念
在深入剖析 Django REST Framework (DRF) 的认证机制之前,必须掌握两个基础概念,因为 DRF 的底层设计高度依赖于它们:
- 面向对象编程 (OOP) 特性:将相关行为封装为类的方法,将状态数据封装为对象的属性。
- 基于类的视图 (CBV):利用 Python 的反射机制,根据 HTTP 请求方法(GET、POST 等)自动路由并执行类中对应的方法。其核心流程为:URL 路由 ->
view()函数 ->dispatch()方法 -> 反射调用具体请求方法。
基础项目搭建
依赖安装与配置
通过 pip 安装 DRF 并在 Django 的 settings.py 中注册应用:
pip install djangorestframework
INSTALLED_APPS = [
# ... 其他应用
'rest_framework',
]
数据模型设计
我们需要两个模型:一个用于存储用户基础信息,另一个用于管理用户登录后生成的身份凭证(Token)。
from django.db import models
class Account(models.Model):
ROLE_CHOICES = (
(1, 'Regular'),
(2, 'Premium'),
(3, 'Enterprise')
)
role = models.IntegerField(choices=ROLE_CHOICES)
username = models.CharField(max_length=50, unique=True)
password_hash = models.CharField(max_length=128)
class AuthToken(models.Model):
account = models.OneToOneField(Account, on_delete=models.CASCADE, related_name='auth_token')
token_key = models.CharField(max_length=128, unique=True)
登录接口实现
实现一个生成 Token 并处理用户登录的视图。此处使用 SHA-256 结合时间戳来生成唯一的凭证字符串。
import hashlib
import time
from django.http import JsonResponse
from rest_framework.views import APIView
from .models import Account, AuthToken
def generate_token_hash(username):
timestamp = str(time.time())
raw_string = f"{username}{timestamp}"
return hashlib.sha256(raw_string.encode('utf-8')).hexdigest()
class LoginAPIView(APIView):
# 登录接口本身不需要身份验证
authentication_classes = []
def post(self, request, *args, **kwargs):
response_data = {'status': 200, 'message': 'Success', 'data': None}
try:
uname = request.data.get('username')
pwd = request.data.get('password')
account = Account.objects.filter(username=uname, password_hash=pwd).first()
if not account:
response_data['status'] = 401
response_data['message'] = 'Invalid credentials'
return JsonResponse(response_data)
new_token = generate_token_hash(uname)
AuthToken.objects.update_or_create(
account=account,
defaults={'token_key': new_token}
)
response_data['data'] = {'token': new_token}
except Exception as e:
response_data['status'] = 500
response_data['message'] = str(e)
return JsonResponse(response_data)
通过 Postman 发送 POST 请求验证登录逻辑。若凭证正确,数据库中将生成或更新对应的 Token 记录;若凭证错误,则返回 401 状态提示。
自定义认证类实现
在需要保护的接口(如订单查询)中引入认证逻辑。我们需要编写一个自定义认证类,并将其配置到视图中。
from rest_framework.authentication import BaseAuthentication
from rest_framework.exceptions import AuthenticationFailed
from .models import AuthToken
class CustomTokenAuthentication(BaseAuthentication):
def authenticate(self, request):
# 尝试从查询参数或请求头中获取 token
token = request.query_params.get('token')
if not token:
return None # 返回 None 表示当前认证类不处理,交由下一个认证类处理
auth_record = AuthToken.objects.filter(token_key=token).first()
if not auth_record:
raise AuthenticationFailed('Invalid or expired token.')
# 返回一个元组,分别赋值给 request.user 和 request.auth
return (auth_record.account, auth_record)
def authenticate_header(self, request):
return 'Token'
在业务视图中应用该认证类:
class OrderAPIView(APIView):
authentication_classes = [CustomTokenAuthentication]
def get(self, request, *args, **kwargs):
# 此时 request.user 和 request.auth 已被正确赋值
orders = {'1': {'item': 'apple', 'cost': 15}}
return JsonResponse({'status': 200, 'data': orders})
当客户端未携带有效 Token 发起 GET 请求时,DRF 会拦截请求并返回认证失败的错误信息。
DRF 认证源码深度剖析
为了彻底理解上述自定义认证是如何生效的,我们需要追踪 DRF 的底层执行流。核心流程始于 APIView 的 dispatch() 方法。
1. 请求分发与 Request 封装
当请求到达 dispatch() 时,DRF 首先会对 Django 原生的 HttpRequest 进行增强封装,生成 DRF 专属的 Request 对象。
def dispatch(self, request, *args, **kwargs):
self.args = args
self.kwargs = kwargs
# 1. 封装原生 request,注入解析器、认证器等组件
request = self.initialize_request(request, *args, **kwargs)
self.request = request
try:
# 2. 执行初始化操作,包含认证、权限和节流检查
self.initial(request, *args, **kwargs)
# 3. 路由到具体的 HTTP 方法处理器 (get, post 等)
handler = getattr(self, request.method.lower(), self.http_method_not_allowed)
response = handler(request, *args, **kwargs)
except Exception as exc:
response = self.handle_exception(exc)
self.response = self.finalize_response(request, response, *args, **kwargs)
return self.response
initialize_request() 方法的核心作用是实例化配置好的认证类,并将其注入到新的 Request 对象中:
def initialize_request(self, request, *args, **kwargs):
parser_context = self.get_parser_context(request)
return Request(
request,
parsers=self.get_parsers(),
authenticators=self.get_authenticators(), # 获取认证器实例列表
negotiator=self.get_content_negotiator(),
parser_context=parser_context
)
get_authenticators() 通过列表推导式,将 authentication_classes 中配置的类实例化:
def get_authenticators(self):
return [auth() for auth in self.authentication_classes]
2. 触发认证逻辑
在 dispatch() 中,self.initial() 被调用,该方法内部会依次执行认证、权限和节流检查。认证的核心入口是 perform_authentication():
def initial(self, request, *args, **kwargs):
# ... 版本控制与内容协商逻辑 ...
# 触发认证机制
self.perform_authentication(request)
self.check_permissions(request)
self.check_throttles(request)
perform_authentication() 的实现非常简洁,它仅仅访问了 request.user 属性。这正是 DRF 的精妙之处——利用 Python 的 @property 装饰器实现懒加载:
def perform_authentication(self, request):
request.user
3. Request.user 属性与 _authenticate 方法
当访问 request.user 时,会触发 Request 类中的 user 属性方法。如果尚未进行过认证,它将调用内部的 _authenticate() 方法:
@property
def user(self):
if not hasattr(self, '_user'):
with wrap_attributeerrors():
self._authenticate()
return self._user
_authenticate() 是认证流程的真正执行者。它会遍历所有注入的认证器实例,依次调用它们的 authenticate() 方法:
def _authenticate(self):
for authenticator in self.authenticators:
try:
user_auth_tuple = authenticator.authenticate(self)
except exceptions.APIException:
self._not_authenticated()
raise
# 如果返回元组,则认证成功,绑定 user 和 auth
if user_auth_tuple is not None:
self._authenticator = authenticator
self.user, self.auth = user_auth_tuple
return
# 如果所有认证器都返回 None,则视为匿名用户
self._not_authenticated()
这就解释了为什么我们在自定义认证类中需要返回 (user, token) 元组,或者在跳过当前认证时返回 None。
4. 全局与局部配置策略
DRF 允许在视图级别(局部)或全局设置中定义认证类。如果在视图中未显式声明 authentication_classes,DRF 会回退到全局配置 api_settings.DEFAULT_AUTHENTICATION_CLASSES。
在 settings.py 中配置全局默认认证策略:
REST_FRAMEWORK = {
'DEFAULT_AUTHENTICATION_CLASSES': [
'myapp.authentication.CustomTokenAuthentication',
]
}
若某个特定接口(如登录或注册)需要豁免全局认证,只需在对应的视图类中将 authentication_classes 设置为空列表即可:
class PublicAPIView(APIView):
authentication_classes = []
# ...
内置认证基类 BaseAuthentication
在编写自定义认证逻辑时,强烈建议继承 DRF 提供的 BaseAuthentication 基类。该基类定义了认证器必须遵循的接口契约:
class BaseAuthentication:
def authenticate(self, request):
"""
验证请求并返回 (user, token) 元组。
子类必须重写此方法。
"""
raise NotImplementedError(".authenticate() must be overridden.")
def authenticate_header(self, request):
"""
返回一个字符串,用作 401 未认证响应中 `WWW-Authenticate` 头的值。
如果返回 None,则认证失败时将返回 403 权限拒绝响应。
"""
pass
通过继承此类,不仅能确保代码符合 DRF 的规范,还能在认证失败时正确构建 HTTP 响应头,从而提升 API 的标准化程度。