Revert "Add ability to sign bundled APEX into device signing process."

#!/usr/bin/env python

#

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

from __future__ import print_function

import logging

import os.path

import re

import tempfile

import zipfile

import common

logger = logging.getLogger(__name__)

OPTIONS = common.OPTIONS

BLOCK_SIZE = common.BLOCK_SIZE

class ApexInfoError(Exception):

  """An Exception raised during Apex Information command."""

  def __init__(self, message):

    Exception.__init__(self, message)

class ApexSigningError(Exception):

  """An Exception raised during Apex Payload signing."""

  def __init__(self, message):

    Exception.__init__(self, message)

class ApexPayloadSignerHelper(object):

  """An Apex Payload Signer helper class."""

  def __init__(self, apex_payload_data):

    self.avbtool = 'avbtool'

    self.apex_payload_file = tempfile.NamedTemporaryFile()

    self.apex_payload_file.write(apex_payload_data)

    self.apex_payload_file.flush()

    self.payload_image_info_helper = ApexInfoHelper()

    self.payload_image_info_helper.SetImageInfo(self.avbtool,

                                                self.apex_payload_file.name)

  def GetPayLoadImageInfo(self):

    return self.payload_image_info_helper.GetImageInfo()

  def StripExistingAVBMetaSignature(self, image_path):

    cmd = [self.avbtool, "erase_footer", "--image", image_path]

    proc = common.Run(cmd)

    output, _ = proc.communicate()

    if proc.returncode != 0:

      raise ApexInfoError(

          "Unable to erase VBMeta footer:\n{}".format(output))

  def Verify(self, image_key):

    cmd = [self.avbtool, "verify_image",

           "--image", self.apex_payload_file.name,

           "--key", image_key]

    proc = common.Run(cmd)

    output, _ = proc.communicate()

    if proc.returncode != 0:

      raise ApexSigningError(

          "Unable to validate Image Signing:\n{}".format(output))

  def Sign(self, package_name, image_key, signing_args=None):

    # Strip any existing VBMeta signature/footers.

    try:

      self.StripExistingAVBMetaSignature(self.apex_payload_file.name)

    except ApexInfoError:

      print ("Unable to strip existing VBMeta Signature.")

      raise

    algorithm = self.GetPayLoadImageInfo().get('Algorithm',

                                               OPTIONS.avb_algorithms.get('apex'))

    assert algorithm, 'Missing AVB signing algorithm for %s' % (package_name,)

    salt = self.GetPayLoadImageInfo().get('Salt')

    cmd = [self.avbtool, "add_hashtree_footer",

           "--do_not_generate_fec",

           "--algorithm", algorithm,

           "--key", image_key,

           "--prop", "apex.key:%s" % package_name,

           "--image", self.apex_payload_file.name,

           "--salt", salt]

    if signing_args:

      cmd.extend([signing_args])

    proc = common.Run(cmd)

    output, _ = proc.communicate()

    if proc.returncode != 0:

      raise ApexSigningError(

          "Unable to sign Apex image:\n{}".format(output))

    # Verify the Signed Image with specified public key.

    print ("Verifying %s" % package_name)

    self.Verify(image_key)

    data = None

    with open(self.apex_payload_file.name) as af:

      data = af.read()

    return data

class ApexInfoHelper(object):

  """An Apex Info helper."""

  def __init__(self):

    self.image_info = {}

  def SetImageInfo(self, avbtool, image_path):

    if not os.path.exists(image_path):

      raise ApexInfoError(

          "Unable to find Image: %s" % image_path)

    cmd = [avbtool, "info_image", "--image", image_path]

    # Extract the Algorithm and Salt information from image.

    regex = re.compile('^\s*(?P<key>Algorithm|Salt)\:\s*(?P<value>.*?)$')

    proc = common.Run(cmd)

    for line in iter(proc.stdout.readline, b''):

      _item = regex.match(line)

      if _item:

        item_dict = _item.groupdict()

        self.image_info[item_dict['key']] = item_dict['value']

    output, _ = proc.communicate()

    if proc.returncode != 0:

      raise ApexInfoError(

          "Unable to fetch information:\n{}".format(output))

  def GetImageInfo(self):

    return self.image_info

      Replace the veritykeyid in BOOT/cmdline of input_target_file_zip

      with keyid of the cert pointed by <path_to_X509_PEM_cert_file>.

  --apex_payload_key <name=key>

      Add a mapping for apex package name to payload signing key.

  --avb_{apex,boot,system,vendor,dtbo,vbmeta}_algorithm <algorithm>

  --avb_{boot,system,vendor,dtbo,vbmeta}_algorithm <algorithm>

  --avb_{boot,system,vendor,dtbo,vbmeta}_key <key>

      Use the specified algorithm (e.g. SHA256_RSA4096) and the key to AVB-sign

      the specified image. Otherwise it uses the existing values in info dict.

  --avb_{apex,boot,system,vendor,dtbo,vbmeta}_extra_args <args>

  --avb_{boot,system,vendor,dtbo,vbmeta}_extra_args <args>

      Specify any additional args that are needed to AVB-sign the image

      (e.g. "--signing_helper /path/to/helper"). The args will be appended to

      the existing ones in info dict.

import stat

import subprocess

import sys

import time

import tempfile

import zipfile

from xml.etree import ElementTree

import add_img_to_target_files

import apex_utils

import common

OPTIONS = common.OPTIONS

OPTIONS.extra_apks = {}

OPTIONS.apex_payload_keys = {}

OPTIONS.skip_apks_with_path_prefix = set()

OPTIONS.key_map = {}

OPTIONS.rebuild_recovery = False

  return data

def SignApex(data, name, image_key_path, container_key, pw,

             platform_api_level, codename_to_api_level_map):

  """Signing an Apex is a two step process.

  1: Extract and sign the image_payload.img with image_key.

  2: Sign the overall APK Container with container key.

  """

  unsigned_apex_payload = tempfile.NamedTemporaryFile()

  signed_apex_payload = tempfile.NamedTemporaryFile()

  unsigned_apex_payload.write(data)

  unsigned_apex_payload.flush()

  unsigned_apex_payload_zip = zipfile.ZipFile(

      unsigned_apex_payload.name, "r")

  signed_apex_payload_zip = zipfile.ZipFile(

      signed_apex_payload.name, "w",

      compression=zipfile.ZIP_DEFLATED,

      allowZip64=False)

  for info in unsigned_apex_payload_zip.infolist():

    filename = info.filename

    data = unsigned_apex_payload_zip.read(filename)

    out_info = copy.copy(info)

    if filename == 'apex_payload.img':

      package_name = re.sub('.apex', '', name)

      apex_sign_helper = apex_utils.ApexPayloadSignerHelper(

          data)

      extra_signing_args=OPTIONS.avb_extra_args.get('apex')

      key_path = '%s_pub.pem' % image_key_path

      signed_apex_payload_image = apex_sign_helper.Sign(

          package_name,

          key_path,

          extra_signing_args)

      common.ZipWriteStr(signed_apex_payload_zip,

                         out_info,

                         signed_apex_payload_image)

    else:

      common.ZipWriteStr(signed_apex_payload_zip,

                         out_info,

                         data)

  common.ZipClose(unsigned_apex_payload_zip)

  common.ZipClose(signed_apex_payload_zip)

  signed_container = tempfile.NamedTemporaryFile()

  signed_aligned_container = tempfile.NamedTemporaryFile()

  common.SignFile(

      signed_apex_payload.name,

      signed_container.name,

      container_key,

      pw,

      min_api_level=None,

      codename_to_api_level_map=codename_to_api_level_map)

  cmd = ['zipalign', '-f', '4096',

         signed_container.name,

         signed_aligned_container.name]

  proc = common.Run(cmd)

  output, _ = proc.communicate()

  if proc.returncode != 0:

    raise ApexInfoError(

      "Unable to zipalign apex container:\n{}".format(output))

  data = signed_aligned_container.read()

  signed_container.close()

  signed_aligned_container.close()

  signed_apex_payload.close()

  return data

def ProcessTargetFiles(input_tf_zip, output_tf_zip, misc_info,

                       apk_key_map, key_passwords, platform_api_level,

  for info in input_tf_zip.infolist():

    filename = info.filename

    if filename.startswith("IMAGES/"):

      continue

    elif filename == "META/care_map.pb" or filename == "META/care_map.txt":

      pass

    # Sign Bundled APEX files.

    elif filename.startswith("SYSTEM/apex") and filename.endswith(".apex"):

      name = os.path.basename(filename)

      # TODO(baligh): Read Apex Image and Container keys from

      # META/apexkeys.txt

      if name in OPTIONS.apex_payload_keys and name in OPTIONS.extra_apks:

        image_key_path = OPTIONS.apex_payload_keys[name]

        container_key_path = OPTIONS.extra_apks[name]

        print("    signing: %-*s (%s)" % (maxsize, name, container_key_path))

        print("           : %-*s (%s)" % (maxsize, 'image_payload', image_key_path))

        signed_data = SignApex(data,

                               name,

                               image_key_path,

                               container_key_path,

                               key_passwords[container_key_path],

                               platform_api_level,

                               codename_to_api_level_map)

        common.ZipWriteStr(output_tf_zip, out_info, signed_data)

      else:

        print("    signing: %-*s (NOT SIGNING)" % (maxsize, name))

        common.ZipWriteStr(output_tf_zip, out_info, data)

    elif ((os.path.dirname(filename) == "SYSTEM/etc/security/apex") and

          filename != 'SYSTEM/etc/security/apex/'):

      # AVBPubkey is used to validate APEX payload_image at load time.

      apex_name = '%s.apex' % os.path.basename(filename)

      if apex_name in OPTIONS.extra_apks:

        avb_pubkey_path = '%s.avbpubkey' % OPTIONS.apex_payload_keys[apex_name]

        print ("Replacing AVB PubKey for %s with %s" % (

            apex_name, avb_pubkey_path))

        with open(avb_pubkey_path) as cf:

          common.ZipWriteStr(output_tf_zip, out_info, cf.read())

      else:

        print ("Not Replacing AVB Pubkey for %s" % apex_name)

        common.ZipWriteStr(output_tf_zip, out_info, data)

    # A non-APK file; copy it verbatim.

    else:

      common.ZipWriteStr(output_tf_zip, out_info, data)

  key_mapping_options = []

  def option_handler(o, a):

    if o == "--apex_payload_key":

      apex_name, key = a.split("=")

      OPTIONS.apex_payload_keys[apex_name] = key

    elif o in ("-e", "--extra_apks"):

    if o in ("-e", "--extra_apks"):

      names, key = a.split("=")

      names = names.split(",")

      for n in names:

      OPTIONS.avb_extra_args['vbmeta'] = a

    elif o == "--avb_boot_key":

      OPTIONS.avb_keys['boot'] = a

    elif o == "--avb_apex_algorithm":

      OPTIONS.avb_algorithms['apex'] = a

    elif o == "--avb_boot_algorithm":

      OPTIONS.avb_algorithms['boot'] = a

    elif o == "--avb_boot_extra_args":

      OPTIONS.avb_algorithms['vendor'] = a

    elif o == "--avb_vendor_extra_args":

      OPTIONS.avb_extra_args['vendor'] = a

    elif o == "--avb_apex_extra_args":

      OPTIONS.avb_extra_args['apex'] = a

    else:

      return False

    return True

      extra_opts="e:d:k:ot:",

      extra_long_opts=[

          "extra_apks=",

          "apex_payload_key=",

          "skip_apks_with_path_prefix=",

          "default_key_mappings=",

          "key_mapping=",

          "avb_vbmeta_algorithm=",

          "avb_vbmeta_key=",

          "avb_vbmeta_extra_args=",

          "avb_apex_algorithm=",

          "avb_boot_algorithm=",

          "avb_boot_key=",

          "avb_boot_extra_args=",

          "avb_vendor_algorithm=",

          "avb_vendor_key=",

          "avb_vendor_extra_args=",

          "avb_apex_extra_args=",

      ],

      extra_option_handler=option_handler)