首页 > 编程语言 >DRF-Authention组件源码分析及扩展

DRF-Authention组件源码分析及扩展

时间:2024-10-27 16:03:00浏览次数:1  
标签:return self request Authention token 源码 user 认证 DRF

drf 认证组件

1、认证组件源码执行流程

在该图中,我把与认证组件无关的代码都删除了,只留下了认证的代码,方便解析。每行注释的开头数字即代表了执行顺序

image
image

注意事项:
  1. 第5步中的self.authenticators获取流程:
    • 首先,明确self代表的是request对象。则self.authenticators相当于去request对象中寻找authenticators的值。
    • 其次,Request在实例化得到request对象时,在参数中可传入authenticators的值
    • 通过下图流程分析:self.authenticators其实就是在视图类中定义的authentication_classes(也可在drf的配置中默认设置)

image

image

  1. authentication_classes必须是一个列表(可迭代对象)

  2. 由于源码中指定了会执行authenticate方法,因此编写认证类时,认证函数名必须是authenticate,return时应该返回一个元组,且类需继承自BaseAuthentication

2. 实践:编写一个token认证类

假设token存储于数据库。另注意:实际工作中编写代码最好采用白名单模式

class UnAuthentication(BaseAuthentication): # 默认继承BaseAuthentication
    def authenticate(self, request):  # 认证函数名固定为authenticate
        # 获取token
        token = request.query_params.get("token")

        # 未传token
        if not token:
            return None

        # 数据库不存在该token
        user_object = models.UserInfo.objects.filter(token=token, status=1).first()
        if not user_object:
            return None

        # token过期
        if datetime.datetime.now() > user_object.token_expiry_date:
            return None

        return user_object, token  # 认证成功,返回元组

3. 扩展点:

从认证失败后的执行过程中可发现:即使认证失败了程序也不会报错,且会继续执行后面的视图函数:

# --------认证失败----------
    def _not_authenticated(self):
        self._authenticator = None

        if api_settings.UNAUTHENTICATED_USER:
            self.user = api_settings.UNAUTHENTICATED_USER()
        else:
            self.user = None

        if api_settings.UNAUTHENTICATED_TOKEN:
            self.auth = api_settings.UNAUTHENTICATED_TOKEN()
        else:
            self.auth = None




那么第一个需求来了,如果我需要在认证失败后,立即让程序报错并停止执行后面的代码,该如何重写呢?
(仅供参考)答案:直接在认证失败后主动抛出错误即可:

class TokenAuthentication(BaseAuthentication):
    def authenticate(self, request):
        # 获取token
        token = request.query_params.get("token")

        # 未传token
        if not token:
            raise AuthenticationFailed({"code": return_code.AUTH_FAILED, "error": "认证失败,token不存在"})

        # 数据库不存在该token
        user_object = models.UserInfo.objects.filter(token=token, status=1).first()
        if not user_object:
            raise AuthenticationFailed({"code": return_code.AUTH_FAILED, "error": "认证失败123"})

        # token过期
        if datetime.datetime.now() > user_object.token_expiry_date:
            raise AuthenticationFailed({"code": return_code.AUTH_OVERDUE, "error": "认证过期"})

        return user_object, token




第二个需求:如果前端传token时不规范,有时在请求头里传,有时在查询参数中传....,那么该如何应用认证组件进行校验呢?
(仅供参考)答案:既然源码中对认证类列表进行循环了,则直接编写多个认证类即可(有多个认证类时,建议不要直接raise,否则可能导致后面的认证类无法校验)

class QueryParamsAuthentication(BaseAuthentication):
    def authenticate(self, request):
        # 判断token是否正确

        # 1. 从查询参数获取token
        token = request.query_params.get("token")
        if not token:
            return

        # 2. 去数据库获取该token
        user_obj = models.UserInfo.objects.filter(token=token).first()

        # 有对象则认证成功,返回  request.user = 用户对象; request.auth = token
        if user_obj:
            return user_obj, token
        return

    def authenticate_header(self, request):
        # return 'Basic realm="API"'
        return "API"


class HeaderAuthentication(BaseAuthentication):
    def authenticate(self, request):
        # 判断token是否正确

        # 1. 从请求头获取token
        token = request.META.get("HTTP_AUTHORIZATION")
        if not token:
            return

        # 2. 去数据库获取该token
        user_obj = models.UserInfo.objects.filter(token=token).first()

        # 有对象则认证成功,返回user和auth
        if user_obj:
            return user_obj, token
        return

    def authenticate_header(self, request):
        # return 'Basic realm="API"'
        return "API"





如果有多个认证类,则建议在最后一个认证类中默认抛出错误即可

class QueryParamsAuthentication(BaseAuthentication):
    def authenticate(self, request):
        # 判断token是否正确

        # 1. 从查询参数获取token
        token = request.query_params.get("token")
        if not token:
            return

        # 2. 去数据库获取该token
        user_obj = models.UserInfo.objects.filter(token=token).first()

        # 有对象则认证成功,返回  request.user = 用户对象; request.auth = token
        if user_obj:
            return user_obj, token
        return

    def authenticate_header(self, request):
        # return 'Basic realm="API"'
        return "API"


class HeaderAuthentication(BaseAuthentication):
    def authenticate(self, request):
        # 判断token是否正确

        # 1. 从请求头获取token
        token = request.META.get("HTTP_AUTHORIZATION")
        if not token:
            return

        # 2. 去数据库获取该token
        user_obj = models.UserInfo.objects.filter(token=token).first()

        # 有对象则认证成功,返回user和auth
        if user_obj:
            return user_obj, token
        return

    def authenticate_header(self, request):
        # return 'Basic realm="API"'
        return "API"


class NoAuthentication(BaseAuthentication):
    def authenticate(self, request):
        raise AuthenticationFailed({"code": 10001, "detail": "认证失败"})

标签:return,self,request,Authention,token,源码,user,认证,DRF
From: https://www.cnblogs.com/harry6/p/18468795

相关文章