Merge "Move non-AB OTA generation code to a separate file"

    srcs: [

        "edify_generator.py",

        "ota_from_target_files.py",

        "non_ab_ota.py",

        "target_files_diff.py",

        "ota_utils.py",

    ],

    libs: [

        "releasetools_check_target_files_vintf",

  raise ValueError('{} is not a valid directory or zip file'.format(inp))

def CheckVintfIfTrebleEnabled(target_files, target_info):

  """Checks compatibility info of the input target files.

  Metadata used for compatibility verification is retrieved from target_zip.

  Compatibility should only be checked for devices that have enabled

  Treble support.

  Args:

    target_files: Path to zip file containing the source files to be included

        for OTA. Can also be the path to extracted directory.

    target_info: The BuildInfo instance that holds the target build info.

  """

  # Will only proceed if the target has enabled the Treble support (as well as

  # having a /vendor partition).

  if not HasTrebleEnabled(target_files, target_info):

    return

  # Skip adding the compatibility package as a workaround for b/114240221. The

  # compatibility will always fail on devices without qualified kernels.

  if OPTIONS.skip_compatibility_check:

    return

  if not CheckVintf(target_files, target_info):

    raise RuntimeError("VINTF compatibility check failed")

def HasTrebleEnabled(target_files, target_info):

  def HasVendorPartition(target_files):

    if os.path.isdir(target_files):

      return os.path.isdir(os.path.join(target_files, "VENDOR"))

    if zipfile.is_zipfile(target_files):

      return HasPartition(zipfile.ZipFile(target_files), "vendor")

    raise ValueError("Unknown target_files argument")

  return (HasVendorPartition(target_files) and

          target_info.GetBuildProp("ro.treble.enabled") == "true")

def HasPartition(target_files_zip, partition):

  try:

    target_files_zip.getinfo(partition.upper() + "/")

    return True

  except KeyError:

    return False

def main(argv):

  args = common.ParseOptions(argv, __doc__)

# Copyright (C) 2020 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.

import copy

import itertools

import os

import zipfile

from common import (ZipDelete, ZipClose, OPTIONS, MakeTempFile,

                    ZipWriteStr, BuildInfo, LoadDictionaryFromFile,

                    SignFile, PARTITIONS_WITH_CARE_MAP, PartitionBuildProps)

METADATA_NAME = 'META-INF/com/android/metadata'

UNZIP_PATTERN = ['IMAGES/*', 'META/*', 'OTA/*', 'RADIO/*']

def FinalizeMetadata(metadata, input_file, output_file, needed_property_files):

  """Finalizes the metadata and signs an A/B OTA package.

  In order to stream an A/B OTA package, we need 'ota-streaming-property-files'

  that contains the offsets and sizes for the ZIP entries. An example

  property-files string is as follows.

    "payload.bin:679:343,payload_properties.txt:378:45,metadata:69:379"

  OTA server can pass down this string, in addition to the package URL, to the

  system update client. System update client can then fetch individual ZIP

  entries (ZIP_STORED) directly at the given offset of the URL.

  Args:

    metadata: The metadata dict for the package.

    input_file: The input ZIP filename that doesn't contain the package METADATA

        entry yet.

    output_file: The final output ZIP filename.

    needed_property_files: The list of PropertyFiles' to be generated.

  """

  def ComputeAllPropertyFiles(input_file, needed_property_files):

    # Write the current metadata entry with placeholders.

    with zipfile.ZipFile(input_file) as input_zip:

      for property_files in needed_property_files:

        metadata[property_files.name] = property_files.Compute(input_zip)

      namelist = input_zip.namelist()

    if METADATA_NAME in namelist:

      ZipDelete(input_file, METADATA_NAME)

    output_zip = zipfile.ZipFile(input_file, 'a')

    WriteMetadata(metadata, output_zip)

    ZipClose(output_zip)

    if OPTIONS.no_signing:

      return input_file

    prelim_signing = MakeTempFile(suffix='.zip')

    SignOutput(input_file, prelim_signing)

    return prelim_signing

  def FinalizeAllPropertyFiles(prelim_signing, needed_property_files):

    with zipfile.ZipFile(prelim_signing) as prelim_signing_zip:

      for property_files in needed_property_files:

        metadata[property_files.name] = property_files.Finalize(

            prelim_signing_zip, len(metadata[property_files.name]))

  # SignOutput(), which in turn calls signapk.jar, will possibly reorder the ZIP

  # entries, as well as padding the entry headers. We do a preliminary signing

  # (with an incomplete metadata entry) to allow that to happen. Then compute

  # the ZIP entry offsets, write back the final metadata and do the final

  # signing.

  prelim_signing = ComputeAllPropertyFiles(input_file, needed_property_files)

  try:

    FinalizeAllPropertyFiles(prelim_signing, needed_property_files)

  except PropertyFiles.InsufficientSpaceException:

    # Even with the preliminary signing, the entry orders may change

    # dramatically, which leads to insufficiently reserved space during the

    # first call to ComputeAllPropertyFiles(). In that case, we redo all the

    # preliminary signing works, based on the already ordered ZIP entries, to

    # address the issue.

    prelim_signing = ComputeAllPropertyFiles(

        prelim_signing, needed_property_files)

    FinalizeAllPropertyFiles(prelim_signing, needed_property_files)

  # Replace the METADATA entry.

  ZipDelete(prelim_signing, METADATA_NAME)

  output_zip = zipfile.ZipFile(prelim_signing, 'a')

  WriteMetadata(metadata, output_zip)

  ZipClose(output_zip)

  # Re-sign the package after updating the metadata entry.

  if OPTIONS.no_signing:

    output_file = prelim_signing

  else:

    SignOutput(prelim_signing, output_file)

  # Reopen the final signed zip to double check the streaming metadata.

  with zipfile.ZipFile(output_file) as output_zip:

    for property_files in needed_property_files:

      property_files.Verify(output_zip, metadata[property_files.name].strip())

  # If requested, dump the metadata to a separate file.

  output_metadata_path = OPTIONS.output_metadata_path

  if output_metadata_path:

    WriteMetadata(metadata, output_metadata_path)

def WriteMetadata(metadata, output):

  """Writes the metadata to the zip archive or a file.

  Args:

    metadata: The metadata dict for the package.

    output: A ZipFile object or a string of the output file path.

  """

  value = "".join(["%s=%s\n" % kv for kv in sorted(metadata.items())])

  if isinstance(output, zipfile.ZipFile):

    ZipWriteStr(output, METADATA_NAME, value,

                compress_type=zipfile.ZIP_STORED)

    return

  with open(output, 'w') as f:

    f.write(value)

def GetPackageMetadata(target_info, source_info=None):

  """Generates and returns the metadata dict.

  It generates a dict() that contains the info to be written into an OTA

  package (META-INF/com/android/metadata). It also handles the detection of

  downgrade / data wipe based on the global options.

  Args:

    target_info: The BuildInfo instance that holds the target build info.

    source_info: The BuildInfo instance that holds the source build info, or

        None if generating full OTA.

  Returns:

    A dict to be written into package metadata entry.

  """

  assert isinstance(target_info, BuildInfo)

  assert source_info is None or isinstance(source_info, BuildInfo)

  separator = '|'

  boot_variable_values = {}

  if OPTIONS.boot_variable_file:

    d = LoadDictionaryFromFile(OPTIONS.boot_variable_file)

    for key, values in d.items():

      boot_variable_values[key] = [val.strip() for val in values.split(',')]

  post_build_devices, post_build_fingerprints = \

      CalculateRuntimeDevicesAndFingerprints(target_info, boot_variable_values)

  metadata = {

      'post-build': separator.join(sorted(post_build_fingerprints)),

      'post-build-incremental': target_info.GetBuildProp(

          'ro.build.version.incremental'),

      'post-sdk-level': target_info.GetBuildProp(

          'ro.build.version.sdk'),

      'post-security-patch-level': target_info.GetBuildProp(

          'ro.build.version.security_patch'),

  }

  if target_info.is_ab and not OPTIONS.force_non_ab:

    metadata['ota-type'] = 'AB'

    metadata['ota-required-cache'] = '0'

  else:

    metadata['ota-type'] = 'BLOCK'

  if OPTIONS.wipe_user_data:

    metadata['ota-wipe'] = 'yes'

  if OPTIONS.retrofit_dynamic_partitions:

    metadata['ota-retrofit-dynamic-partitions'] = 'yes'

  is_incremental = source_info is not None

  if is_incremental:

    pre_build_devices, pre_build_fingerprints = \

        CalculateRuntimeDevicesAndFingerprints(source_info,

                                               boot_variable_values)

    metadata['pre-build'] = separator.join(sorted(pre_build_fingerprints))

    metadata['pre-build-incremental'] = source_info.GetBuildProp(

        'ro.build.version.incremental')

    metadata['pre-device'] = separator.join(sorted(pre_build_devices))

  else:

    metadata['pre-device'] = separator.join(sorted(post_build_devices))

  # Use the actual post-timestamp, even for a downgrade case.

  metadata['post-timestamp'] = target_info.GetBuildProp('ro.build.date.utc')

  # Detect downgrades and set up downgrade flags accordingly.

  if is_incremental:

    HandleDowngradeMetadata(metadata, target_info, source_info)

  return metadata

def HandleDowngradeMetadata(metadata, target_info, source_info):

  # Only incremental OTAs are allowed to reach here.

  assert OPTIONS.incremental_source is not None

  post_timestamp = target_info.GetBuildProp("ro.build.date.utc")

  pre_timestamp = source_info.GetBuildProp("ro.build.date.utc")

  is_downgrade = int(post_timestamp) < int(pre_timestamp)

  if OPTIONS.downgrade:

    if not is_downgrade:

      raise RuntimeError(

          "--downgrade or --override_timestamp specified but no downgrade "

          "detected: pre: %s, post: %s" % (pre_timestamp, post_timestamp))

    metadata["ota-downgrade"] = "yes"

  else:

    if is_downgrade:

      raise RuntimeError(

          "Downgrade detected based on timestamp check: pre: %s, post: %s. "

          "Need to specify --override_timestamp OR --downgrade to allow "

          "building the incremental." % (pre_timestamp, post_timestamp))

def CalculateRuntimeDevicesAndFingerprints(build_info, boot_variable_values):

  """Returns a tuple of sets for runtime devices and fingerprints"""

  device_names = {build_info.device}

  fingerprints = {build_info.fingerprint}

  if not boot_variable_values:

    return device_names, fingerprints

  # Calculate all possible combinations of the values for the boot variables.

  keys = boot_variable_values.keys()

  value_list = boot_variable_values.values()

  combinations = [dict(zip(keys, values))

                  for values in itertools.product(*value_list)]

  for placeholder_values in combinations:

    # Reload the info_dict as some build properties may change their values

    # based on the value of ro.boot* properties.

    info_dict = copy.deepcopy(build_info.info_dict)

    for partition in PARTITIONS_WITH_CARE_MAP:

      partition_prop_key = "{}.build.prop".format(partition)

      input_file = info_dict[partition_prop_key].input_file

      if isinstance(input_file, zipfile.ZipFile):

        with zipfile.ZipFile(input_file.filename) as input_zip:

          info_dict[partition_prop_key] = \

              PartitionBuildProps.FromInputFile(input_zip, partition,

                                                placeholder_values)

      else:

        info_dict[partition_prop_key] = \

            PartitionBuildProps.FromInputFile(input_file, partition,

                                              placeholder_values)

    info_dict["build.prop"] = info_dict["system.build.prop"]

    new_build_info = BuildInfo(info_dict, build_info.oem_dicts)

    device_names.add(new_build_info.device)

    fingerprints.add(new_build_info.fingerprint)

  return device_names, fingerprints

class PropertyFiles(object):

  """A class that computes the property-files string for an OTA package.

  A property-files string is a comma-separated string that contains the

  offset/size info for an OTA package. The entries, which must be ZIP_STORED,

  can be fetched directly with the package URL along with the offset/size info.

  These strings can be used for streaming A/B OTAs, or allowing an updater to

  download package metadata entry directly, without paying the cost of

  downloading entire package.

  Computing the final property-files string requires two passes. Because doing

  the whole package signing (with signapk.jar) will possibly reorder the ZIP

  entries, which may in turn invalidate earlier computed ZIP entry offset/size

  values.

  This class provides functions to be called for each pass. The general flow is

  as follows.

    property_files = PropertyFiles()

    # The first pass, which writes placeholders before doing initial signing.

    property_files.Compute()

    SignOutput()

    # The second pass, by replacing the placeholders with actual data.

    property_files.Finalize()

    SignOutput()

  And the caller can additionally verify the final result.

    property_files.Verify()

  """

  def __init__(self):

    self.name = None

    self.required = ()

    self.optional = ()

  def Compute(self, input_zip):

    """Computes and returns a property-files string with placeholders.

    We reserve extra space for the offset and size of the metadata entry itself,

    although we don't know the final values until the package gets signed.

    Args:

      input_zip: The input ZIP file.

    Returns:

      A string with placeholders for the metadata offset/size info, e.g.

      "payload.bin:679:343,payload_properties.txt:378:45,metadata:        ".

    """

    return self.GetPropertyFilesString(input_zip, reserve_space=True)

  class InsufficientSpaceException(Exception):

    pass

  def Finalize(self, input_zip, reserved_length):

    """Finalizes a property-files string with actual METADATA offset/size info.

    The input ZIP file has been signed, with the ZIP entries in the desired

    place (signapk.jar will possibly reorder the ZIP entries). Now we compute

    the ZIP entry offsets and construct the property-files string with actual

    data. Note that during this process, we must pad the property-files string

    to the reserved length, so that the METADATA entry size remains the same.

    Otherwise the entries' offsets and sizes may change again.

    Args:

      input_zip: The input ZIP file.

      reserved_length: The reserved length of the property-files string during

          the call to Compute(). The final string must be no more than this

          size.

    Returns:

      A property-files string including the metadata offset/size info, e.g.

      "payload.bin:679:343,payload_properties.txt:378:45,metadata:69:379  ".

    Raises:

      InsufficientSpaceException: If the reserved length is insufficient to hold

          the final string.

    """

    result = self.GetPropertyFilesString(input_zip, reserve_space=False)

    if len(result) > reserved_length:

      raise self.InsufficientSpaceException(

          'Insufficient reserved space: reserved={}, actual={}'.format(

              reserved_length, len(result)))

    result += ' ' * (reserved_length - len(result))

    return result

  def Verify(self, input_zip, expected):

    """Verifies the input ZIP file contains the expected property-files string.

    Args:

      input_zip: The input ZIP file.

      expected: The property-files string that's computed from Finalize().

    Raises:

      AssertionError: On finding a mismatch.

    """

    actual = self.GetPropertyFilesString(input_zip)

    assert actual == expected, \

        "Mismatching streaming metadata: {} vs {}.".format(actual, expected)

  def GetPropertyFilesString(self, zip_file, reserve_space=False):

    """

    Constructs the property-files string per request.

    Args:

      zip_file: The input ZIP file.

      reserved_length: The reserved length of the property-files string.

    Returns:

      A property-files string including the metadata offset/size info, e.g.

      "payload.bin:679:343,payload_properties.txt:378:45,metadata:     ".

    """

    def ComputeEntryOffsetSize(name):

      """Computes the zip entry offset and size."""

      info = zip_file.getinfo(name)

      offset = info.header_offset

      offset += zipfile.sizeFileHeader

      offset += len(info.extra) + len(info.filename)

      size = info.file_size

      return '%s:%d:%d' % (os.path.basename(name), offset, size)

    tokens = []

    tokens.extend(self._GetPrecomputed(zip_file))

    for entry in self.required:

      tokens.append(ComputeEntryOffsetSize(entry))

    for entry in self.optional:

      if entry in zip_file.namelist():

        tokens.append(ComputeEntryOffsetSize(entry))

    # 'META-INF/com/android/metadata' is required. We don't know its actual

    # offset and length (as well as the values for other entries). So we reserve

    # 15-byte as a placeholder ('offset:length'), which is sufficient to cover

    # the space for metadata entry. Because 'offset' allows a max of 10-digit

    # (i.e. ~9 GiB), with a max of 4-digit for the length. Note that all the

    # reserved space serves the metadata entry only.

    if reserve_space:

      tokens.append('metadata:' + ' ' * 15)

    else:

      tokens.append(ComputeEntryOffsetSize(METADATA_NAME))

    return ','.join(tokens)

  def _GetPrecomputed(self, input_zip):

    """Computes the additional tokens to be included into the property-files.

    This applies to tokens without actual ZIP entries, such as

    payload_metadata.bin. We want to expose the offset/size to updaters, so

    that they can download the payload metadata directly with the info.

    Args:

      input_zip: The input zip file.

    Returns:

      A list of strings (tokens) to be added to the property-files string.

    """

    # pylint: disable=no-self-use

    # pylint: disable=unused-argument

    return []

def SignOutput(temp_zip_name, output_zip_name):

  pw = OPTIONS.key_passwords[OPTIONS.package_key]

  SignFile(temp_zip_name, output_zip_name, OPTIONS.package_key, pw,

           whole_file=True)