API认证及数据加密-阿里云开发者社区

API认证

API认证加密过程：

客户端和服务端保留一个相同的随机字符串，如：key = 'uiakjsdfasjdf898';
将随机字符串key与当前时间戳进行拼接，如：uiakjsdfasjdf898|1540192216171，将拼接生成的新字符串通过md5()加密；
客户端将第2步生成的md5值和当前时间戳做为URL参数传给服务端，URL如：http://127.0.0.1:8000/test/?sign=fb7005761539b0d18d130455a9de9914&ctime=1540192216171；
服务端接到客户端传递过来的md5和时间戳后，用服务器端保留的随机字符串key（此字符串与客户端保留的字符串一致）和客户端传过来的时间，并使用与第2步相同的算法生成一个md5值，然后将服务端的md5值与客户端传递过来的md5值做比较，如果相同，则证明此客户端是受信任的；

注意：不要认为至此认证就结束了，此认证机制还存在以下两点漏洞：

如果客户端在向服务端发起请求时，URL被黑客拿到，黑客也可以利用此URL向服务端发起请求；
解决思路：在服务端维护一个md5值的字典，将每次认证过的md5值存储起来，下次请求来时判断客户端传过来的md5值是否已经存在，如果存在就拒绝访问；

就算解决了第1点问题，还有一个问题：客户端在向服务端发起请求时，URL被黑客拿到，如果黑客的请求比受信客户端的请求先到达服务端，那么受信客户端的请求就会被拒；
解决思路：在服务端接收到请求时也生成一个当前时间戳，并与客户端传递过来的时间戳做比较，如果时间差大与3秒（此时间可根据实际情况自定义），则拒绝访问；

客户端代码：

import requests
import time
import hashlib

# 生成md5值
def gen_sign(ctime):
    key = 'uiakjsdfasjdf898'
    val = '%s|%s' %(key,ctime,)
    obj = hashlib.md5()
    obj.update(val.encode('utf-8'))
    return obj.hexdigest()

# 客户端发起post请求
ctime = int(time.time() * 1000)
result = requests.post(
    url='http://127.0.0.1:8000/test/',
    params={'sign':gen_sign(ctime),'ctime':ctime},    # url参数
    data='adfasdfasdfasdfasdf'                        # 发送给服务端的数据，
)

print(result.text)
print(result.url)
print(result.ok)

输入结果：

{"status":true,"data":666}
http://127.0.0.1:8000/test/?sign=b5abce59bd5776db5e4a107b403775c5&ctime=1540194227854
True

服务端代码：
定义一个API认证的类，在dispatch()方法中实现API认证，需要经过认证的视图类需要继承这个认证类。

import hashlib
import time
from django.conf import settings
from rest_framework.views import APIView
from rest_framework.response import Response
from utils.security import decrypt

key = 'uiakjsdfasjdf898'

def gen_sign(ctime):
    val = '%s|%s' %(key, ctime,)
    obj = hashlib.md5()
    obj.update(val.encode('utf-8'))
    return obj.hexdigest()

SIGN_RECORD = {}
# 定义一个API认证的类，在`dispatch()`方法中实现API认证，需要经过认证的视图类需要继承这个认证类。
class APIAuthView(APIView):
    def dispatch(self, request, *args, **kwargs):
        client_sign = request.GET.get('sign')  # 客户端签名
        client_ctime = int(request.GET.get('ctime'))  # 客户端时间
        server_time = int(time.time() * 1000)  # 服务端时间

        # 请求时间大于3秒，拒绝访问
        if server_time - client_ctime > 3000:
            return Response({'status': False, 'error': '你在路上的时间太久了'})
        
        # 客户端URL携带的MD5如果已经验证过，则拒绝访问
        if client_sign in SIGN_RECORD:
            return Response({'status': False, 'error': '签名已经被使用过了'})

        # 如果客户端的md5值与服务端的md5值不相同，则说明说请求url被篡改，拒绝访问
        server_sign = gen_sign(client_ctime)
        if server_sign != client_sign:
            return Response({'status': False, 'error': '签名错误'})

        SIGN_RECORD[client_sign] = client_ctime
        return super().dispatch(request, *args, **kwargs)

# 需要经过API认证的视图类需要继承认证类`APIAuthView`
class TestView(APIAuthView):
    def post(self,request):
        print(request.data)
        print(request.url)
        return Response({'status':True,'data':666})

注意：此示例中的服务端代码是一个Django程序，在验证示例时，还需要在urls.py文件是添加如下代码：
from django.contrib import admin
from api.views import TestView
from django.urls import path

urlpatterns = [
   path('admin/', admin.site.urls),
   path('test/',TestView.as_view()),
]

数据加密

数据加密算法这里我用RSA。

安装rsa模块

pip3 install rsa

生成一对公钥和私钥

# ######### 1. 生成公钥私钥 #########
pub_key_obj, priv_key_obj = rsa.newkeys(1024)   # 128 - 11 = 117，只能对117个字符加密
# 公钥字符串
pub_key_str = pub_key_obj.save_pkcs1()
pub_key_code = base64.standard_b64encode(pub_key_str)   # 对公钥再次进行base64编码
print(pub_key_code)
# 私钥字符串
priv_key_str = priv_key_obj.save_pkcs1()
priv_key_code = base64.standard_b64encode(priv_key_str)    # 对私钥再次进行base64编码
print(priv_key_code)

注意：公钥用来对数据进行加密，私钥用对加密后的数据进行解密；所以客户端生的私钥需要发送到服务端并保存起来；

输出结果：

# 公钥字符串 ，字节类型
b'LS0tLS1CRUdJTiBSU0EgUFVCTElDIEtFWS0tLS0tCk1JR0pBb0dCQUtiTGluemJqMkttb1hOUVRvVlBtV2JzVDVqV3F6cm1scjJRU09HR0o0TnVzM0FFMDhiL0RESHgKaW5BTkN1djRVcVB2M3FlWWJiKzRhR3cvMXhZaTJNekVDL2h1cWQwZXdLMk9ha1Y1aWwvZEpMdlB3SHJLN2IrZQpSQnhZaUoyOUh3QWhUemJEaFcvQjBUaGh0M0dmVThPQjVnWVNhS29MTTJndlpxMURIb1kvQWdNQkFBRT0KLS0tLS1FTkQgUlNBIFBVQkxJQyBLRVktLS0tLQo='

# 私钥字符串 ，字节类型
b'LS0tLS1CRUdJTiBSU0EgUFJJVkFURSBLRVktLS0tLQpNSUlDWUFJQkFBS0JnUUNteTRwODI0OWlwcUZ6VUU2RlQ1bG03RStZMXFzNjVwYTlrRWpoaGllRGJyTndCTlBHCi93d3g4WXB3RFFycitGS2o3OTZubUcyL3VHaHNQOWNXSXRqTXhBdjRicW5kSHNDdGptcEZlWXBmM1NTN3o4QjYKeXUyL25rUWNXSWlkdlI4QUlVODJ3NFZ2d2RFNFliZHhuMVBEZ2VZR0VtaXFDek5vTDJhdFF4NkdQd0lEQVFBQgpBb0dBWjBuVVVNMkdWWWpxb2dZeEdjelpLaXRjZjBFd2VDRWpaL0Jac1k3cUdUSU1YR29nMnpKRjB3Zkl1dXJZCndKZmVWVGJOb3V0NXl5ZmZRbW1sdkwwNE5WZ2FRZFM5eHZSUmtUbkZ5WFZja2x5eTFFSklNQ1JhdXd4U3JadEgKOGRvUC9RSE93dm1IeXVvNDRaT1A4d3o1T1lwRitvSnpWYlZEWW9EdDlCMkJrd0VDUlFEVWFoSGJqTERzTEVkUApyQW4rWEs1UFQrUEFxQjhuSDcvTTBRZ2s3MURoMlpuZzdpZlZiVWhBWE4zVmxBeU5tcytQZTBWbUdvUDBNc1BUCjdEbVdDQldaUk92MWp3STlBTWtGSWppeGNhMW9aRUJJd1cyaFJXVndlN2grYmhHWVhzMk5BZEloOWpZUW85bGgKcGJJRnJ3YnRVQWg0VkVLenROMWFyeVIydk1rekFZcnNVUUpGQUtUVm52LzV3TDIxYXExSCt3VlpsS2JWZnc3ZApLRGVyS3FMZFAyMnlETmtHaktRQkRBWlNaTFFWbk13RnRHd2F5NkV6YnRwYUR6WHNRd3pzam85L3ZJc1E4ZTYvCkFqd2swUWJpZ1VWRHNFSGtNQzhWQ0J0d3A3aFJJdjYveER0Z3hEbVlKZFkxTXJqL29FMjduQ1RoVE9lQ2xaOUIKRkM3RTk4M3FETUVvekdtMDZ2RUNSQ05UaGFTT05aWHVUZnFDRXVtdm9YYTFJcDg1NTdYaXFQU2ZlbXNXYXZmcApFOWJDZy9URUEwa2dzeFd1c0RjVzQ5S2IyVlhWekJrUUJsWWxiRElqdmhCV3czbVUKLS0tLS1FTkQgUlNBIFBSSVZBVEUgS0VZLS0tLS0K'

数据加密：

# # ######### 2. 加密 #########
def encrypt(pub_key_code,value):
    key_str = base64.standard_b64decode(pub_key_code)
    pk = rsa.PublicKey.load_pkcs1(key_str)
    result = rsa.encrypt(value.encode('utf-8'), pk)
    return result

data = encrypt(pub_key_code,'zff')   # “zff”是要被加密的数据
print(data)
print(len(data))

输出结果：

# 加密后的数据
b'GJy\xfc\x82R\xc6N\xebo_\xdad\xbf\x93\xe8\xb9\xd0y\xee\x0b\x9b\xe6\xad\x9f\x07\xcf4\x7f\x8f\x1b\xb4\xc4f\xa0\x01\xc8z\xf9\xd1\x89\xbd\xc0\x1b\xdfhi\x93\x14\x84\x1d\x15\xaa3y"2,\xeb\x1aP\xda2)\'\x98\x90R\xab\xf3\xfcV:\xf7\x12\xcd\xf2d}\xb4NZ\x021\xd6\xce=\x9e\xdf\x07\x0eD\xbc\xf8\xb9\xcdO\xe1\xb0R\x0e\x1cg\xc0X1"\xab\xcd\x18K\xc2\x9bn\xbb\xda{\xd1\xec\x1d\xbai\'\xef\x86\xa85C'
128

不管多大的数据，加密后的长度都是128个字符；

数据解密

# # ######### 3. 解密 #########
def decrypt(priv_key_code,value):
    key_str = base64.standard_b64decode(priv_key_code)
    pk = rsa.PrivateKey.load_pkcs1(key_str)
    val = rsa.decrypt(value, pk)
    return val

origin = decrypt(priv_key_code,data)
print(origin)

输出结果：

# 解密后数据，字节类型
b'zff'

注意：加密长度为1024字节时，可加密的字符长度为117？117是这么计算得来的：8字节=1字符，1024字节=128字符，加密算法本身会占用11个字符，所以128 - 11 = 117；以此类推：当加密长度为2048字节时，可加密的字符长度为 2048 / 8 - 11 = 245。
但是当我们要发送的数据达到几KB，甚至几MB时，怎么办呢？看看接下来的，大数据加密！

大字符串数据加密

对于大字符串加密的思路：

以加密长度为1024字节为例，将大字符串的数据分割成若干个小字符串，每段字符串长度为117，然后再循环对小字符串加密，然后再将加密后的小字符串拼接成大字符串传给服务端；
服务端将加密的大字符串分割成每段128个字符长度，进行分段解密，然后再将解密后的小字符串拼接成元始数据。

import rsa
import base64
 

# ######### 1. 生成公钥私钥 #########
pub_key_obj, priv_key_obj = rsa.newkeys(1024) # 128 - 11 = 117
# 公钥字符串
pub_key_str = pub_key_obj.save_pkcs1()
pub_key_code = base64.standard_b64encode(pub_key_str)
print(pub_key_code)
# 私钥字符串
priv_key_str = priv_key_obj.save_pkcs1()
priv_key_code = base64.standard_b64encode(priv_key_str)
print(priv_key_code)

#
# # # ######### 2. 加密 #########
def encrypt(pub_key_code,value):
    key_str = base64.standard_b64decode(pub_key_code)
    pk = rsa.PublicKey.load_pkcs1(key_str)
    value_bytes = value.encode('utf-8')
    data_list = []
    for i in range(0,len(value_bytes),117):
        chunk = value_bytes[i:i+117]
        result = rsa.encrypt(chunk, pk)
        data_list.append(result)

    return b''.join(data_list)

data = encrypt(pub_key_code,'zff'*1000)
print(len(data),data)


# # # ######### 3. 解密 #########
def decrypt(priv_key_code,bytes_value):
    key_str = base64.standard_b64decode(priv_key_code)
    pk = rsa.PrivateKey.load_pkcs1(key_str)
    result = []
    for i in range(0,len(bytes_value),128):
        chunk = bytes_value[i:i+128]
        val = rsa.decrypt(chunk, pk)
        result.append(val)
    return b''.join(result)

origin = decrypt(priv_key_code,data)
origin_str = origin.decode('utf-8')
print(origin_str)

API认证及数据加密

API认证

数据加密

生成一对公钥和私钥

数据加密：

数据解密

大字符串数据加密

热门文章

最新文章

相关课程

相关电子书

相关实验场景