前言

如果我们不用使用

drf

那套认证规则，我们想自定义认证类，那么我们首先要知道，

drf

本身是如何定义认证规则的，也就是要查看它的源码是如何写的
 

源码分析

源码的入口在

APIView.py

文件下的

dispatch

方法下的

self.initial

方法中的

self.perform_authentication(request)

，点击查看后如下

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

的

user

方法，

request

代表的是

drf

的

Request

，所以我们进入

drf

的

Request

类中查找

user

方法属性，源码如下:

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

上述代码的意思是：返回与当前请求关联的用户，由提供给请求的身份验证类进行身份验证。如果没有用户，我们需要通过

_authenticate

方法验证，我们查看下它的源码

def _authenticate(self):
"""
尝试依次使用每个身份验证实例对请求进行身份验证。
"""
for authenticator in self.authenticators:
try:
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()

我们可以看到

self.authenticators

验证器其实是调用父类

APIView

的

authenticators

，

APIView

的

authenticators

在源码

initialize_request

方法下的

get_authenticators

，我们查看源码

def get_authenticators(self):
"""
Instantiates and returns the list of authenticators that this view can use.
"""
return [auth() for auth in self.authentication_classes]

再点击

authentication_classes

查看

authentication_classes = api_settings.DEFAULT_AUTHENTICATION_CLASSES

我们就知道了

drf

默认的认证器在

settings

文件下的

DEFAULT_AUTHENTICATION_CLASSES

类下面

'DEFAULT_AUTHENTICATION_CLASSES': [
'rest_framework.authentication.SessionAuthentication',
'rest_framework.authentication.BasicAuthentication'
],

我们发现

drf

默认有2个认证类一个基础的认证，另一个

session

认证，这两个认证类都继承自

BaseAuthentication

，我们来看下源码

class BaseAuthentication:
"""
所有的认证类都继承自BaseAuthentication.
"""

def authenticate(self, request):
"""
认证请求返回一个二元组(user, token)，并且此方法必须重写，否则抛出异常
"""
raise NotImplementedError(".authenticate() must be overridden.")

def authenticate_header(self, request):

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

接下来我们看下

BasicAuthentication

如何写的，后续我们依葫芦画瓢

class BasicAuthentication(BaseAuthentication):
"""
针对用户名密码的 HTTP 基本身份验证
"""
www_authenticate_realm = 'api'

def authenticate(self, request):
"""
如果使用 HTTP 基本身份验证提供了正确的用户名和密码，则返回“User”。否则返回“None”。
"""
# 获取请求头中`HTTP_AUTHORIZATION`，并进行分割
auth = get_authorization_header(request).split()

# 如果没有auth或者auth的第一个索引值的小写不等于basic，则返回None
if not auth or auth[0].lower() != b'basic':
return None

# auth列表的长度必须等于2，格式['basic', 'abc.xyz.123']
# 如果auth的长度等于1，则抛出异常
if len(auth) == 1:
msg = _('Invalid basic header. No credentials provided.')
raise exceptions.AuthenticationFailed(msg)
# 如果长度大于2，也抛出异常
elif len(auth) > 2:
msg = _('Invalid basic header. Credentials string should not contain spaces.')
raise exceptions.AuthenticationFailed(msg)

try:
try:
# auth[1]解码格式为utf-8
auth_decoded = base64.b64decode(auth[1]).decode('utf-8')
except UnicodeDecodeError:
auth_decoded = base64.b64decode(auth[1]).decode('latin-1')
auth_parts = auth_decoded.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 password
with 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.Authen
ad0
ticationFailed(_('User inactive or deleted.'))

return (user, None)

def authenticate_header(self, request):
return 'Basic realm="%s"' % self.www_authenticate_realm

 

自定义认证类

1. 创建继承BaseAuthentication的认证类
2. 实现authenticate方法
3. 实现体根据认证规则 确定 游客 正常用户 非法用户
4. 进行全局或局部配置（一般采用全局配置）

认证规则

1. 没有认证信息，返回

   None

   （游客）
2. 有认证信息认证失败，抛异常（非法用户）
3. 有认证信息认证成功，返回用户和认证信息的元组（合法用户）

我们创建一个文件夹

authentications

，写入如下代码

from rest_framework.authentication import BaseAuthentication
from rest_framework.exceptions import AuthenticationFailed
from api.models import User

class MyAuthentications(BaseAuthentication):

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

# 处理游客
if auth is None:
return None

auth_list = auth.split()
if not len(auth_list) == 2 and auth_list[0].lower() == "auth":
raise AuthenticationFailed("认证信息有误，非法用户")
# 合法的用户还需要从auth_list[1]中解析出来
# 注：假设一种情况，信息为xx.yy.zz，就可以解析出admin用户：实际开发，该逻辑一定是校验用户的正常逻辑
if auth_list[1] != 'xx.yy.zz':  # 校验失败
raise AuthenticationFailed("用户校验失败，非法用户")

user = User.objects.filter(username='jkc').first()
print(user)

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

return user, None

然后在

settings.py

中配置全局的自定义认证类

REST_FRAMEWORK = {
'DEFAULT_AUTHENTICATION_CLASSES': [
'api.authentications.MyAuthentications'
],
}

最后写入视图函数

class TestView(APIView):
def get(self, request, *args, **kwargs):
return APIResponse(data_msg="drf get ok")

然后我们访问视图，在

headers

中不传

Authorization

代表游客，游客可以访问成功

{
"statusCode": 0,
"message": "drf get ok"
}

接着我们在请求头中只传

auth

访问视图会抛出异常信息

{
"detail": "认证信息有误，非法用户"
}
然后我们在请求头中传入错误的认证，
auth 111






访问视图会抛出异常信息

{
"detail": "用户校验失败，非法用户"
}
最后我们在请求头中传入正确的认证，
auth xx.yy.zz
，这次会得到正确的返回结果

{
"statusCode": 0,
"message": "drf get ok"
}
以上的测试，就代表我们自定义的认证类起作用了