認證與權限頻率組件
身份驗證是將傳入請求與一組標識憑據（例如請求來自的用戶或其簽名的令牌）相關聯的機制。然后 權限 和 限制 組件決定是否拒絕這個請求。
簡單來說就是：
- 認證確定了你是誰
- 權限確定你能不能訪問某個接口
- 限制確定你訪問某個接口的頻率
一、認證組件
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)?
)
