#!/usr/bin/env python3 # -*- coding: utf-8 -*- # Copyright (c) 2021 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. """ The tool for making updater package. positional arguments: target_package Target package file path. update_package Update package file path. optional arguments: -h, --help show this help message and exit -s SOURCE_PACKAGE, --source_package SOURCE_PACKAGE Source package file path. -nz, --no_zip No zip mode, which means to output the update package without zip. -pf PARTITION_FILE, --partition_file PARTITION_FILE Variable partition mode, Partition list file path. -sa {ECC,RSA}, --signing_algorithm {ECC,RSA} The signing algorithms supported by the tool include ['ECC', 'RSA']. -ha {sha256,sha384}, --hash_algorithm {sha256,sha384} The hash algorithms supported by the tool include ['sha256', 'sha384']. -pk PRIVATE_KEY, --private_key PRIVATE_KEY Private key file path. -nl2, --not_l2 Not L2 mode, Distinguish between L1 and L2. -sl {256,384}, --signing_length {256,384} The signing content length supported by the tool include ['256', '384']. -xp, --xml_path XML file path. -sc, --sd_card SD Card mode, Create update package for SD Card. -su, --stream_update Stream update mode, Create update package for stream update. -ab, --ab_partition_update Ab partition update mode, Create update package for ab partition update. """ import filecmp import os import sys import argparse import subprocess import tempfile import hashlib import xmltodict import patch_package_process import math from gigraph_process import GigraphProcess from image_class import FullUpdateImage from image_class import IncUpdateImage from transfers_manager import TransfersManager from log_exception import UPDATE_LOGGER from script_generator import PreludeScript from script_generator import VerseScript from script_generator import RefrainScript from script_generator import EndingScript from update_package import build_update_package from unpack_updater_package import UnpackPackage from utils import OPTIONS_MANAGER from utils import UPDATER_CONFIG from utils import parse_partition_file_xml from utils import unzip_package from utils import clear_resource from utils import PRODUCT from utils import XML_FILE_PATH from utils import get_update_info from utils import SCRIPT_KEY_LIST from utils import PER_BLOCK_SIZE from utils import E2FSDROID_PATH from utils import MAXIMUM_RECURSION_DEPTH from utils import VERSE_SCRIPT_EVENT from utils import INC_IMAGE_EVENT from utils import DIFF_EXE_PATH from utils import PARTITION_CHANGE_EVENT from utils import DECOUPLED_EVENT from utils import get_update_config_softversion from vendor_script import create_vendor_script_class from create_chunk import CreateChunk from concurrent.futures import ThreadPoolExecutor sys.setrecursionlimit(MAXIMUM_RECURSION_DEPTH) def type_check(arg): """ Argument check, which is used to check whether the specified arg is a file. :param arg: the arg to check :return: Check result, which is False if the arg is invalid. """ if arg is not None and not os.path.exists(arg): UPDATE_LOGGER.print_log( "FileNotFoundError, path: %s" % arg, UPDATE_LOGGER.ERROR_LOG) return False return arg def private_key_check(arg): """ Argument check, which is used to check whether the specified arg is a private_key. :param arg: The arg to check. :return: Check result, which is False if the arg is invalid. """ if arg != "ON_SERVER" and not os.path.isfile(arg): UPDATE_LOGGER.print_log( "FileNotFoundError, path: %s" % arg, UPDATE_LOGGER.ERROR_LOG) return False return arg def check_update_package(arg): """ Argument check, which is used to check whether the update package path exists. :param arg: The arg to check. :return: Check result """ make_dir_path = None if os.path.exists(arg): if os.path.isfile(arg): UPDATE_LOGGER.print_log( "Update package must be a dir path, not a file path. " "path: %s" % arg, UPDATE_LOGGER.ERROR_LOG) return False else: try: UPDATE_LOGGER.print_log( "Update package path does not exist. The dir will be created!" "path: %s" % arg, UPDATE_LOGGER.WARNING_LOG) os.makedirs(arg) make_dir_path = arg except OSError: UPDATE_LOGGER.print_log( "Make update package path dir failed! " "path: %s" % arg, UPDATE_LOGGER.ERROR_LOG) return False if make_dir_path is not None: OPTIONS_MANAGER.make_dir_path = make_dir_path OPTIONS_MANAGER.update_package = arg return arg def unpack_check(arg): """ Argument check, which is used to check whether the update package path exists. :param arg: The arg to check. :return: Check result """ unpack_package = os.path.join(OPTIONS_MANAGER.update_package, arg) if not os.path.isfile(unpack_package): UPDATE_LOGGER.print_log( "FileNotFoundError, path: %s" % unpack_package, UPDATE_LOGGER.ERROR_LOG) OPTIONS_MANAGER.unpack_package_path = None return False OPTIONS_MANAGER.unpack_package_path = unpack_package return arg def create_entrance_args(): """ Arguments for the tool to create an update package :return source_package : source version package target_package : target version package update_package : update package output path no_zip : whether to enable the update package zip function. partition_file : partition table XML file signing_algorithm : signature algorithm (ECC and RSA (default)) private_key : path of the private key file """ parser = OPTIONS_MANAGER.parser parser.description = "Tool for creating update package." parser.add_argument("-unpack", "--unpack_package", type=unpack_check, default=None, help="Unpack updater package.") parser.add_argument("-s", "--source_package", type=type_check, default=None, help="Source package file path.") parser.add_argument("target_package", type=type_check, help="Target package file path.") parser.add_argument("update_package", type=check_update_package, help="Update package file path.") parser.add_argument("-nz", "--no_zip", action='store_true', help="No zip mode, Output update package without zip.") parser.add_argument("-pf", "--partition_file", default=None, help="Variable partition mode, " "Partition list file path.") parser.add_argument("-sa", "--signing_algorithm", default='RSA', choices=['ECC', 'RSA'], help="The signing algorithm " "supported by the tool include ['ECC', 'RSA'].") parser.add_argument("-ha", "--hash_algorithm", default='sha256', choices=['sha256', 'sha384'], help="The hash algorithm " "supported by the tool include " "['sha256', 'sha384'].") parser.add_argument("-pk", "--private_key", type=private_key_check, default=None, help="Private key file path.") parser.add_argument("-nl2", "--not_l2", action='store_true', help="Not L2 mode, Distinguish between L1 and L2.") parser.add_argument("-sl", "--signing_length", default='256', choices=['256', '384'], help="The signing content length " "supported by the tool include " "['256', '384'].") parser.add_argument("-xp", "--xml_path", type=private_key_check, default=None, help="XML file path.") parser.add_argument("-sc", "--sd_card", action='store_true', help="SD Card mode, " "Create update package for SD Card.") parser.add_argument("-su", "--stream_update", action='store_true', help="Stream update mode, " "Create update package for stream update.") parser.add_argument("-ab", "--ab_partition_update", action='store_true', help="Ab partition update mode, " "Create update package for ab partition update.") def parse_args(): args = OPTIONS_MANAGER.parser.parse_args() OPTIONS_MANAGER.source_package = args.source_package OPTIONS_MANAGER.target_package = args.target_package OPTIONS_MANAGER.update_package = args.update_package OPTIONS_MANAGER.no_zip = args.no_zip OPTIONS_MANAGER.partition_file = args.partition_file OPTIONS_MANAGER.signing_algorithm = args.signing_algorithm OPTIONS_MANAGER.hash_algorithm = args.hash_algorithm OPTIONS_MANAGER.private_key = args.private_key OPTIONS_MANAGER.not_l2 = args.not_l2 OPTIONS_MANAGER.signing_length = int(args.signing_length) OPTIONS_MANAGER.xml_path = args.xml_path OPTIONS_MANAGER.sd_card = args.sd_card OPTIONS_MANAGER.stream_update = args.stream_update OPTIONS_MANAGER.ab_partition_update = args.ab_partition_update def get_args(): ret_args = \ [OPTIONS_MANAGER.source_package, OPTIONS_MANAGER.target_package, OPTIONS_MANAGER.update_package, OPTIONS_MANAGER.no_zip, OPTIONS_MANAGER.not_l2, OPTIONS_MANAGER.partition_file, OPTIONS_MANAGER.signing_algorithm, OPTIONS_MANAGER.hash_algorithm, OPTIONS_MANAGER.private_key] return ret_args def get_script_obj(): """ Obtain Opera script object :return: """ script_obj_list = create_vendor_script_class() if script_obj_list == [None] * len(SCRIPT_KEY_LIST): prelude_script = PreludeScript() verse_script = VerseScript() refrain_script = RefrainScript() ending_script = EndingScript() generate_verse_script = \ OPTIONS_MANAGER.init.invoke_event(VERSE_SCRIPT_EVENT) if generate_verse_script: verse_script = generate_verse_script() else: UPDATE_LOGGER.print_log( "Get vendor extension object completed!" "The vendor extension script will be generated.") prelude_script = script_obj_list[0] verse_script = script_obj_list[1] refrain_script = script_obj_list[2] ending_script = script_obj_list[3] return prelude_script, verse_script, refrain_script, ending_script def get_source_package_path(source_package): """ get_source_package_path. :param source_package: source package path :return: """ if os.path.isdir(source_package): OPTIONS_MANAGER.source_package_dir = source_package elif source_package.endswith('.zip'): # Decompress the source package. tmp_dir_obj, unzip_dir = unzip_package(source_package) if tmp_dir_obj is False or unzip_dir is False: clear_resource(err_clear=True) return False OPTIONS_MANAGER.source_package_dir = unzip_dir OPTIONS_MANAGER.source_package_temp_obj = tmp_dir_obj else: UPDATE_LOGGER.print_log("Input Update Package type exception!" "path: %s" % source_package, UPDATE_LOGGER.ERROR_LOG) clear_resource(err_clear=True) return False return True def check_incremental_args(no_zip, partition_file, source_package, incremental_img_list): """ When the incremental list is not empty, incremental processing is required. In this case, check related arguments. :param no_zip: no zip mode :param partition_file: :param source_package: :param incremental_img_list: :return: """ if "boot" in incremental_img_list: UPDATE_LOGGER.print_log("boot cannot be incrementally processed!", UPDATE_LOGGER.ERROR_LOG) clear_resource(err_clear=True) return False if source_package is None: UPDATE_LOGGER.print_log("The source package is missing, " "cannot be incrementally processed!", UPDATE_LOGGER.ERROR_LOG) clear_resource(err_clear=True) return False if no_zip: UPDATE_LOGGER.print_log("No ZIP mode, cannot be incrementally processed!", UPDATE_LOGGER.ERROR_LOG) clear_resource(err_clear=True) return False if partition_file is not None: UPDATE_LOGGER.print_log("Partition file is not None, " "cannot be incrementally processed!", UPDATE_LOGGER.ERROR_LOG) clear_resource(err_clear=True) return False if not get_source_package_path(source_package): return False partition_change = OPTIONS_MANAGER.init.invoke_event(PARTITION_CHANGE_EVENT) if callable(partition_change) and partition_change() is False: return False UPDATE_LOGGER.print_log("Partition interception check finish.", UPDATE_LOGGER.INFO_LOG) xml_path = '' if OPTIONS_MANAGER.source_package_dir is not False: xml_path = os.path.join(OPTIONS_MANAGER.source_package_dir, UPDATER_CONFIG, XML_FILE_PATH) if OPTIONS_MANAGER.source_package_dir is False: OPTIONS_MANAGER.source_package_temp_obj = None OPTIONS_MANAGER.source_package_dir = None if os.path.exists(xml_path): with open(xml_path, 'r') as xml_file: xml_str = xml_file.read() else: UPDATE_LOGGER.print_log("XML file does not exist! xml path: %s" % xml_path, UPDATE_LOGGER.ERROR_LOG) return False xml_content_dict = xmltodict.parse(xml_str, encoding='utf-8') package_dict = xml_content_dict.get('package', {}) get_update_config_softversion(OPTIONS_MANAGER.source_package_dir, package_dict.get('head', {})) head_dict = package_dict.get('head', {}).get('info') OPTIONS_MANAGER.source_package_version = head_dict.get("@softVersion") no_need_version_check_res = OPTIONS_MANAGER.init.invoke_event(DECOUPLED_EVENT) if no_need_version_check_res is False: if check_package_version(OPTIONS_MANAGER.target_package_version, OPTIONS_MANAGER.source_package_version) is False: clear_resource(err_clear=True) return False return True def check_userdata_image(): """ Check the userdata image. Updating this image is prohibited. :return: """ if 'userdata' in OPTIONS_MANAGER.full_img_list or \ 'userdata' in OPTIONS_MANAGER.incremental_img_list: UPDATE_LOGGER.print_log( "userdata image does not participate in update!" "Please check xml config, path: %s!" % os.path.join(OPTIONS_MANAGER.target_package_config_dir, XML_FILE_PATH), UPDATE_LOGGER.ERROR_LOG) clear_resource(err_clear=True) return False return True def check_images_list(): """ Check full_img_list and incremental_img_list. If their lengths are 0, an error will be logged. :return: """ if len(OPTIONS_MANAGER.full_img_list) == 0 and \ len(OPTIONS_MANAGER.incremental_img_list) == 0: UPDATE_LOGGER.print_log( "The image list is empty!" "Please check xml config, path: %s!" % os.path.join(OPTIONS_MANAGER.target_package_config_dir, XML_FILE_PATH), UPDATE_LOGGER.ERROR_LOG) clear_resource(err_clear=True) return False return True def check_target_package_path(target_package): """ Check the target_package path. :param target_package: target package path :return: """ if os.path.isdir(target_package): OPTIONS_MANAGER.target_package_dir = target_package temp_dir_list = os.listdir(target_package) if UPDATER_CONFIG in temp_dir_list: OPTIONS_MANAGER.target_package_config_dir = \ os.path.join(target_package, UPDATER_CONFIG) else: UPDATE_LOGGER.print_log( "Exception's target package path! path: %s" % target_package, UPDATE_LOGGER.ERROR_LOG) return False elif target_package.endswith('.zip'): # Decompress the target package. tmp_dir_obj, unzip_dir = unzip_package(target_package) if tmp_dir_obj is False or unzip_dir is False: clear_resource(err_clear=True) return False OPTIONS_MANAGER.target_package_dir = unzip_dir OPTIONS_MANAGER.target_package_temp_obj = tmp_dir_obj OPTIONS_MANAGER.target_package_config_dir = \ os.path.join(unzip_dir, UPDATER_CONFIG) else: UPDATE_LOGGER.print_log( "Input Update Package type exception! path: %s" % target_package, UPDATE_LOGGER.ERROR_LOG) clear_resource(err_clear=True) return False return True def check_miss_private_key(private_key): """ Check private key. :param private_key: :return: """ if private_key is None: UPDATE_LOGGER.print_log( "Private key is None, update package cannot be signed! " "Please specify the signature private key by -pk.", UPDATE_LOGGER.ERROR_LOG) clear_resource(err_clear=True) return False return True def check_package_version(target_ver, source_ver): """ target_ver: target version source_ver: source version return: """ try: target_num = ''.join(target_ver.split(' ')[-1].replace('.', '')) source_num = ''.join(source_ver.split(' ')[-1].replace('.', '')) if int(target_num) <= int(source_num): UPDATE_LOGGER.print_log( 'Target package version %s <= Source package version!' 'Unable to make updater package!', UPDATE_LOGGER.ERROR_LOG) return False except ValueError: UPDATE_LOGGER.print_log('your package version number is not compliant.' 'Please check your package version number!', UPDATE_LOGGER.ERROR_LOG) return False return True def generate_image_map_file(image_path, map_path, image_name): """ :param image_path: image path :param map_path: image map file path :param image_name: image name :return: """ if not os.path.exists(image_path): UPDATE_LOGGER.print_log("The source %s.img file is missing from the" "source package, cannot be incrementally processed. " % image_name, UPDATE_LOGGER.ERROR_LOG) return False cmd = [E2FSDROID_PATH, "-B", map_path, "-a", "/%s" % image_name, image_path, "-e"] res = subprocess.Popen(cmd, shell=False, stdout=subprocess.PIPE, stderr=subprocess.PIPE) _, err = res.communicate(timeout=300) if res.returncode != 0: UPDATE_LOGGER.print_log("%s.map generate failed, reason:%s" % (image_name, err.decode()), UPDATE_LOGGER.ERROR_LOG) return False UPDATE_LOGGER.print_log("%s.map generate success" % image_name, UPDATE_LOGGER.INFO_LOG) return True def get_file_sha256(update_package): sha256obj = hashlib.sha256() maxbuf = 8192 with open(update_package, 'rb') as package_file: while True: buf = package_file.read(maxbuf) if not buf: break sha256obj.update(buf) hash_value = sha256obj.hexdigest() return str(hash_value).upper() def write_image_patch_script(partition, src_image_path, tgt_image_path, script_check_cmd_list, script_write_cmd_list, verse_script): """ Add command content to the script. :param partition: image name :param script_check_cmd_list: incremental check command list :param script_write_cmd_list: incremental write command list :param verse_script: verse script object :return: """ src_sha = get_file_sha256(src_image_path) src_size = os.path.getsize(src_image_path) tgt_sha = get_file_sha256(tgt_image_path) tgt_size = os.path.getsize(tgt_image_path) sha_check_cmd = verse_script.image_sha_check(partition, src_size, src_sha, tgt_size, tgt_sha) first_block_check_cmd = verse_script.first_block_check(partition) abort_cmd = verse_script.abort(partition) cmd = 'if ({sha_check_cmd} != 0)' \ '{{\n {abort_cmd}}}\n'.format( sha_check_cmd=sha_check_cmd, abort_cmd=abort_cmd) script_check_cmd_list.append(cmd) image_patch_cmd = verse_script.image_patch(partition, os.path.getsize(src_image_path), get_file_sha256(src_image_path), os.path.getsize(tgt_image_path), get_file_sha256(tgt_image_path)) cmd = '%s_WRITE_FLAG%s' % (partition, image_patch_cmd) script_write_cmd_list.append(cmd) return True def increment_image_diff_processing( partition, src_image_path, tgt_image_path, script_check_cmd_list, script_write_cmd_list, verse_script): """ Incremental image processing :param verse_script: verse script :param incremental_img_list: incremental image list :param source_package_dir: source package path :param target_package_dir: target package path :return: """ patch_file_obj = tempfile.NamedTemporaryFile( prefix="%s_patch.dat-" % partition, mode='wb') OPTIONS_MANAGER.incremental_image_file_obj_dict[partition] = patch_file_obj cmd = [DIFF_EXE_PATH] cmd.extend(['-s', src_image_path, '-d', tgt_image_path, '-p', patch_file_obj.name, '-l', '4096']) sub_p = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT) try: output, _ = sub_p.communicate(timeout=1800) except subprocess.TimeoutExpired: sub_p.kill() sub_p.wait() if sub_p.returncode != 0: raise ValueError(output) return write_image_patch_script(partition, src_image_path, tgt_image_path, script_check_cmd_list, script_write_cmd_list, verse_script) def copy_in_ab_process(patch_process, src_image_class, need_copy_blocks, non_continuous_blocks, each_img): """ Handle the process of copying blocks in an AB partition during an update. :param patch_process: The patch process object responsible for managing the update. :param src_image_class: The source image class containing metadata about the image. :param need_copy_blocks: A list to store blocks that need to be copied. :param non_continuous_blocks: A list to store blocks that are non-continuous. :param each_img: The current image being processed. :return: """ if OPTIONS_MANAGER.ab_partition_update: chunk_pkgdiff_list = patch_process.get_chunk_pkgdiff_list() transfer_content = patch_process.get_transfer_content_in_chunk() chunk_new_list = patch_process.get_chunk_new_list() # Parase transfer_content, record all blocks involved in the content no_copy_blocks_in_ab, transfer_content_no_startoffert = parse_transfer_content(transfer_content) # Find the blocks to copy, total blocks includeing zero, excluding total_blocks no_copy_blocks_set = set(no_copy_blocks_in_ab) def process_block(src_block): if src_block in no_copy_blocks_set: return None, src_block # Not to copy return src_block, None # To copy with ThreadPoolExecutor() as executor: results = executor.map(process_block, range(src_image_class.total_blocks)) for to_copy, not_to_copy in results: if to_copy is not None: need_copy_blocks.append(to_copy) if not_to_copy is not None: non_continuous_blocks.append(not_to_copy) # Sort blocks consecutively if len(need_copy_blocks): group_numbers_list = group_numbers(need_copy_blocks) else: group_numbers_list = [] print(f'there is no copy blocks in image {each_img} !') # Add the copy command for ab partition synchronization transfer_content = patch_process.add_ab_copy_content(len(need_copy_blocks), group_numbers_list, transfer_content_no_startoffert) OPTIONS_MANAGER.image_transfer_dict_contents[each_img] = transfer_content OPTIONS_MANAGER.image_patch_dic[each_img] = chunk_pkgdiff_list OPTIONS_MANAGER.image_new_dic[each_img] = chunk_new_list def add_incremental_command(verse_script, script_check_cmd_list, script_write_cmd_list): """ add command for increment_image_progressing to verse_script :param verse_script: verse script :param script_check_cmd_list: verse_script check command list :param script_write_cmd_list: verse_script write command list :return: """ verse_script.add_command("\n# ---- start incremental check here ----\n") for each_check_cmd in script_check_cmd_list: verse_script.add_command(each_check_cmd) verse_script.add_command("\n# ---- start incremental write here ----\n") for each_write_cmd in script_write_cmd_list: verse_script.add_command(each_write_cmd) def increment_image_processing( verse_script, incremental_img_list, source_package_dir, target_package_dir): """ Incremental image processing :param verse_script: verse script :param incremental_img_list: incremental image list :param source_package_dir: source package path :param target_package_dir: target package path :return: """ script_check_cmd_list = [] script_write_cmd_list = [] patch_process = None for each_img_name in OPTIONS_MANAGER.incremental_img_name_list: each_img = each_img_name[:-4] each_src_image_path = os.path.join(source_package_dir, '%s.img' % each_img) each_src_map_path = os.path.join(source_package_dir, '%s.map' % each_img) each_tgt_image_path = os.path.join(target_package_dir, '%s.img' % each_img) each_tgt_map_path = os.path.join(target_package_dir, '%s.map' % each_img) # This will store tuples of (start, end, length) for continuous ranges non_continuous_blocks = [] need_copy_blocks = [] check_make_map_path(each_img) # Call the new function to process image maps continue_processing, src_generate_map, tgt_generate_map = process_image_maps( each_img, each_src_image_path, each_src_map_path, each_tgt_image_path, each_tgt_map_path) if not continue_processing: continue # get the large of target image if not get_large_of_target_image(each_tgt_image_path, each_img): return False if not src_generate_map or not tgt_generate_map: if not handle_no_map_generation(each_img, each_src_image_path, each_tgt_image_path, \ script_check_cmd_list, script_write_cmd_list, verse_script): return False continue inc_image = OPTIONS_MANAGER.init.invoke_event(INC_IMAGE_EVENT) if inc_image: src_image_class, tgt_image_class = inc_image(each_src_image_path, each_src_map_path, each_tgt_image_path, each_tgt_map_path) else: src_image_class = IncUpdateImage(each_src_image_path, each_src_map_path) tgt_image_class = IncUpdateImage(each_tgt_image_path, each_tgt_map_path) transfers_manager = TransfersManager(each_img, tgt_image_class, src_image_class) transfers_manager.find_process_needs() actions_list = transfers_manager.get_action_list() graph_process = GigraphProcess(actions_list, src_image_class, tgt_image_class) # Streaming update does not need to handle stash and free commands if not OPTIONS_MANAGER.ab_partition_update: graph_process.stash_process() actions_list = graph_process.actions_list patch_process = patch_package_process.PatchProcess(each_img, tgt_image_class, src_image_class, actions_list) patch_process.patch_process(each_tgt_image_path) # Add copy command for ab partition copy_in_ab_process(patch_process, src_image_class, need_copy_blocks, non_continuous_blocks, each_img) patch_process.write_script(each_img, script_check_cmd_list, script_write_cmd_list, verse_script) OPTIONS_MANAGER.incremental_block_file_obj_dict[each_img] = patch_process.package_patch_zip if not OPTIONS_MANAGER.stream_update: if not check_patch_file(patch_process): UPDATE_LOGGER.print_log('Verify the incremental result failed!', UPDATE_LOGGER.ERROR_LOG) raise RuntimeError add_incremental_command(verse_script, script_check_cmd_list, script_write_cmd_list) return True def get_large_of_target_image(each_tgt_image_path, each_img): """ Reads the target image content and stores it in OPTIONS_MANAGER.diff_image_new_data. :param each_tgt_image_path: The path to the target image. :param each_img: The name of the image (without extension). :return: True if successful, False otherwise. """ try: with open(each_tgt_image_path, 'rb') as f: target_content = f.read() OPTIONS_MANAGER.diff_image_new_data[each_img] = target_content return True except Exception as e: print(f"Error reading target image {each_img}: {e}") return False def process_image_maps(each_img, each_src_image_path, each_src_map_path, each_tgt_image_path, each_tgt_map_path): """ Process source and target image maps for incremental updates. :param each_img: The image name without extension :param each_src_image_path: Path to the source image :param each_src_map_path: Path to the source map :param each_tgt_image_path: Path to the target image :param each_tgt_map_path: Path to the target map :return: A tuple (continue_processing, src_generate_map, tgt_generate_map) - continue_processing: Whether to continue processing the current image - src_generate_map: Whether the source map was successfully generated - tgt_generate_map: Whether the target map was successfully generated """ # Check if source and target images are identical if filecmp.cmp(each_src_image_path, each_tgt_image_path): UPDATE_LOGGER.print_log( "Source Image is the same as Target Image! src image path: %s, tgt image path: %s" % (each_src_image_path, each_tgt_image_path), UPDATE_LOGGER.INFO_LOG) OPTIONS_MANAGER.incremental_img_list.remove(each_img) return False, False, False # Stop further processing for this image # Initialize map generation flags src_generate_map = True tgt_generate_map = True # Generate source map if it does not exist if not os.path.exists(each_src_map_path): src_generate_map = generate_image_map_file(each_src_image_path, each_src_map_path, each_img) # Generate target map if it does not exist if not os.path.exists(each_tgt_map_path): tgt_generate_map = generate_image_map_file(each_tgt_image_path, each_tgt_map_path, each_img) return True, src_generate_map, tgt_generate_map def handle_no_map_generation(each_img, each_src_image_path, each_tgt_image_path, script_check_cmd_list, script_write_cmd_list, verse_script): if OPTIONS_MANAGER.stream_update: print(f'do no map process:{each_img}') OPTIONS_MANAGER.no_map_image_exist = True OPTIONS_MANAGER.no_map_file_list.append(each_img) # Directly cut the target image into new commands if not os.path.exists(each_tgt_image_path): UPDATE_LOGGER.print_log( "The file is missing " "from the target package, " "the component: %s cannot be full update processed. " % each_tgt_image_path) return False with open(each_tgt_image_path, 'rb') as f: tartget_no_map_content = f.read() chunk, block_sets = split_image_file(each_img, tartget_no_map_content) OPTIONS_MANAGER.image_chunk[each_img] = chunk OPTIONS_MANAGER.image_block_sets[each_img] = block_sets return True # If it is not a streaming update and cannot generate map file,directly diff the image elif increment_image_diff_processing(each_img, each_src_image_path, each_tgt_image_path, script_check_cmd_list, script_write_cmd_list, verse_script) is True: return True UPDATE_LOGGER.print_log("increment_image_diff_processing %s failed" % each_img) clear_resource(err_clear=True) return False def check_patch_file(patch_process): new_dat_file_obj, patch_dat_file_obj, transfer_list_file_obj = \ patch_process.package_patch_zip.get_file_obj() with open(transfer_list_file_obj.name) as f_t: num = 0 diff_str = None diff_num = 0 for line in f_t: if line.startswith('new '): each_line_list = \ line.strip().replace("new ", "").split(",")[1:] for idx in range(0, len(each_line_list), 2): num += \ int(each_line_list[idx + 1]) - int(each_line_list[idx]) continue if line.startswith('bsdiff ') or line.startswith('pkgdiff '): diff_str = line if diff_str: diff_list = diff_str.split('\n')[0].split(' ') diff_num = int(diff_list[1]) + int(diff_list[2]) check_flag = \ (os.path.getsize(new_dat_file_obj.name) == num * PER_BLOCK_SIZE) and \ (os.path.getsize(patch_dat_file_obj.name) == diff_num) return check_flag def check_make_map_path(each_img): """ If env does not exist, the command for map generation does not exist in the environment variable, and False will be returned. """ try: cmd = [E2FSDROID_PATH, " -h"] subprocess.Popen(cmd, shell=False, stdout=subprocess.PIPE, stderr=subprocess.STDOUT) except FileNotFoundError: UPDATE_LOGGER.print_log( "Command not found, need check the env! " "Make %s.map failed!" % each_img, UPDATE_LOGGER.ERROR_LOG) clear_resource(err_clear=True) raise RuntimeError return True def incremental_processing(no_zip, partition_file, source_package, verse_script): """ Incremental processing. :param no_zip: no zip mode :param partition_file: partition xml file path :param source_package: source package path :param verse_script: verse script obj :return : processing result """ if len(OPTIONS_MANAGER.incremental_img_list) != 0: if check_incremental_args(no_zip, partition_file, source_package, OPTIONS_MANAGER.incremental_img_list) \ is False: return False if increment_image_processing( verse_script, OPTIONS_MANAGER.incremental_img_list, OPTIONS_MANAGER.source_package_dir, OPTIONS_MANAGER.target_package_dir) is False: return False else: if source_package is not None: UPDATE_LOGGER.print_log( "There is no incremental image, " "the - S parameter is not required!", UPDATE_LOGGER.ERROR_LOG) raise RuntimeError def check_args(private_key, source_package, target_package, update_package): """ Input args check. :param private_key: private key path :param source_package: source package path :param target_package: target package path :param update_package: output package path :return : Judgment result """ if source_package is False or private_key is False or \ target_package is False or update_package is False: return False if check_miss_private_key(private_key) is False: return False if check_target_package_path(target_package) is False: return False if get_update_info() is False: return False if check_images_list() is False: return False return True def unpack_package_processing(): if OPTIONS_MANAGER.unpack_package_path: package = UnpackPackage() if not package.unpack_package(): UPDATE_LOGGER.print_log("Unpack update package.bin failed!", UPDATE_LOGGER.ERROR_LOG) clear_resource(err_clear=True) sys.exit(1) UPDATE_LOGGER.print_log("Unpack update package.bin success!") clear_resource() sys.exit(0) def group_numbers(arr): """ Groups and sorts a list of consecutive numbers, returning a list of block ranges. :param arr: A list of integers representing block numbers. :return: A list of integers representing the start and end of each block range. """ result = [] current_group = [arr[0]] transfer_content_blocks = [] for i in range(1, len(arr)): if arr[i] == arr[i - 1] + 1: current_group.append(arr[i]) else: result.append({ 'min': min(current_group), 'max': max(current_group), 'length': len(current_group) }) transfer_content_blocks.append(min(current_group)) transfer_content_blocks.append(max(current_group) + 1) current_group = [arr[i]] result.append({ 'min': min(current_group), 'max': max(current_group), 'length': len(current_group) }) transfer_content_blocks.append(min(current_group)) transfer_content_blocks.append(max(current_group) + 1) return transfer_content_blocks def parse_transfer_content(content): """ Parses the transfer content to extract block information and modified lines. :param content: The string content to be parsed, typically containing block data. :return: A tuple containing: - all_blocks: A list of all extracted block numbers. - modified_content: A modified version of the original content with adjustments made. """ all_blocks = [] modified_lines = [] lines = content.splitlines() for line in lines[4:]: if 'pkgdiff' in line: modified_line, blocks = handle_pkgdiff(line) modified_lines.append(modified_line) all_blocks.extend(blocks) elif 'zero' in line or 'erase' in line: blocks = handle_zero_erase(line) all_blocks.extend(blocks) modified_lines.append(line) elif 'move' in line or 'new' in line: blocks = handle_move_new(line) all_blocks.extend(blocks) modified_lines.append(line) elif 'free' in line: print('no add free cmd') modified_lines.append(line) else: raise ValueError(f"no parse transfer_list info, got {lines}.") modified_content = '\n'.join(lines[:4] + modified_lines) return all_blocks, modified_content def handle_pkgdiff(line): """ Handles the 'pkgdiff' line type. :param line: The line to process. :return: A tuple containing the modified line and the list of blocks. """ parts = line.replace(',', ' ').split() # Change the start offset address of pkgdiff to all zeros modified_line = line.replace(line.split()[1], '0', 1) length = int(parts[5]) if length % 2 != 0: raise ValueError(f"Length must be even, got {length}.") # Extract the range values of a pair of blocks pairs = list(map(int, parts[6:length + 6])) blocks = [] # List all blocks in a single pair of block range values for i in range(0, len(pairs), 2): start = pairs[i] print(f'start:{start}') end = pairs[i + 1] print(f'end:{end}') blocks.extend(range(start, end)) print(f'len(blocks):{len(blocks)}') return modified_line, blocks def handle_zero_erase(line): """ Handles the 'zero' or 'erase' line type. :param line: The line to process. :return: The list of blocks extracted from the line. """ parts = line.replace(',', ' ').split() length = int(parts[1]) print(f'lenght:{length}') if length % 2 != 0: raise ValueError(f"Length must be even, got {length}.") pairs = list(map(int, parts[2:length + 2])) blocks = [] for i in range(0, len(pairs), 2): start = pairs[i] print(f'start:{start}') end = pairs[i + 1] print(f'end:{end}') blocks.extend(range(start, end)) print(f'len(blocks):{len(blocks)}') return blocks def handle_move_new(line): """ Handles the 'move' or 'new' line type. :param line: The line to process. :return: The list of blocks extracted from the line. """ parts = line.replace(',', ' ').split() length = int(parts[2]) print(f'lenght:{length}') if length % 2 != 0: raise ValueError(f"Length must be even, got {length}.") pairs = list(map(int, parts[3:length + 3])) blocks = [] for i in range(0, len(pairs), 2): start = pairs[i] print(f'start:{start}') end = pairs[i + 1] print(f'end:{end}') blocks.extend(range(start, end)) print(f'len(blocks):{len(blocks)}') return blocks def split_image_file(each_img, full_image_data): """ Splits the full image data into smaller chunks. :param each_img: The image to be split (not used in the current implementation). :param full_image_data: The complete image data to be split. :return: A tuple containing two lists: - chunks: A list of data chunks. - block_sets: A list of corresponding block sets for each chunk. """ # get the total block number of the image total_blocks = math.ceil(len(full_image_data) / 4096) print(f'total tgt blocks:{total_blocks}') # step 1：get the total size of the image data max_chunk_size = OPTIONS_MANAGER.chunk_limit * 4096 total_size = len(full_image_data) print(f'total size:{total_size}') chunks = [] block_sets = [] # step 2：cut the image data into fixed block size num = math.ceil(total_size / max_chunk_size) print(f'num:{num}') for i in range(num): start_index = i * max_chunk_size end_index = min(start_index + max_chunk_size, total_size) chunk = full_image_data[start_index:end_index] # step 3：record the corresponding block set block_set = list(range(math.ceil(start_index / 4096), math.ceil(end_index / 4096))) chunks.append(chunk) block_sets.append(block_set) print(f"block {i + 1}: size = {len(chunk)} bytes, block sets = {block_set}") print(f'total size:{total_size}') print(f'total tgt blocks:{total_blocks}') return chunks, block_sets create_entrance_args() def main(): """ Entry function. """ parse_args() OPTIONS_MANAGER.product = PRODUCT source_package, target_package, update_package, no_zip, not_l2, \ partition_file, signing_algorithm, hash_algorithm, private_key = get_args() if not_l2: no_zip = True # Unpack updater package unpack_package_processing() if OPTIONS_MANAGER.sd_card: if source_package is not None or OPTIONS_MANAGER.xml_path is not None or partition_file is not None: UPDATE_LOGGER.print_log("SDCard updater:-S/-xp/-pf parameter is not required!", UPDATE_LOGGER.ERROR_LOG) raise RuntimeError if check_args(private_key, source_package, target_package, update_package) is False: clear_resource(err_clear=True) return if not OPTIONS_MANAGER.sd_card: if check_userdata_image() is False: clear_resource(err_clear=True) return # Create a Script object. prelude_script, verse_script, refrain_script, ending_script = get_script_obj() # Create partition. if partition_file is not None: verse_script.add_command("\n# ---- do updater partitions ----\n") updater_partitions_cmd = verse_script.updater_partitions() verse_script.add_command(updater_partitions_cmd) partition_file_obj, partitions_list, partitions_file_path_list = parse_partition_file_xml(partition_file) if partition_file_obj is False: clear_resource(err_clear=True) return False OPTIONS_MANAGER.partition_file_obj = partition_file_obj OPTIONS_MANAGER.full_img_list = partitions_list OPTIONS_MANAGER.full_image_path_list = partitions_file_path_list # Incremental processing if incremental_processing(no_zip, partition_file, source_package, verse_script) is False: clear_resource(err_clear=True) return # Full processing if len(OPTIONS_MANAGER.full_img_list) != 0: verse_script.add_command("\n# ---- full image ----\n") full_update_image = FullUpdateImage(OPTIONS_MANAGER.target_package_dir, OPTIONS_MANAGER.full_img_list, OPTIONS_MANAGER.full_img_name_list, verse_script, OPTIONS_MANAGER.full_image_path_list, no_zip=OPTIONS_MANAGER.no_zip) full_image_content_len_list, full_image_file_obj_list = full_update_image.update_full_image() if full_image_content_len_list is False or full_image_file_obj_list is False: clear_resource(err_clear=True) return # Full streaming process if OPTIONS_MANAGER.stream_update: for each_img_name in OPTIONS_MANAGER.full_img_name_list: print(f'full streaming process! current image name is:{each_img_name}') each_img = each_img_name[:-4] chunks, block_sets = split_image_file(each_img, OPTIONS_MANAGER.full_image_new_data[each_img]) OPTIONS_MANAGER.full_chunk[each_img] = chunks OPTIONS_MANAGER.full_block_sets[each_img] = block_sets OPTIONS_MANAGER.full_image_content_len_list, OPTIONS_MANAGER.full_image_file_obj_list = \ full_image_content_len_list, full_image_file_obj_list # Generate the update package. if build_update_package( no_zip, update_package, prelude_script, verse_script, refrain_script, ending_script) is False: clear_resource(err_clear=True) return # Clear resources. clear_resource() if __name__ == '__main__': main()