Skip to content

E
EmbeddedDevice_SafeUpgrade
Commit 8ee22668 authored by aohui.li's avatar aohui.li
Browse files
Options

优化升级的验证字段

parent 282350fa
Hide whitespace changes
Inline Side-by-side
Showing with 338 additions and 306 deletions
at32_gateway_safe_update.py
View file @ 8ee22668
'''
@Modified: 2024-5-30
@Brief: Update Image Safely
'''
# !usr/bin/python3
import paho.mqtt.client as mqtt
import os
import threading
import json
import time
import sys
import zlib
#----------------------------------------------------------------------------------------#
# 测试平台
mqttConfig = {
'mqttServer': 'www.witium.com.cn',
'mqttPort': 12883,
'userName': 'witcd',
'password': 'Witium37774020'
}
# 正式平台
# mqttConfig = {
# 'mqttServer': '116.62.127.242',
# 'mqttPort': 1883,
# 'userName': 'witcd',
# 'password': 'Witium37774020'
# }
# # 颖泰平台
# mqttConfig = {
# 'mqttServer': '112.17.190.17',
# 'mqttPort': 1883,
# 'userName': 'witcd',
# 'password': 'Witium37774020'
# }
dev_type = 'WTG902F'
updateCheckInTopic = 'WT/%s/Update/CheckIn' % dev_type
updateResponseTopic = 'WT/%s/Update/Response' % dev_type
updateTopic = 'WT/%s/{0}/Update' % dev_type
updateList = [{'sn': '21126112', 'package': 0, 'time': time.time(), 'send': True}]
NewImage = "Gateway.bin"
cmd7Topic = 'WT/{}/{}/BackDoor'.format(dev_type,updateList[0]['sn'])
cmd7ResTopic = 'WT/{}/{}/BackDoorResp'.format(dev_type,updateList[0]['sn'])
CheckPassFlag = False
#----------------------------------------------------------------------------------------#
StampingMark = "WTG"
FIXED_LOCATION = 0x200
FIXED_STAMPING_SIZE = 37
FIXED_HEAD = 3
#----------------------------------------------------------------------------------------#
def get_Image_Size(file_path):
file_size = 0
with open(file_path, 'rb') as file:
file_data = file.read()
file_size = len(file_data)
return file_size
def read_stamping(fiel_path):
with open(file_path, 'rb') as file:
file.seek(FIXED_LOCATION)
content = file.read(FIXED_STAMPING_SIZE)
print("Content--->{}".format(content))
return content
def HAS_STAMPING(content):
try:
if StampingMark in content.decode()[:FIXED_HEAD]:
return True
else:
return False
except:
print("NO STAMPING HERE, EXIT!")
return False
def main_client_on_connect(client, userdata, flags, rc):
log('Main client connected with result code ' + str(rc))
log('Main client subscribe ' + updateCheckInTopic + ' topic')
client.subscribe(updateCheckInTopic)
log('Main client subscribe ' + updateResponseTopic + ' topic')
client.subscribe(updateResponseTopic)
log('Main client subscribe ' + cmd7ResTopic + ' topic')
client.subscribe(cmd7ResTopic)
def main_client_on_message(client, userdata, msg):
global updateList
global CheckPassFlag
if msg.topic == updateCheckInTopic and CheckPassFlag == True:
try:
jsonMsg = json.loads(msg.payload.decode())
if jsonMsg['gId'] not in ['20010801']:
return
updateDict = {}
if 'version' in jsonMsg:
if checkVersion(jsonMsg['version']):
updateDict['sn'] = jsonMsg['gId']
updateDict['package'] = 0
updateDict['time'] = time.time()
updateDict['send'] = True
updateList.append(updateDict)
else:
updateDict['sn'] = jsonMsg['gId']
updateDict['package'] = 0
updateDict['time'] = time.time()
updateDict['send'] = True
updateList.append(updateDict)
except:
log('Not a json data')
elif msg.topic == updateResponseTopic and CheckPassFlag == True:
try:
responseMsg = msg.payload.decode().split()
for each in updateList:
if responseMsg[1] == each['sn']:
if responseMsg[2] == '0':
each['package'] = int(responseMsg[0]) + 1
each['time'] = time.time()
each['send'] = True
else:
each['send'] = True
except:
log('Error')
elif msg.topic == cmd7ResTopic:
try:
Cmd7Res = json.loads(msg.payload)
if 'stamping' in Cmd7Res:
print("检查本地镜像")
UpdataImageStamping = read_stamping(file_path)
UpdataImageStamping = UpdataImageStamping.decode()
if UpdataImageStamping == Cmd7Res['stamping']:
print("Romote Stamping {}".format(Cmd7Res['stamping']))
print("Local Stamping {}".format(UpdataImageStamping))
print("Equal")
CheckPassFlag = True
else:
print("Romote Stamping {}".format(Cmd7Res['stamping']))
print("Local Stamping {}".format(UpdataImageStamping))
print("exit")
except:
log("Error Cmd 7 Res")
def checkVersion(oldVersion):
if 'Beta' in oldVersion and 'Beta' in tdlVersion or 'Beta' not in oldVersion and 'Beta' not in tdlVersion:
if oldVersion < tdlVersion:
return True
else:
return False
elif 'Beta' in oldVersion and 'Beta' not in tdlVersion:
if oldVersion[0:5] < tdlVersion:
return True
else:
return False
elif 'Beta' not in oldVersion and 'Beta' in tdlVersion:
if oldVersion < tdlVersion[0:5]:
return True
else:
return False
return False
def createCheckCode(checkData):
sumDat = 0
codeBytes = b''
for each in checkData:
sumDat += each
sumHex = hex(sumDat).lstrip('0x')
while len(sumHex) != 8:
sumHex = '0' + sumHex
codeBytes += bytes((int(sumHex[0:2], 16),))
codeBytes += bytes((int(sumHex[2:4], 16),))
codeBytes += bytes((int(sumHex[4:6], 16),))
codeBytes += bytes((int(sumHex[6:8], 16),))
return codeBytes
def update():
global fileData
global updateList
global mainClient
global packageSize
global oldTime
data = [{"cmd":7,"ResquestStamping":1}]
data = json.dumps(data, separators=(',', ':'))
print(data)
while CheckPassFlag == False:
print("waiting for check Pass Flag...")
mainClient.publish(cmd7Topic, data)
time.sleep(5)
startPackage = bytes((0,)) + b'Witium_Update' + bytes((packageSize,))
startPackage += createCheckCode(startPackage)
while True:
if len(updateList) == 0:
time.sleep(1)
else:
curTime = time.time()
if curTime - oldTime > 30:
oldTime = time.time()
log('Current List:' + str(updateList))
for eachModule in updateList:
if eachModule['send'] == True:
eachModule['send'] = False
if eachModule['package'] == 0:
mainClient.publish(updateTopic.format(eachModule['sn']), startPackage)
else:
if eachModule['package'] != packageSize:
sendData = bytes((eachModule['package'],)) + fileData[
(eachModule['package'] - 1) * 1024:eachModule[
'package'] * 1024]
else:
sendData = bytes((eachModule['package'],)) + fileData[(eachModule['package'] - 1) * 1024:]
sendData += createCheckCode(sendData)
mainClient.publish(updateTopic.format(eachModule['sn']), sendData)
log('Send the ' + str(eachModule['package']) + ' package to ' + eachModule['sn'])
if eachModule['package'] == packageSize:
updateList.remove(eachModule)
log('Update success ' + eachModule['sn'])
nowTime = time.time()
if nowTime - eachModule['time'] > 30:
updateList.remove(eachModule)
log('Update time out ' + eachModule['sn'])
def log(logData):
with open('log.txt', 'a+') as logFile:
logFile.write(logData + '\n')
print(logData)
#----------------------------------------------------------------------------------------#
path = os.path.abspath(os.path.dirname(os.path.dirname(__file__)))
resourcesPath = path + os.sep + 'scripts' + os.sep
curFileList = os.listdir(os.getcwd())
oldTime = time.time()
if not os.path.exists('log.txt'):
with open('log.txt', 'w'):
pass
file_path = resourcesPath + NewImage
print("Update Image Path [{}]".format(file_path))
size = get_Image_Size(file_path)
print(f"文件大小：{get_Image_Size(file_path)} bytes")
if HAS_STAMPING(read_stamping(file_path)) == True:
print("STAMPING IS EXIST.")
print("\n\n\n\n\n")
else:
print("STAMPING IS NOT EXIST.")
exit(0)
with open(file_path, 'rb') as file:
fileData = file.read()
packageSize = (len(fileData) - 1) // 1024 + 1
log('packageSize: ' + str(packageSize))
if not file_path:
log('Update file has not found.')
else:
mainClient = mqtt.Client()
mainClient.username_pw_set(mqttConfig['userName'], mqttConfig['password'])
mainClient.on_connect = main_client_on_connect
mainClient.on_message = main_client_on_message
mainClient.connect(mqttConfig['mqttServer'], mqttConfig['mqttPort'], 60)
thread = threading.Thread(target=update)
thread.setDaemon(True)
thread.start()
mainClient.loop_forever()
'''
@Modified: 2024-5-30
@Brief: Update Image Safely
'''
# !usr/bin/python3
import paho.mqtt.client as mqtt
import os
import threading
import json
import time
import re
#----------------------------------------------------------------------------------------#
# 测试平台
mqttConfig = {
'mqttServer': 'www.witium.com.cn',
'mqttPort': 12883,
'userName': 'witcd',
'password': 'Witium37774020'
}
# 正式平台
# mqttConfig = {
# 'mqttServer': '116.62.127.242',
# 'mqttPort': 1883,
# 'userName': 'witcd',
# 'password': 'Witium37774020'
# }
# # 颖泰平台
# mqttConfig = {
# 'mqttServer': '112.17.190.17',
# 'mqttPort': 1883,
# 'userName': 'witcd',
# 'password': 'Witium37774020'
# }
dev_type = 'WTG906F'
updateCheckInTopic = 'WT/%s/Update/CheckIn' % dev_type
updateResponseTopic = 'WT/%s/Update/Response' % dev_type
updateTopic = 'WT/%s/{0}/Update' % dev_type
updateList = [{'sn': '21126112', 'package': 0, 'time': time.time(), 'send': True}]
NewImage = "BIN/v2.35.bin"
cmd7Topic = 'WT/{}/{}/BackDoor'.format(dev_type,updateList[0]['sn'])
cmd7ResTopic = 'WT/{}/{}/BackDoorResp'.format(dev_type,updateList[0]['sn'])
CheckPassFlag = False
#----------------------------------------------------------------------------------------#
StampingMark = "WTG"
FIXED_LOCATION = 0x200
FIXED_STAMPING_SIZE = 37
FIXED_HEAD = 3
#----------------------------------------------------------------------------------------#
def extract_characters_around(pattern, text, before=1, after=2):
matches = re.finditer(pattern, text)
for match in matches:
start = match.start()
before_chars = text[start-before-1:start] if before > 0 else ''
target_char = text[start]
after_chars = text[start+1:start+after+1] if after > 0 else ''
return before_chars+"_"+after_chars
def get_substrings_around(main_str, sub_str, before=0, after=10):
start = main_str.find(sub_str)
if start == -1:
return None, None # 子字符串未找到
# 提取前一段和后一段字符串
before_str = main_str[0:start]
after_str = main_str[start + len(sub_str):-1]
return before_str, after_str
def get_Image_Size(file_path):
file_size = 0
with open(file_path, 'rb') as file:
file_data = file.read()
file_size = len(file_data)
return file_size
def read_stamping(fiel_path):
with open(file_path, 'rb') as file:
file.seek(FIXED_LOCATION)
content = file.read(FIXED_STAMPING_SIZE)
print("Content--->{}".format(content))
return content
def HAS_STAMPING(content):
try:
if StampingMark in content.decode()[:FIXED_HEAD]:
return True
else:
return False
except:
print("NO STAMPING HERE, EXIT!")
return False
def main_client_on_connect(client, userdata, flags, rc):
log('Main client connected with result code ' + str(rc))
log('Main client subscribe ' + updateCheckInTopic + ' topic')
client.subscribe(updateCheckInTopic)
log('Main client subscribe ' + updateResponseTopic + ' topic')
client.subscribe(updateResponseTopic)
log('Main client subscribe ' + cmd7ResTopic + ' topic')
client.subscribe(cmd7ResTopic)
def main_client_on_message(client, userdata, msg):
global updateList
global CheckPassFlag
if msg.topic == updateCheckInTopic and CheckPassFlag == True:
try:
jsonMsg = json.loads(msg.payload.decode())
if jsonMsg['gId'] not in ['20010801']:
return
updateDict = {}
if 'version' in jsonMsg:
if checkVersion(jsonMsg['version']):
updateDict['sn'] = jsonMsg['gId']
updateDict['package'] = 0
updateDict['time'] = time.time()
updateDict['send'] = True
updateList.append(updateDict)
else:
updateDict['sn'] = jsonMsg['gId']
updateDict['package'] = 0
updateDict['time'] = time.time()
updateDict['send'] = True
updateList.append(updateDict)
except:
log('Not a json data')
elif msg.topic == updateResponseTopic and CheckPassFlag == True:
try:
responseMsg = msg.payload.decode().split()
for each in updateList:
if responseMsg[1] == each['sn']:
if responseMsg[2] == '0':
each['package'] = int(responseMsg[0]) + 1
each['time'] = time.time()
each['send'] = True
else:
each['send'] = True
except:
log('Error')
elif msg.topic == cmd7ResTopic:
try:
Cmd7Res = json.loads(msg.payload)
if 'stamping' in Cmd7Res:
LocalImage = read_stamping(file_path)
LocalImage = LocalImage.decode()
local_version = extract_characters_around("_", str(LocalImage))
print("\nLocal 镜像版本：{}".format(local_version))
Local_before, Local_after = get_substrings_around(LocalImage, local_version)
RemoteImage = Cmd7Res['stamping']
RemoteImage = RemoteImage
Remote_version = extract_characters_around("_", str(RemoteImage))
print("\nRemote 镜像版本：{}".format(Remote_version))
Remote_before, Remote_after = get_substrings_around(RemoteImage, Remote_version)
if Local_before == Remote_before and Local_after == Remote_after:
CheckPassFlag = True
print("Equal")
else:
print(LocalImage)
print(RemoteImage)
print("Before {}|{}".format(Local_before, Remote_before))
print("After {}|{}".format(Local_after, Remote_after))
print("Fail")
except Exception as e:
log("Error Cmd 7 Res")
print("Exception-> {}".format(e))
def checkVersion(oldVersion):
if 'Beta' in oldVersion and 'Beta' in tdlVersion or 'Beta' not in oldVersion and 'Beta' not in tdlVersion:
if oldVersion < tdlVersion:
return True
else:
return False
elif 'Beta' in oldVersion and 'Beta' not in tdlVersion:
if oldVersion[0:5] < tdlVersion:
return True
else:
return False
elif 'Beta' not in oldVersion and 'Beta' in tdlVersion:
if oldVersion < tdlVersion[0:5]:
return True
else:
return False
return False
def createCheckCode(checkData):
sumDat = 0
codeBytes = b''
for each in checkData:
sumDat += each
sumHex = hex(sumDat).lstrip('0x')
while len(sumHex) != 8:
sumHex = '0' + sumHex
codeBytes += bytes((int(sumHex[0:2], 16),))
codeBytes += bytes((int(sumHex[2:4], 16),))
codeBytes += bytes((int(sumHex[4:6], 16),))
codeBytes += bytes((int(sumHex[6:8], 16),))
return codeBytes
def update():
global fileData
global updateList
global mainClient
global packageSize
global oldTime
data = [{"cmd":7,"ResquestStamping":1}]
data = json.dumps(data, separators=(',', ':'))
print(data)
while CheckPassFlag == False:
print("waiting for check Pass Flag...")
mainClient.publish(cmd7Topic, data)
time.sleep(5)
startPackage = bytes((0,)) + b'Witium_Update' + bytes((packageSize,))
startPackage += createCheckCode(startPackage)
while True:
if len(updateList) == 0:
time.sleep(1)
else:
curTime = time.time()
if curTime - oldTime > 30:
oldTime = time.time()
log('Current List:' + str(updateList))
for eachModule in updateList:
if eachModule['send'] == True:
eachModule['send'] = False
if eachModule['package'] == 0:
mainClient.publish(updateTopic.format(eachModule['sn']), startPackage)
else:
if eachModule['package'] != packageSize:
sendData = bytes((eachModule['package'],)) + fileData[
(eachModule['package'] - 1) * 1024:eachModule[
'package'] * 1024]
else:
sendData = bytes((eachModule['package'],)) + fileData[(eachModule['package'] - 1) * 1024:]
sendData += createCheckCode(sendData)
mainClient.publish(updateTopic.format(eachModule['sn']), sendData)
log('Send the ' + str(eachModule['package']) + ' package to ' + eachModule['sn'])
if eachModule['package'] == packageSize:
updateList.remove(eachModule)
log('Update success ' + eachModule['sn'])
nowTime = time.time()
if nowTime - eachModule['time'] > 30:
updateList.remove(eachModule)
log('Update time out ' + eachModule['sn'])
def log(logData):
with open('log.txt', 'a+') as logFile:
logFile.write(logData + '\n')
print(logData)
#----------------------------------------------------------------------------------------#
path = os.path.abspath(os.path.dirname(os.path.dirname(__file__)))
resourcesPath = path + os.sep + 'safe_Upgrade_Python_Scripts' + os.sep
curFileList = os.listdir(os.getcwd())
oldTime = time.time()
if not os.path.exists('log.txt'):
with open('log.txt', 'w'):
pass
file_path = resourcesPath + NewImage
print("Update Image Path [{}]".format(file_path))
size = get_Image_Size(file_path)
print(f"文件大小：{get_Image_Size(file_path)} bytes")
if HAS_STAMPING(read_stamping(file_path)) == True:
print("STAMPING IS EXIST.")
print("\n\n\n\n\n")
else:
print("STAMPING IS NOT EXIST.")
exit(0)
with open(file_path, 'rb') as file:
fileData = file.read()
packageSize = (len(fileData) - 1) // 1024 + 1
log('packageSize: ' + str(packageSize))
if not file_path:
log('Update file has not found.')
else:
mainClient = mqtt.Client(mqtt.CallbackAPIVersion.VERSION1)
mainClient.username_pw_set(mqttConfig['userName'], mqttConfig['password'])
mainClient.on_connect = main_client_on_connect
mainClient.on_message = main_client_on_message
mainClient.connect(mqttConfig['mqttServer'], mqttConfig['mqttPort'], 60)
thread = threading.Thread(target=update)
thread.setDaemon(True)
thread.start()
mainClient.loop_forever()
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment