《Web接口开发与自动化测试基于Python语言》--第11章

第11章 接口的安全机制

本章将介绍接口的几种常用的安全机制。

11.1 用户认证

接口测试工具的User Auth/Authorization选项，是包含在request请求中的。

11.1.1 开发带Auth接口

为了练习与安全有关的接口开发，下面重新在sign应用下创建views_if_sec.py视图文件：

#! /usr/bin/python
# -*- coding:utf-8 -*-

from django.contrib import auth as django_auth
import base64

# 用户认证
def user_auth(request):
get_http_auth = request.META.get('HTTP_AUTHORIZATION', b'')
auth = get_http_auth.split()
try:
auth_parts = base64.b64decode(auth[1]).decode('utf-8').partition(':')
except IndexError:
return "null"
username, password = auth_parts[0], auth_parts[2]
user = django_auth.authenticate(username=username, password=password)
if user is not None:
django_auth.login(request, user)
return "success"
else:
return "fail"

对上述代码进行分析：

request.META是一个字典，包含了本次HTTP请求的Header信息，eg：用户认证、IP地址、用户Agent（通常是浏览器的名称和版本号）等；HTTP_AUTHORIZATION用于获取HTTP认证数据，如果为空，将得到一个空的bytes对象

当客户端传输的认证数据为：admin/admin123456，这里得到的数据为：Basic YWRtaW46YWRtaW4xMjM0NTY=

通过split()方法将其拆分成list，拆分后的数据为：[‘Basic’, ‘YWRtaW46YWRtaW4xMjM0NTY=’]

接下来，取出list中的加密串，通过base64对加密串进行解码，通过decode()方法以UTF-8编码对字符串进行解码，partition()方法以冒号“:”为分隔符对字符串进行分隔，得到的数据为：(‘admin’, ‘:’, ‘admin123456’)

然后通过try…except…进行异常处理，如果获取不到Auth数据，则抛IndexError类型的异常，函数返回“null”字符串

最后，取出auth_parts元组中对应认证的username和password，最终的数据是：admin、admin123456

最后的最后，调用Django的认证模块，对得到的Auth信息进行验证，若成功则返回“success”，失败则返回“fail”

在发布会查询接口调用上面的user_auth函数：

from django.http import JsonResponse

# 查询发布会接口-
1a342
-增加用户认证
def get_event_list(request):
auth_result = user_auth(request)    #调用用户认证函数
if auth_result == "null":
return JsonResponse({'status':10011, 'message':'user auth null'})

if auth_result == "fail":
return JsonResponse({'status':10012, 'message':'user auth fail'})

eid = request.GET.get("eid", "")    # 发布会id
name = request.GET.get("name", "")  # 发布会名称

if eid == '' and name == '':
return JsonResponse({'status':10021, 'message':'parameter error'})

if eid != '':
event = {}
try:
result = Event.objects.get(id=eid)
except ObjectDoesNotExist:
return JsonResponse({'status':10022, 'message':'query result is empty'})
else:
event['name'] = result.name
event['limit'] = result.limit
event['status'] = result.status
event['address'] = result.address
event['start_time'] = result.start_time
return JsonResponse({'status':200, 'message':'success', 'data':event})
if name != '':
datas = []
results = Event.objects.filter(name__contains=name)
if results:
for r in results:
event = {}
event['name'] = r.name
event['limit'] = r.limit
event['status'] = r.status
event['address'] = r.address
event['start_time'] = r.start_time
datas.append(event)
return JsonResponse({'status':200, 'message':'success', 'data':datas})
else:
return JsonResponse({'status':10022, 'message':'query result is empty'})

这样就完成了在接口中增加认证机制的功能，只有通过认证才能继续测试接口全部内容。

11.1.2 接口文档

注意： 在urls.py中增加该接口，代码如下：

#!/usr/bin/python
# -*- coding:utf-8 -*-

from django.conf.urls import url
from sign import views_if, views_if_sec

urlpatterns = [
......
# ex: /api/get_event_list/
url(r'^get_event_list', views_if.get_event_list, name='get_event_list'),
url(r'sec_get_event_list', views_if_sec.get_event_list, name='sec_get_event_list'),
......

11.1.3 接口测试用例

编写对应的测试用例：sec_test_case.py

Ps：Requests库的get()和post()方法均提供auth参数用于设置用户签名。

#!/usr/bin/python
# -*- coding:utf-8 -*-

import unittest
import requests

class GetEventListTest(unittest.TestCase):
"""查询发布会信息带用户认证"""

def setUp(self):
self.base_url = "http://10.18.214.88:8000/api/sec_get_event_list/"

def test_get_event_list_auth_null(self):
"""auth为空"""
r = requests.get(self.base_url, params={'eid':1})
result = r.json()
self.assertEqual(result['status'], 10011)
self.assertEqual(result['message'], 'user auth null')

if __name__ == '__main__':
unittest.main()

没有把全部的case写完，就先试一下效果，结果运行测试，返回的结果却是错误：

E
======================================================================
ERROR: test_get_event_list_auth_null (__main__.GetEventListTest)
auth为空
----------------------------------------------------------------------
Traceback (most recent call last):
File "sec_test_case.py", line 27, in test_get_event_list_auth_null
result = r.json()
File "/usr/local/lib/python2.7/dist-packages/requests/models.py", line 892, in json
return complexjson.loads(self.text, **kwargs)
File "/usr/lib/python2.7/dist-packages/simplejson/__init__.py", line 516, in loads
return _default_decoder.decode(s)
File "/usr/lib/python2.7/dist-packages/simplejson/decoder.py", line 370, in decode
obj, end = self.raw_decode(s)
File "/usr/lib/python2.7/dist-packages/simplejson/decoder.py", line 400, in raw_decode
return self.scan_once(s, idx=_w(s, idx).end())
JSONDecodeError: Expecting value: line 2 column 1 (char 1)

----------------------------------------------------------------------
Ran 1 test in 0.112s

FAILED (errors=1)

观察django那边返回的错误，请求的API返回了500错误，奇怪哪里出错了，但是在测试用例的返回信息里，实在是没看出哪里有误啊，不过好在django那的提示信息比较有用：

Internal Server Error: /api/sec_get_event_list/
Traceback (most recent call last):
File "/usr/local/lib/python2.7/dist-packages/django/core/handlers/exception.py", line 39, in inner
response = get_response(request)
File "/usr/local/lib/python2.7/dist-packages/django/core/handlers/base.py", line 187, in _get_response
response = self.process_exception_by_middleware(e, request)
File "/usr/local/lib/python2.7/dist-packages/django/core/handlers/base.py", line 185, in _get_response
response = wrapped_callback(request, *callback_args, **callback_kwargs)
File "/home/csg/guest/sign/views_if_sec.py", line 27, in get_event_list
return JsonResponse({'status':10011, 'message':'user auth null'})
NameError: global name 'JsonResponse' is not defined
[27/Sep/2017 14:05:03] "GET /api/sec_get_event_list/?eid=1 HTTP/1.1" 500 61568

从这里就很明显了，原来我们在views_if_sec.py里没有定义JsonResponse，好吧我的错，忘记导入了，补充到该文件，重新运行测试用例，成功：

.
----------------------------------------------------------------------
Ran 1 test in 0.016s

OK

将测试用例补充完整即可：

#!/usr/bin/python
# -*- coding:utf-8 -*-

#########################################################
# (C) 2000-2017 NSFOCUS Corporation. All rights Reserved#
#########################################################

"""
Function:
Create Time: 2017年09月26日 星期二 22时50分33秒
Author: zhaoxinzhen@intra.nsfocus.com
"""

import unittest
import requests

class GetEventListTest(unittest.TestCase):
"""查询发布会信息带用户认证"""

def setUp(self):
self.base_url = "http://10.18.214.88:8000/api/sec_get_event_list/"

def test_get_event_list_auth_null(self):
"""auth为空"""
r = requests.get(self.base_url, params={'eid':1})
result = r.json()
self.assertEqual(result['status'], 10011)
self.assertEqual(result['message'], 'user auth null')

def test_get_event_list_auth_error(self):
"""auth错误"""
auth_user = ('abc', '123')
r = requests.get(self.base_url, auth=auth_user, params={'eid':1})
result = r.json()
self.assertEqual(result['status'], 10012)
self.assertEqual(result['message'], 'user auth fail')

def test_get_event_list_eid_null(self):
"""eid参数为空"""
auth_user = ('admin', 'admin123456')
r = requests.get(self.base_url, auth=auth_user, params={'eid':''})
result = r.json()
self.assertEqual(result['status'], 10021)
self.assertEqual(result['message'], 'parameter error')

def test_get_event_list_eid_success(self):
"""根据eid查询结果成功"""
auth_user = ('admin', 'admin123456')
r = requests.get(self.base_url, auth=auth_user, params={'eid':1})
result = r.json()
self.assertEqual(result['status'], 200)
self.assertEqual(result['message'], 'success')
self.assertEqual(result['data']['name'], u'小米5发布会')
self.assertEqual(result['data']['address'], u'北京国家会议中心')

if __name__ == '__main__':
unittest.main()

11.2 数字签名

在使用HTTP/SOAP协议传输数据时，签名作为其中一个参数，有着重要的作用：

鉴权： 通过客户端的密钥和服务端的密钥匹配。

如何理解鉴权呢，举个例子：

向接口传参：http://127.0.0.1:8000/api/?a=1&b=2

假设签名的密钥为：@admin123

加上签名之后的接口传参为：http://127.0.0.1:8000/api/?a=1&b=2&sign=@admin123

显然这样明文传输sign参数是不安全的，一般会通过加密算法进行加密，eg：MD5

>>> import hashlib
>>> md5 = hashlib.md5()
>>> sign_str = "@admin123"
>>> sign_bytes_utf8 = sign_str.encode(encoding="utf-8")
>>> md5.update(sign_bytes_utf8)
>>> md5.hexdigest()
'4b9db269c5f978e1264480b0a7619eea'
>>>

似曾相识啊，目前正在测试的第三方应用接口就是使用的这种数字签名方式。

将“@admin123”通过MD5加密之后得到：4b9db269c5f978e1264480b0a7619eea

单独作为鉴权，带签名的接口为：http://127.0.0.1:8000/api/?a=1&b=2&sign=4b9db269c5f978e1264480b0a7619eea

使用MD5加密算法，好处是不可逆，当服务器接收到参数后，同样需要对“@admin123”进行MD5加密，然后与调用者传来的sign加密字符串对比是否一致，从而来鉴别调用者是否有权访问接口。

MD5 Message-Digest Algorithm 5 消息摘要加密算法第五版，用于确保信息传输的完整一致，是计算机广泛使用的杂凑算法之一，主流编程语言已经普遍支持MD5实现。

数据防篡改： 参数是明文传输，将接口参数及密钥生成加密字符串，将加密字符串作为签名。

eg：http://127.0.0.1:8000/api/?a=1&b=2

假设签名的密钥为：@admin123

签名的明文为：a=1&b=2&api_key=@admin123

通过MD5算法将整个接口参数（a=1&b=2&api_key=@admin123）生成加密字符串：786bfe32ae1d3764f208e03ca0bfaf13

所以，作为数据防篡改，带签名的接口为：

http://127.0.0.1:8000/api/?a=1&b=2&sign=786bfe32ae1d3764f208e03ca0bfaf13

对整个接口的参数做了加密，所以，只要任意一个参数发生变化，签名的验证就会失败。

好处是：加强了鉴权和对数据完整性的保护；

坏处是：MD5加密不可逆，服务器端必须知道客户端的接口参数和值，否则签名的验证就会失败。但实际上接口在设计时，服务器端是不知道客户端的请求参数值的。eg：嘉宾手机号的查询，服务器不知道调用者传的手机号具体是什么，只是通过数据库来查询该号码是否存在，那么就不能使用全参数加密的方式。

11.2.1 开发接口

编辑：…/guest/sign/views_if_sec.py视图文件

import time, hashlib

# 用户签名+时间戳
def user_sign(request):
if request.method == "POST":
client_time = request.POST.get('time', '')  # 客户端时间戳
client_sign = request.POST.get('sign', '')  # 客户端签名
else:
return "error"

if client_time == '' or client_sign == '':
return "sign null"

# 服务器时间
now_time = time.time()  # 当前时间戳
server_time = str(now_time).split('.')[0]

# 获取时间差
time_difference = int(server_time) - int(client_time)
if time_difference >= 60:
return "timeout"

# 签名检查
md5 = hashlib.md5()
sign_str = client_time + "&Guest-Bugmaster"
sign_bytes_utf8 = sign_str.encode(encoding="utf-8")
md5.update(sign_bytes_utf8)
server_sign = md5.hexdigest()

if server_sign != client_sign:
return "sign fail"
else:
return "sign success"

对上述代码进行分析：

创建user_sign()函数，处理签名参数；

首先，通过POST方法获取两个参数client_time和client_sign，如果客户端请求方法不是POST，那么函数返回error错误；

然后，判断两个参数均不能为空，如果为空，则返回sign null错误；

然后，对时间戳进行判断，对客户端的时间戳和服务器端的时间戳进行判断，如果时间戳大于60，则返回超时timeout错误；

最后，对签名进行检查，如果签名检查通过，则返回成功，否则返回签名验证失败。

注意： 之所以将时间戳用“.”split，是因为python3的时间戳精度较高，但我们只需要小数点的前10位。

将用户签名功能应用到添加发布会接口中，编辑：…/guest/sign/views_if_sec.py视图文件：

def add_event(request):
sign_result = user_sign(request)
if sign_result == "error":
return JsonResponse({'status':10011, 'message':'request error'})
elif sign_result == "sign null":
return JsonResponse({'status':10022, 'message':'user sign null'})
elif sign_result == "timeout":
return JsonResponse({'status':10013, 'message':'user sign timeout'})
elif sign_result == "sign fail":
return JsonResponse({'status':10014, 'message':'user sign error'})

eid = request.POST.get('eid', '')
name = request.POST.get('name', '')
limit = request.POST.get('limit', '')
status = request.POST.get('status', '')
address = request.POST.get('address', '')
start_time = request.POST.get('start_time', '')

if eid == '' or name == '' or limit == '' or address == '' or start_time == '':
return JsonResponse({'status':10021, 'message':'parameter error'})

result = Event.objects.filter(id=eid)

if result:
return JsonResponse({'status':10022, 'message':'event id already exists'})

result = Event.objects.filter(name=name)

if result:
return JsonResponse({'status':10023, 'message':'event name already exists'})

if status == '':
status = 1
try:
Event.objects.create(id=eid, name=name, limit=limit, address=address, status=int(status), start_time=start_time)
except ValidationError as e:
error = 'start_time format error. It must be in YYYY-MM-DD HH:MM:SS format.'
return JsonResponse({'status':10024, 'message':error})

return JsonResponse({'status':200, 'message':'add event success'})

调用user_sign()函数处理用户签名，根据函数返回字符串，将相应的处理结果返回给客户端，当用户签名验证通过后，接下来的处理过程和之前是一样的。

11.2.2 接口文档

11.2.3 接口用例

由于接口中加入了时间戳和md5加密算法，所以一般的接口测试工具无法模拟，此时就凸显出通过代码方式测试接口的优越性了。

添加接口测试用例：add_event_test.py

#!/usr/bin/python
# -*- coding:utf-8 -*-

import unittest, requests, hashlib
from time import time

class AddEventTest(unittest.TestCase):

def setUp(self):
self.base_url = "http://127.0.0.1:8000/api/sec_add_event/"
# app_key
self.api_key = "&Guest-Bugmaster"
# 当前时间
now_time = time()
self.client_time = str(now_time).split('.')[0]
# sign
md5 = hashlib.md5()
sign_str = self.client_time + self.api_key
sign_bytes_utf8 = sign_str.encode(encoding="utf-8")
md5.update(sign_bytes_utf8)
self.sign_md5 = md5.hexdigest()

def test_add_event_request_error(self):
'''请求方法错误'''
r = requests.get(self.base_url)
result = r.json()
self.assertEqual(result['status']. 10011)
self.assertEqual(result['message'], 'request error')

def test_add_event_sign_null(self):
'''签名参数为空'''
payload = {'eid':1, 'limit':'', 'address':'', 'start_time':'',
'time':'', 'sign':''}
r = requests.post(self.base_url, data=payload)
result = r.json()
self.assertEqual(result['status'], 10012)
self.assertEqual(result['message'], 'user sign null')

def test_add_event_time_out(self):
'''请求超时'''
now_time = str(int(self.client_time) - 61)
payload = {'eid':1, 'limit':'', 'address':'', 'start_time':'',
'time':now_time, 'sign':'abc'}
r = requests.post(self.base_url, data=payload)
result = r.json()
self.assertEqual(result['status'], 10013)
self.assertEqual(result['message'], 'user sign timeout')

def test_add_event_sign_error(self):
'''签名错误'''
payload = {'eid':1, 'limit':'', 'address':'', 'start_time':'',
'time':self.client_time, 'sign':'abc'}
r = requests.post(self.base_url, data=payload)
self.assertEqual(result['status'], 10014)
self.assertEqual(result['message'], 'user sign error')

def test_add_event_success(self):
'''添加成功'''
payload = {'eid':21, 'name':'一加5手机发布会', 'limit':2000,
'address':'深圳宝体', 'start_time':'2017-05-10 12:00:00',
'time':self.client_time, 'sign':self.sign_md5}
r = requests.post(self.base_url, data=payload)
result = r.json()
self.assertEqual(result['status'], 200)
self.assertEqual(result['message'], 'add event success')

if __name__ == '__main__':
unittest.main()

11.3 接口加密

以AES加密算法为例。

11.3.1 PyCrypto库

PyCrypto库，是一个免费的加密算法库，支持常见的DES、AES加密，以及MD5、SHA等各种HASH运算。

PyCrypto在Windows系统中安装需要依赖于“vcvarsall.bat”文件，需要安装庞大的Visual Studio才能解决，所以建议还是使用Linux系统来进行学习和使用。

通过下面的例子来演示PyCrypto库的强大：

eg1

SHA-256算法属于密码SHA-2系列哈希，它产生了一个消息的256位摘要。哈希值用作表示大量数据的固定大小的唯一值，数据的少量更改会在哈希值中产生不可预知的大量更改，从而验证数据的安全。

>>> from Crypto.Hash import SHA256
>>> hash = SHA256.new()
>>> hash.update(b'message')
>>> hash.digest()    #使用digest()方法加密
'\xabS\n\x13\xe4Y\x14\x98+y\xf9\xb7\xe3\xfb\xa9\x94\xcf\xd1\xf3\xfb"\xf7\x1c\xea\x1a\xfb\xf0+F\x0cm\x1d'
>>> hash.hexdigest()    #使用hexdigest()方法加密
'ab530a13e45914982b79f9b7e3fba994cfd1f3fb22f71cea1afbf02b460c6d1d'
>>>

通过digest()方法可以对字符串“message”进行加密，当然，通过hexdigest()方法也可以将其转换为16进制的加密字符串。

eg2

AES是Advanced Encryption Standard的缩写，即高级加密标准，是目前非常流行的加密算法。

>>> from Crypto.Cipher import AES
>>> obj = AES.new('This is a key123', AES.MODE_CBC, 'This is an IV456')
>>> message = "The answer is no"
>>> ciphertext = obj.encrypt(message)    #加密
>>> ciphertext
'\xd6\x83\x8dd!VT\x92\xaa`A\x05\xe0\x9b\x8b\xf1'
>>> obj2 = AES.new('This is a key123', AES.MODE_CBC, 'This is an IV456')
>>> obj2.decrypt(ciphertext)    #解密
'The answer is no'
>>>

加密的过程：

“This is a key123”为key，长度有着严格的要求，必须为16、24或32位，否则会抛出异常：“ValueError: AES key must be either 16, 24 or 32 bytes long”。

“This is an IV456”为VI，长度要求更加严格，只能为16位，否则将抛出异常：“ValueError: IV must be 16 bytes long”。

通过encrypt()方法对message字符串进行加密，得到：’\xd6\x83\x8dd!VT\x92\xaa`A\x05\xe0\x9b\x8b\xf1’

解密的过程：

要想对加密字符串进行解密，则必须知道加密时所使用的key和VI，通过decrypt()方法对加密字符串解密得到：“The answer is no”。

eg3

此外，Crypto还提供了一个强大的随机算法。

>>> from Crypto.Random import random
>>> random.choice(['dogs', 'cats', 'bears'])
'bears'
>>>

11.3.2 AES加密接口开发

将AES加密算法应用到接口开发中，先从编写测试用例开始，因为加密的过程是在客户端进行的，也就是在测试用例当中进行。

编写接口测试用例文件：Interface_AES_test.py

#! /usr/bin/python
# -*- coding:utf-8 -*-

from Crypto.Cipher import AES
import base64
import requests
import unittest
import json

class AESTest(unittest.TestCase):
"""AES加密后的接口测试用例"""

def setUp(self):
"""初始化测试参数"""
BS = 16
self.pad = lambda s: s + (BS - len(s) % BS) * chr(BS - len(s) % BS)    # 使用lambda定义匿名函数来对字符串进行补足，使其长度变为16、24、32位
self.base_url = "http://127.0.0.1:8000/api/sec_get_guest_list/"
self.app_key = "W7v4D60fds2Cmk2U"    # 定义好app_key，app_key是密钥，只能告诉给合法的接口调用者

def encryptBase64(self, src):
return base64.urlsafe_b64encode(src)

def encryptAES(self, src, key):
"""生成AES密文"""
iv = b"1172311105789011"    # 定义好iv，iv也是保密的，必须为16位
cryptor = AES.new(key, AES.MODE_CBC, iv)
ciphertext = cryptor.encrypt(self.pad(src))    # 通过encrypt()方法对src（JSON格式的接口参数）生成加密字符串
return self.encryptBase64(ciphertext)    # 通过encrypt()方法生成的加密字符串太长不适合传输，于是，通过base64模块的urlsafe_b64encode()方法对AES加密字符串进行二次加密

def test_case_interface(self):
"""测试AES加密的接口"""
payload = {'eid':1, 'phone': '18011001100'}    # 使用字典格式来存放接口参数
# 加密过程
encoded = self.encryptAES(json.dumps(payload), self.app_key).decode()    # 通过json.dumps()方法将payload字典转化为JSON格式，和app_key一起做为encryptAES()方法的参数，用于生成AES加密字符串

r = requests.post(self.base_url, data = {"data": encoded})    # 将加密后的字符串作为data参数发送到接口请求
result = r.json()
self.assertEqual(result['status'], 200)
self.assertEqual(result['message'], 'success')

if __name__ == '__main__':
unittest.main()

对上述代码进行分析，见备注。

注意：

encrypt()方法要求被加密的字符串长度必须是16、24、32位，如果直接生成可能会引发异常：“ValueError: Input strings must be a multiple of 16 in length”，可是被加密字符串的长度是可控的，因为接口参数的个数和长度是不固定的，所以，为了解决这个问题，还需要对字符串的长度进行处理，使它的长度符合encrypt()的需求；

通过encrypt()方法加密后的字符串是这样的：

b'>_\x80\xlfi\x97\x8f\x94~\xeaE\……'

通过urlsafe_b64encode()方法加密后的字符串是这样的：

b'gouBbuKWEeY5w……'

当服务器接收到加密的接口参数后，需要再警告一系列的过程解密：

编辑接口文件：views_if_sec.py

#! /usr/bin/python
# -*- coding:utf-8 -*-

import json
from Crypto.Cipher import AES

# AES加密算法
BS = 16
unpad = lambda s: s[0: - ord(s[-1])]

def decryptBase64(src):
return base64.urlsafe_b64decode(src)    # 通过urlsafe_b64decode()方法对base64加密字符串进行解密

def decryptAES(src, key):
"""
解析AES密文
"""
src = decryptBase64(src)    # 调用decryptBase64()方法，将base64加密字符串解密为AES加密字符串
iv = b'1172311105789011'
cryptor = AES.new(key, AES.MODE_CBC, iv)
text = cryptor.decrypt(src).decode()    # 通过decrypt()对AES加密字符串进行解密
return unpad(text)    # 通过unpad匿名函数对字符串的长度进行还原

def aes_encryption(request):

app_key = 'W7v4D60fds2Cmk2U'    # 服务器端与合法客户端约定的密钥app_key

if request.method == "POST":    # 判断客户端请求方法是否为POST，通过POST.get()方法接收data参数
data = request.POST.get("data", "")
else:
return "error"    # 如果请求方法不为POST，则函数返回“error”字符串

# 解密
decode = decryptAES(data, app_key)    # 调用decryptAES()函数解密，传参数字符串和app_key
# 转化为字典
dict_data = json.loads(decode)    # 将解密后的字符串通过json.loads()方法转化成字典，并作为函数的返回值
return dict_data

在查询嘉宾列表的接口中调用aes_encryption()函数进行AES加密字符串的解密，继续编辑views_if_sec.py文件：

# 嘉宾查询接口---AES算法
def get_guest_list(request):
dict_data = aes_encryption(request)

if dict_data == "error":
return JsonResponse({'status':10011, 'message':'request error'})

# 取出对应的发布会id和嘉宾手机号
eid = dict_data['eid']
phone = dict_data['phone']

if eid == '':
return JsonResponse({'status':10021, 'message':'eid cannot be empty'})

if eid != '' and phone == '':
datas = []
results = Guest.objects.filter(event_id=eid)
if results:
for r in results:
guest = {}
guest['realname'] = r.realname
guest['phone'] = r.phone
guest['email'] = r.email
guest['sign'] = r.sign
datas.append(guest)
return JsonResponse({'status':200, 'message':'success', 'data':datas})
else:
return JsonResponse({'status':10022, 'message':'query result is empty'})

if eid != '' and phone != '':
guest = {}
try:
result = Guest.objects.get(phone=phone, event_id=eid)
except ObjectDoesNotExist:
return JsonResponse({'status':10022, 'message':'query result is empty'})
else:
guest['realname'] = result.realname
guest['phone'] = result.phone
guest['email'] = result.email
guest['sign'] = result.sign
return JsonResponse({'status':200, 'message':'success', 'data':guest})

如果aes_encryption()函数返回“error”，则说明该接口的方法调用错误，返回客户端“request error”，否则，取出解密字符串（字典）中的eid 和phone的参数进行查询嘉宾列表的处理。

11.3.3 接口文档

增加了加密后的查询嘉宾接口文档：

11.3.4 补充接口测试用例

最后，补充查询嘉宾接口的测试用例：

Interface_AES_test.py

#! /usr/bin/python
# -*- coding:utf-8 -*-

from Crypto.Cipher import AES
import base64
import requests
import unittest
import json

class AESTest(unittest.TestCase):
"""AES加密后的接口测试用例"""

def setUp(self):
"""初始化测试参数"""
BS = 16
self.pad = lambda s: s + (BS - len(s) % BS) * chr(BS - len(s) % BS)
self.base_url = "http://10.18.214.88:8000/api/sec_get_guest_list/"
self.app_key = "W7v4D60fds2Cmk2U"

def encryptBase64(self, src):
return base64.urlsafe_b64encode(src)

def encryptAES(self, src, key):
"""生成AES密文"""
iv = b"1172311105789011"
cryptor = AES.new(key, AES.MODE_CBC, iv)
ciphertext = cryptor.encrypt(self.pad(src))
return self.encryptBase64(ciphertext)

def test_case_interface(self):
"""测试AES加密的接口"""
payload = {'eid':1, 'phone': '18011001100'}
# 加密
encoded = self.encryptAES(json.dumps(payload), self.app_key).decode()

r = requests.post(self.base_url, data = {"data": encoded})
result = r.json()
self.assertEqual(result['status'], 200)
self.assertEqual(result['message'], 'success')

def test_get_guest_list_request_error(self):
"""测试嘉宾查询接口：eid为空"""
payload = {'eid': '', 'phone': ''}
encoded = self.encryptAES(json.dumps(payload), self.app_key).decode()
r = requests.post(self.base_url, data={"data": encoded})
result = r.json()
self.assertEqual(result['status'], 10011)
self.assertEqual(result['message'], 'request error')

def test_get_guest_list_eid_null(self):
"""测试嘉宾查询接口：phone为空"""
payload = {'eid': '', 'phone': '18011001100'}
encoded = self.encryptAES(json.dumps(payload), self.app_key).decode()
r = requests.post(self.base_url, data={"data": encoded})
result = r.json()
self.assertEqual(result['status'], 10021)
self.assertEqual(result['message'], 'eid cannot be empty')

def test_get_guest_list_eid_error(self):
"""根据eid查询结果为空"""
payload = {'eid': '901', 'phone': ''}
encoded = self.encryptAES(json.dumps(payload), self.app_key).decode()
r = requests.post(self.base_url, data={"data": encoded})
result = r.json()
self.assertEqual(result['status'], 10022)
self.assertEqual(result['message'], 'query result is empty')

def test_get_guest_list_eid_success(self):
"""根据eid查询结果成功"""
payload = {'eid': '1', 'phone': '18011001100'}
encoded = self.encryptAES(json.dumps(payload), self.app_key).decode()
r = requests.post(self.base_url, data={"data": encoded})
result = r.json()
self.assertEqual(result['status'], 200)
self.assertEqual(result['message'], 'success')
self.assertEqual(result['data'][0]['realname'], 'alen')
self.assertEqual(result['data'][0]['phone'], '18011001100')

def test_get_event_list_eid_phone_null(self):
"""根据eid和phone查询结果为空"""
payload = {'eid': '200', 'phone': '10000000000'}
encoded = self.encryptAES(json.dumps(payload), self.app_key).decode()
r = requests.post(self.base_url, data={"data": encoded})
result = r.json()
self.assertEqual(result['status'], 10022)
self.assertEqual(result['message'], 'query result is empty')

def test_get_event_list_eid_phone_success(self):
"""根据eid和phone查询结果成功"""
payload = {'eid': '1', 'phone': '18011001100'}
encoded = self.encryptAES(json.dumps(payload), self.app_key).decode()
r = requests.post(self.base_url, data={"data": encoded})
result = r.json()
self.assertEqual(result['status'], 200)
self.assertEqual(result['message'], 'success')
self.assertEqual(result['data']['realname'], 'alen')
self.assertEqual(result['data']['phone'], '18011001100')

if __name__ == '__main__':
unittest.main()

封装了AES算法的加密算法后，在接口测试用例中调用即可，过程不复杂。

总结

使用MD5方式的相对来说还简单点，AES这种的确相对复杂，书中也只是做了基础的介绍和应用展示，在实际产品的开发过程中，加密环节相对复杂，测试人员，如果要测试带有加密的接口，当然了，如果了解加密过程最好，如果不了解，也不影响测试，只需要通过研发人员获取到关键信息，加入到接口测试用例中即可。