import request
import usys
+import uos
+from ucryptolib import aes
+import hashlib
+import base64
+import ucryptolib
+import urandom as random
+
+
def server_init(server_type = None):
#服务器选择1，防走丢，2大锤,3三基同创,4童信 5云米
if server_type == None:
odata = 0
i+=1
return result.decode('utf-8') #返回utf8的str
+
+# Device resource authorization
+#设备资源授权平台
+SHA1_BLOCK_SIZE = 64
+PAD_BLOCK_SIZE = 16
+SHA256 = uhashlib.sha256
+# 补位
+pad = lambda s: s + chr(PAD_BLOCK_SIZE - len(s) % PAD_BLOCK_SIZE) * (PAD_BLOCK_SIZE - len(s) % PAD_BLOCK_SIZE)
+# 除去补16字节的多余字符
+unpad = lambda s: s[:-s[-1]]
+class device_res_author:
+
+ key = b"MWrMZjdtcPhj41jWuoqFAhTmqyFluqj1"
+ pid = "qAtdDOGXgXYjT2HGFr3a"
+ def __init__(self) -> None:
+ self.deviceId = "876226002022574"
+
+ def HMAC(self,k, m):
+
+ KEY_BLOCK = k + (b'\0' * (SHA1_BLOCK_SIZE - len(k)))
+ KEY_INNER = bytes((x ^ 0x36) for x in KEY_BLOCK)
+ KEY_OUTER = bytes((x ^ 0x5C) for x in KEY_BLOCK)
+ inner_message = KEY_INNER + m
+ outer_message = KEY_OUTER + SHA256(inner_message).digest()
+ return SHA256(outer_message)
+
+
+ def get_hmac_sha256(self,k, m):
+ data = self.HMAC(k, m).digest()
+ # print("HMAC(k, m).digest()：", data)
+ # print("HMAC(k, m).digest()：", ubinascii.hexlify(data))
+ hex_data = ubinascii.hexlify(data)
+ hex_data = hex_data.decode('ascii')
+ return hex_data
+ # 加密函数
+ def aes_CBC_Encrypt(self,data, key, iv): # CBC模式的加密函数，data为明文，key为16字节密钥,iv为16字节的偏移量
+ key = key.encode('utf-8')
+ iv = iv.encode('utf-8') # CBC 模式下的偏移量
+ data = pad(data) # 补位
+ data = data.encode('utf-8')
+ aes = ucryptolib.aes(key,ucryptolib.MODE_CBC,iv)
+ result = base64.b64encode(aes.encrypt(data))
+ return str(result, 'utf-8') # 以字符串的形式返回
+
+ # 解密函数
+ def aes_CBC_Decrypt(self,data, key, iv): # CBC模式的解密函数，data为密文，key为16字节密钥
+ key = key.encode('utf-8')
+ iv = iv.encode('utf-8')
+
+ aes = ucryptolib.aes(key,ucryptolib.MODE_CBC,iv) #CBC模式
+
+ # decrypt AES解密 B64decode为base64 转码
+ result = aes.decrypt(base64.b64decode(data))
+ result = unpad(result) # 除去补16字节的多余字符
+ return str(result, 'utf-8') # 以字符串的形式返回
+ #激活接口 没有密码走激活
+ def dev_active(self):
+ signal = EventMesh.publish("screen_get_sig")
+ if signal == -1 or EventMesh.publish("get_sim_Status") != 1:#网络不正常
+ print("dev_active 无 SIM卡!!!!!!!!\r\n")
+ return
+ # 签名内容： = deviceId+pid+timeStamp
+ dev_pass = ""
+ dev_pass = EventMesh.publish("get_storage_config","dev_pass")
+ print("dev_pass:{}\r\n".format(dev_pass))
+ if dev_pass == None:
+ dev_pass = ""
+ if dev_pass == "" or dev_pass == None:
+ date_tmp = utime.localtime() #获取当前时间
+ print("date_tmp:{}\r\n".format(date_tmp)) #转换时间戳s
+ timeStamp = (utime.mktime(date_tmp))*1000#获取时间戳ms
+ deviceId = self.deviceId #"876226002022574" #以imei号
+ pid = self.pid
+ key = self.key
+ sign_data = "{}{}{}".format(deviceId,pid,timeStamp) #签名内容
+ print("sign_data:{}\r\n".format(sign_data))
+ sha256_hex = self.get_hmac_sha256(key,sign_data)
+ print("sha256_hex:{}\r\n".format(sha256_hex))
+
+ data = {
+ "deviceId":"{}".format(deviceId), #设备名
+ "pid":pid, #产品PID
+ "sign":"{}".format(sha256_hex), #HMACSHA256签名后的数据
+ "timeStamp":timeStamp, #要以ms的形式
+ "signMethod":"HMACSHA256" #签名方式 HMACSHA256
+ }
+ print(ujson.dumps(data))
+ headers_T ={"Content-Type":"application/json"}
+ url = "http://dmp.raisound.com/bz-api/dev/auth/active"
+ data =ujson.dumps(data)
+ response = 0
+ response_result = 0
+ try:
+ response = request.put(url,data=data,headers = headers_T)
+ response_result = 1
+ except Exception as e:
+ print("put error!!!!!\r\n")
+ response_result = 0
+
+ IV = "" #偏移量
+ PASSWORD= "" #服务器返回后的加密内容 密码
+ if response_result == 1: #访问成功ip
+ if response.status_code == 200:
+ content = response.content
+ c = next(content)
+ # "api.xcloudtech.com"
+ # d_server_port = 9000
+ data = ujson.loads(c)
+ # data = response.json()
+ print("data-{}".format(data))
+ if data != None:
+ if "success" in data:
+ if data['success'] == True:
+ if "data" in data:
+ data_tmp = data["data"]
+ if "nonce" in data_tmp: #偏移量
+ IV = data_tmp["nonce"]
+ if "password" in data_tmp:
+ PASSWORD = data_tmp["password"]
+ # EventMesh.publish("set_storage_config",["dev_pass",PASS])
+ if PASSWORD != None and IV != None:
+ if len(PASSWORD) != 0 and len(IV):
+ password = PASSWORD
+ iv = IV
+ #解密 之后保存本地下一次登入做使用
+ decrypt_data = self.aes_CBC_Decrypt(password, self.key, iv) #对服务器下发的密码进行解密
+ print("decrypt_data {}".format(decrypt_data))
+ EventMesh.publish("set_storage_config",["dev_pass",decrypt_data])
+
+
+ print("IV:{}\r\n PASS:{}".format(IV,PASSWORD))
+ return [response_result,PASSWORD,IV]
+ #随机生成16个字节的字符串作为加密的IV 偏移量
+ def generate_random_string(self,length=16):
+ random_str = ""
+ chars = 'abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789'
+ for i in range(length):
+ random_str+=chars[random.randint(0,length)]
+ return random_str
+
+ #设备登入 没有Token 走登入获取token
+ def dev_sign_in(self):
+ signal = EventMesh.publish("screen_get_sig")
+ if signal == -1 or EventMesh.publish("get_sim_Status") != 1:#网络不正常
+ print("dev_sign_in 无 SIM卡!!!!!!!!\r\n")
+ return
+ token = ""
+ token = EventMesh.publish("get_storage_config","dev_token")
+ dev_pass = EventMesh.publish("get_storage_config","dev_pass")
+ print("token:{}\r\n".format(token))
+ if token == None:
+ token = ""
+ if dev_pass == None:
+ dev_pass = ""
+ if len(dev_pass) == 0:#没有密码
+ return
+ if token == None or token == "":
+ if 1: #返回结果成功
+ #获取本地保存的密码
+ decrypt_data=EventMesh.publish("get_storage_config","dev_pass")
+ # decrypt_data = self.aes_CBC_Decrypt(password, self.key, iv) #对服务器下发的密码进行解密
+ print("decrypt_data {}".format(decrypt_data))
+ #随机生成16个字符串 设备生成做加密使用
+ random_string = self.generate_random_string(16) #随机的16个字符串
+ enc_iv = random_string #加密的IV偏移量
+ #设备端进行密码加密
+ encrypt_data = self.aes_CBC_Encrypt(decrypt_data, self.key, enc_iv)
+ print("encrypt_data {}".format(encrypt_data))
+
+ #登入数据
+ deviceId = self.deviceId
+ sign_in_data = {
+ "deviceId":"{}".format(deviceId), #设备ID
+ "password":"{}".format(encrypt_data), #加密的数据
+ "nonce":"{}".format(random_string) #随机的16个字符串
+ }
+ print(ujson.dumps(sign_in_data))
+ sign_url = "http://dmp.raisound.com/bz-api/dev/auth/sign_in" #登入地址
+ headers_T ={"Content-Type":"application/json"}
+ response_result = 0
+ response = 0
+ try:
+ response = request.post(sign_url,data=ujson.dumps(sign_in_data),headers = headers_T)
+ response_result = 1
+ except Exception as e:
+ print("post error!!!!!\r\n")
+ response_result = 0
+ # {'msg': '', 'code': 20001, 'success': True,
+ # 'data': {'deviceId': '876226002022574',
+ # 'token': '95436a9e09fd0ff89feb8825257dca982f'}
+ # }
+ token = ""
+ if response_result == 1:
+ if response.status_code == 200:
+ content = response.content
+ c = next(content)
+ data = ujson.loads(c)
+ print("data-{}".format(data))
+ if data != None:
+ if "success" in data:
+ if data['success'] == True: #登入成功
+ if "data" in data:
+ data_tmp = data["data"]
+ if "token" in data_tmp: #token
+ token = data_tmp["token"]
+ EventMesh.publish("set_storage_config",["dev_token",token])
+ print("token:{}\r\n".format(token))
+ pass
+ #获取新的令牌 更新token
+ def update_dev_token(self):
+ dev_pass = ""
+ dev_pass = EventMesh.publish("get_storage_config","dev_pass")
+
+ if dev_pass == None:
+ dev_pass = ""
+ if len(dev_pass) == 0: #没有密码走激活
+ print("no dev_pass:{}\r\n".format(dev_pass))
+ self.dev_active()
+ token = EventMesh.publish("get_storage_config","dev_token")
+ if token == None:
+ token = ""
+ if len(token) == 0: #没有token走登入
+ print("no dev_token:{}\r\n".format(token))
+ self.dev_sign_in()
+
+ dev_tokenip = "http://dmp.raisound.com/bz-api/dev/auth/token" #设备新令牌更新地址
+ token = EventMesh.publish("get_storage_config","dev_token")
+ if token == None:
+ token = ""
+ if len(token) != 0:
+ response_result=0
+ response = 0
+ headers_T = {"bz-dt":"{}".format(token)}
+ try:
+ response = request.get(dev_tokenip,headers = headers_T)
+ response_result = 1
+ except Exception as e:
+ print("get error!!!!!\r\n")
+ response_result = 0
+ if response_result == 1:
+ if response.status_code == 200:
+ try:
+ headers_data = response.headers
+ print("headers_data!!!!! headers_data:{}\r\n".format(headers_data))
+ if "bz-dt" in headers_data: #bz-dt字段是否在头部数据里面 如果有就需要更新
+ bz_dt = headers_data["bz-dt"]
+ dev_token = bz_dt.replace(" ", "") #去掉所有的空格符
+ print(dev_token)
+ EventMesh.publish("set_storage_config",["dev_token",dev_token])
+ except Exception as e:
+ print("headers_data!!!!! error\r\n")
+
+ pass
+
+device_res_author_t = device_res_author()
+
class SockManager(Abstract):
def __init__(self):
self.sock = None
# self.server_port = server_data[1]
self.server_addr = ""
self.server_port = ""
+ self.dev_res_timer = osTimer()
+ self.dev_res_timerCount = 0
ip_infor = {}
ip_infor = EventMesh.publish("get_new_ip_com")
print("ip_infor1:{} \n\n\n".format(ip_infor))
#utime.sleep_ms(500)
# 发送数据
+ if EventMesh.DEV_RES_MANAGE_AUTHEN_PLATFORM == 1:
+ self.dev_res_timer.stop()
+ self.dev_res_timer.start(10000, 1, self.dev_timer_cb)
self.publish_timer.start(2000, 1, self.publish_timer_cb)
EventMesh.publish(constants.PROTOCOL.LK_SEND)
def publish_timer_cb(self, *args):
self.authen_set()
-
+ def dev_timer_cb(self, *args): #设备资源管理鉴权平台定时器
+ if EventMesh.DEV_RES_MANAGE_AUTHEN_PLATFORM == 1:
+ device_res_author_t.dev_active() #设备激活
+ device_res_author_t.dev_sign_in() #设备登入
+ self.dev_res_timerCount+=1
+ if self.dev_res_timerCount > 6: #开机一分钟内没有获取到密码和token 等下一次更新令牌在走登入 激活流程
+ self.dev_res_timerCount=0
+ self.dev_res_timer.stop()
+
def authen_set(self):
sever_state = EventMesh.publish("get_server_state")#获取修改服务器状态
# print("sever_state{}".format(sever_state))
projects / mirrors / ASR1603_G5.git / commitdiff