Quelle  ota_from_raw_img.py

#!/usr/bin/env python
#
# Copyright (C) 2008 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 series of .img files, produces an OTA package that installs thoese images
"""

import sys
import os
import argparse
import shutil
import subprocess
import tempfile
import logging
import zipfile

import common
import ota_metadata_pb2
from payload_signer import PayloadSigner
from ota_utils import PayloadGenerator, FinalizeMetadata
from ota_signing_utils import AddSigningArgumentParse


logger = logging.getLogger(__name__)


def ResolveBinaryPath(filename, search_path, bin_path):
  if bin_path is not None:
    return bin_path
  if not search_path:
    return filename
  if not os.path.exists(search_path):
    return filename
  path = os.path.join(search_path, "bin", filename)
  if os.path.exists(path):
    return path
  path = os.path.join(search_path, filename)
  if os.path.exists(path):
    return path
  return path


def UpdateDynamicPartitionInfo(contents, in_file):
    with open(in_file, 'r') as fp:
        for line in fp.readlines():
            parts = line.split('=', maxsplit=1)
            if len(parts) != 2:
                continue
            contents[parts[0]] = parts[1]


def WriteDynamicPartitionInfo(in_file, out_fp):
    keyvalues = {
        "virtual_ab": "true",
        "virtual_ab_compression": "true",
        "virtual_ab_compression_method": "none",
        "super_partition_groups": "",
    }
    if in_file is not None:
        UpdateDynamicPartitionInfo(keyvalues, in_file)
    for key in keyvalues:
        line = "{}={}\n".format(key, keyvalues[key])
        out_fp.write(line.encode("utf-8"))
    out_fp.flush()


def main(argv):
  parser = argparse.ArgumentParser(
      prog=argv[0], description="Given a series of .img files, produces a full OTA package that installs thoese images")
  parser.add_argument("images", nargs="+", type=str,
                      help="List of images to generate OTA")
  parser.add_argument("--partition_names", nargs='?', type=str,
                      help="Partition names to install the images, default to basename of the image(no file name extension)")
  parser.add_argument('--output', type=str,
                      help='Paths to output merged ota', required=True)
  parser.add_argument('--max_timestamp', type=int,
                      help='Maximum build timestamp allowed to install this OTA')
  parser.add_argument("--metadata_proto_file", type=str,
                      help="Optional OTA metadata proto to use for signing")
  parser.add_argument("--dynamic_partition_info_file", type=str,
                      help="Optional dynamic partition info file")
  parser.add_argument("--delta_generator_path", type=str,
                      help="Path to delta_generator")
  parser.add_argument("-v", action="store_true",
                      help="Enable verbose logging", dest="verbose")
  AddSigningArgumentParse(parser)

  args = parser.parse_args(argv[1:])
  if args.verbose:
    logger.setLevel(logging.INFO)
  logger.info(args)
  old_imgs = [""] * len(args.images)
  for (i, img) in enumerate(args.images):
    if ":" in img:
      old_imgs[i], args.images[i] = img.split(":", maxsplit=1)

  if not args.partition_names:
    args.partition_names = [os.path.splitext(os.path.basename(path))[
        0] for path in args.images]
  else:
    args.partition_names = args.partition_names.split(",")
  with tempfile.NamedTemporaryFile() as unsigned_payload, tempfile.NamedTemporaryFile() as dynamic_partition_info_file:
    WriteDynamicPartitionInfo(args.dynamic_partition_info_file, dynamic_partition_info_file)
    cmd = [ResolveBinaryPath("delta_generator", args.search_path, args.delta_generator_path)]
    cmd.append("--partition_names=" + ":".join(args.partition_names))
    cmd.append("--dynamic_partition_info_file=" +
               dynamic_partition_info_file.name)
    cmd.append("--old_partitions=" + ":".join(old_imgs))
    cmd.append("--new_partitions=" + ":".join(args.images))
    cmd.append("--out_file=" + unsigned_payload.name)
    cmd.append("--is_partial_update")
    if args.max_timestamp:
      cmd.append("--max_timestamp=" + str(args.max_timestamp))
      cmd.append("--partition_timestamps=boot:" + str(args.max_timestamp))
    logger.info("Running %s", cmd)

    subprocess.check_call(cmd)
    generator = PayloadGenerator()
    generator.payload_file = unsigned_payload.name
    logger.info("Payload size: %d", os.path.getsize(generator.payload_file))

    # Get signing keys
    key_passwords = common.GetKeyPasswords([args.package_key])

    if args.package_key:
      logger.info("Signing payload...")
      signer = PayloadSigner(args.package_key, args.private_key_suffix,
                             key_passwords[args.package_key],
                             payload_signer=args.payload_signer,
                             payload_signer_args=args.payload_signer_args,
                             payload_signer_maximum_signature_size=args.payload_signer_maximum_signature_size)
      generator.payload_file = unsigned_payload.name
      generator.Sign(signer)

    logger.info("Payload size: %d", os.path.getsize(generator.payload_file))

    logger.info("Writing to %s", args.output)
    with zipfile.ZipFile(args.output, "w") as zfp:
      generator.WriteToZip(zfp)

    if args.package_key and args.metadata_proto_file:
      temp_zip = common.MakeTempFile(prefix="temp-", suffix=".zip")
      metadata = ota_metadata_pb2.OtaMetadata()
      with open(args.metadata_proto_file, "rb") as fp:
          metadata.ParseFromString(fp.read())
      shutil.copy(args.output, temp_zip)
      FinalizeMetadata(metadata, temp_zip, args.output, package_key=args.package_key)

if __name__ == "__main__":
  logging.basicConfig()
  main(sys.argv)