platform_build/tools/releasetools/add_img_to_target_files.py
Kelvin Zhang 834f5d458f Store EROFS images uncompressed
Test: th
Change-Id: I53005bee139079afe25da231ca1d42f03fa010a5
2022-02-02 12:52:34 -08:00

1085 lines
39 KiB
Python

#!/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()