| #!/usr/bin/env python |
| # |
| # Copyright (C) 2014 The Android Open Source Project |
| # |
| # 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. |
| |
| """ |
| Given a target-files zipfile that does not contain images (ie, does |
| not have an IMAGES/ top-level subdirectory), produce the images and |
| add them to the zipfile. |
| |
| Usage: add_img_to_target_files [flag] target_files |
| |
| -a (--add_missing) |
| Build and add missing images to "IMAGES/". If this option is |
| not specified, this script will simply exit when "IMAGES/" |
| directory exists in the target file. |
| |
| -r (--rebuild_recovery) |
| Rebuild the recovery patch and write it to the system image. Only |
| meaningful when system image needs to be rebuilt and there're separate |
| boot / recovery images. |
| |
| --replace_verity_private_key |
| Replace the private key used for verity signing. (same as the option |
| in sign_target_files_apks) |
| |
| --replace_verity_public_key |
| Replace the certificate (public key) used for verity verification. (same |
| as the option in sign_target_files_apks) |
| |
| --is_signing |
| Skip building & adding the images for "userdata" and "cache" if we |
| are signing the target files. |
| """ |
| |
| from __future__ import print_function |
| |
| import datetime |
| import logging |
| import os |
| import shlex |
| import shutil |
| import stat |
| import sys |
| import uuid |
| import tempfile |
| import zipfile |
| |
| import build_image |
| import build_super_image |
| import common |
| import verity_utils |
| import ota_metadata_pb2 |
| |
| from apex_utils import GetApexInfoFromTargetFiles |
| from common import AddCareMapForAbOta |
| |
| if sys.hexversion < 0x02070000: |
| print("Python 2.7 or newer is required.", file=sys.stderr) |
| sys.exit(1) |
| |
| logger = logging.getLogger(__name__) |
| |
| OPTIONS = common.OPTIONS |
| OPTIONS.add_missing = False |
| OPTIONS.rebuild_recovery = False |
| OPTIONS.replace_updated_files_list = [] |
| OPTIONS.replace_verity_public_key = False |
| OPTIONS.replace_verity_private_key = False |
| OPTIONS.is_signing = False |
| |
| # Use a fixed timestamp (01/01/2009 00:00:00 UTC) for files when packaging |
| # images. (b/24377993, b/80600931) |
| FIXED_FILE_TIMESTAMP = int(( |
| datetime.datetime(2009, 1, 1, 0, 0, 0, 0, None) - |
| datetime.datetime.utcfromtimestamp(0)).total_seconds()) |
| |
| |
| class OutputFile(object): |
| """A helper class to write a generated file to the given dir or zip. |
| |
| When generating images, we want the outputs to go into the given zip file, or |
| the given dir. |
| |
| Attributes: |
| name: The name of the output file, regardless of the final destination. |
| """ |
| |
| def __init__(self, output_zip, input_dir, *args): |
| # We write the intermediate output file under the given input_dir, even if |
| # the final destination is a zip archive. |
| self.name = os.path.join(input_dir, *args) |
| self._output_zip = output_zip |
| if self._output_zip: |
| self._zip_name = os.path.join(*args) |
| |
| def Write(self, compress_type=None): |
| if self._output_zip: |
| common.ZipWrite(self._output_zip, self.name, |
| self._zip_name, compress_type=compress_type) |
| |
| |
| def AddSystem(output_zip, recovery_img=None, boot_img=None): |
| """Turn the contents of SYSTEM into a system image and store it in |
| output_zip. Returns the name of the system image file.""" |
| |
| img = OutputFile(output_zip, OPTIONS.input_tmp, "IMAGES", "system.img") |
| if os.path.exists(img.name): |
| logger.info("system.img already exists; no need to rebuild...") |
| return img.name |
| |
| def output_sink(fn, data): |
| output_file = os.path.join(OPTIONS.input_tmp, "SYSTEM", fn) |
| with open(output_file, "wb") as ofile: |
| ofile.write(data) |
| |
| if output_zip: |
| arc_name = "SYSTEM/" + fn |
| if arc_name in output_zip.namelist(): |
| OPTIONS.replace_updated_files_list.append(arc_name) |
| else: |
| common.ZipWrite(output_zip, output_file, arc_name) |
| |
| board_uses_vendorimage = OPTIONS.info_dict.get( |
| "board_uses_vendorimage") == "true" |
| |
| if (OPTIONS.rebuild_recovery and not board_uses_vendorimage and |
| recovery_img is not None and boot_img is not None): |
| logger.info("Building new recovery patch on system at system/vendor") |
| common.MakeRecoveryPatch(OPTIONS.input_tmp, output_sink, recovery_img, |
| boot_img, info_dict=OPTIONS.info_dict) |
| |
| block_list = OutputFile(output_zip, OPTIONS.input_tmp, |
| "IMAGES", "system.map") |
| CreateImage(OPTIONS.input_tmp, OPTIONS.info_dict, "system", img, |
| block_list=block_list) |
| return img.name |
| |
| |
| def AddSystemOther(output_zip): |
| """Turn the contents of SYSTEM_OTHER into a system_other image |
| and store it in output_zip.""" |
| |
| img = OutputFile(output_zip, OPTIONS.input_tmp, "IMAGES", "system_other.img") |
| if os.path.exists(img.name): |
| logger.info("system_other.img already exists; no need to rebuild...") |
| return |
| |
| CreateImage(OPTIONS.input_tmp, OPTIONS.info_dict, "system_other", img) |
| |
| |
| def AddVendor(output_zip, recovery_img=None, boot_img=None): |
| """Turn the contents of VENDOR into a vendor image and store in it |
| output_zip.""" |
| |
| img = OutputFile(output_zip, OPTIONS.input_tmp, "IMAGES", "vendor.img") |
| if os.path.exists(img.name): |
| logger.info("vendor.img already exists; no need to rebuild...") |
| return img.name |
| |
| def output_sink(fn, data): |
| ofile = open(os.path.join(OPTIONS.input_tmp, "VENDOR", fn), "w") |
| ofile.write(data) |
| ofile.close() |
| |
| if output_zip: |
| arc_name = "VENDOR/" + fn |
| if arc_name in output_zip.namelist(): |
| OPTIONS.replace_updated_files_list.append(arc_name) |
| else: |
| common.ZipWrite(output_zip, ofile.name, arc_name) |
| |
| board_uses_vendorimage = OPTIONS.info_dict.get( |
| "board_uses_vendorimage") == "true" |
| |
| if (OPTIONS.rebuild_recovery and board_uses_vendorimage and |
| recovery_img is not None and boot_img is not None): |
| logger.info("Building new recovery patch on vendor") |
| common.MakeRecoveryPatch(OPTIONS.input_tmp, output_sink, recovery_img, |
| boot_img, info_dict=OPTIONS.info_dict) |
| |
| block_list = OutputFile(output_zip, OPTIONS.input_tmp, |
| "IMAGES", "vendor.map") |
| CreateImage(OPTIONS.input_tmp, OPTIONS.info_dict, "vendor", img, |
| block_list=block_list) |
| return img.name |
| |
| |
| def AddProduct(output_zip): |
| """Turn the contents of PRODUCT into a product image and store it in |
| output_zip.""" |
| |
| img = OutputFile(output_zip, OPTIONS.input_tmp, "IMAGES", "product.img") |
| if os.path.exists(img.name): |
| logger.info("product.img already exists; no need to rebuild...") |
| return img.name |
| |
| block_list = OutputFile( |
| output_zip, OPTIONS.input_tmp, "IMAGES", "product.map") |
| CreateImage( |
| OPTIONS.input_tmp, OPTIONS.info_dict, "product", img, |
| block_list=block_list) |
| return img.name |
| |
| |
| def AddSystemExt(output_zip): |
| """Turn the contents of SYSTEM_EXT into a system_ext image and store it in |
| output_zip.""" |
| |
| img = OutputFile(output_zip, OPTIONS.input_tmp, "IMAGES", |
| "system_ext.img") |
| if os.path.exists(img.name): |
| logger.info("system_ext.img already exists; no need to rebuild...") |
| return img.name |
| |
| block_list = OutputFile( |
| output_zip, OPTIONS.input_tmp, "IMAGES", "system_ext.map") |
| CreateImage( |
| OPTIONS.input_tmp, OPTIONS.info_dict, "system_ext", img, |
| block_list=block_list) |
| return img.name |
| |
| |
| def AddOdm(output_zip): |
| """Turn the contents of ODM into an odm image and store it in output_zip.""" |
| |
| img = OutputFile(output_zip, OPTIONS.input_tmp, "IMAGES", "odm.img") |
| if os.path.exists(img.name): |
| logger.info("odm.img already exists; no need to rebuild...") |
| return img.name |
| |
| block_list = OutputFile( |
| output_zip, OPTIONS.input_tmp, "IMAGES", "odm.map") |
| CreateImage( |
| OPTIONS.input_tmp, OPTIONS.info_dict, "odm", img, |
| block_list=block_list) |
| return img.name |
| |
| |
| def AddVendorDlkm(output_zip): |
| """Turn the contents of VENDOR_DLKM into an vendor_dlkm image and store it in output_zip.""" |
| |
| img = OutputFile(output_zip, OPTIONS.input_tmp, "IMAGES", "vendor_dlkm.img") |
| if os.path.exists(img.name): |
| logger.info("vendor_dlkm.img already exists; no need to rebuild...") |
| return img.name |
| |
| block_list = OutputFile( |
| output_zip, OPTIONS.input_tmp, "IMAGES", "vendor_dlkm.map") |
| CreateImage( |
| OPTIONS.input_tmp, OPTIONS.info_dict, "vendor_dlkm", img, |
| block_list=block_list) |
| return img.name |
| |
| |
| def AddOdmDlkm(output_zip): |
| """Turn the contents of OdmDlkm into an odm_dlkm image and store it in output_zip.""" |
| |
| img = OutputFile(output_zip, OPTIONS.input_tmp, "IMAGES", "odm_dlkm.img") |
| if os.path.exists(img.name): |
| logger.info("odm_dlkm.img already exists; no need to rebuild...") |
| return img.name |
| |
| block_list = OutputFile( |
| output_zip, OPTIONS.input_tmp, "IMAGES", "odm_dlkm.map") |
| CreateImage( |
| OPTIONS.input_tmp, OPTIONS.info_dict, "odm_dlkm", img, |
| block_list=block_list) |
| return img.name |
| |
| def AddSystemDlkm(output_zip): |
| """Turn the contents of SystemDlkm into an system_dlkm image and store it in output_zip.""" |
| |
| img = OutputFile(output_zip, OPTIONS.input_tmp, "IMAGES", "system_dlkm.img") |
| if os.path.exists(img.name): |
| logger.info("system_dlkm.img already exists; no need to rebuild...") |
| return img.name |
| |
| block_list = OutputFile( |
| output_zip, OPTIONS.input_tmp, "IMAGES", "system_dlkm.map") |
| CreateImage( |
| OPTIONS.input_tmp, OPTIONS.info_dict, "system_dlkm", img, |
| block_list=block_list) |
| return img.name |
| |
| |
| def AddDtbo(output_zip): |
| """Adds the DTBO image. |
| |
| Uses the image under IMAGES/ if it already exists. Otherwise looks for the |
| image under PREBUILT_IMAGES/, signs it as needed, and returns the image name. |
| """ |
| img = OutputFile(output_zip, OPTIONS.input_tmp, "IMAGES", "dtbo.img") |
| if os.path.exists(img.name): |
| logger.info("dtbo.img already exists; no need to rebuild...") |
| return img.name |
| |
| dtbo_prebuilt_path = os.path.join( |
| OPTIONS.input_tmp, "PREBUILT_IMAGES", "dtbo.img") |
| assert os.path.exists(dtbo_prebuilt_path) |
| shutil.copy(dtbo_prebuilt_path, img.name) |
| |
| # AVB-sign the image as needed. |
| if OPTIONS.info_dict.get("avb_enable") == "true": |
| # Signing requires +w |
| os.chmod(img.name, os.stat(img.name).st_mode | stat.S_IWUSR) |
| |
| avbtool = OPTIONS.info_dict["avb_avbtool"] |
| part_size = OPTIONS.info_dict["dtbo_size"] |
| # The AVB hash footer will be replaced if already present. |
| cmd = [avbtool, "add_hash_footer", "--image", img.name, |
| "--partition_size", str(part_size), "--partition_name", "dtbo"] |
| common.AppendAVBSigningArgs(cmd, "dtbo") |
| args = OPTIONS.info_dict.get("avb_dtbo_add_hash_footer_args") |
| if args and args.strip(): |
| cmd.extend(shlex.split(args)) |
| common.RunAndCheckOutput(cmd) |
| |
| img.Write() |
| return img.name |
| |
| |
| def AddPvmfw(output_zip): |
| """Adds the pvmfw image. |
| |
| Uses the image under IMAGES/ if it already exists. Otherwise looks for the |
| image under PREBUILT_IMAGES/, signs it as needed, and returns the image name. |
| """ |
| img = OutputFile(output_zip, OPTIONS.input_tmp, "IMAGES", "pvmfw.img") |
| if os.path.exists(img.name): |
| logger.info("pvmfw.img already exists; no need to rebuild...") |
| return img.name |
| |
| pvmfw_prebuilt_path = os.path.join( |
| OPTIONS.input_tmp, "PREBUILT_IMAGES", "pvmfw.img") |
| assert os.path.exists(pvmfw_prebuilt_path) |
| shutil.copy(pvmfw_prebuilt_path, img.name) |
| |
| # AVB-sign the image as needed. |
| if OPTIONS.info_dict.get("avb_enable") == "true": |
| # Signing requires +w |
| os.chmod(img.name, os.stat(img.name).st_mode | stat.S_IWUSR) |
| |
| avbtool = OPTIONS.info_dict["avb_avbtool"] |
| part_size = OPTIONS.info_dict["pvmfw_size"] |
| # The AVB hash footer will be replaced if already present. |
| cmd = [avbtool, "add_hash_footer", "--image", img.name, |
| "--partition_size", str(part_size), "--partition_name", "pvmfw"] |
| common.AppendAVBSigningArgs(cmd, "pvmfw") |
| args = OPTIONS.info_dict.get("avb_pvmfw_add_hash_footer_args") |
| if args and args.strip(): |
| cmd.extend(shlex.split(args)) |
| common.RunAndCheckOutput(cmd) |
| |
| img.Write() |
| return img.name |
| |
| |
| def AddCustomImages(output_zip, partition_name): |
| """Adds and signs custom images in IMAGES/. |
| |
| Args: |
| output_zip: The output zip file (needs to be already open), or None to |
| write images to OPTIONS.input_tmp/. |
| |
| Uses the image under IMAGES/ if it already exists. Otherwise looks for the |
| image under PREBUILT_IMAGES/, signs it as needed, and returns the image name. |
| |
| Raises: |
| AssertionError: If image can't be found. |
| """ |
| |
| key_path = OPTIONS.info_dict.get("avb_{}_key_path".format(partition_name)) |
| algorithm = OPTIONS.info_dict.get("avb_{}_algorithm".format(partition_name)) |
| extra_args = OPTIONS.info_dict.get( |
| "avb_{}_add_hashtree_footer_args".format(partition_name)) |
| partition_size = OPTIONS.info_dict.get( |
| "avb_{}_partition_size".format(partition_name)) |
| |
| builder = verity_utils.CreateCustomImageBuilder( |
| OPTIONS.info_dict, partition_name, partition_size, |
| key_path, algorithm, extra_args) |
| |
| for img_name in OPTIONS.info_dict.get( |
| "avb_{}_image_list".format(partition_name)).split(): |
| custom_image = OutputFile( |
| output_zip, OPTIONS.input_tmp, "IMAGES", img_name) |
| if os.path.exists(custom_image.name): |
| continue |
| |
| custom_image_prebuilt_path = os.path.join( |
| OPTIONS.input_tmp, "PREBUILT_IMAGES", img_name) |
| assert os.path.exists(custom_image_prebuilt_path), \ |
| "Failed to find %s at %s" % (img_name, custom_image_prebuilt_path) |
| |
| shutil.copy(custom_image_prebuilt_path, custom_image.name) |
| |
| if builder is not None: |
| builder.Build(custom_image.name) |
| |
| custom_image.Write() |
| |
| default = os.path.join(OPTIONS.input_tmp, "IMAGES", partition_name + ".img") |
| assert os.path.exists(default), \ |
| "There should be one %s.img" % (partition_name) |
| return default |
| |
| |
| def CreateImage(input_dir, info_dict, what, output_file, block_list=None): |
| logger.info("creating %s.img...", what) |
| |
| image_props = build_image.ImagePropFromGlobalDict(info_dict, what) |
| image_props["timestamp"] = FIXED_FILE_TIMESTAMP |
| |
| if what == "system": |
| fs_config_prefix = "" |
| else: |
| fs_config_prefix = what + "_" |
| |
| fs_config = os.path.join( |
| input_dir, "META/" + fs_config_prefix + "filesystem_config.txt") |
| if not os.path.exists(fs_config): |
| fs_config = None |
| |
| # Override values loaded from info_dict. |
| if fs_config: |
| image_props["fs_config"] = fs_config |
| if block_list: |
| image_props["block_list"] = block_list.name |
| |
| # Use repeatable ext4 FS UUID and hash_seed UUID (based on partition name and |
| # build fingerprint). Also use the legacy build id, because the vbmeta digest |
| # isn't available at this point. |
| build_info = common.BuildInfo(info_dict, use_legacy_id=True) |
| uuid_seed = what + "-" + build_info.GetPartitionFingerprint(what) |
| image_props["uuid"] = str(uuid.uuid5(uuid.NAMESPACE_URL, uuid_seed)) |
| hash_seed = "hash_seed-" + uuid_seed |
| image_props["hash_seed"] = str(uuid.uuid5(uuid.NAMESPACE_URL, hash_seed)) |
| |
| build_image.BuildImage( |
| os.path.join(input_dir, what.upper()), image_props, output_file.name) |
| |
| output_file.Write() |
| if block_list: |
| block_list.Write() |
| |
| # Set the '_image_size' for given image size. |
| is_verity_partition = "verity_block_device" in image_props |
| verity_supported = (image_props.get("verity") == "true" or |
| image_props.get("avb_enable") == "true") |
| is_avb_enable = image_props.get("avb_hashtree_enable") == "true" |
| if verity_supported and (is_verity_partition or is_avb_enable): |
| image_size = image_props.get("image_size") |
| if image_size: |
| image_size_key = what + "_image_size" |
| info_dict[image_size_key] = int(image_size) |
| |
| use_dynamic_size = ( |
| info_dict.get("use_dynamic_partition_size") == "true" and |
| what in shlex.split(info_dict.get("dynamic_partition_list", "").strip())) |
| if use_dynamic_size: |
| info_dict.update(build_image.GlobalDictFromImageProp(image_props, what)) |
| |
| |
| def AddUserdata(output_zip): |
| """Create a userdata image and store it in output_zip. |
| |
| In most case we just create and store an empty userdata.img; |
| But the invoker can also request to create userdata.img with real |
| data from the target files, by setting "userdata_img_with_data=true" |
| in OPTIONS.info_dict. |
| """ |
| |
| img = OutputFile(output_zip, OPTIONS.input_tmp, "IMAGES", "userdata.img") |
| if os.path.exists(img.name): |
| logger.info("userdata.img already exists; no need to rebuild...") |
| return |
| |
| # Skip userdata.img if no size. |
| image_props = build_image.ImagePropFromGlobalDict(OPTIONS.info_dict, "data") |
| if not image_props.get("partition_size"): |
| return |
| |
| logger.info("creating userdata.img...") |
| |
| image_props["timestamp"] = FIXED_FILE_TIMESTAMP |
| |
| if OPTIONS.info_dict.get("userdata_img_with_data") == "true": |
| user_dir = os.path.join(OPTIONS.input_tmp, "DATA") |
| else: |
| user_dir = common.MakeTempDir() |
| |
| build_image.BuildImage(user_dir, image_props, img.name) |
| |
| common.CheckSize(img.name, "userdata.img", OPTIONS.info_dict) |
| # Always use compression for useradata image. |
| # As it's likely huge and consist of lots of 0s. |
| img.Write(zipfile.ZIP_DEFLATED) |
| |
| |
| def AddVBMeta(output_zip, partitions, name, needed_partitions): |
| """Creates a VBMeta image and stores it in output_zip. |
| |
| It generates the requested VBMeta image. The requested image could be for |
| top-level or chained VBMeta image, which is determined based on the name. |
| |
| Args: |
| output_zip: The output zip file, which needs to be already open. |
| partitions: A dict that's keyed by partition names with image paths as |
| values. Only valid partition names are accepted, as partitions listed |
| in common.AVB_PARTITIONS and custom partitions listed in |
| OPTIONS.info_dict.get("avb_custom_images_partition_list") |
| name: Name of the VBMeta partition, e.g. 'vbmeta', 'vbmeta_system'. |
| needed_partitions: Partitions whose descriptors should be included into the |
| generated VBMeta image. |
| |
| Returns: |
| Path to the created image. |
| |
| Raises: |
| AssertionError: On invalid input args. |
| """ |
| assert needed_partitions, "Needed partitions must be specified" |
| |
| img = OutputFile( |
| output_zip, OPTIONS.input_tmp, "IMAGES", "{}.img".format(name)) |
| if os.path.exists(img.name): |
| logger.info("%s.img already exists; not rebuilding...", name) |
| return img.name |
| |
| common.BuildVBMeta(img.name, partitions, name, needed_partitions) |
| img.Write() |
| return img.name |
| |
| |
| def AddPartitionTable(output_zip): |
| """Create a partition table image and store it in output_zip.""" |
| |
| img = OutputFile( |
| output_zip, OPTIONS.input_tmp, "IMAGES", "partition-table.img") |
| bpt = OutputFile( |
| output_zip, OPTIONS.input_tmp, "META", "partition-table.bpt") |
| |
| # use BPTTOOL from environ, or "bpttool" if empty or not set. |
| bpttool = os.getenv("BPTTOOL") or "bpttool" |
| cmd = [bpttool, "make_table", "--output_json", bpt.name, |
| "--output_gpt", img.name] |
| input_files_str = OPTIONS.info_dict["board_bpt_input_files"] |
| input_files = input_files_str.split(" ") |
| for i in input_files: |
| cmd.extend(["--input", i]) |
| disk_size = OPTIONS.info_dict.get("board_bpt_disk_size") |
| if disk_size: |
| cmd.extend(["--disk_size", disk_size]) |
| args = OPTIONS.info_dict.get("board_bpt_make_table_args") |
| if args: |
| cmd.extend(shlex.split(args)) |
| common.RunAndCheckOutput(cmd) |
| |
| img.Write() |
| bpt.Write() |
| |
| |
| def AddCache(output_zip): |
| """Create an empty cache image and store it in output_zip.""" |
| |
| img = OutputFile(output_zip, OPTIONS.input_tmp, "IMAGES", "cache.img") |
| if os.path.exists(img.name): |
| logger.info("cache.img already exists; no need to rebuild...") |
| return |
| |
| image_props = build_image.ImagePropFromGlobalDict(OPTIONS.info_dict, "cache") |
| # The build system has to explicitly request for cache.img. |
| if "fs_type" not in image_props: |
| return |
| |
| logger.info("creating cache.img...") |
| |
| image_props["timestamp"] = FIXED_FILE_TIMESTAMP |
| |
| user_dir = common.MakeTempDir() |
| build_image.BuildImage(user_dir, image_props, img.name) |
| |
| common.CheckSize(img.name, "cache.img", OPTIONS.info_dict) |
| img.Write() |
| |
| |
| def CheckAbOtaImages(output_zip, ab_partitions): |
| """Checks that all the listed A/B partitions have their images available. |
| |
| The images need to be available under IMAGES/ or RADIO/, with the former takes |
| a priority. |
| |
| Args: |
| output_zip: The output zip file (needs to be already open), or None to |
| find images in OPTIONS.input_tmp/. |
| ab_partitions: The list of A/B partitions. |
| |
| Raises: |
| AssertionError: If it can't find an image. |
| """ |
| for partition in ab_partitions: |
| img_name = partition + ".img" |
| |
| # Assert that the image is present under IMAGES/ now. |
| if output_zip: |
| # Zip spec says: All slashes MUST be forward slashes. |
| images_path = "IMAGES/" + img_name |
| radio_path = "RADIO/" + img_name |
| available = (images_path in output_zip.namelist() or |
| radio_path in output_zip.namelist()) |
| else: |
| images_path = os.path.join(OPTIONS.input_tmp, "IMAGES", img_name) |
| radio_path = os.path.join(OPTIONS.input_tmp, "RADIO", img_name) |
| available = os.path.exists(images_path) or os.path.exists(radio_path) |
| |
| assert available, "Failed to find " + img_name |
| |
| |
| def AddPackRadioImages(output_zip, images): |
| """Copies images listed in META/pack_radioimages.txt from RADIO/ to IMAGES/. |
| |
| Args: |
| output_zip: The output zip file (needs to be already open), or None to |
| write images to OPTIONS.input_tmp/. |
| images: A list of image names. |
| |
| Raises: |
| AssertionError: If a listed image can't be found. |
| """ |
| for image in images: |
| img_name = image.strip() |
| _, ext = os.path.splitext(img_name) |
| if not ext: |
| img_name += ".img" |
| |
| prebuilt_path = os.path.join(OPTIONS.input_tmp, "IMAGES", img_name) |
| if os.path.exists(prebuilt_path): |
| logger.info("%s already exists, no need to overwrite...", img_name) |
| continue |
| |
| img_radio_path = os.path.join(OPTIONS.input_tmp, "RADIO", img_name) |
| assert os.path.exists(img_radio_path), \ |
| "Failed to find %s at %s" % (img_name, img_radio_path) |
| |
| if output_zip: |
| common.ZipWrite(output_zip, img_radio_path, "IMAGES/" + img_name) |
| else: |
| shutil.copy(img_radio_path, prebuilt_path) |
| |
| |
| def AddSuperEmpty(output_zip): |
| """Create a super_empty.img and store it in output_zip.""" |
| |
| img = OutputFile(output_zip, OPTIONS.input_tmp, "IMAGES", "super_empty.img") |
| build_super_image.BuildSuperImage(OPTIONS.info_dict, img.name) |
| img.Write() |
| |
| |
| def AddSuperSplit(output_zip): |
| """Create split super_*.img and store it in output_zip.""" |
| |
| outdir = os.path.join(OPTIONS.input_tmp, "OTA") |
| built = build_super_image.BuildSuperImage(OPTIONS.input_tmp, outdir) |
| |
| if built: |
| for dev in OPTIONS.info_dict['super_block_devices'].strip().split(): |
| img = OutputFile(output_zip, OPTIONS.input_tmp, "OTA", |
| "super_" + dev + ".img") |
| img.Write() |
| |
| |
| def ReplaceUpdatedFiles(zip_filename, files_list): |
| """Updates all the ZIP entries listed in files_list. |
| |
| For now the list includes META/care_map.pb, and the related files under |
| SYSTEM/ after rebuilding recovery. |
| """ |
| common.ZipDelete(zip_filename, files_list) |
| output_zip = zipfile.ZipFile(zip_filename, "a", |
| compression=zipfile.ZIP_DEFLATED, |
| allowZip64=True) |
| for item in files_list: |
| file_path = os.path.join(OPTIONS.input_tmp, item) |
| assert os.path.exists(file_path) |
| common.ZipWrite(output_zip, file_path, arcname=item) |
| common.ZipClose(output_zip) |
| |
| |
| def HasPartition(partition_name): |
| """Determines if the target files archive should build a given partition.""" |
| |
| return ((os.path.isdir( |
| os.path.join(OPTIONS.input_tmp, partition_name.upper())) and |
| OPTIONS.info_dict.get( |
| "building_{}_image".format(partition_name)) == "true") or |
| os.path.exists( |
| os.path.join(OPTIONS.input_tmp, "IMAGES", |
| "{}.img".format(partition_name)))) |
| |
| |
| def AddApexInfo(output_zip): |
| apex_infos = GetApexInfoFromTargetFiles(OPTIONS.input_tmp, 'system', |
| compressed_only=False) |
| apex_metadata_proto = ota_metadata_pb2.ApexMetadata() |
| apex_metadata_proto.apex_info.extend(apex_infos) |
| apex_info_bytes = apex_metadata_proto.SerializeToString() |
| |
| output_file = os.path.join(OPTIONS.input_tmp, "META", "apex_info.pb") |
| with open(output_file, "wb") as ofile: |
| ofile.write(apex_info_bytes) |
| if output_zip: |
| arc_name = "META/apex_info.pb" |
| if arc_name in output_zip.namelist(): |
| OPTIONS.replace_updated_files_list.append(arc_name) |
| else: |
| common.ZipWrite(output_zip, output_file, arc_name) |
| |
| |
| def AddVbmetaDigest(output_zip): |
| """Write the vbmeta digest to the output dir and zipfile.""" |
| |
| # Calculate the vbmeta digest and put the result in to META/ |
| boot_images = OPTIONS.info_dict.get("boot_images") |
| # Disable the digest calculation if the target_file is used as a container |
| # for boot images. A boot container might contain boot-5.4.img, boot-5.10.img |
| # etc., instead of just a boot.img and will fail in vbmeta digest calculation. |
| boot_container = boot_images and ( |
| len(boot_images.split()) >= 2 or boot_images.split()[0] != 'boot.img') |
| if (OPTIONS.info_dict.get("avb_enable") == "true" and not boot_container and |
| OPTIONS.info_dict.get("avb_building_vbmeta_image") == "true"): |
| avbtool = OPTIONS.info_dict["avb_avbtool"] |
| digest = verity_utils.CalculateVbmetaDigest(OPTIONS.input_tmp, avbtool) |
| vbmeta_digest_txt = os.path.join(OPTIONS.input_tmp, "META", |
| "vbmeta_digest.txt") |
| with open(vbmeta_digest_txt, 'w') as f: |
| f.write(digest) |
| # writes to the output zipfile |
| if output_zip: |
| arc_name = "META/vbmeta_digest.txt" |
| if arc_name in output_zip.namelist(): |
| OPTIONS.replace_updated_files_list.append(arc_name) |
| else: |
| common.ZipWriteStr(output_zip, arc_name, digest) |
| |
| |
| def AddImagesToTargetFiles(filename): |
| """Creates and adds images (boot/recovery/system/...) to a target_files.zip. |
| |
| It works with either a zip file (zip mode), or a directory that contains the |
| files to be packed into a target_files.zip (dir mode). The latter is used when |
| being called from build/make/core/Makefile. |
| |
| The images will be created under IMAGES/ in the input target_files.zip. |
| |
| Args: |
| filename: the target_files.zip, or the zip root directory. |
| """ |
| if os.path.isdir(filename): |
| OPTIONS.input_tmp = os.path.abspath(filename) |
| else: |
| OPTIONS.input_tmp = common.UnzipTemp(filename) |
| |
| if not OPTIONS.add_missing: |
| if os.path.isdir(os.path.join(OPTIONS.input_tmp, "IMAGES")): |
| logger.warning("target_files appears to already contain images.") |
| sys.exit(1) |
| |
| OPTIONS.info_dict = common.LoadInfoDict(OPTIONS.input_tmp, repacking=True) |
| |
| has_recovery = OPTIONS.info_dict.get("no_recovery") != "true" |
| has_boot = OPTIONS.info_dict.get("no_boot") != "true" |
| has_init_boot = OPTIONS.info_dict.get("init_boot") == "true" |
| has_vendor_boot = OPTIONS.info_dict.get("vendor_boot") == "true" |
| |
| # {vendor,odm,product,system_ext,vendor_dlkm,odm_dlkm, system_dlkm, system, system_other}.img |
| # can be built from source, or dropped into target_files.zip as a prebuilt blob. |
| has_vendor = HasPartition("vendor") |
| has_odm = HasPartition("odm") |
| has_vendor_dlkm = HasPartition("vendor_dlkm") |
| has_odm_dlkm = HasPartition("odm_dlkm") |
| has_system_dlkm = HasPartition("system_dlkm") |
| has_product = HasPartition("product") |
| has_system_ext = HasPartition("system_ext") |
| has_system = HasPartition("system") |
| has_system_other = HasPartition("system_other") |
| has_userdata = OPTIONS.info_dict.get("building_userdata_image") == "true" |
| has_cache = OPTIONS.info_dict.get("building_cache_image") == "true" |
| |
| # Set up the output destination. It writes to the given directory for dir |
| # mode; otherwise appends to the given ZIP. |
| if os.path.isdir(filename): |
| output_zip = None |
| else: |
| output_zip = zipfile.ZipFile(filename, "a", |
| compression=zipfile.ZIP_DEFLATED, |
| allowZip64=True) |
| |
| # Always make input_tmp/IMAGES available, since we may stage boot / recovery |
| # images there even under zip mode. The directory will be cleaned up as part |
| # of OPTIONS.input_tmp. |
| images_dir = os.path.join(OPTIONS.input_tmp, "IMAGES") |
| if not os.path.isdir(images_dir): |
| os.makedirs(images_dir) |
| |
| # A map between partition names and their paths, which could be used when |
| # generating AVB vbmeta image. |
| partitions = {} |
| |
| def banner(s): |
| logger.info("\n\n++++ %s ++++\n\n", s) |
| |
| boot_image = None |
| if has_boot: |
| banner("boot") |
| boot_images = OPTIONS.info_dict.get("boot_images") |
| if boot_images is None: |
| boot_images = "boot.img" |
| for index, b in enumerate(boot_images.split()): |
| # common.GetBootableImage() returns the image directly if present. |
| boot_image = common.GetBootableImage( |
| "IMAGES/" + b, b, OPTIONS.input_tmp, "BOOT") |
| # boot.img may be unavailable in some targets (e.g. aosp_arm64). |
| if boot_image: |
| boot_image_path = os.path.join(OPTIONS.input_tmp, "IMAGES", b) |
| # Although multiple boot images can be generated, include the image |
| # descriptor of only the first boot image in vbmeta |
| if index == 0: |
| partitions['boot'] = boot_image_path |
| if not os.path.exists(boot_image_path): |
| boot_image.WriteToDir(OPTIONS.input_tmp) |
| if output_zip: |
| boot_image.AddToZip(output_zip) |
| |
| if has_init_boot: |
| banner("init_boot") |
| init_boot_image = common.GetBootableImage( |
| "IMAGES/init_boot.img", "init_boot.img", OPTIONS.input_tmp, "INIT_BOOT") |
| if init_boot_image: |
| partitions['init_boot'] = os.path.join( |
| OPTIONS.input_tmp, "IMAGES", "init_boot.img") |
| if not os.path.exists(partitions['init_boot']): |
| init_boot_image.WriteToDir(OPTIONS.input_tmp) |
| if output_zip: |
| init_boot_image.AddToZip(output_zip) |
| |
| if has_vendor_boot: |
| banner("vendor_boot") |
| vendor_boot_image = common.GetVendorBootImage( |
| "IMAGES/vendor_boot.img", "vendor_boot.img", OPTIONS.input_tmp, |
| "VENDOR_BOOT") |
| if vendor_boot_image: |
| partitions['vendor_boot'] = os.path.join(OPTIONS.input_tmp, "IMAGES", |
| "vendor_boot.img") |
| if not os.path.exists(partitions['vendor_boot']): |
| vendor_boot_image.WriteToDir(OPTIONS.input_tmp) |
| if output_zip: |
| vendor_boot_image.AddToZip(output_zip) |
| |
| recovery_image = None |
| if has_recovery: |
| banner("recovery") |
| recovery_image = common.GetBootableImage( |
| "IMAGES/recovery.img", "recovery.img", OPTIONS.input_tmp, "RECOVERY") |
| assert recovery_image, "Failed to create recovery.img." |
| partitions['recovery'] = os.path.join( |
| OPTIONS.input_tmp, "IMAGES", "recovery.img") |
| if not os.path.exists(partitions['recovery']): |
| recovery_image.WriteToDir(OPTIONS.input_tmp) |
| if output_zip: |
| recovery_image.AddToZip(output_zip) |
| |
| banner("recovery (two-step image)") |
| # The special recovery.img for two-step package use. |
| recovery_two_step_image = common.GetBootableImage( |
| "OTA/recovery-two-step.img", "recovery-two-step.img", |
| OPTIONS.input_tmp, "RECOVERY", two_step_image=True) |
| assert recovery_two_step_image, "Failed to create recovery-two-step.img." |
| recovery_two_step_image_path = os.path.join( |
| OPTIONS.input_tmp, "OTA", "recovery-two-step.img") |
| if not os.path.exists(recovery_two_step_image_path): |
| recovery_two_step_image.WriteToDir(OPTIONS.input_tmp) |
| if output_zip: |
| recovery_two_step_image.AddToZip(output_zip) |
| |
| def add_partition(partition, has_partition, add_func, add_args): |
| if has_partition: |
| banner(partition) |
| partitions[partition] = add_func(output_zip, *add_args) |
| |
| add_partition_calls = ( |
| ("system", has_system, AddSystem, [recovery_image, boot_image]), |
| ("vendor", has_vendor, AddVendor, [recovery_image, boot_image]), |
| ("product", has_product, AddProduct, []), |
| ("system_ext", has_system_ext, AddSystemExt, []), |
| ("odm", has_odm, AddOdm, []), |
| ("vendor_dlkm", has_vendor_dlkm, AddVendorDlkm, []), |
| ("odm_dlkm", has_odm_dlkm, AddOdmDlkm, []), |
| ("system_dlkm", has_system_dlkm, AddSystemDlkm, []), |
| ("system_other", has_system_other, AddSystemOther, []), |
| ) |
| for call in add_partition_calls: |
| add_partition(*call) |
| |
| AddApexInfo(output_zip) |
| |
| if not OPTIONS.is_signing: |
| banner("userdata") |
| AddUserdata(output_zip) |
| banner("cache") |
| AddCache(output_zip) |
| |
| if OPTIONS.info_dict.get("board_bpt_enable") == "true": |
| banner("partition-table") |
| AddPartitionTable(output_zip) |
| |
| add_partition("dtbo", |
| OPTIONS.info_dict.get("has_dtbo") == "true", AddDtbo, []) |
| add_partition("pvmfw", |
| OPTIONS.info_dict.get("has_pvmfw") == "true", AddPvmfw, []) |
| |
| # Custom images. |
| custom_partitions = OPTIONS.info_dict.get( |
| "avb_custom_images_partition_list", "").strip().split() |
| for partition_name in custom_partitions: |
| partition_name = partition_name.strip() |
| banner("custom images for " + partition_name) |
| partitions[partition_name] = AddCustomImages(output_zip, partition_name) |
| |
| if OPTIONS.info_dict.get("avb_enable") == "true": |
| # vbmeta_partitions includes the partitions that should be included into |
| # top-level vbmeta.img, which are the ones that are not included in any |
| # chained VBMeta image plus the chained VBMeta images themselves. |
| # Currently custom_partitions are all chained to VBMeta image. |
| vbmeta_partitions = common.AVB_PARTITIONS[:] + tuple(custom_partitions) |
| |
| vbmeta_system = OPTIONS.info_dict.get("avb_vbmeta_system", "").strip() |
| if vbmeta_system: |
| banner("vbmeta_system") |
| partitions["vbmeta_system"] = AddVBMeta( |
| output_zip, partitions, "vbmeta_system", vbmeta_system.split()) |
| vbmeta_partitions = [ |
| item for item in vbmeta_partitions |
| if item not in vbmeta_system.split()] |
| vbmeta_partitions.append("vbmeta_system") |
| |
| vbmeta_vendor = OPTIONS.info_dict.get("avb_vbmeta_vendor", "").strip() |
| if vbmeta_vendor: |
| banner("vbmeta_vendor") |
| partitions["vbmeta_vendor"] = AddVBMeta( |
| output_zip, partitions, "vbmeta_vendor", vbmeta_vendor.split()) |
| vbmeta_partitions = [ |
| item for item in vbmeta_partitions |
| if item not in vbmeta_vendor.split()] |
| vbmeta_partitions.append("vbmeta_vendor") |
| |
| if OPTIONS.info_dict.get("avb_building_vbmeta_image") == "true": |
| banner("vbmeta") |
| AddVBMeta(output_zip, partitions, "vbmeta", vbmeta_partitions) |
| |
| if OPTIONS.info_dict.get("use_dynamic_partitions") == "true": |
| if OPTIONS.info_dict.get("build_super_empty_partition") == "true": |
| banner("super_empty") |
| AddSuperEmpty(output_zip) |
| |
| if OPTIONS.info_dict.get("build_super_partition") == "true": |
| if OPTIONS.info_dict.get( |
| "build_retrofit_dynamic_partitions_ota_package") == "true": |
| banner("super split images") |
| AddSuperSplit(output_zip) |
| |
| banner("radio") |
| ab_partitions_txt = os.path.join(OPTIONS.input_tmp, "META", |
| "ab_partitions.txt") |
| if os.path.exists(ab_partitions_txt): |
| with open(ab_partitions_txt) as f: |
| ab_partitions = f.read().splitlines() |
| |
| # For devices using A/B update, make sure we have all the needed images |
| # ready under IMAGES/ or RADIO/. |
| CheckAbOtaImages(output_zip, ab_partitions) |
| |
| # Generate care_map.pb for ab_partitions, then write this file to |
| # target_files package. |
| output_care_map = os.path.join(OPTIONS.input_tmp, "META", "care_map.pb") |
| AddCareMapForAbOta(output_zip if output_zip else output_care_map, |
| ab_partitions, partitions) |
| |
| # Radio images that need to be packed into IMAGES/, and product-img.zip. |
| pack_radioimages_txt = os.path.join( |
| OPTIONS.input_tmp, "META", "pack_radioimages.txt") |
| if os.path.exists(pack_radioimages_txt): |
| with open(pack_radioimages_txt) as f: |
| AddPackRadioImages(output_zip, f.readlines()) |
| |
| AddVbmetaDigest(output_zip) |
| |
| if output_zip: |
| common.ZipClose(output_zip) |
| if OPTIONS.replace_updated_files_list: |
| ReplaceUpdatedFiles(output_zip.filename, |
| OPTIONS.replace_updated_files_list) |
| |
| |
| def OptimizeCompressedEntries(zipfile_path): |
| """Convert files that do not compress well to uncompressed storage |
| |
| EROFS images tend to be compressed already, so compressing them again |
| yields little space savings. Leaving them uncompressed will make |
| downstream tooling's job easier, and save compute time. |
| """ |
| if not zipfile.is_zipfile(zipfile_path): |
| return |
| entries_to_store = [] |
| with tempfile.TemporaryDirectory() as tmpdir: |
| with zipfile.ZipFile(zipfile_path, "r", allowZip64=True) as zfp: |
| for zinfo in zfp.filelist: |
| if not zinfo.filename.startswith("IMAGES/") and not zinfo.filename.startswith("META"): |
| pass |
| # Don't try to store userdata.img uncompressed, it's usually huge. |
| if zinfo.filename.endswith("userdata.img"): |
| pass |
| if zinfo.compress_size > zinfo.file_size * 0.80 and zinfo.compress_type != zipfile.ZIP_STORED: |
| entries_to_store.append(zinfo) |
| zfp.extract(zinfo, tmpdir) |
| # Remove these entries, then re-add them as ZIP_STORED |
| common.RunAndCheckOutput( |
| ["zip", "-d", zipfile_path] + [entry.filename for entry in entries_to_store]) |
| with zipfile.ZipFile(zipfile_path, "a", allowZip64=True) as zfp: |
| for entry in entries_to_store: |
| zfp.write(os.path.join(tmpdir, entry.filename), entry.filename, compress_type=zipfile.ZIP_STORED) |
| |
| |
| def main(argv): |
| def option_handler(o, a): |
| if o in ("-a", "--add_missing"): |
| OPTIONS.add_missing = True |
| elif o in ("-r", "--rebuild_recovery",): |
| OPTIONS.rebuild_recovery = True |
| elif o == "--replace_verity_private_key": |
| OPTIONS.replace_verity_private_key = (True, a) |
| elif o == "--replace_verity_public_key": |
| OPTIONS.replace_verity_public_key = (True, a) |
| elif o == "--is_signing": |
| OPTIONS.is_signing = True |
| else: |
| return False |
| return True |
| |
| args = common.ParseOptions( |
| argv, __doc__, extra_opts="ar", |
| extra_long_opts=["add_missing", "rebuild_recovery", |
| "replace_verity_public_key=", |
| "replace_verity_private_key=", |
| "is_signing"], |
| extra_option_handler=option_handler) |
| |
| if len(args) != 1: |
| common.Usage(__doc__) |
| sys.exit(1) |
| |
| common.InitLogging() |
| |
| AddImagesToTargetFiles(args[0]) |
| OptimizeCompressedEntries(args[0]) |
| logger.info("done.") |
| |
| |
| if __name__ == '__main__': |
| try: |
| common.CloseInheritedPipes() |
| main(sys.argv[1:]) |
| finally: |
| common.Cleanup() |
