时间戳

首页 > 技术文章 > django的rest framework框架——认证、权限、节流控制

django的rest framework框架——认证、权限、节流控制

yanlin-10 2019-01-09 21:27 原文

一、登录认证示例

模拟用户登录，获取token，当用户访问订单或用户中心时，判断用户携带正确的token，则允许查看订单和用户信息，否则抛出异常：

from django.conf.urls import url
from django.contrib import admin
from api import views

urlpatterns = [
    url(r'^admin/', admin.site.urls),
    url(r'^api/v1/auth/$', views.AuthView.as_view()),
    url(r'^api/v1/order/$', views.OrderView.as_view()),
    url(r'^api/v1/userInfo/$', views.UserInfoView.as_view()),
]
urls.py
from django.db import models


class UserInfo(models.Model):
    user_type_choices = (
        (1, "普通用户"),
        (2, "vip"),
        (3, "svip"),
    )
    user_type = models.IntegerField(choices=user_type_choices)
    username = models.CharField(max_length=32, unique=True)
    password = models.CharField(max_length=64)


class UserToken(models.Model):
    user = models.OneToOneField(to="UserInfo")
    token = models.CharField(max_length=64)


class Order(models.Model):
    id = models.AutoField(primary_key=True)
    name = models.CharField(max_length=32)
    price = models.DecimalField(max_digits=5, decimal_places=2)
    create_time = models.DateTimeField(auto_now_add=True)
    user = models.ForeignKey(to="UserInfo", on_delete=models.CASCADE, null=True, blank=True)
models.py
import hashlib
import time

from django.http import JsonResponse
from rest_framework.views import APIView
from rest_framework import exceptions

from api import models


def md5(user):
    """生成token"""
    ctime = str(time.time())  # 当前时间
    m = hashlib.md5(bytes(user, encoding="utf-8"))
    m.update(bytes(ctime, encoding="utf-8"))
    return m.hexdigest()


class Authtication(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)  # rest framework会将这两个字段赋值给request，以供后续操作使用

    def authenticate_header(self, request):
        pass


class AuthView(APIView):
    """登录"""
    def post(self, request, *args, **kwargs):
        res = {"code": 1000, "msg": None}
        try:
            # 从请求中获取用户登录信息
            user = request._request.POST.get("username")
            pwd = request._request.POST.get("password")
            # 到数据库获取用户信息
            user_obj = models.UserInfo.objects.filter(username=user, password=pwd).first()
            # 如果获取到对象则说明认证成功，为登录用户创建token，如果认证失败则返回错误信息
            if not user_obj:
                res["code"] = 1001
                res["msg"] = "用户名或密码错误"
            else:
                token = md5(user)
                # 将token存入数据库：如果数据库存在token就更新，不存在就创建
                models.UserToken.objects.update_or_create(user=user_obj, defaults={"token": token})
                res["token"] = token
        except Exception as e:
            res["code"] = 1002
            res["msg"] = e
        return JsonResponse(res)


class OrderView(APIView):
    """订单"""
    authentication_classes = [Authtication]

    def get(self, request, *args, **kwargs):
        res = {"code": 1000, "msg": None, "data": None}
        try:
            orders = models.Order.objects.filter(user=request.user).values("id", "name", "price", "create_time", "user__username")
            res["data"] = list(orders)
        except Exception as e:
            res["code"] = 1002
            res["msg"] = e
        return JsonResponse(res)


class UserInfoView(APIView):
    """用户中心"""
    authentication_classes = [Authtication]

    def get(self, request, *args, **kwargs):
        res = {"code": 1000, "msg": None, "data": None}
        try:
            # print(request.user)  # 用户对象
            # print(request.auth)  # 认证对象
            user = models.UserInfo.objects.filter(id=request.auth.user_id).values("id", "username", "password")
            res["data"] = list(user)
        except Exception as e:
            res["code"] = 1002
            res["msg"] = e
        return JsonResponse(res)
views.py

二、rest framework认证流程源码

rest framework的request.py：

_not_authenticated()方法的处理流程：

三、rest framework配置

1、如何将之前写在视图里面的 authentication_classes 写入rest framework的配置文件中：

rest framework的配置信息在rest framework的settings.py里面：

rest framework的views.py：

新建一个utils文件，新建auth.py文件，将自定义的认证类写到这个文件里面：

代码：

from rest_framework import exceptions

from api import models


class Authtication(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)  # rest framework会将这两个字段赋值给request，以供后续操作使用

    def authenticate_header(self, request):
        pass
auth.py

再去项目的settings里面设置这个认证类的路径：

REST_FRAMEWORK = {
    "DEFAULT_AUTHENTICATION_CLASSES": ["api.utils.auth.Authtication",]
}
settings.py

 这样相当于做了一个全局配置，就不用在每个视图里面再去设置认证类了：

import hashlib
import time

from django.http import JsonResponse
from rest_framework.views import APIView
# from rest_framework import exceptions

from api import models


def md5(user):
    """生成token"""
    ctime = str(time.time())  # 当前时间
    m = hashlib.md5(bytes(user, encoding="utf-8"))
    m.update(bytes(ctime, encoding="utf-8"))
    return m.hexdigest()


class AuthView(APIView):
    """登录"""
    authentication_classes = []
    def post(self, request, *args, **kwargs):
        res = {"code": 1000, "msg": None}
        try:
            # 从请求中获取用户登录信息
            user = request._request.POST.get("username")
            pwd = request._request.POST.get("password")
            # 到数据库获取用户信息
            user_obj = models.UserInfo.objects.filter(username=user, password=pwd).first()
            # 如果获取到对象则说明认证成功，为登录用户创建token，如果认证失败则返回错误信息
            if not user_obj:
                res["code"] = 1001
                res["msg"] = "用户名或密码错误"
            else:
                token = md5(user)
                # 将token存入数据库：如果数据库存在token就更新，不存在就创建
                models.UserToken.objects.update_or_create(user=user_obj, defaults={"token": token})
                res["token"] = token
        except Exception as e:
            res["code"] = 1002
            res["msg"] = e
        return JsonResponse(res)


class OrderView(APIView):
    """订单"""
    # authentication_classes = [Authtication]

    def get(self, request, *args, **kwargs):
        res = {"code": 1000, "msg": None, "data": None}
        try:
            orders = models.Order.objects.filter(user=request.user).values("id", "name", "price", "create_time", "user__username")
            res["data"] = list(orders)
        except Exception as e:
            res["code"] = 1002
            res["msg"] = e
        return JsonResponse(res)


class UserInfoView(APIView):
    """用户中心"""
    # authentication_classes = [Authtication]

    def get(self, request, *args, **kwargs):
        res = {"code": 1000, "msg": None, "data": None}
        try:
            # print(request.user)  # 用户对象
            # print(request.auth)  # 认证对象
            user = models.UserInfo.objects.filter(id=request.auth.user_id).values("id", "username", "password")
            res["data"] = list(user)
        except Exception as e:
            res["code"] = 1002
            res["msg"] = e
        return JsonResponse(res)
views.py

2、匿名用户配置

看源码我们知道，当认证方法返回None时，rest framework默认是从它的配置文件中读取匿名用户来赋值给self.user：

当读取到“UNAUTHENTICATED_USER”这个值时，就会使用这个值，所有我们可以在项目的配置文件中对这个值进行设置：

这样，当认证方法返回None时，self.user="匿名用户"

那个UNAUTHENTICATED_TOKEN也是同样的设置方法。

四、rest framework内置的认证类

在rest framework的authentication.py中：

from rest_framework.authentication import BaseAuthentication


class Authtication(BaseAuthentication):
    """自定制认证"""
    def authenticate(self, request):
        ......

    def authenticate_header(self, request):
        ......
View Code

BasicAuthentication认证类：是采用浏览器对用户名和密码进行base64加密，然后通过请求头发送给服务端，服务端获取请求头，对之前加密的用户名和密码进行解密，再到数据库进行校验。

五、rest framework权限使用

需求：给不同的视图设置不同的访问权限，如设置svip用户可以查看所有订单，普通用户和vip用户可以查看所有用户信息

1、源码实现流程

如果has_permission()返回True，则表示有权访问,否则无权访问。

 

2、权限控制的实现（局部）

创建权限类，在视图中使用

新建permission.py文件，写相关的权限控制类：

class MyPermissionSvip(object):
    """svip 访问权限控制"""
    message = "只有svip用户才能访问"  # 设置无权访问消息内容
    def has_permission(self, request, view):
        if request.user.user_type != 3:  # 如果用户类型不是svip 则拒绝访问
            return False
        return True


class MyPermissionOrdinaryAndVip(object):
    """普通用户和vip 访问权限控制"""
    def has_permission(self, request, view):
        if request.user.user_type == 3:  # 如果用户类型是svip 则拒绝访问
            return False
        return True
permission.py
import hashlib
import time

from django.http import JsonResponse
from rest_framework.views import APIView

from api import models
from api.utils.permission import MyPermissionSvip, MyPermissionOrdinaryAndVip


def md5(user):
    """生成token"""
    ctime = str(time.time())  # 当前时间
    m = hashlib.md5(bytes(user, encoding="utf-8"))
    m.update(bytes(ctime, encoding="utf-8"))
    return m.hexdigest()


class AuthView(APIView):
    """登录"""
    authentication_classes = []

    def post(self, request, *args, **kwargs):
        res = {"code": 1000, "msg": None}
        try:
            # 从请求中获取用户登录信息
            user = request._request.POST.get("username")
            pwd = request._request.POST.get("password")
            # 到数据库获取用户信息
            user_obj = models.UserInfo.objects.filter(username=user, password=pwd).first()
            # 如果获取到对象则说明认证成功，为登录用户创建token，如果认证失败则返回错误信息
            if not user_obj:
                res["code"] = 1001
                res["msg"] = "用户名或密码错误"
            else:
                token = md5(user)
                # 将token存入数据库：如果数据库存在token就更新，不存在就创建
                models.UserToken.objects.update_or_create(user=user_obj, defaults={"token": token})
                res["token"] = token
        except Exception as e:
            res["code"] = 1002
            res["msg"] = e
        return JsonResponse(res)


class OrderView(APIView):
    """订单"""
    # authentication_classes = [Authtication]  # 认证类（局部）
    permission_classes = [MyPermissionSvip,]  # 权限控制类（局部）

    def get(self, request, *args, **kwargs):
        res = {"code": 1000, "msg": None, "data": None}
        try:
            orders = models.Order.objects.all().values("id", "name", "price", "create_time", "user__username")
            res["data"] = list(orders)
        except Exception as e:
            res["code"] = 1002
            res["msg"] = e
        return JsonResponse(res)


class UserInfoView(APIView):
    """用户中心"""
    # authentication_classes = [Authtication]
    permission_classes = [MyPermissionOrdinaryAndVip]

    def get(self, request, *args, **kwargs):
        res = {"code": 1000, "msg": None, "data": None}
        try:
            # print(request.user)  # 用户对象
            # print(request.auth)  # 认证对象
            user = models.UserInfo.objects.all().values("id", "username", "password")
            res["data"] = list(user)
        except Exception as e:
            res["code"] = 1002
            res["msg"] = e
        return JsonResponse(res)
views.py

 3、权限控制的实现（全局）

 在settings中导入权限类的路径来实现全局控制,这样就不需要在每个视图中设置permission_classes了。

 

import hashlib
import time

from django.http import JsonResponse
from rest_framework.views import APIView

from api import models
from api.utils.permission import MyPermissionSvip, MyPermissionOrdinaryAndVip


def md5(user):
    """生成token"""
    ctime = str(time.time())  # 当前时间
    m = hashlib.md5(bytes(user, encoding="utf-8"))
    m.update(bytes(ctime, encoding="utf-8"))
    return m.hexdigest()


class AuthView(APIView):
    """登录"""
    authentication_classes = []
    permission_classes = []

    def post(self, request, *args, **kwargs):
        res = {"code": 1000, "msg": None}
        try:
            # 从请求中获取用户登录信息
            user = request._request.POST.get("username")
            pwd = request._request.POST.get("password")
            # 到数据库获取用户信息
            user_obj = models.UserInfo.objects.filter(username=user, password=pwd).first()
            # 如果获取到对象则说明认证成功，为登录用户创建token，如果认证失败则返回错误信息
            if not user_obj:
                res["code"] = 1001
                res["msg"] = "用户名或密码错误"
            else:
                token = md5(user)
                # 将token存入数据库：如果数据库存在token就更新，不存在就创建
                models.UserToken.objects.update_or_create(user=user_obj, defaults={"token": token})
                res["token"] = token
        except Exception as e:
            res["code"] = 1002
            res["msg"] = e
        return JsonResponse(res)


class OrderView(APIView):
    """订单"""
    # authentication_classes = [Authtication]  # 认证类（局部）
    # permission_classes = [MyPermissionSvip,]  # 权限控制类（局部）

    def get(self, request, *args, **kwargs):
        res = {"code": 1000, "msg": None, "data": None}
        try:
            orders = models.Order.objects.all().values("id", "name", "price", "create_time", "user__username")
            res["data"] = list(orders)
        except Exception as e:
            res["code"] = 1002
            res["msg"] = e
        return JsonResponse(res)


class UserInfoView(APIView):
    """用户中心"""
    # authentication_classes = [Authtication]
    permission_classes = [MyPermissionOrdinaryAndVip]

    def get(self, request, *args, **kwargs):
        res = {"code": 1000, "msg": None, "data": None}
        try:
            # print(request.user)  # 用户对象
            # print(request.auth)  # 认证对象
            user = models.UserInfo.objects.all().values("id", "username", "password")
            res["data"] = list(user)
        except Exception as e:
            res["code"] = 1002
            res["msg"] = e
        return JsonResponse(res)
views.py

 

六、rest framework内置的权限类

from rest_framework.permissions import BasePermission

在rest framework的permissions.py源码中：

建议在自定义权限控制类时，继承这个BasePermission类：

 

七、rest framework的访问频率控制

如：限制某个用户1分钟只能访问多少次

1、源码流程

如果allow_request()返回True，表示可以访问，否则表示频率太高，不能访问

2、需求：对用户登录进行频率控制

新建文件：

代码：

import time
from rest_framework.throttling import BaseThrottle


VISIT_RECORD = {}  # 存储用户访问记录


class VisitThrotlle(BaseThrottle):
    """用户登录访问频率控制"""
    def __init__(self):
        self.history = None

    def allow_request(self, request, view):
        # 获取用户IP
        # remote_addr = request.META.get('REMOTE_ADDR')
        remote_addr = self.get_ident(request)  # 也可以通过继承父类方法来获取IP
        
        ctime = time.time()  # 用户访问时间
        # 判断用户是否可以访问 如果该IP还没有访问过，直接放行；
        # 如果该IP已经存在于访问记录中，判断其访问频率是否达到上限
        if remote_addr not in VISIT_RECORD:
            VISIT_RECORD[remote_addr] = [ctime]
            return True
        self.history = VISIT_RECORD.get(remote_addr)  # 获取访问历史时间列表
        # 当前访问时间与访问记录中的时间进行比较，如果当前时间是在一分钟之后访问的，就删掉访问记录中的时间
        while self.history and self.history[-1] < ctime-60:
            self.history.pop()
        # 控制一分钟内允许访问3次
        if len(self.history) < 3:
            self.history.insert(0, ctime)  # 将最近的一次访问时间插入到列表第一个位置
            return True
        return False

    def wait(self):
        # 可以返回None，也可以返回时间，提示用户还要等多少秒就可以访问了
        ctime = time.time()
        return 60 - (ctime - self.history[-1])
throtlle.py

在views.py中引入：

import hashlib
import time

from django.http import JsonResponse
from rest_framework.views import APIView

from api import models
from api.utils.permission import MyPermissionSvip, MyPermissionOrdinaryAndVip
from api.utils.throtlle import VisitThrotlle


def md5(user):
    """生成token"""
    ctime = str(time.time())  # 当前时间
    m = hashlib.md5(bytes(user, encoding="utf-8"))
    m.update(bytes(ctime, encoding="utf-8"))
    return m.hexdigest()


class AuthView(APIView):
    """登录"""
    authentication_classes = []
    permission_classes = []
    throttle_classes = [VisitThrotlle]  # 访问频率控制

    def post(self, request, *args, **kwargs):
        res = {"code": 1000, "msg": None}
        try:
            # 从请求中获取用户登录信息
            user = request._request.POST.get("username")
            pwd = request._request.POST.get("password")
            # 到数据库获取用户信息
            user_obj = models.UserInfo.objects.filter(username=user, password=pwd).first()
            # 如果获取到对象则说明认证成功，为登录用户创建token，如果认证失败则返回错误信息
            if not user_obj:
                res["code"] = 1001
                res["msg"] = "用户名或密码错误"
            else:
                token = md5(user)
                # 将token存入数据库：如果数据库存在token就更新，不存在就创建
                models.UserToken.objects.update_or_create(user=user_obj, defaults={"token": token})
                res["token"] = token
        except Exception as e:
            res["code"] = 1002
            res["msg"] = e
        return JsonResponse(res)


class OrderView(APIView):
    """订单"""
    # authentication_classes = [Authtication]  # 认证类（局部）
    # permission_classes = [MyPermissionSvip,]  # 权限控制类（局部）

    def get(self, request, *args, **kwargs):
        res = {"code": 1000, "msg": None, "data": None}
        try:
            orders = models.Order.objects.all().values("id", "name", "price", "create_time", "user__username")
            res["data"] = list(orders)
        except Exception as e:
            res["code"] = 1002
            res["msg"] = e
        return JsonResponse(res)


class UserInfoView(APIView):
    """用户中心"""
    # authentication_classes = [Authtication]
    permission_classes = [MyPermissionOrdinaryAndVip]

    def get(self, request, *args, **kwargs):
        res = {"code": 1000, "msg": None, "data": None}
        try:
            # print(request.user)  # 用户对象
            # print(request.auth)  # 认证对象
            user = models.UserInfo.objects.all().values("id", "username", "password")
            res["data"] = list(user)
        except Exception as e:
            res["code"] = 1002
            res["msg"] = e
        return JsonResponse(res)
views.py

3、频率控制也可以做全局设置，方法与权限控制相同

4、内置控制频率的类

在rest_framework的throttling.py中:

1、示例1

在throtlle.py中继承SimpleRateThrottle类：

from rest_framework.throttling import BaseThrottle, SimpleRateThrottle


class VisitThrotlle(SimpleRateThrottle):
    """用户登录访问频率控制"""
    scope = "throtlle_rate"  # 定义一个key，从配置文件中获取访问频次

    def get_cache_key(self, request, view):
        # 程序回去Django的缓存中获取key，此时我们重写这个方法，给他返回一个用户IP作为key
        return self.get_ident(request)
throtlle.py

这个实现效果和上面自己写的一样。

2、示例2 

对登录用户做频率控制

from rest_framework.throttling import BaseThrottle, SimpleRateThrottle


class UserThrotlle(SimpleRateThrottle):
    """对已登录用户进行访问频率控制"""
    scope = "user_throtlle_rate"  # 定义一个key，从配置文件中获取访问频次

    def get_cache_key(self, request, view):
        # 返回一个用户唯一标志，如用户名
        return request.user.username


class VisitThrotlle(SimpleRateThrottle):
    """匿名用户登录访问频率控制"""
    scope = "throtlle_rate"  # 定义一个key，从配置文件中获取访问频次

    def get_cache_key(self, request, view):
        # 程序回去Django的缓存中获取key，此时我们重写这个方法，给他返回一个用户IP作为key
        return self.get_ident(request)
throtlle.py

setting.py：

 

推荐阅读