Django REST framework 认证、权限和频率组件

认证与权限频率组件

身份验证是将传入请求与一组标识凭据（例如请求来自的用户或其签名的令牌）相关联的机制。然后权限和限制组件决定是否拒绝这个请求。

简单来说就是：

认证确定了你是谁
权限确定你能不能访问某个接口
限制确定你访问某个接口的频率

一、认证组件

REST framework 提供了一些开箱即用的身份验证方案，并且还允许你实现自定义方案。

自定义Token认证

定义一个用户表和一个保存用户Token的表：

class UserInfo(models.Model):username = models.CharField(max_length=16)password = models.CharField(max_length=32)type = models.SmallIntegerField(choices=((1, '普通用户'), (2, 'VIP用户')),default=1)class Token(models.Model):user = models.OneToOneField(to='UserInfo')token_code = models.CharField(max_length=128)

定义一个登录视图：

def get_random_token(username):"""根据用户名和时间戳生成随机token:param username::return:"""import hashlib, timetimestamp = str(time.time())m = hashlib.md5(bytes(username, encoding="utf8"))m.update(bytes(timestamp, encoding="utf8"))return m.hexdigest()class LoginView(APIView):"""校验用户名密码是否正确从而生成token的视图"""def post(self, request):res = {"code": 0}print(request.data)username = request.data.get("username")password = request.data.get("password")user = models.UserInfo.objects.filter(username=username, password=password).first()if user:# 如果用户名密码正确token = get_random_token(username)models.Token.objects.update_or_create(defaults={"token_code": token}, user=user)res["token"] = tokenelse:res["code"] = 1res["error"] = "用户名或密码错误"return Response(res)

定义一个认证类

from rest_framework.authentication import BaseAuthentication
from rest_framework.exceptions import AuthenticationFailedclass MyAuth(BaseAuthentication):def authenticate(self, request):     # 必须实现authenticate方法，返回（认证之后的用户，认证的obj）if request.method in ["POST", "PUT", "DELETE"]:request_token = request.data.get("token", None)if not request_token:raise AuthenticationFailed('缺少token')token_obj = models.Token.objects.filter(token_code=request_token).first()if not token_obj:raise AuthenticationFailed('无效的token')return token_obj.user.username, Noneelse:return None, None

视图级别认证

class CommentViewSet(ModelViewSet):queryset = models.Comment.objects.all()serializer_class = app01_serializers.CommentSerializerauthentication_classes = [MyAuth, ]

全局级别认证

# 在settings.py中配置
REST_FRAMEWORK = {"DEFAULT_AUTHENTICATION_CLASSES": ["app01.utils.MyAuth", ]
}

二、权限组件

只有VIP用户才能看的内容。

自定义一个权限类

# 自定义权限
class MyPermission(BasePermission):message = 'VIP用户才能访问'def has_permission(self, request, view):"""必须实现has_permission，有权限返回True，无权限返回False"""# 因为在进行权限判断之前已经做了认证判断，所以这里可以直接拿到request.userif request.user and request.user.type == 2:  # 如果是VIP用户return Trueelse:return False

视图级别配置

class CommentViewSet(ModelViewSet):queryset = models.Comment.objects.all()serializer_class = app01_serializers.CommentSerializerauthentication_classes = [MyAuth, ]permission_classes = [MyPermission, ]

全局级别设置

# 在settings.py中设置rest framework相关配置项
REST_FRAMEWORK = {"DEFAULT_AUTHENTICATION_CLASSES": ["app01.utils.MyAuth", ],"DEFAULT_PERMISSION_CLASSES": ["app01.utils.MyPermission", ]
}

三、频率限制组件

DRF内置了基本的限制类，首先我们自己动手写一个限制类，熟悉下限制组件的执行过程。

自定义限制类

VISIT_RECORD = {}
# 自定义限制
class MyThrottle(object):def __init__(self):self.history = Nonedef allow_request(self, request, view): """必须实现allow_request，允许访问返回True，否则返回False自定义频率限制60秒内只能访问三次"""# 获取用户IPip = request.META.get("REMOTE_ADDR")timestamp = time.time()if ip not in VISIT_RECORD:VISIT_RECORD[ip] = [timestamp, ]return Truehistory = VISIT_RECORD[ip]self.history = historyhistory.insert(0, timestamp)while history and history[-1] < timestamp - 60:history.pop()if len(history) > 3:return Falseelse:return Truedef wait(self):"""限制时间还剩多少"""timestamp = time.time()return 60 - (timestamp - self.history[-1])

视图使用

class CommentViewSet(ModelViewSet):queryset = models.Comment.objects.all()serializer_class = app01_serializers.CommentSerializerthrottle_classes = [MyThrottle, ]

全局使用

# 在settings.py中设置rest framework相关配置项
REST_FRAMEWORK = {"DEFAULT_AUTHENTICATION_CLASSES": ["app01.utils.MyAuth", ],"DEFAULT_PERMISSION_CLASSES": ["app01.utils.MyPermission", ],"DEFAULT_THROTTLE_CLASSES": ["app01.utils.MyThrottle", ]
}

使用内置限制类

from rest_framework.throttling import SimpleRateThrottleclass VisitThrottle(SimpleRateThrottle):scope = "xxx"def get_cache_key(self, request, view):return self.get_ident(request)

全局配置

# 在settings.py中设置rest framework相关配置项
REST_FRAMEWORK = {"DEFAULT_AUTHENTICATION_CLASSES": ["app01.utils.MyAuth", ],# "DEFAULT_PERMISSION_CLASSES": ["app01.utils.MyPermission", ]"DEFAULT_THROTTLE_CLASSES": ["app01.utils.VisitThrottle", ],"DEFAULT_THROTTLE_RATES": {"xxx": "5/m",}
}

认证类源码

############################ authentication.py ####################################
from __future__ import unicode_literalsimport base64
import binasciifrom django.contrib.auth import authenticate, get_user_model
from django.middleware.csrf import CsrfViewMiddleware
from django.utils.six import text_type
from django.utils.translation import ugettext_lazy as _from rest_framework import HTTP_HEADER_ENCODING, exceptionsdef get_authorization_header(request):"""Return request's 'Authorization:' header, as a bytestring.Hide some test client ickyness where the header can be unicode."""auth = request.META.get('HTTP_AUTHORIZATION', b'')if isinstance(auth, text_type):# Work around django test client oddnessauth = auth.encode(HTTP_HEADER_ENCODING)return authclass CSRFCheck(CsrfViewMiddleware):def _reject(self, request, reason):# Return the failure reason instead of an HttpResponsereturn reasonclass BaseAuthentication(object):"""All authentication classes should extend 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 theauthentication scheme should return `403 Permission Denied` responses."""passclass BasicAuthentication(BaseAuthentication):"""HTTP Basic authentication against username/password."""www_authenticate_realm = 'api'def authenticate(self, request):"""Returns a `User` if a correct username and password have been suppliedusing HTTP Basic authentication.  Otherwise returns `None`."""auth = get_authorization_header(request).split()if not auth or auth[0].lower() != b'basic':return Noneif len(auth) == 1:msg = _('Invalid basic header. No credentials provided.')raise exceptions.AuthenticationFailed(msg)elif len(auth) > 2:msg = _('Invalid basic header. Credentials string should not contain spaces.')raise exceptions.AuthenticationFailed(msg)try:auth_parts = base64.b64decode(auth[1]).decode(HTTP_HEADER_ENCODING).partition(':')except (TypeError, UnicodeDecodeError, binascii.Error):msg = _('Invalid basic header. Credentials not correctly base64 encoded.')raise exceptions.AuthenticationFailed(msg)userid, password = auth_parts[0], auth_parts[2]return self.authenticate_credentials(userid, password, request)def authenticate_credentials(self, userid, password, request=None):"""Authenticate the userid and password against username and passwordwith optional request for context."""credentials = {get_user_model().USERNAME_FIELD: userid,'password': password}user = authenticate(request=request, **credentials)if user is None:raise exceptions.AuthenticationFailed(_('Invalid username/password.'))if not user.is_active:raise exceptions.AuthenticationFailed(_('User inactive or deleted.'))return (user, None)def authenticate_header(self, request):return 'Basic realm="%s"' % self.www_authenticate_realmclass SessionAuthentication(BaseAuthentication):"""Use Django's session framework for authentication."""def authenticate(self, request):"""Returns a `User` if the request session currently has a logged in user.Otherwise returns `None`."""# Get the session-based user from the underlying HttpRequest objectuser = getattr(request._request, 'user', None)# Unauthenticated, CSRF validation not requiredif not user or not user.is_active:return Noneself.enforce_csrf(request)# CSRF passed with authenticated userreturn (user, None)def enforce_csrf(self, request):"""Enforce CSRF validation for session based authentication."""check = CSRFCheck()# populates request.META['CSRF_COOKIE'], which is used in process_view()check.process_request(request)reason = check.process_view(request, None, (), {})if reason:# CSRF failed, bail with explicit error messageraise exceptions.PermissionDenied('CSRF Failed: %s' % reason)class TokenAuthentication(BaseAuthentication):"""Simple token based authentication.Clients should authenticate by passing the token key in the "Authorization"HTTP header, prepended with the string "Token ".  For example:Authorization: Token 401f7ac837da42b97f613d789819ff93537bee6a"""keyword = 'Token'model = Nonedef get_model(self):if self.model is not None:return self.modelfrom rest_framework.authtoken.models import Tokenreturn Token"""A custom token model may be used, but must have the following properties.* key -- The string identifying the token* user -- The user to which the token belongs"""def authenticate(self, request):auth = get_authorization_header(request).split()if not auth or auth[0].lower() != self.keyword.lower().encode():return Noneif len(auth) == 1:msg = _('Invalid token header. No credentials provided.')raise exceptions.AuthenticationFailed(msg)elif len(auth) > 2:msg = _('Invalid token header. Token string should not contain spaces.')raise exceptions.AuthenticationFailed(msg)try:token = auth[1].decode()except UnicodeError:msg = _('Invalid token header. Token string should not contain invalid characters.')raise exceptions.AuthenticationFailed(msg)return self.authenticate_credentials(token)def authenticate_credentials(self, key):model = self.get_model()try:token = model.objects.select_related('user').get(key=key)except model.DoesNotExist:raise exceptions.AuthenticationFailed(_('Invalid token.'))if not token.user.is_active:raise exceptions.AuthenticationFailed(_('User inactive or deleted.'))return (token.user, token)def authenticate_header(self, request):return self.keywordclass RemoteUserAuthentication(BaseAuthentication):"""REMOTE_USER authentication.To use this, set up your web server to perform authentication, which willset the REMOTE_USER environment variable. You will need to have'django.contrib.auth.backends.RemoteUserBackend in yourAUTHENTICATION_BACKENDS setting"""# Name of request header to grab username from.  This will be the key as# used in the request.META dictionary, i.e. the normalization of headers to# all uppercase and the addition of "HTTP_" prefix apply.header = "REMOTE_USER"def authenticate(self, request):user = authenticate(remote_user=request.META.get(self.header))if user and user.is_active:return (user, None)