雪泥鴻爪

dog2's blog

Django REST Framework 学习笔记（十一）：认证组件

Posted on In Disqus:
Symbols count in article: 9k Reading time ≈ 8 mins.
  • Title(EN): Django REST Framework Learning Notes (11): Authentication
  • Author: dog2

基本信息

  • 源码 rest_framework.authentication
  • 官方文档 API Guide - Authentication
  • 本文demo代码Github
    • 自定义认证类
    • jwt & simple-jwt

源码分析

入口

从rest_framework.views.APIView的dispath(self, request, *args, **kwargs)下手 ，dispath方法内 self.initial(request, *args, **kwargs) 进入三大认证

  • 认证组件 self.perform_authentication(request)
    • 校验用户：游客、合法用户、非法用户
    • 游客：代表校验通过，直接进入下一步校验（权限校验）
    • 合法用户：代表校验通过，将用户存储在request.user中，再进入下一步校验（权限校验）
    • 非法用户：代表校验失败，抛出异常，返回403权限异常结果
  • 权限组件 self.check_permissions(request)
    • 校验用户权限：必须登录、所有用户、登录之后读写，游客只读、自定义用户角色
    • 认证通过：可以进入下一步校验（频率认证）
    • 认证失败：抛出异常，返回403权限异常结果
  • 频率组件 self.check_throttles(request)
    • 限制视图接口被访问的频率次数 - 限制的条件(IP、id、唯一键)、频率周期时间(s、m、h)、频率的次数（3/s）
    • 没有达到限次：正常访问接口
    • 达到限次：限制时间内不能访问，限制时间达到后，可以重新访问

本文介绍认证组件。

rest_framework.views.APIView.initial()

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
def initial(self, request, *args, **kwargs):
        """
        Runs anything that needs to occur prior to calling the method handler.
        """
        self.format_kwarg = self.get_format_suffix(**kwargs)

        # Perform content negotiation and store the accepted info on the request
        neg = self.perform_content_negotiation(request)
        request.accepted_renderer, request.accepted_media_type = neg

        # Determine the API version, if versioning is in use.
        version, scheme = self.determine_version(request, *args, **kwargs)
        request.version, request.versioning_scheme = version, scheme

        # Ensure that the incoming request is permitted
        self.perform_authentication(request)  #认证
        self.check_permissions(request)  #权限
        self.check_throttles(request)  #频率

rest_framework.views.APIView.perform_authentication()

1
2
3
4
5
6
7
8
9
def perform_authentication(self, request):
        """
        Perform authentication on the incoming request.

        Note that if you override this and simply 'pass', then authentication
        will instead be performed lazily, the first time either
        `request.user` or `request.auth` is accessed.
        """
        request.user

rest_framework.request.Request.user

1
2
3
4
5
6
7
8
9
10
@property
def user(self):
    """
    Returns the user associated with the current request, as authenticated
    by the authentication classes provided to the request.
    """
    if not hasattr(self, '_user'):
        with wrap_attributeerrors():
            self._authenticate()
    return self._user

rest_framework.request.Request._authenticate()

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
# 做认证
def _authenticate(self):  #这里的self就是request
    # 遍历拿到一个个认证器，进行认证
    # self.authenticators配置的一堆认证类产生的认证类对象组成的 list
　　　　 # 即：[auth() for auth in self.authentication_classes]
    for authenticator in self.authenticators:
        try:
            # 认证器(对象)调用认证方法authenticate(认证类对象self, request请求对象)
            # 返回值：登陆的用户与认证的信息组成的 tuple
            # 该方法被try包裹，代表该方法会抛异常，抛异常就代表认证失败
            user_auth_tuple = authenticator.authenticate(self)
        except exceptions.APIException:
            self._not_authenticated()
            raise

        # 返回值的处理
        if user_auth_tuple is not None:
            self._authenticator = authenticator
            # 如何有返回值，就将 登陆用户 与 登陆认证 分别保存到 request.user、request.auth
            self.user, self.auth = user_auth_tuple
            return
    # 如果返回值user_auth_tuple为空，代表认证通过，但是没有 登陆用户 与 登陆认证信息，代表游客
    self._not_authenticated()

自定义认证类

认证组件一般都是自定义的，不会使用原始的

方法

从源码的settings文件可以看出，认证类需要继承BasicAuthentication(在authentication.py文件)

1
2
3
4
DEFAULT_AUTHENTICATION_CLASSES': [
    'rest_framework.authentication.SessionAuthentication',  #会重新开启CSRF认证
    'rest_framework.authentication.BasicAuthentication'
]

具体流程如下：

  1. 创建继承BaseAuthentication的认证类
  2. 重写authenticate方法
  3. 实现体根据认证规则 确定游客、非法用户、合法用户 （根据自己的认证规则）
    • 没有认证信息返回None(游客)
    • 有认证信息认证失败抛异常(非法用户)
    • 有认证信息认证成功返回用户与认证信息元组(合法用户)
  4. 进行全局或局部配置

示例代码

models.py

1
2
3
4
5
6
7
8
9
10
11
12
13
from django.db import models
from django.contrib.auth.models import AbstractUser

class User(AbstractUser):
    mobile = models.CharField(max_length=11,unique=True)

    class Meta:
        db_table = 'user'
        verbose_name = '用户表'
        verbose_name_plural = verbose_name

    def __str__(self):
        return self.username

utils.authentications.py

Source Code 

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
# 1）继承BaseAuthentication类
# 2）重写authenticate(self, request)方法，自定义认证规则
# 3）自定义认证规则基于的条件：
#       没有认证信息返回None(游客)
#       有认证信息认证失败抛异常(非法用户)
#       有认证信息认证成功返回用户与认证信息元组(合法用户)
# 4）完全视图类的全局(settings文件中)或局部(确切的视图类)配置

from rest_framework.authentication import BaseAuthentication
from rest_framework.exceptions import AuthenticationFailed   #异常接收
from app01 import models

#继承BaseAuthentication
class MyAuthentication(BaseAuthentication):
    """
        同前台请求头拿认证信息auth（获取认证的字段要与前台约定）
        没有auth是游客，返回None
        有auth进行校验
            失败是非法用户，抛出异常
            成功是合法用户，返回 (用户, 认证信息)
    """
    def authenticate(self, request):   #重写authenticate方法
　　　　#前台在请求头携带认证信息，
　　　　#且默认规范用 Authorization 字段携带认证信息，
　　　　#后台固定在请求对象的META字段中 HTTP_AUTHORIZATION 获取
        #认证信息auth
        auth = request.META.get('HTTP_AUTHORIZATION',None)#处理游客
        if auth is None:
            return None
        #设置认证字段小规则(两段式):"auth 认证字符串" 在BasicAuthentication类中有规则范式
        auth_list = auth.split()     #校验是否还是非法用户,不是两段，第一段不是auth就是非法用户
        if not (len(auth_list) == 2 and auth_list[0].lower() == 'auth'):
            raise AuthenticationFailed('认证信息有误，非法用户')  #抛异常

        #校验认证信息第二段从auth_list[1]中解析出来
        # 注：假设一种情况，信息为abc.123.xyz，就可以解析出admin用户；实际开发，该逻辑一定是校验用户的正常逻辑
        if auth_list[1] != 'abc.123.xyz': #校验失败
            raise AuthenticationFailed('信息错误，非法用户')

        #最后再去数据库校验是否有此用户
        user = models.User.objects.filter(username='admin').first()
        if not user:
            raise AuthenticationFailed('用户数据有误，非法用户')

        return (user,None)

settings.py

在settings文件中配置自定义认证组件

1
2
3
4
5
6
REST_FRAMEWORK = {
    # 认证类配置
    'DEFAULT_AUTHENTICATION_CLASSES': [
        'utils.authentications.MyAuthentication',
    ],
}

views.py

1
2
3
4
5
6
7
8
9
10
from rest_framework.views import APIView
from utils.response import APIResponse

class TestAPIView(APIView):
    def get(self, request, *args, **kwargs):
        # 如果通过了认证组件，request.user就一定有值
        # 游客：AnonymousUser
        # 用户：User表中的具体用户对象
        print(request.user)
        return APIResponse(0, 'test get ok')

urls.py

1
2
3
4
5
6
from django.urls import path
from . import views

urlpatterns = [
    path('test/', views.TestAPIView.as_view()),
]

Postman测试

使用Postman的get请求，在自定义认证组件获取用户，在views视图通过request.user能打印出来

Token认证

JWT

RESTful API里使用最普遍的就是基于json的token认证了，即Json Web Token（JWT）。JWT 是一个开放标准(RFC 7519)，它定义了一种用于简洁，自包含的用于通信双方之间以 JSON 对象的形式安全传递信息的方法。JWT 可以使用 HMAC 算法或者是 RSA 的公钥密钥对进行签名。

优点

  1. 服务器不要存储token，token交给每一个客户端自己存储，服务器压力小
  2. 服务器存储的是 签发和校验token 两段算法，签发认证的效率高
  3. 算法完成各集群服务器同步成本低，路由项目完成集群部署（适应高并发）

JWT格式

  1. jwt token采用三段式：头部.载荷.签名
  2. 每一部分都是一个json字典加密形参的字符串
  3. 头部和载荷采用的是base64可逆加密（前台后台都可以解密）
  4. 签名采用hash256不可逆加密（后台校验采用碰撞校验）
  5. 各部分字典的内容：
    • 头部：基础信息 - 公司信息、项目组信息、可逆加密采用的算法
    • 载荷：有用但非私密的信息 - 用户可公开信息、过期时间
    • 签名：头部+载荷+秘钥 不可逆加密后的结果
    • 注：服务器jwt签名加密秘钥一定不能泄露

签发与校验

  • 签发token：固定的头部信息加密。当前的登陆用户与过期时间加密。头部+载荷+秘钥生成不可逆加密
  • 校验token：头部可校验也可以不校验，载荷校验出用户与过期时间，头部+载荷+秘钥完成碰撞检测校验token是否被篡改

DRF自带的JWT认证

DRF的源码里是实现了token认证模块了的，不过存在一些不足，因此一般不推荐使用。

一般使用的是DRF的两个jwt插件：

  • jango-rest-framework-jwt: 曾经很常用，但后来作者已不再维护，因此也不推荐使用
  • django-rest-framework-simplejwt: DRF官方文档推荐使用的插件

DRF插件 django-rest-framework-jwt

示例代码如下

settings.py

此处设置的是全局配置

1
2
3
4
5
6
7
REST_FRAMEWORK = {
    # 认证类配置
    'DEFAULT_AUTHENTICATION_CLASSES': [
        'utils.authentications.MyAuthentication',
        'rest_framework_jwt.authentication.JSONWebTokenAuthentication', # jwt全局配置
    ],
}

models.py

代码同上

views.py

此处设置的是局部配置

1
2
3
4
5
6
7
8
9
10
11
from rest_framework.views import APIView
from utils.response import APIResponse
# 必须登录后才能访问 - 通过了认证权限组件
from rest_framework.permissions import IsAuthenticated
from rest_framework_jwt.authentication import JSONWebTokenAuthentication

class UserDetail(APIView):
    authentication_classes = [JSONWebTokenAuthentication]  # 局部配置jwt-token校验request.user
    permission_classes = [IsAuthenticated]  # 结合权限组件筛选掉游客
    def get(self, request, *args, **kwargs):
        return APIResponse(results={'username': request.user.username})

url.py

1
2
3
4
5
6
7
8
# ObtainJSONWebToken视图类就是通过username和password得到user对象然后签发token（生成token）
from rest_framework_jwt.views import ObtainJSONWebToken, obtain_jwt_token

from . import views

urlpatterns = [
    path('jwt/', obtain_jwt_token),
    path('user/detail/', views.UserDetail.as_view()),

DRF插件 django-rest-framework-simplejwt

示例代码如下

settings.py

此处设置的是全局配置

1
2
3
4
5
6
7
REST_FRAMEWORK = {
    # 认证类配置
    'DEFAULT_AUTHENTICATION_CLASSES': [
        'utils.authentications.MyAuthentication',
        'rest_framework_simplejwt.authentication.JWTAuthentication', # 全局配置
    ],
}

此外，还有simplejwt的一些细节配置：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
# Reference :
# 1. https://django-rest-framework-simplejwt.readthedocs.io/en/latest/settings.html
# 2. https://juejin.im/post/5d23f43df265da1bcc196749#heading-15
SIMPLE_JWT = {
    'ACCESS_TOKEN_LIFETIME': timedelta(minutes=5),
    'REFRESH_TOKEN_LIFETIME': timedelta(days=1),
    'ROTATE_REFRESH_TOKENS': False,
    'BLACKLIST_AFTER_ROTATION': True,

    'ALGORITHM': 'HS256',
    'SIGNING_KEY': SECRET_KEY,
    'VERIFYING_KEY': None,

    'AUTH_HEADER_TYPES': ('Bearer',),
    'USER_ID_FIELD': 'id',
    'USER_ID_CLAIM': 'user_id',

    'AUTH_TOKEN_CLASSES': ('rest_framework_simplejwt.tokens.AccessToken',),
    'TOKEN_TYPE_CLAIM': 'token_type',

    'JTI_CLAIM': 'jti',

    'SLIDING_TOKEN_REFRESH_EXP_CLAIM': 'refresh_exp',
    'SLIDING_TOKEN_LIFETIME': timedelta(minutes=5),
    'SLIDING_TOKEN_REFRESH_LIFETIME': timedelta(days=1),
}

models.py

代码同上

views.py

此处设置的是局部配置

1
2
3
4
5
6
7
8
9
10
11
from rest_framework.views import APIView
from utils.response import APIResponse
# 必须登录后才能访问 - 通过了认证权限组件
from rest_framework.permissions import IsAuthenticated
from rest_framework_simplejwt.authentication import JWTAuthentication

class UserDetail2(APIView):
    authentication_classes = [JWTAuthentication]  # 局部配置simplejwt-token校验request.user
    permission_classes = [IsAuthenticated]  # 结合权限组件筛选掉游客
    def get(self, request, *args, **kwargs):
        return APIResponse(results={'username': request.user.username})

url.py

1
2
3
4
5
6
7
8
9
# ObtainJSONWebToken视图类就是通过username和password得到user对象然后签发token（生成token）
from rest_framework_jwt.views import ObtainJSONWebToken, obtain_jwt_token

from . import views

urlpatterns = [
    path('simple_jwt/', TokenObtainPairView.as_view(), name='token_obtain_pair'),
    path('simple_jwt/refresh/', TokenRefreshView.as_view(), name='token_refresh'),
    path('user/detail2/', views.UserDetail2.as_view()),

参考链接 & 扩展阅读

【DRF】用户注册登录及JWT JSON Web Token 入门教程