Python进阶42-drf框架(四)

• 三大认证准备工作
• 三大认证介绍
• 权限六表分析
• 六表数据测试
• 认证组件
  • 自定义认证类
• 权限组件
  • 自定义权限类
• 频率组件
  • 自定义频率类

-曾老湿, 江湖人称曾老大。

---

-多年互联网运维工作经验，曾负责过大规模集群架构自动化运维管理工作。 -擅长Web集群架构与自动化运维，曾负责国内某大型金融公司运维工作。 -devops项目经理兼DBA。 -开发过一套自动化运维平台（功能如下）： 1)整合了各个公有云API，自主创建云主机。 2)ELK自动化收集日志功能。 3)Saltstack自动化运维统一配置管理工具。 4)Git、Jenkins自动化代码上线及自动化测试平台。 5)堡垒机，连接Linux、Windows平台及日志审计。 6)SQL执行及审批流程。 7)慢查询日志分析web界面。

---

三大认证准备工作

---

创建项目



---

创建数据库

代码语言：javascript复制

mysql> create database dg_proj2;

---

设置数据库

代码语言：javascript复制

DATABASES = {
    'default': {
        'ENGINE': 'django.db.backends.mysql',
        'NAME': 'dg_proj2',
        'HOST':'10.0.0.51',
        'USER':'zls',
        'PASSWORD':'123'
        # 'NAME': os.path.join(BASE_DIR, 'db.sqlite3'),
    }
}

---

使用pymysql

代码语言：javascript复制

import pymysql
pymysql.install_as_MySQLdb()

---

设置时区

代码语言：javascript复制

LANGUAGE_CODE = 'zh-hans'

TIME_ZONE = 'Asia/Shanghai'

USE_I18N = True

USE_L10N = True

USE_TZ = False

---

路由分发

代码语言：javascript复制

from django.conf.urls import url, include
from django.contrib import admin

from django.views.static import serve
from django.conf import settings
urlpatterns = [
    url(r'^admin/', admin.site.urls),
    # 路由分发
    url(r'^api/', include('api.urls')),
]

---

设置Media及路由

代码语言：javascript复制

MEDIA_URL = '/media/'
MEDIA_ROOT = os.path.join(BASE_DIR,'media')

代码语言：javascript复制

from django.conf.urls import url, include
from django.contrib import admin

from django.views.static import serve
from django.conf import settings
urlpatterns = [
    url(r'^admin/', admin.site.urls),
    # 路由分发
    url(r'^api/', include('api.urls')),

    url(r'^media/(?P<path>.*)', serve, {'document_root': settings.MEDIA_ROOT}),
]

---

自定义exception

代码语言：javascript复制

REST_FRAMEWORK = {
    'EXCEPTION_HANDLER': 'utils.exception.exception_handler',
}

---

自定义Response

代码语言：javascript复制

from rest_framework.response import Response

"""
Response({
    'status':0,
    'msg':'ok',
    'results':[],
    'token':'', # 有 这样的额外的key-value数据结果
},status=http_status,headers=headers,exceptio=True|False)


## 我们想要的结果，就是在APIResponse上加上括号就出来最简单的结果
APIResponse() => Respone({'status':0, 'msg':'ok',})
"""


class APIResponse(Response):
    def __init__(self, data_status=0, data_msg='ok', results=None, http_status=None, headers=None, exception=False,
                 **kwargs):
        data = {
            'status': data_status,
            'msg': data_msg
        }

        ## results可能是False、0等数据，这些数据某些情况下，也会作为合法数据返回
        if results is not None:
            data['results'] = results

        if kwargs is not None:
            for k, v in kwargs.items():
                setattr(data, k, v)  ## 反射，相当于：data[k] = v
        super().__init__(data=data, status=http_status, headers=headers, exception=exception)

---

注册drf

代码语言：javascript复制

INSTALLED_APPS = [
    'django.contrib.admin',
    'django.contrib.auth',
    'django.contrib.contenttypes',
    'django.contrib.sessions',
    'django.contrib.messages',
    'django.contrib.staticfiles',
    'api.apps.ApiConfig',
    'rest_framework',
]

三大认证介绍

代码语言：javascript复制

"""
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)
    
"""

权限六表分析

基于用户权限访问的认证： RBAC （Role Based Acess Control）

自己简单了解：基于auth的认证规则

Django框架采用的是RBAC认证规则，RBAC认证规则通常分为三表规则、五表规则，Django采用的是六表规则

三表规则： 1.用户表 2.角色表 3.权限表

五表规则： 1.用户表 2.角色表 3.权限表 4.用户角色关联表 5.角色权限关联表

Django六表规则： 1.用户表 2.角色表 3.权限表 4.用户角色关联表 5.角色权限关联表 6.用户权限关联表

---

自定义权限六表

settings.py

代码语言：javascript复制

## 配置自定义user表
AUTH_USER_MODEL = 'api.User'

models.py

代码语言：javascript复制

from django.db import models

## 用户表：角色groups     权限user_permissions
## 角色表：用户user_set   权限permissions
## 权限表：用户user_set   角色group_set

from django.contrib.auth.models import AbstractUser

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

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

    def __str__(self):
        return self.username

数据库迁移

代码语言：javascript复制

MacBook-pro:dg_proj_n driverzeng$ python3 manage.py makemigrations
MacBook-pro:dg_proj_n driverzeng$ python3 manage.py migrate



六表数据测试

---

创建超级用户

代码语言：javascript复制

MacBook-pro:dg_proj_n driverzeng$ python3 manage.py createsuperuser
Username: admin
Email address:
Password:   admin12345
Password (again): admin12345
Superuser created successfully.

---

Django脚本文件

代码语言：javascript复制

import os,django

os.environ.setdefault("DJANGO_SETTINGS_MODULE", "dg_proj_n.settings")
django.setup()

---

六表数据关联测试

代码语言：javascript复制

import os,django

os.environ.setdefault("DJANGO_SETTINGS_MODULE", "dg_proj_n.settings")
django.setup()

from api import models
user = models.User.objects.first()
print(user.username)
print(user.groups.first().name)
print(user.user_permissions.first().name)

from django.contrib.auth.models import Group

group = Group.objects.first()

print(group.name)
print(group.user_set.first().username)
print(group.permissions.first().name)

from django.contrib.auth.models import Permission
p_16 = Permission.objects.filter(pk=15).first()
print(p_16.user_set.first().username)
p_17 = Permission.objects.filter(pk=17).first()
print(p_17.group_set.first().name)



认证组件

---

源码分析

代码语言：javascript复制

"""
#################### 认证组件 ####################
    Request类的 方法属性 user 的get方法 => self._authenticate() 完成认证
    
    认证的细则：
    # 做认证
    def _authenticate(self):
        # 遍历拿到一个个认证器，进行认证
        # self.authenticators配置的一堆认证类产生的认证类对象组成的 list
        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()

#################### 权限组件 ########################
    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)
                )
"""

自定义认证类

在api目录下创建authentications.py文件

代码语言：javascript复制

## 自定义认证类

# 1.继承BaseAuthentication
# 2.重写authenticate(self,request)方法，自定义认证规则
# 3.认证规则基于：
#   - 没有认证信息返回None（游客）
#   - 有认证信息认证失败抛异常（非法用户）
#   - 有认证信息认证成功返回用户与认证信息元组（合法用户）
# 4.完成视图类的配置：
#   - 全局（settings文件中）
#   - 局部（确切的视图类）
from rest_framework.exceptions import AuthenticationFailed
from rest_framework.authentication import BaseAuthentication
from . import models


class MyAuthentication(BaseAuthentication):
    def authenticate(self, request):
        # 前台在请求头中携带认证信息，且默认规范用Authorization字段携带认证信息
        # 后台固定在请求对象的META字典中获取
        auth = request.META.get('HTTP_AUTHORIZATION', None)

        # 处理游客
        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='admin').first()

        if not user:
            raise AuthenticationFailed('用户数据有误：非法用户')

        return (user, None)

全局配置

代码语言：javascript复制

REST_FRAMEWORK = {
    'EXCEPTION_HANDLER': 'utils.exception.exception_handler',
    
    ### 配置自己写的认证类
    'DEFAULT_AUTHENTICATION_CLASSES': [
        'api.authentications.MyAuthentication',
    ]
}

视图测试

代码语言：javascript复制

from rest_framework.views import APIView
from rest_framework.generics import GenericAPIView
from rest_framework.viewsets import GenericViewSet, ViewSet
from utils.response import APIResponse


class TestView(APIView):
    def get(self, request, *args, **kwargs):
        return APIResponse(0, 'Test Get OK')

路由设置

代码语言：javascript复制

from django.conf.urls import url
from . import views

urlpatterns = [
    url(r'^test/$', views.TestView.as_view())
]

没有带认证信息







如果认证通过，后台就可以从request.user中拿到用户

代码语言：javascript复制

from rest_framework.views import APIView
from rest_framework.generics import GenericAPIView
from rest_framework.viewsets import GenericViewSet, ViewSet
from utils.response import APIResponse


class TestView(APIView):
    def get(self, request, *args, **kwargs):
        print(request.user)
        return APIResponse(0, 'Test Get OK')



---

认证组件的小结

代码语言：javascript复制

"""
1) 创建继承BaseAuthentication的认证类
2) 实现authenticate方法
3) 实现体根据认证规则 确定游客、非法用户、合法用户
4) 进行全局或局部配置

认证规则
i.没有认证信息返回None(游客)
ii.有认证信息认证失败抛异常(非法用户)
iii.有认证信息认证成功返回用户与认证信息元组(合法用户)
"""

权限组件

代码语言：javascript复制

"""
1）AllowAny：
    认证规则全部返还True：return True
        游客与登陆用户都有所有权限

2) IsAuthenticated：
    认证规则必须有登陆的合法用户：return bool(request.user and request.user.is_authenticated)
        游客没有任何权限，登陆用户才有权限
    
3) IsAdminUser：
    认证规则必须是后台管理用户：return bool(request.user and request.user.is_staff)
        游客没有任何权限，登陆用户才有权限

4) IsAuthenticatedOrReadOnly
    认证规则必须是只读请求或是合法用户：
        return bool(
            request.method in SAFE_METHODS or
            request.user and
            request.user.is_authenticated
        )
        游客只读，合法用户无限制
"""

---

登录后才能通过认证：视图

代码语言：javascript复制

from rest_framework.views import APIView
from rest_framework.generics import GenericAPIView
from rest_framework.viewsets import GenericViewSet, ViewSet
from utils.response import APIResponse


class TestView(APIView):
    def get(self, request, *args, **kwargs):
        print(request.user)
        return APIResponse(0, 'Test Get OK')

## 只有登录后才能访问
from rest_framework.permissions import IsAuthenticated
class TestAuthenticatedAPIView(APIView):
    ## 局部认证
    permission_classes = [IsAuthenticated]
    def get(self, request, *args, **kwargs):
        # print(request.user)
        return APIResponse(0, '登陆成功 访问接口 OK')

---

登录后才能通过认证：路由

代码语言：javascript复制

from django.conf.urls import url
from . import views

urlpatterns = [
    url(r'^test/$', views.TestView.as_view()),
    url(r'^test1/$', views.TestAuthenticatedAPIView.as_view())
]





---

游客只读：视图

代码语言：javascript复制

from rest_framework.views import APIView
from rest_framework.generics import GenericAPIView
from rest_framework.viewsets import GenericViewSet, ViewSet
from utils.response import APIResponse


class TestView(APIView):
    def get(self, request, *args, **kwargs):
        print(request.user)
        return APIResponse(0, 'Test Get OK')

## 只有登录后才能访问
from rest_framework.permissions import IsAuthenticated
class TestAuthenticatedAPIView(APIView):
    ## 局部认证
    permission_classes = [IsAuthenticated]
    def get(self, request, *args, **kwargs):
        # print(request.user)
        return APIResponse(0, '登陆成功 访问接口 OK')

## 游客只读，增删改查必须登录
from rest_framework.permissions import IsAuthenticatedOrReadOnly
class TestAuthenticatedOrReadOnlyAPIView(APIView):
    ## 局部认证
    permission_classes = [IsAuthenticatedOrReadOnly]
    def get(self, request, *args, **kwargs):
        return APIResponse(0, '未登录，游客访问，读取 OK')

    def post(self,request,*args,**kwargs):
        return APIResponse(0, '已登录，登录用户，读写 OK')

---

游客只读：路由

代码语言：javascript复制

from django.conf.urls import url
from . import views

urlpatterns = [
    url(r'^test/$', views.TestView.as_view()),
    url(r'^test1/$', views.TestAuthenticatedAPIView.as_view()),
    url(r'^test2/$', views.TestAuthenticatedOrReadOnlyAPIView.as_view()),
]





自定义权限类

需求：游客只读，登录用户只读，管理员可以增删改查。

---

权限类文件

在api目录下创建，permissions.py文件，在里面定义权限类

代码语言：javascript复制

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.objects.filter(name='管理员').first()
        groups = request.user.groups.all()
        r2 = group and groups
        r3 = group in groups
        return r1 or (r2 and r3)

---

视图

代码语言：javascript复制

from rest_framework.views import APIView
from rest_framework.generics import GenericAPIView
from rest_framework.viewsets import GenericViewSet, ViewSet
from utils.response import APIResponse
from .permissions import MyPermission


class TestView(APIView):
    def get(self, request, *args, **kwargs):
        print(request.user)
        return APIResponse(0, 'Test Get OK')

## 需求：游客只读，登录用户只读，管理员可以增删改查。

from rest_framework.permissions import IsAuthenticatedOrReadOnly
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')

---

路由

代码语言：javascript复制

from django.conf.urls import url
from . import views

urlpatterns = [
    url(r'^test/$', views.TestView.as_view()),
    url(r'^test1/$', views.TestAuthenticatedAPIView.as_view()),
    url(r'^test2/$', views.TestAuthenticatedOrReadOnlyAPIView.as_view()),
    url(r'^test3/$', views.TestAdminOrReadOnlyAPIView.as_view()),
]









---

自定义权限类小结

代码语言：javascript复制

"""
1) 创建继承BasePermission的权限类
2) 实现has_permission方法
3) 实现体根据权限规则 确定有无权限
4) 进行全局或局部配置

认证规则
i.满足设置的用户条件，代表有权限，返回True
ii.不满足设置的用户条件，代表有权限，返回False
"""

频率组件

---

频率类源码

入口

代码语言：javascript复制

# 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)

自定义频率类

需求：短信接口 1/min 频率限制

---

步骤

代码语言：javascript复制

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

---

throttles 频率类

首先先在api目录下创建一个throttles.py

代码语言：javascript复制

from rest_framework.throttling import SimpleRateThrottle


class SMSRateThrottle(SimpleRateThrottle):
    scope = 'sms'

    ## 只对提交手机号的get方法进行限制
    def get_cache_key(self, request, view):
        mobile = request.query_params.get('mobile')
        ## 没有手机号 ，就不做 频率限制
        if not mobile:
            return None
        ## 返回可以根据手机号动态变化，且不易重复的字符串，作为操作缓存的key
        return 'throttle_%(scope)s_%(ident)s' % {'scope': self.scope, 'ident': mobile}

---

全局配置

settings.py配置

代码语言：javascript复制

REST_FRAMEWORK = {
    # 全局配置异常模块
    'EXCEPTION_HANDLER': 'utils.exception.exception_handler',
    # 认证类配置
    'DEFAULT_AUTHENTICATION_CLASSES': [
        'rest_framework.authentication.SessionAuthentication',
        'rest_framework.authentication.BasicAuthentication',
        'api.authentications.MyAuthentication',
    ],
    # 权限类配置
    'DEFAULT_PERMISSION_CLASSES': [
        'rest_framework.permissions.AllowAny',
    ],
    # 频率类配置
    'DEFAULT_THROTTLE_CLASSES': [],
    # 频率限制条件配置
    'DEFAULT_THROTTLE_RATES': {
        'user': '3/min',
        'anon': None,
        'sms': '3/min'
    },
}

---

局部配置

创建view.py视图

一般我们都会走局部配置

代码语言：javascript复制

from rest_framework.views import APIView
from rest_framework.generics import GenericAPIView
from rest_framework.viewsets import GenericViewSet, ViewSet
from utils.response import APIResponse
from .permissions import MyPermission
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')

---

路由

代码语言：javascript复制

from django.conf.urls import url
from . import views

urlpatterns = [
    url(r'^test/$', views.TestView.as_view()),
    url(r'^test1/$', views.TestAuthenticatedAPIView.as_view()),
    url(r'^test2/$', views.TestAuthenticatedOrReadOnlyAPIView.as_view()),
    url(r'^test3/$', views.TestAdminOrReadOnlyAPIView.as_view()),
    url(r'^sms/$', views.TestSMSAPIView.as_view()),
]





python 运维 django 自动化 api

0 人点赞