1、API验证分析
API三关验证
客户端和服务端中都存放一份相同的随机字符串，客户端发请求的时候把随机字符串和当前时间进行MD5加密，同时带着当前时间通过请求头发送到API，进入三关验证。
第一关是时间验证 (验证服务器当前时间和客户端发送过来的时间，超过10s后，验证不通过)
第二关是MD5规则验证(服务端把自己的密钥同客户端发送过来的时间进行MD5加密，进行密文的比较)
第三关是访问列表验证(从访问列表中查询是否存在，如果存在，验证不通过，否则把当前值存到列表中，并设置超时时间)，这里的时间可以设置成2S
逻辑图1
逻辑图2
原理：
1、客户端与服务器都存放着用于验证的Token字段，值字段无论通过什么方法外部的黑客都是无法获取的。
2、客户端把本地的用户名+时间戳+Token的组合进行MD5加密生成一段新的md5-token
3、客户端访问的时候携带：用户名、时间戳、生成的一段新的md5-token
4、服务端收到请求后，先判断用户名、时间戳是否合法（时间戳和当前时间不能大于2分钟）
5、用户名和时间戳都合法了，在去Redis中判断是否有当前用户的key判断是否在这2分钟有访问过，我这里设置Redis2分钟过期，即合法用户第一次访问后，把他的用户名加入到Redis中作为key：MD5作为vlaue存储。如果存在说明是在2分钟内访问的。拒绝
6、以上都通过之后说明是合法用户和合法请求，然后在判断MD5值是否相等如果相同认证通过，执行相关Veiws然后并返回相关的状态或数据
2、HouseStark代码
#_*_coding:utf-8_*_
from core import info_collection
from conf import settings
import urllib.request,sys,os,json,datetime
import urllib.parse
from core import api_token
class ArgvHandler(object):
def __init__(self,argv_list):
self.argvs = argv_list
self.parse_argv()
def parse_argv(self):
if len(self.argvs) >1:
if hasattr(self,self.argvs[1]):
func = getattr(self,self.argvs[1])
func()
else:
self.help_msg()
else:
self.help_msg()
def help_msg(self):
msg = '''
collect_data 收集资产数据
run_forever 未实现
get_asset_id 获取资产id
report_asset 汇报资产数据到服务器
'''
print(msg)
def collect_data(self):
obj = info_collection.InfoCollection()
asset_data = obj.collect() #收集
print(asset_data)
def run_forever(self):
pass
def __attach_token(self,url_str):
'''generate md5 by token_id and username,and attach it on the url request'''
user = settings.Params['auth']['user']
token_id = settings.Params['auth']['token']
md5_token,timestamp = api_token.get_token(user,token_id)
url_arg_str = "user=%s×tamp=%s&token=%s" %(user,timestamp,md5_token)
if "?" in url_str:#already has arg
new_url = url_str + "&" + url_arg_str
else:
new_url = url_str + "?" + url_arg_str
return new_url
#print(url_arg_str)
def __submit_data(self,action_type,data,method):
'''
send data to server
:param action_type: url
:param data: 具体要发送的数据
:param method: get/post
:return:
'''
if action_type in settings.Params['urls']:
if type(settings.Params['port']) is int:
url = "http://%s:%s%s" %(settings.Params['server'],settings.Params['port'],settings.Params['urls'][action_type])
else:
url = "http://%s%s" %(settings.Params['server'],settings.Params['urls'][action_type])
url = self.__attach_token(url)
print('Connecting [%s], it may take a minute' % url)
if method == "get":
args = ""
for k,v in data.items():
args += "&%s=%s" %(k,v)
args = args[1:]
url_with_args = "%s?%s" %(url,args)
print(url_with_args)
try:
req = urllib.request.urlopen(url_with_args,timeout=settings.Params['request_timeout'])
#req_data =urlopen(req,timeout=settings.Params['request_timeout'])
#callback = req_data.read()
callback = req.read()
print("-->server response:",callback)
return callback
except urllib.URLError as e:
sys.exit("\033[31;1m%s\033[0m"%e)
elif method == "post":
try:
data_encode = urllib.parse.urlencode(data).encode()
req = urllib.request.urlopen(url=url,data=data_encode,timeout=settings.Params['request_timeout'])
#res_data = urllib.urlopen(req,timeout=settings.Params['request_timeout'])
callback = req.read()
callback = json.loads(callback.decode())
print("\033[31;1m[%s]:[%s]\033[0m response:\n%s" %(method,url,callback) )
return callback
except Exception as e:
sys.exit("\033[31;1m%s\033[0m"%e)
else:
raise KeyError
#def __get_asset_id_by_sn(self,sn):
# return self.__submit_data("get_asset_id_by_sn",{"sn":sn},"get")
def load_asset_id(self,sn=None):
asset_id_file = settings.Params['asset_id']
has_asset_id = False
if os.path.isfile(asset_id_file):
asset_id = open(asset_id_file).read().strip()
if asset_id.isdigit():
return asset_id
else:
has_asset_id = False
else:
has_asset_id = False
def __update_asset_id(self,new_asset_id):
asset_id_file = settings.Params['asset_id']
f = open(asset_id_file,"w",encoding="utf-8")
f.write(str(new_asset_id))
f.close()
def report_asset(self):
obj = info_collection.InfoCollection()
asset_data = obj.collect()
asset_id = self.load_asset_id(asset_data["sn"])
if asset_id: #reported to server before
asset_data["asset_id"] = asset_id
post_url = "asset_report"
else:#first time report to server
'''report to another url,this will put the asset into approval waiting zone, when the asset is approved ,this request returns
asset's ID'''
asset_data["asset_id"] = None
post_url = "asset_report_with_no_id"
data = {"asset_data": json.dumps(asset_data)}
response = self.__submit_data(post_url,data,method="post")
if "asset_id" in response:
self.__update_asset_id(response["asset_id"])
self.log_record(response)
def log_record(self,log,action_type=None):
f = open(settings.Params["log_file"],"ab")
if log is str:
pass
if type(log) is dict:
if "info" in log:
for msg in log["info"]:
log_format = "%s\tINFO\t%s\n" %(datetime.datetime.now().strftime("%Y-%m-%d-%H:%M:%S"),msg)
#print msg
f.write(log_format.encode())
if "error" in log:
for msg in log["error"]:
log_format = "%s\tERROR\t%s\n" %(datetime.datetime.now().strftime("%Y-%m-%d-%H:%M:%S"),msg)
f.write(log_format.encode())
if "warning" in log:
for msg in log["warning"]:
log_format = "%s\tWARNING\t%s\n" %(datetime.datetime.now().strftime("%Y-%m-%d-%H:%M:%S"),msg)
f.write(log_format.encode())
f.close()
一次请求包含POST和GET
部分代码
def __attach_token(self,url_str):
'''generate md5 by token_id and username,and attach it on the url request'''
user = settings.Params['auth']['user']
token_id = settings.Params['auth']['token']
md5_token,timestamp = api_token.get_token(user,token_id)
url_arg_str = "user=%s×tamp=%s&token=%s" %(user,timestamp,md5_token)
if "?" in url_str:#already has arg
new_url = url_str + "&" + url_arg_str
else:
new_url = url_str + "?" + url_arg_str
return new_url
客户端截图
3、settings代码
#_*_coding:utf8_*_
import os
BaseDir = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
Params = {
#"server": "192.168.12.22",
"server": "127.0.0.1",
"port":8000,
'request_timeout':30,
"urls":{
"asset_report_with_no_id":"/asset/report/asset_with_no_asset_id/", #新资产待批准区
"asset_report":"/asset/report/", #正式资产表接口
},
'asset_id': '%s/var/.asset_id' % BaseDir,
'log_file': '%s/logs/run_log' % BaseDir,
'auth':{
'user':'jack',
'token': 'agbc!232'
},
}
4、api_token代码
#_*_coding:utf-8_*_
import hashlib,time
def get_token(username,token_id):
timestamp = int(time.time())
md5_format_str = "%s\n%s\n%s" %(username,timestamp,token_id)
obj = hashlib.md5()
obj.update(md5_format_str.encode())
print("token format:[%s]" % md5_format_str)
print("token :[%s]" % obj.hexdigest())
return obj.hexdigest()[10:17], timestamp
if __name__ =='__main__':
print(get_token('alex','test') )
为什么要切片
主要是因为太长了
5、API认证测试
合法认证测试
用户名密码截图
客户端截图
服务器控台截图
更改token后验证
后台修改token
客户端截图
