#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# Copyright (c) 2023 Huawei Device Co., Ltd.
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""
Description : Generate the update.bin hash check data
"""
import os
import struct
import hashlib
import enum
from log_exception import UPDATE_LOGGER
# hash data sample
# hash info module:
# hash info:1 32 3 4194304
# hash value module:
# /version_list (32bytes)
# 1 176
# 1 176 bf10259a1fc1b2f780a49ce6XXXXXXXX
# hash sign module:
# hash sign:45ef8ec12e56e3b82c9a05XXXXXX
HashType = enum.Enum('HashType', ('SHA256', 'SHA384', 'SHA512'))
HashAlgo = {HashType.SHA256 : hashlib.sha256,
HashType.SHA384 : hashlib.sha384,
HashType.SHA512 : hashlib.sha512}
HASH_TYPE_SIZE = 2
HASH_LENGTH_SIZE = 4
HASH_TLV_SIZE = HASH_TYPE_SIZE + HASH_LENGTH_SIZE
UPGRADE_HASHINFO_SIZE = 10
HASH_DATA_HEADER_SIZE = 38
HASH_DATA_ADDR_SIZE = 8
COMPONENT_NAME_SIZE = 32
# hash block size
HASH_BLOCK_SIZE = 4 * 1024 * 1024
"""
Format
H: unsigned short
I: unsigned int
B: unsigned char
s: char[]
"""
HASH_TLV_FMT = "<HI"
HASH_INFO_FMT = "<3HI"
HASH_DATA_HEADER_FMT = "<32sHI"
HASH_DATA_ADDR_FMT = "<2I"
class CreateHash(object):
"""
Create the component hash data
"""
def __init__(self, hash_type, count):
self.hashinfo_tlv_type = 0x06
self.hashdata_tlv_type = 0x07
self.sign_tlv_type = 0x08
self.hash_type = hash_type
self.hash_digest_size = HashAlgo[hash_type]().digest_size
self.component_num = count
self.block_size = HASH_BLOCK_SIZE
self.hashinfo_value = bytes()
self.hashdata = bytes()
self.signdata = bytes()
self.hashdata_list = []
def write_hashinfo(self):
try:
hashinfo_tlv = struct.pack(HASH_TLV_FMT, self.hashinfo_tlv_type, UPGRADE_HASHINFO_SIZE)
hashinfo_header = struct.pack(HASH_INFO_FMT, self.hash_type.value, self.hash_digest_size,
self.component_num, self.block_size)
except struct.error:
UPDATE_LOGGER.print_log("Pack fail!", log_type=UPDATE_LOGGER.ERROR_LOG)
return False
# write hashinfo
self.hashinfo_value = hashinfo_tlv + hashinfo_header
return True
def write_hashdata(self):
try:
hashdata_len = len(self.hashdata)
hashdata_tlv = struct.pack(HASH_TLV_FMT, self.hashdata_tlv_type, hashdata_len)
except struct.error:
UPDATE_LOGGER.print_log("Pack fail!", log_type=UPDATE_LOGGER.ERROR_LOG)
return False
UPDATE_LOGGER.print_log("Write hashdata hash len %d" % hashdata_len)
# write hashdata
self.hashdata = hashdata_tlv + self.hashdata
UPDATE_LOGGER.print_log("Write hashdata hash tlv complete")
return True
def write_signdata(self, signdata):
try:
signdata_len = len(signdata)
signdata_tlv = struct.pack(HASH_TLV_FMT, self.sign_tlv_type, signdata_len)
except struct.error:
UPDATE_LOGGER.print_log("Pack fail!", log_type=UPDATE_LOGGER.ERROR_LOG)
return False
# write signdata
self.signdata = signdata_tlv + signdata
UPDATE_LOGGER.print_log("Write hashdata sign tlv complete")
return True
def calculate_hash_data(self, data):
hash_algo = HashAlgo[self.hash_type]()
hash_algo.update(data)
return hash_algo.digest()
def write_component_hash_data(self, component):
UPDATE_LOGGER.print_log("calc component hash")
try:
with open(component.file_path, "rb") as component_file:
component_len = os.path.getsize(component.file_path)
block_num = component_len // HASH_BLOCK_SIZE
component_name = component.component_addr.decode().ljust(COMPONENT_NAME_SIZE, "\0")
UPDATE_LOGGER.print_log(
"calc component hash component name:%s %d" % (component_name, len(component_name)))
total_block = block_num + 1 if component_len % HASH_BLOCK_SIZE > 0 else block_num
self.hashdata += struct.pack(HASH_DATA_HEADER_FMT, component_name.encode(),
total_block, component_len)
UPDATE_LOGGER.print_log("calc component hash block_num:%d" % total_block)
write_len = 0
for i in range(0, block_num):
component_file.seek(write_len)
component_data = component_file.read(HASH_BLOCK_SIZE)
write_len += HASH_BLOCK_SIZE
self.hashdata += struct.pack(HASH_DATA_ADDR_FMT, (i * HASH_BLOCK_SIZE if i != 0 else 0),
write_len - 1) + self.calculate_hash_data(component_data)
if component_len - write_len > 0 :
component_file.seek(write_len)
component_data = component_file.read(component_len - write_len)
self.hashdata += struct.pack(HASH_DATA_ADDR_FMT, (write_len if write_len != 0 else 0),
component_len - 1) + self.calculate_hash_data(component_data)
except (struct.error, IOError):
return False
UPDATE_LOGGER.print_log("calc component hash complete ComponentSize:%d" % component_len)
return True
def parse_hashinfo(self, data):
# parse hashinfo
hash_type_value = 0
try:
hash_type_value, self.hash_digest_size, self.component_num, self.block_size = \
struct.unpack(HASH_INFO_FMT, data[:UPGRADE_HASHINFO_SIZE])
self.hash_type = HashType(hash_type_value)
except struct.error:
return False
UPDATE_LOGGER.print_log("parese hashinfo complete, %d %d %d %d" % (hash_type_value,
self.hash_digest_size, self.component_num, self.block_size))
return True
def parse_hashdata(self, data):
offset = 0
try:
for i in range(0, self.component_num):
img_name, hash_num, img_size = struct.unpack(HASH_DATA_HEADER_FMT,
data[offset: HASH_DATA_HEADER_SIZE + offset])
UPDATE_LOGGER.print_log("parese hashinfo complete, %s %d %d" % (img_name,
hash_num, img_size))
offset += HASH_DATA_HEADER_SIZE
self.hashdata_list.append((img_name.decode(), hash_num, img_size))
for j in range(0, hash_num):
hash_data_star, hash_data_end = struct.unpack(HASH_DATA_ADDR_FMT,
data[offset: HASH_DATA_ADDR_SIZE + offset])
hash_data = data[HASH_DATA_ADDR_SIZE + offset:HASH_DATA_ADDR_SIZE + self.hash_digest_size + offset]
offset += (HASH_DATA_ADDR_SIZE + self.hash_digest_size)
self.hashdata_list.append((hash_data_star, hash_data_end, hash_data))
except struct.error:
return False
UPDATE_LOGGER.print_log("parese hashdata complete")
return True
def parse_signdata(self, data):
# parse signdata
self.signdata = data
UPDATE_LOGGER.print_log("parese hashdata sign complete")
return True
def parse_print_hashdata(self, save_path):
hash_check_fd = os.open(os.path.join(save_path + "hash_check_file_parse"), os.O_RDWR | os.O_CREAT, 0o755)
with os.fdopen(hash_check_fd, "wb+") as hash_check_file_p:
hash_check_file_p.write(("hash info:").encode())
hash_check_file_p.write(("%s %s %s %s\n" % (
HashType(self.hash_type.value).name, str(self.hash_digest_size),
str(self.component_num), str(self.block_size))).encode())
offset = 0
for i in range(0, self.component_num):
hash_check_file_p.write(("%s\n" % (self.hashdata_list[offset][0])).encode())
hash_check_file_p.write(("%s %s\n" % (
str(self.hashdata_list[offset][1]), str(self.hashdata_list[offset][2]))).encode())
for j in range(0, self.hashdata_list[offset][1]):
index = offset + 1
hashdata_hexstr = "".join("%02x" % b for b in self.hashdata_list[j + index][2])
hash_check_file_p.write(("%s" % (
str(self.hashdata_list[j + index][0]), str(self.hashdata_list[j + index][1]),
hashdata_hexstr)).encode())
offset += (1 + self.hashdata_list[offset][1])
signdata_hexstr = "".join("%02x" % b for b in self.signdata)
hash_check_file_p.write(("hash sign:").encode())
hash_check_file_p.write(signdata_hexstr.encode())