API验证

时间：2023-03-09 03:36:29
API验证

API验证说明

API验证:

    a. 发令牌: 静态

        PS: 隐患 key被别人获取

    b. 动态令牌

        PS: (问题越严重)用户生成的每个令牌被黑客获取到,都会破解

    c. 高级版本

        PS: 黑客网速快,会窃取, so要对数据加密

    d. 终极版本

特点:

   为什么要用API ？

       - 数据在传输过程中,保证数据安全

   你是如何设计的 ？

       - Tornado 中的加密Cookie类似

       - 创建动态key  md5(key + time)|time (Tornado中也是这么做)

       - 限制

         - 第一关: 时间

         - 第二关: 算法规则

         - 第三关: 已访问的记录

       PS: 黑客网速快,会窃取, so要对数据加密

1、发令牌: 静态（服务器、客户端有相同的静态key）

###客户端

import requests

key = "asdfasdfasdfasdf098712sdfs"

response = requests.get("http://127.0.0.1:8000/api/asset.html",headers={'OpenKey':key})

print(response.text)
 
###服务端

def api_check(request):

    if request.method == "GET":

        #################静态################

        server_key = "adsfasdfzxcvzw241wdsaddsfsdf"   # 应放在settings

        client_key = request.META.get("HTTP_OPENKEY")  # META里的Openkey字段会自动修改，需提前看META里的字段

        if server_key != client_key:

            return HttpResponse("key错误，你是非法用户")

        return HttpResponse("欢迎光临")
2、动态令牌
### 客户端

import requests

import time

import hashlib

ctime = time.time()

key = "adsfasdfzxcvzw241wdsaddsfsdf"

new_key = "%s|%s" % (key,ctime,)

m = hashlib.md5()

m.update(bytes(new_key,encoding="utf-8"))

md5_key = m.hexdigest()  # key进行hash

new_time_key = "%s|%s" % (md5_key,ctime,)

response = requests.get("http://127.0.0.1:8000/api/asset.html",headers={"openkey":new_time_key})

print(response.text)
#### 服务端

def api_check(request):

    if request.method == "GET":

        server_key = "adsfasdfzxcvzw241wdsaddsfsdf"   # 应放在settings

        client_md5_time_key = request.META.get("HTTP_OPENKEY")

        client_md5_key, client_ctime = client_md5_time_key.split("|")

        temp = "%s|%s" % (server_key, client_ctime)

        m = hashlib.md5()

        m.update(bytes(temp, encoding='utf-8'))

        server_md5_key = m.hexdigest()

        if server_md5_key != client_md5_key:

            return HttpResponse("验证失败")

        return HttpResponse("欢迎光临，访问成功")

3、高级版本

（1）、时间限制，超过自定义时间（自定义10s）不让访问

（2）、hash限制，比对server和client的hash值

（3）、记录自定义时间内已访问的hash_key，禁止重复访问

### 客户端

import requests

import time

import hashlib

ctime = time.time()

key = "adsfasdfzxcvzw241wdsaddsfsdf"

new_key = "%s|%s" % (key,ctime,)

m = hashlib.md5()

m.update(bytes(new_key,encoding="utf-8"))

md5_key = m.hexdigest()  # key进行hash

new_time_key = "%s|%s" % (md5_key,ctime,)

response = requests.get("http://127.0.0.1:8000/api/asset.html",headers={"openkey":new_time_key})

print(response.text)
### 服务端（装饰器版本）

api_key_record = {}  # 创建全局的字段，记录自定义时间内已访问的hash_key

def API_check(func):

    def wrapper(request,*args,**kwarg):

        if request.method == "GET":

            import hashlib

            # ####################API认证##################

            client_md5_time_key = request.META.get("HTTP_OPENKEY")

            client_md5_key, client_ctime = client_md5_time_key.split("|")

            client_ctime = float(client_ctime)

            server_time = time.time()

            # 第一关

            if server_time - client_ctime > 10:

                return HttpResponse("【第一关】时间有点长")

            # 第二关

            tmp = "%s|%s" % (settings.AUTH_KEY, str(client_ctime))

            m = hashlib.md5()

            m.update(bytes(tmp, encoding="utf-8"))

            server_md5_key = m.hexdigest()

            if server_md5_key != client_md5_key:

                return HttpResponse("【第二关】规则不正确，验证码错误")

            for k in list(api_key_record.keys()):

                v = api_key_record[k]

                if server_time > v:

                    api_key_record.pop(k)

            # 第三关

            if client_md5_time_key in api_key_record:

                return HttpResponse("【第三关】有人已经来过了")

            else:

                api_key_record[client_md5_time_key] = client_ctime + 10

        # ####################API认证##################

        ret = func(request,*args,**kwarg)

        return ret

    return wrapper

@API_check

def asset(request):

    。。。。。。

     # 主函数

4、终极版本（加密）

from Crypto.Cipher import AES

from lib.conf.config import settings

def encrypt(message):

    """

    数据加密

    :param message:

    :return:

    """

    key = settings.DATA_KEY

    cipher = AES.new(key, AES.MODE_CBC, key)

    ba_data = bytearray(message,encoding='utf-8')

    v1 = len(ba_data)

    v2 = v1 % 16

    if v2 == 0:

        v3 = 16

    else:

        v3 = 16 - v2

    for i in range(v3):

        ba_data.append(v3)

    final_data = ba_data.decode('utf-8')

    msg = cipher.encrypt(final_data) # 要加密的字符串，必须是16个字节或16个字节的倍数

    return msg

def decrypt(msg):

    """

    数据解密

    :param message:

    :return:

    """

    from Crypto.Cipher import AES

    key = settings.DATA_KEY

    cipher = AES.new(key, AES.MODE_CBC, key)

    result = cipher.decrypt(msg) # result = b'\xe8\xa6\x81\xe5\x8a\xa0\xe5\xaf\x86\xe5\x8a\xa0\xe5\xaf\x86\xe5\x8a\xa0sdfsd\t\t\t\t\t\t\t\t\t'

    data = result[0:-result[-1]]

    return str(data,encoding='utf-8')

def auth():  # hash（key+时间）

    """

    API验证

    :return:

    """

    import time

    import requests

    import hashlib

    ctime = time.time()

    key = "asdfasdfasdfasdf098712sdfs"

    new_key = "%s|%s" %(key,ctime,)

    m = hashlib.md5()

    m.update(bytes(new_key,encoding='utf-8'))  #里面是字节数据

    md5_key = m.hexdigest()                    #返回值是字符窜类型

    md5_time_key = "%s|%s" %(md5_key,ctime)

    return md5_time_key

lib/utils.py

from Crypto.Cipher import AES

import requests

import json

from lib.utils import encrypt

from lib.utils import auth

#对数据加密字典

v1 = encrypt(json.dumps({"k1":"v1"}))      #获取的是加密后的字节

print(v1)

response = requests.post(

    url="http://127.0.0.1:8000/api/asset.html",

    headers={'OpenKey':auth(),'content-type':'application/json'},

    data=v1

)

print(response.text)

客户端

import json

import hashlib

from django.shortcuts import render,HttpResponse

from repository import models

from django.conf import settings

from api.service import PluginManager

import time

import json

from Crypto.Cipher import AES

api_key_record ={

    "76942d662d98ebe3b920a7b791bf5040|1501510243.92804":1501510243.92804,

}

def decrypt(msg):

    key = b'dfdsdfsasdfdsdfs'

    cipher = AES.new(key, AES.MODE_CBC, key)

    result = cipher.decrypt(msg)  # 把加密后的字节解密成不加密的字节

    data = result[0:-result[-1]]

    return str(data, encoding='utf-8')

def outer(func):

    def wrapper(request):

        client_md5_time_key = request.META.get("HTTP_OPENKEY")

        client_md5_key, client_ctime = client_md5_time_key.split("|")

        client_ctime = float(client_ctime)

        server_ctime = time.time()

        # 第一关 时间关

        if server_ctime - client_ctime > 30:

            return HttpResponse("第一关  小伙子,别虎我,太长了")

        # 第二关 客户端时间和服务端key加密和 客户端的密钥对比

        temp = "%s|%s" % (settings.AUTH_KEY, client_ctime)

        m = hashlib.md5()

        m.update(bytes(temp, encoding='utf-8'))

        server_md5_key = m.hexdigest()

        if server_md5_key != client_md5_key:

            return HttpResponse("第二关   规则正确")

        # 以后基于memcache,目前先写入内存删除超过10s的值

        for k in list(api_key_record.keys()):

            v = api_key_record[k]

            if server_ctime > v:

                del api_key_record[k]

        # 第三关 判断字典里是否有之前访问的key,如果有不通过,没有加入字典

        if client_md5_time_key in api_key_record:

            return HttpResponse("第三关  已经有人来过了")

        else:

            api_key_record[client_md5_time_key] = client_ctime + 10

            obj = func(request)

            return obj

    return wrapper

@outer

def asset(request):

    if request.method == 'GET':

        ys = '重要的不能被闲杂人等看的数据'

        return HttpResponse(ys)

    elif request.method == 'POST':

        server_info = decrypt(request.body)

        server_info = json.loads(server_info)

        # # 新资产信息

        # server_info = json.loads(request.body.decode('utf-8'))

        hostname = server_info['basic']['data']['hostname']

        # 老资产信息

        server_obj = models.Server.objects.filter(hostname=hostname).first()

        if not server_obj:

            return HttpResponse('当前主机名在资产中未录入')

        PluginManager(server_info,server_obj,hostname).exec_plugin()

        return HttpResponse("...")

服务端

相关文章