1.1 api认证原理介绍
1、api认证原理：客户端生成秘钥
1) 客户端与服务器端有一个相同的字符串作为auth_key
2) 客户端使用encryption=”auth_key|time.time()”用auth_key和客户端时间生成md5秘钥
3) 客户端将”encryption|time.time()” 将生成的秘钥和发送请求的时间一起发送给服务器
2、api认证原理：服务器验证秘钥三步曲
1、时间验证： 如果请求到达服务器端时间距离客户端发送请求的时间大于2秒，验证失败
2、是否存在： 这个秘钥第一次过来时会加入列表中，如果已经在列表中证明已经来过，验证失败
3、秘钥正确： 只有前两关都通过，在通过服务器的auth_key与客户端发送的时间生成秘钥看是否相等
1.2 api认证插件使用
1、测试API认证插件步骤
1、创建任意django项目，如ApiAuth，无需指定静态文件等，只需创建app01
2、在app01中使用CBV创建AssetView视图函数处理get和post请求，并重写dispatch方法，不使用csrf规则
3、在app01同级目录下创建utils目录，并创建认证文件auth.py
4、在AssetView的get和post函数中使用自己定义的@method_decorator(auth.api_auth)进行api认证
5、客户端访问时只用使用服务器端指定的字符串，和发送请求的时间生成秘钥，验证通过才能访问
2、测试代码
import time
import hashlib
from django.http import JsonResponseASSET_AUTH_KEY = '299095cc-1330-11e5-b06a-a45e60bec08b'
ASSET_AUTH_HEADER_NAME = 'HTTP_AUTH_KEY'
ASSET_AUTH_TIME = 2ENCRYPT_LIST = [
# {'time': 1516180308.850576, 'encrypt': '25bd4a6454329e7633b4db5c810c44ac'}
]def api_auth_method(request):
# 获取客户端header中传入的验证字符串：{'auth-key': '7b01e466ca23c9a3fb8c3a12b0214635|1516179535.832950'}
auth_key = request.META.get('HTTP_AUTH_KEY')
if not auth_key:
return False
sp = auth_key.split('|')
if len(sp) != 2:
return False
encrypt, timestamp = sp # encrypt = 客户端发送时间 + 授权字符串 的md5值
# timestamp 是客户端发送的时间
timestamp = float(timestamp)
limit_timestamp = time.time() - ASSET_AUTH_TIME if limit_timestamp > timestamp: # 如果当前时间比请求时间的差大于2 秒(验证失败)
return False
ha = hashlib.md5(ASSET_AUTH_KEY.encode('utf-8'))
# 授权字符串 + 客户端发送请求那一刻时间 的md5值与客户端发送的值是否相等
ha.update(bytes("%s|%f" % (ASSET_AUTH_KEY, timestamp), encoding='utf-8'))
result = ha.hexdigest()
if encrypt != result:
return False exist = False
del_keys = []
for k, v in enumerate(ENCRYPT_LIST):
# v = {'time': 1516180308.850576, 'encrypt': '25bd4a6454329e7633b4db5c810c44ac'}
m = v['time']
n = v['encrypt']
if m < limit_timestamp: # 时间已过期
del_keys.append(k) # 将ENCRYPT_LIST列表中对应要删除的索引加入到del_keys列表中
continue
if n == encrypt:
exist = True
for k in reversed(del_keys): # 将ENCRYPT_LIST列表中已到的秘钥删除
del ENCRYPT_LIST[k] if exist:
return False
ENCRYPT_LIST.append({'encrypt': encrypt, 'time': timestamp})
return Truedef api_auth(func):
def inner(request, *args, **kwargs):
if not api_auth_method(request): # 调用api_auth_method方法验证客户端秘钥是否有效
return JsonResponse({'code': 1001, 'message': 'API授权失败'}, json_dumps_params={'ensure_ascii': False})
return func(request, *args, **kwargs) return inner
Project/utils/auth.py 用于api认证的装饰器(插件代码)
from django.conf.urls import url
from django.contrib import admin
from app01 import viewsurlpatterns = [
url(r'^admin/', admin.site.urls),
url(r'^asset/', views.AssetView.as_view()),
]
urls.py
from django.shortcuts import HttpResponse
from django.utils.decorators import method_decorator
from django.views import View
from django.views.decorators.csrf import csrf_exempt
import jsonfrom utils import authclass AssetView(View): @method_decorator(csrf_exempt) # 指定Home这个视图函数不使用csrf
def dispatch(self, request, *args, **kwargs):
result = super(AssetView,self).dispatch(request, *args, **kwargs)
return result @method_decorator(auth.api_auth)
def get(self,request):
return HttpResponse('api认证成功') @method_decorator(auth.api_auth)
def post(self,request):
print(request.POST)
print(request.body)
return HttpResponse(json.dumps({'status':'ok'}))
views.py使用插件进行API验证
import hashlib,time,requests,json
key = '299095cc-1330-11e5-b06a-a45e60bec08b' # 与服务器端加密字符串相同# auth_key函数用来生成验证秘钥
def auth_key():
ha = hashlib.md5(key.encode('utf-8'))
time_span = time.time()
ha.update(bytes("%s|%f" % (key, time_span), encoding='utf-8'))
encryption = ha.hexdigest()
result = "%s|%f" % (encryption, time_span)
return {'auth-key': result}####################### 1、发送get请求 ###########################
headers = {}
headers.update(auth_key())
response_get = requests.get( # 获取今天未采集资产的主机名列表
url='http://127.0.0.1:8000/asset/',
headers=headers,
data={'user': 'alex', 'pwd': ''}, # 通过请求体传递数据：post方式
)print('get:',response_get.text)####################### 2、发送post请求 ###########################
headers = {}
headers.update(auth_key())
response_post = requests.post( # 获取今天未采集资产的主机名列表
url='http://127.0.0.1:8000/asset/',
headers=headers,
json={'user': 'tom', 'pwd': ''}, # 通过请求体传递数据：post方式
)print('post:',response_post.text)
在任意位置创建客户端test.py文件进行测试
3、api认证精简版原理解析
from django.shortcuts import HttpResponse
from django.utils.decorators import method_decorator
from django.views import View
from django.views.decorators.csrf import csrf_exempt
import json,time,hashlibAPPID = "afafdsafdsfsdfdsa"visited = []class AssetView(View): @method_decorator(csrf_exempt) # 指定Home这个视图函数不使用csrf
def dispatch(self, request, *args, **kwargs):
result = super(AssetView,self).dispatch(request, *args, **kwargs)
return result def get(self,request):
# req_appid = "d23672549e553b031756be33b1314573|1516174765.7812178"
req_appid = self.request.META.get('HTTP_APPID',None)
v, client_time = req_appid.split('|') current_time = time.time()
float_client_time = float(client_time) # 第一关：如果请求过来的时间距当前时间大于10秒，验证失败
if current_time - 10 > float_client_time:
return HttpResponse("验证失败") # 第二关：如果这个请求已经来过一次了，验证失败
if req_appid in visited:
return HttpResponse("验证失败") m = hashlib.md5()
m.update(bytes(APPID + client_time, encoding='utf-8'))
new_appid = m.hexdigest() # 第三关：判断v是否等于 APPID + client_time 的md5值
if new_appid == v:
visited.append(new_appid)
return HttpResponse('...')
else:
return HttpResponse('去你的吧') def post(self,request): return HttpResponse(json.dumps({'status':'ok'}))
服务器端：views.py
appid = "afafdsafdsfsdfdsa"import requests,hashlib,timecurrent_time = str(time.time())m = hashlib.md5()
m.update(bytes(appid + current_time,encoding='utf-8'))
new_appid = m.hexdigest()new_new_id = "%s|%s"%(new_appid,current_time)print(new_new_id)response = requests.get('http://127.0.0.1:8000/asset/',
# params={'appid':appid},
headers={'appid':new_new_id},
)response.encoding = 'utf-8'print('response',response.text)response2 = requests.post('http://127.0.0.1:8000/asset/',
# params={'appid':appid},
headers={'appid':new_new_id},
)
print(response2.text)
客户端test.py
