drf框架

drf框架 - 三大认证组件 | 认证组件 | 权限组件 | 频率组件

RBAC

基于用户权限访问控制的认证 - Role-Based Access Control

Django框架采用的是RBAC认证规则，RBAC认证规则通常会分为 三表规则、五表规则，Django采用的是六表规则
# 三表：用户表、角色表、权限表
# 五表：用户表、角色表、权限表、用户角色关系表、角色权限关系表
# 六表：用户表、角色表、权限表、用户角色关系表、角色权限关系表、用户权限关系表

django六表跨表查询
# 用户表 -> 角色表 groups，用户表 -> 权限 user_permissions
# 角色表 -> 用户表 user_set，角色表 -> 权限 permissions
# 权限表 -> 用户表user_set，权限表 -> 角色表group_set


'''
重点补充：一、如果自定义User表后，再另一个项目中采用原生User表，完成数据库迁移时，可能失败
    1）卸载Django重新装
    2）将django.contrib下面的admin、auth下的数据库迁移记录文件清空
 二、1）outh认证的六表必须在第一次数据库迁移前确定，第一次数据库迁移完成
 	2）完成数据库迁移，出现了outh的用户表迁移异常，需要删除数据库迁移文件 user表所在自定义应用下的、admin组件下的、auth组件下的
'''

三大认证组件

源码分析

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

img

一.认证组件

  • 认证组件源码

def perform_authentication(self, request):
        # 调用 Request类的 方法属性 user 的get方法 => self._authenticate() 完成认证
        request.user
        
    认证的细则：
    # 做认证
    def _authenticate(self):
        # 遍历拿到一个个认证器，进行认证
        # self.authenticators配置的一堆认证类产生的认证类对象组成的 list
        for authenticator in self.authenticators: # request二次封装时添加了认证器list
            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()
def initialize_request(self, request, *args, **kwargs):
        """
        Returns the initial request object.
        """
        parser_context = self.get_parser_context(request)
        return Request(
            request,
            parsers=self.get_parsers(),
            authenticators=self.get_authenticators(),
            # 二次封装request时,添加了authenticators 认证
            negotiator=self.get_content_negotiator(),
            parser_context=parser_context
        )
    
def get_content_negotiator(self):
        """
        Instantiate and return the content negotiation class to use.
        """
        if not getattr(self, '_negotiator', None):
            self._negotiator = self.content_negotiation_class()
        return self._negotiator

class APIView(view):

content_negotiation_class = api_settings.DEFAULT_CONTENT_NEGOTIATION_CLASS

drf de settings.py中:

'DEFAULT_CONTENT_NEGOTIATION_CLASS': 'rest_framework.negotiation.DefaultContentNegotiation',

自定义认证类

  • 1.继承BaseAuthentication类

  • 2.重写authenticate(self, request)方法，自定义认证规则

  • 3.认证规则基于的条件:

    ​ - 没有认证信息返回None(游客)
    ​ - 有认证信息认证失败抛异常(非法用户)
    ​ - 有认证信息认证成功返回用户与认证信息的元素(合法用户)

  • 4.完成视图类的全局(settings.py)或局部(指定的视图类)配置

使用

authentications.py 文件中

from rest_framework.authentication import BaseAuthentication
from rest_framework.exceptions import AuthenticationFailed
from . import models
class MyAuthentication(BaseAuthentication):
    def authenticate(self, request):
        # 获取认证字符串
        auth = request.META.get('HTTP_AUTHORIZATION', None)  # 请求头的信息被wsgiref转为str
        print(auth)  # auth abc.123.xyz
        # 处理游客
        if auth is None:
            return None
​
        # 设置认证字段小规则(两段式) : auth 认证字符串
        auth_list = auth.split()  # 按空格切分
        # 校验合法用户还是非法用户
        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='admin123').first()
        if not user:
            raise AuthenticationFailed('用户数据有误，非法用户')
        return (user, None)

 补充:
        # 请求头中会携带认证信息（获取认证的字段要与前台约定）
        # 且默认规范用 Authorization 字段为键 值是认证字符串
        # 后台固定在请求对象的META字段中 HTTP_AUTHORIZATION 获取

settings.py 文件中

REST_FRAMEWORK = {
    # 认证类配置
    'DEFAULT_AUTHENTICATION_CLASSES': [
        # drf自带的认证 SessionAuthentication与BasicAuthentication
        # 'rest_framework.authentication.SessionAuthentication',  # 一般不用,因为会重新启动csrf
        # 'rest_framework.authentication.BasicAuthentication',
　　　　　　# 全局配置
        'app01.authentications.MyAuthentication',  
    ],
}


# 局部配置
class MyAPIView(APIView):
        authentication_calsses = []
        pass

二.权限组件

  • 权限组件源码

权限组件入口
    self.check_permissions(request)
    认证细则：
    def check_permissions(self, request):
        # 遍历权限对象列表得到一个个权限对象(权限器)，进行权限认证
        for permission in self.get_permissions():
            # 权限类一定有一个has_permission权限方法，用来做权限认证的
            # 参数：权限对象self、请求对象request、视图类对象
            # 返回值：有权限返回True，无权限返回False
            if not permission.has_permission(request, self):
                self.permission_denied(
                    request, message=getattr(permission, 'message', None)
                )

def get_permissions(self):

        return [permission() for permission in self.permission_classes]
drf 中的settings.py 中
'DEFAULT_PERMISSION_CLASSES': [
        'rest_framework.permissions.AllowAny',  # 默认的权限认证配置
    ],

# 根据配置路径找到 permissions 

#权限组件自带的四个权限,默认全局配置的是AllowAny,
    - 可以在settings.py中更换全局配置
    - 可以在视图类中针对性的局部配置
    - 可以自定义
class AllowAny(BasePermission):
    # 游客与登陆用户都有所有权限
    def has_permission(self, request, view):
        return True
​
class IsAuthenticated(BasePermission):
    # 游客没有任何权限，登陆用户才有权限
    def has_permission(self, request, view):
        return bool(request.user and request.user.is_authenticated)
​
class IsAdminUser(BasePermission):
    # 游客没有任何权限，登陆用户并且是管理员才有权限
    def has_permission(self, request, view):
        return bool(request.user and request.user.is_staff)
​
class IsAuthenticatedOrReadOnly(BasePermission):
    # 认证规则必须是只读请求或是合法用户: 游客只读，合法用户无限制
    def has_permission(self, request, view):
        return bool(
            request.method in SAFE_METHODS or 
            request.user and
            request.user.is_authenticated
        )

自定义权限类

1) 创建继承BasePermission的权限类
2) 重写has_permission方法
3) 实现体根据权限规则 确定有无权限
4) 进行全局或局部配置
​
认证规则
i.满足设置的用户条件，代表有权限，返回True
ii.不满足设置的用户条件，代表没有权限，返回False

permissions.py 文件中

from rest_framework.permissions import BasePermission
from django.contrib.auth.models import Group
class MyPermission(BasePermission):
    def has_permission(self, request, view):
        # 只读接口判断
        r1 = request.method in ('GET', 'HEAD', 'OPTIONS')
        # group为有权限的分组
        group = Group.objects.filter(name='管理员').first()
        # groups为当前用户所属的所有分组
        groups = request.user.groups.all()
        r2 = group and groups
        r3 = group in groups
        # 读接口大家都有权限，写接口必须为指定分组下的登陆用户
        return r1 or (r2 and r3)

view.py 文件中

# 游客只读，登录用户只读，只有登录用户属于 管理员 分组，才可以增删改
from utils.permissions import MyPermission
class TestAdminOrReadOnlyAPIView(APIView):
    permission_classes = [MyPermission]  # 局部配置
    # 所有用户都可以访问
    def get(self, request, *args, **kwargs):
        return APIResponse(0, '自定义读 OK')
    # 必须是 自定义“管理员”分组 下的用户
    def post(self, request, *args, **kwargs):
        return APIResponse(0, '自定义写 OK')

settings.py 文件中

REST_FRAMEWORK = {
    # 权限类配置
    'DEFAULT_PERMISSION_CLASSES': [
        'rest_framework.permissions.AllowAny',
        'rest_framework.permissions.IsAuthenticated',
        'rest_framework.permissions.IsAdminUser',
        'rest_framework.permissions.IsAuthenticatedOrReadOnly',
        
        # 全局配置自定义的
        'rest_framework.permissions.MyPermission',
    ],
}

三.频率组件

频率组件源码

# 1）APIView的dispath方法中的 self.initial(request, *args, **kwargs) 点进去
# 2）self.check_throttles(request) 进行频率认证
​
# 频率组件核心源码分析
def check_throttles(self, request):
    throttle_durations = []
    # 1）遍历配置的频率认证类，初始化得到一个个频率认证类对象（会调用频率认证类的 __init__() 方法）
    # 2）频率认证类对象调用 allow_request 方法，判断是否限次（没有限次可访问，限次不可访问）
    # 3）频率认证类对象在限次后，调用 wait 方法，获取还需等待多长时间可以进行下一次访问
    # 注：频率认证类都是继承 SimpleRateThrottle 类
    for throttle in self.get_throttles():
        if not throttle.allow_request(request, self):
            # 只要频率限制了，allow_request 返回False了，才会调用wait
            throttle_durations.append(throttle.wait())
​
            if throttle_durations:
                # Filter out `None` values which may happen in case of config / rate
                # changes, see #1438
                durations = [
                    duration for duration in throttle_durations
                    if duration is not None
                ]
​
                duration = max(durations, default=None)
                self.throttled(request, duration)

class SimpleRateThrottle(BaseThrottle):
　　def __init__(self):
     if not getattr(self, 'rate', None):
        # 得到settings配置的 次数/时间 赋值给rate
        self.rate = self.get_rate()
    # 将切分后的 '次数/时间' 解压赋值 num_requests=次数,duration=时间
    self.num_requests, self.duration = self.parse_rate(self.rate)
　　# 自定义频率限制时 需要我们实现的方法
　　def get_cache_key(self, request, view):
      raise NotImplementedError('.get_cache_key() must be overridden')

源码分析：初始化方法、判断是否有权限方法、计数等待时间方法

自定义频率类

# 1) 自定义一个继承 SimpleRateThrottle 类 的频率类
# 2) 设置一个 scope 类属性，属性值为任意见名知意的字符串
# 3) 在settings配置文件中，配置drf的DEFAULT_THROTTLE_RATES，格式为 {scope字符串: '次数/时间'}
# 4) 在自定义频率类中重写 get_cache_key 方法
    # 限制的对象返回 与限制信息有关的字符串
    # 不限制的对象返回 None (只能放回None，不能是False或是''等)


# 1) 自定义一个继承 SimpleRateThrottle 类 的频率类
# 2) 设置一个 scope 类属性，属性值为任意见名知意的字符串
# 3) 在settings配置文件中，配置drf的DEFAULT_THROTTLE_RATES，格式为 {scope字符串: '次数/时间'}
# 4) 在自定义频率类中重写 get_cache_key 方法
    # 限制的对象返回 与限制信息有关的字符串
    # 不限制的对象返回 None (只能放回None，不能是False或是''等)

throttles.py 文件中

from rest_framework.throttling import SimpleRateThrottle
​
class SMSRateThrottle(SimpleRateThrottle):
    # scope就是settings配置中限次字典中的key
    # SMSRateThrottle在实例化时会调用SimpleRateThrottle的__init__方法,通过scope获取对应的值
    # 键scope 对应的 value在源码中会按 '/' 切分成 '次数'和'时间'
    scope = 'sms'
​
    def get_cache_key(self, request, view):
        # 这里针对提交手机号的get 请求进行限制
        mobile = request.query_params.get('mobile')
        # 没有手机号不限制
        if not mobile:
            return True
        # 返回可以根据手机号动态变化，且不易重复的字符串，作为操作缓存的key
        return 'throttle_%(scope)s_%(ident)s' % {'scope': self.scope, 'ident': mobile}

settings.py 文件中配置

REST_FRAMEWORK = {
    # 频率限制条件配置
    'DEFAULT_THROTTLE_RATES': {
        'sms': '1/min'
    },
}

views.py 文件中 局部使用

from . throttles import SMSRateThrottle
class TestSMSAPIView(APIView):
    # 局部配置使用自定义频率限制
    throttle_classes = [SMSRateThrottle]
    def get(self, request, *args, **kwargs):
        return APIResponse(0, 'get 获取验证码 OK')
    def post(self, request, *args, **kwargs):
        return APIResponse(0, 'post 获取验证码  OK')
【推广】 免费学中医，健康全家人
原文地址：https://www.cnblogs.com/wwbplus/p/11933061.html