首页 > 文章 > rest framework 认证

rest framework 认证

一、简单认证示例

需求:

  • 用户名密码正确:没有 token 则产生一个 token,有 token 则更新,返回登录成功;
  • 若用户名或密码错误,返回错误信息。

1、models.py

from django.db import models


class UserInfo(models.Model):
    USER_TYPE = (
        (1, '普通用户'),
        (2, 'VIP'),
        (3, 'SVIP')
    )

    user_type = models.IntegerField(choices=USER_TYPE)
    username = models.CharField(max_length=32)
    password = models.CharField(max_length=64)


class UserToken(models.Model):
    user = models.OneToOneField(UserInfo, on_delete=models.CASCADE)
    token = models.CharField(max_length=64)

2、urls.py

from django.contrib import admin
from django.urls import path
from app.views import IndexView

urlpatterns = [
    path('admin/', admin.site.urls),
    path('index/', IndexView.as_view()),
]

3、views.py

from django.shortcuts import render, HttpResponse
from django.views import View
import json
import hashlib
import time
from app import models


def md5(user):
    """加密"""
    ctime = str(time.time())
    m = hashlib.md5(bytes(user, encoding='utf-8'))
    m.update(bytes(ctime, encoding='utf-8'))
    return m.hexdigest()


class IndexView(View):
    def post(self, request, *args, **kwargs):
        ret = {'code': 1000, 'msg': None}
        try:
            user = request.POST.get('username')
            password = request.POST.get('password')
            obj = models.UserInfo.objects.filter(username=user, password=password).first()
            if not obj:
                ret['code'] = 10001
                ret['msg'] = '用户名或密码错误'

             # 为用户创建token    
            token = md5(user)
            
            # 存在就更新,不存在就创建
            o = models.UserToken.objects.update_or_create(user=obj, defaults={'token': token})
        except Exception as e:
            ret['code'] = 1002
            ret['msg'] = '请求异常'
        return HttpResponse(json.dumps(ret))

利用 postman 工具模拟发送 post 请求:

分享图片

二、rest framework 认证

基于上面的例子,我们用 rest framework 给每个类都实现认证功能。

1、新建一个 auth.py 脚本文件,app/utils/auth.py

from rest_framework import exceptions
from app import models


class MyAuthentication(object):
    """认证类"""
    def authenticate(self, request):
        token = request._request.GET.get('token')       # 获取 token
        token_obj = models.UserToken.objects.filter(token=token).first()
        if not token_obj:
            raise exceptions.AuthenticationFailed('用户认证失败')
        return (token_obj.user, token_obj)  # 返回一个元组

    def authenticate_header(self, val):
        pass

2、project/urls.py

from django.contrib import admin
from django.urls import path
from app.views import IndexView, OrderView

urlpatterns = [
    path('admin/', admin.site.urls),
    path('index/', IndexView.as_view()),
    path('order/', OrderView.as_view()),
]

3、app/views.py

from django.shortcuts import render, HttpResponse
from rest_framework.views import APIView
import hashlib
import time
from app import models
from django.http import JsonResponse
from .utils.auth import MyAuthentication        # 导入认证的类

def md5(user):
    """加密"""
    ctime = str(time.time())
    m = hashlib.md5(bytes(user, encoding='utf-8'))
    m.update(bytes(ctime, encoding='utf-8'))
    return m.hexdigest()

# 订单信息
ORDER_DICT = {
    1: {
        'name': 'rose',
        'age': 18,
        'gender': 'female'
    },

    2: {
        'name': 'tom',
        'age': 19,
        'gender': 'male'
    },
}


class OrderView(APIView):
    """订单管理"""
    authentication_classes = [MyAuthentication, ]  # 添加认证
    ret = {'code': 1000, 'msg': None, 'data': None, }

    def get(self, request, *args, **kwargs):
        self.ret['data'] = ORDER_DICT
        return JsonResponse(self.ret)

    def post(self, request, *args, **kwargs):
        try:
            user = request._request.POST.get('username')
            pwd = request._request.POST.get('password')
            obj = models.UserInfo.objects.filter(username=user, password=pwd).first()
            if not obj:
                self.ret['code'] = 1001
                self.ret['msg'] = '用户名或密码错误'

            token = md5(user)
            models.UserToken.objects.update_or_create(user=obj, defaults={'token': token})
            self.ret['token'] = token
        except Exception as e:
            self.ret['code'] = 1002
            self.ret['msg'] = '请求异常'

        return JsonResponse(self.ret)

在视图中我们新添加了一个订单类 OrderView,现在我们用 postman 发送 get 请求:

分享图片

发现认证失败,这是因为我们在认证时候获取 URL 中的 token,若过来的请求没有携带 token 就会认证失败。下面再来看看携带 token 的样子:

分享图片

三、分析 DRF 源码实现认证

在这里我们将通过分析 drf 的源码来分析它是怎么实现认证的,又是怎么来编辑认证类的,下面是源码大致流程:

封装原生request 对象

1、drf 使用的 CBV 模式,所有请求过来,首先执行 dispatch() 方法,restframe work 对 dispatch() 方法增加了一些其他功能 restframework/views.py

def dispatch(self, request, *args, **kwargs):
    """
        `.dispatch()` is pretty much the same as Django's regular dispatch,
        but with extra hooks for startup, finalize, and exception handling.
        """
    self.args = args
    self.kwargs = kwargs
    # 对原生的 request 对象进行加工,丰富了
    # request= Request(request,parsers=self.get_parsers(),authenticators=self.get_authenticators(),negotiator=self.get_content_negotiator(),parser_context=parser_context)
    # 第一个参数为原生的 request 对象,
    request = self.initialize_request(request, *args, **kwargs)
    self.request = request
    self.headers = self.default_response_headers  # deprecate?

    try:
        # 认证
        self.initial(request, *args, **kwargs)

        # Get the appropriate handler method
        if request.method.lower() in self.http_method_names:
            handler = getattr(self, request.method.lower(),
                              self.http_method_not_allowed)
            else:
                handler = 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

从上面源码可以看到,在 dispatch() 方法中,主要做了两件事:

  • self.initialize_request(request, *args, **kwargs):对原生 request 对象进行了封装,增加了一些其他功能
  • self.initial(request, *args, **kwargs):执行认证功能

2、initialize_request()

下面我们来看看 initialize_request() 方法怎么封装原生 request 对象的:

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(),   # 这个函数将返回一个对象列表
        negotiator=self.get_content_negotiator(),
        parser_context=parser_context
    )

3、get_authenticators()

使用列表生成式循环 self.authentication_classes 中的对象,并执行,返回一个对象列表:

def get_authenticators(self):
    """
    Instantiates and returns the list of authenticators that this view can use.
    """

    return [auth() for auth in self.authentication_classes]

4、authentication_classes

api_settings 会从 settings 中匹配 DEFAULT_AUTHENTICATION_CLASSES 字段,所有可用通过在 settings 中设置 DEFAULT_AUTHENTICATION_CLASSES,可用实现全局认证:

class APIView(View):

    # The following policies may be set at either globally, or per-view.
    renderer_classes = api_settings.DEFAULT_RENDERER_CLASSES
    parser_classes = api_settings.DEFAULT_PARSER_CLASSES
    # 这句,从 setting 中找 DEFAULT_AUTHENTICATION_CLASSES 字段
    authentication_classes = api_settings.DEFAULT_AUTHENTICATION_CLASSES
    ...

认证

下面我们来分析 dispatch() 中的另一个方法:initial(request, *args, **kwargs)(认证)

1、initial()

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)

2、perform_authentication()

这里面就调用了 user:

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        # 这个 request 是原生 request 对象

3、user()

user() 是一个静态方法,因此调用它是不用加括号:

@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

4、_authenticate()

循环所有 authenticator 对象:

def _authenticate(self):
    """
    Attempt to authenticate the request using each authentication instance
    in turn.
    循环认证类的所有对象
    这里分三种情况
    1.如果authenticate方法抛出异常,self._not_authenticated()执行
    2.有返回值,必须是元组:(request.user,request.auth)
    3.返回None,表示当前认证不处理,等下一个认证来处理
    """
    for authenticator in self.authenticators:
        try:
            # 执行认证类的authenticate方法
            user_auth_tuple = authenticator.authenticate(self)
        except exceptions.APIException:
            self._not_authenticated()
            raise

        if user_auth_tuple is not None:
            self._authenticator = authenticator
            # 返回一个元组
            self.user, self.auth = user_auth_tuple
            return

    self._not_authenticated()

authenticate() 返回的是一个元组:

user_auth_tuple = return (token_obj.user, token_obj)
self.user, self.auth = (token_obj.user, token_obj)
# 那么相当于,即将用户对象封装到 request 对象中,可以通过 request 访问用户相关信息:
request.user = token_obj.user
request.auth = token_obj

若没有返回值则执行 _not_authenticated() 方法,将返回一个匿名用户,认证失败:

def _not_authenticated(self):
    """
    Set authenticator, user & authtoken representing an unauthenticated request.

    Defaults are None, AnonymousUser(匿名) & None.
    """
    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

总结

通过一系列的认证,发现 rest framework 最终执行的是 authenticate() 方法来认证,如果在我们自定义认证类时,重写 authenticate() 方法,那么就默认会执行我们自己定义的 authenticate() 方法。


自定义认证类

app/utils/auth.py

通过获取 URL 中 token 来认证:

from rest_framework import exceptions
from app import models


class MyAuthentication(object):
    def authenticate(self, request):
        token = request._request.GET.get('token')
        token_obj = models.UserToken.objects.filter(token=token).first()
        if not token_obj:
            raise exceptions.AuthenticationFailed('用户认证失败')
        return (token_obj.user, token_obj)

    def authenticate_header(self, val):
        pass

四、配置文件

1、在封装原生 request 对象的时候,我们发现 api_settings 是从 settings 中读取的配置文件:

class APIView(View):

    # The following policies may be set at either globally, or per-view.
    renderer_classes = api_settings.DEFAULT_RENDERER_CLASSES
    parser_classes = api_settings.DEFAULT_PARSER_CLASSES
    authentication_classes = api_settings.DEFAULT_AUTHENTICATION_CLASSES

2、api_settings

从 settings 中加载 REST_FRAMEWORK,从而找到 DEFAULT_AUTHENTICATION_CLASSES

api_settings = APISettings(None, DEFAULTS, IMPORT_STRINGS)


def reload_api_settings(*args, **kwargs):
    setting = kwargs['setting']
    # 从 settings 中加载 REST_FRAMEWORK
    if setting == 'REST_FRAMEWORK':
        api_settings.reload()

3、基于此我们在 settings 中配置认证类,从而使得每个视图类都有认证功能,而不需要单独添加 authentication_classes = [MyAuthentication, ]

# 设置全局认证
REST_FRAMEWORK = {
    "DEFAULT_AUTHENTICATION_CLASSES": ['app.utils.auth.MyAuthentication', ]     # 认证类的路径
}

**如果某个视图类不需要认证,可以单独设置 authentication_classes = []**

五、其它内置认证类

rest framework 还提供了一些其他的内置认证类 rest_framework/authentication.py

分享图片

类 BaseAuthentication

BaseAuthentication 就实现了两个方法,因此我们自定义认证类时继承 BaseAuthentication 可以不写 authenticate_header() 方法,但是一定要重写 authenticate() 方法,否则会报错。

class BaseAuthentication(object):
    """
    All authentication classes should extend BaseAuthentication.(所有的认证类都应该继承 BaseAuthentication)
    """

    def authenticate(self, request):
        """
        Authenticate the request and return a two-tuple of (user, token).
        """
        # 不重写会报错
        raise NotImplementedError(".authenticate() must be overridden.")

    def authenticate_header(self, request):
        """
        Return a string to be used as the value of the `WWW-Authenticate`
        header in a `401 Unauthenticated` response, or `None` if the
        authentication scheme should return `403 Permission Denied` responses.
        """
        pass

匿名用户

如果允许匿名用户(即没有登录的用户)访问,可以修改自定义认证类为 app/utils/auth.py

from rest_framework import exceptions
from app import models
from rest_framework.authentication import BaseAuthentication


class FirstAuthentication(BaseAuthentication):
    """不返回值,则会执行 _not_authenticated() 方法,返回一个匿名用户"""
    def authenticate(self, request):
        pass
    def authenticate_header(self, request):
        pass


class MyAuthentication(BaseAuthentication):
    def authenticate(self, request):
        token = request._request.GET.get('token')
        token_obj = models.UserToken.objects.filter(token=token).first()
        if not token_obj:
            raise exceptions.AuthenticationFailed('用户认证失败')
        return (token_obj.user, token_obj)

    def authenticate_header(self, val):
        pass

2、settings.py

# 设置全局认证
REST_FRAMEWORK = {
    "DEFAULT_AUTHENTICATION_CLASSES": ['app.utils.auth.FirstAuthentication', 'app.utils.auth.MyAuthentication', ],
}

访问:http 时首先会以 `认证,将返回一个匿名用户AnonymousUser`,当然你也可以设置为中文:

# 设置全局认证
REST_FRAMEWORK = {
    "DEFAULT_AUTHENTICATION_CLASSES": ['app.utils.auth.FirstAuthentication', 'app.utils.auth.MyAuthentication', ],
    "UNAUTHENTICATED_USER": lambda: '匿名用户'
}

你也可设置为 None,Token 也可以设置:

"UNAUTHENTICATED_TOKEN": None

分享图片

总结

自定义认证类需要实现以下几步:

  • 自定义一个类,继承 BaseAuthentication,类中必须实现 authenticate() 方法
  • 若全局需要认证,配置 settings 即可
  • 若局部某个类视图不需要认证,可以在其中添加 authentication_classes = []
  • 若匿名用户也允许访问,可以定义一个类,不返回值
时间:2019-06-24 11:39:35阅读(33)
联系我们 - 留言反馈
© 2017 版权所有 鲁ICP备17052893号