Merge changes Idc7776be,I409fe30a

from __future__ import print_function

import base64

import collections

import copy

import errno

import re

import shlex

import shutil

import string

import subprocess

import sys

import tempfile

    kwargs: Any additional args to be passed to subprocess.Popen(), such as env,

        stdin, etc. stdout and stderr will default to subprocess.PIPE and

        subprocess.STDOUT respectively unless caller specifies any of them.

        universal_newlines will default to True, as most of the users in

        releasetools expect string output.

  Returns:

    A subprocess.Popen object.

  if 'stdout' not in kwargs and 'stderr' not in kwargs:

    kwargs['stdout'] = subprocess.PIPE

    kwargs['stderr'] = subprocess.STDOUT

  if 'universal_newlines' not in kwargs:

    kwargs['universal_newlines'] = True

  # Don't log any if caller explicitly says so.

  if verbose != False:

    logger.info("  Running: \"%s\"", " ".join(args))

  def read_helper(fn):

    if isinstance(input_file, zipfile.ZipFile):

      return input_file.read(fn)

      return input_file.read(fn).decode()

    else:

      path = os.path.join(input_file, *fn.split("/"))

      try:

  # system. Other areas assume system is always at "/system" so point /system

  # at /.

  if system_root_image:

    assert not d.has_key("/system") and d.has_key("/")

    assert '/system' not in d and '/' in d

    d["/system"] = d["/"]

  return d

    # filename listed in system.map may contain an additional leading slash

    # (i.e. "//system/framework/am.jar"). Using lstrip to get consistent

    # results.

    arcname = string.replace(entry, which, which.upper(), 1).lstrip('/')

    arcname = entry.replace(which, which.upper(), 1).lstrip('/')

    # Special handling another case, where files not under /system

    # (e.g. "/sbin/charger") are packed under ROOT/ in a target_files.zip.

    if basename:

      installed_files.add(basename)

  for line in tf_zip.read("META/apkcerts.txt").split("\n"):

  for line in tf_zip.read('META/apkcerts.txt').decode().split('\n'):

    line = line.strip()

    if not line:

      continue

      if not first:

        print("key file %s still missing some passwords." % (self.pwfile,))

        if sys.version_info[0] >= 3:

          raw_input = input  # pylint: disable=redefined-builtin

        answer = raw_input("try to edit again? [y]> ").strip()

        if answer and answer[0] not in 'yY':

          raise RuntimeError("key passwords unavailable")

  This gives the same result as `openssl x509 -in <filename> -outform DER`.

  Returns:

    The decoded certificate string.

    The decoded certificate bytes.

  """

  cert_buffer = []

  save = False

      cert_buffer.append(line)

    if "--BEGIN CERTIFICATE--" in line:

      save = True

  cert = "".join(cert_buffer).decode('base64')

  cert = base64.b64decode("".join(cert_buffer))

  return cert

  logger.info("putting script in %s", sh_location)

  output_sink(sh_location, sh)

  output_sink(sh_location, sh.encode())

class DynamicPartitionUpdate(object):

  def test_ExtractPublicKey(self):

    cert = os.path.join(self.testdata_dir, 'testkey.x509.pem')

    pubkey = os.path.join(self.testdata_dir, 'testkey.pubkey.pem')

    with open(pubkey, 'rb') as pubkey_fp:

    with open(pubkey) as pubkey_fp:

      self.assertEqual(pubkey_fp.read(), common.ExtractPublicKey(cert))

  def test_ExtractPublicKey_invalidInput(self):

  def test_ExtractAvbPublicKey(self):

    privkey = os.path.join(self.testdata_dir, 'testkey.key')

    pubkey = os.path.join(self.testdata_dir, 'testkey.pubkey.pem')

    with open(common.ExtractAvbPublicKey(privkey)) as privkey_fp, \

        open(common.ExtractAvbPublicKey(pubkey)) as pubkey_fp:

    with open(common.ExtractAvbPublicKey(privkey), 'rb') as privkey_fp, \

        open(common.ExtractAvbPublicKey(pubkey), 'rb') as pubkey_fp:

      self.assertEqual(privkey_fp.read(), pubkey_fp.read())

  def test_ParseCertificate(self):

    cert = os.path.join(self.testdata_dir, 'testkey.x509.pem')

    cmd = ['openssl', 'x509', '-in', cert, '-outform', 'DER']

    proc = common.Run(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE)

    proc = common.Run(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE,

                      universal_newlines=False)

    expected, _ = proc.communicate()

    self.assertEqual(0, proc.returncode)

    target_files = common.MakeTempFile(prefix='target_files-', suffix='.zip')

    with zipfile.ZipFile(target_files, 'w') as target_files_zip:

      info_values = ''.join(

          ['{}={}\n'.format(k, v) for k, v in sorted(info_dict.iteritems())])

          ['{}={}\n'.format(k, v) for k, v in sorted(info_dict.items())])

      common.ZipWriteStr(target_files_zip, 'META/misc_info.txt', info_values)

      FSTAB_TEMPLATE = "/dev/block/system {} ext4 ro,barrier=1 defaults"

    loc = os.path.join(self._tempdir, prefix, name)

    if not os.path.exists(os.path.dirname(loc)):

      os.makedirs(os.path.dirname(loc))

    with open(loc, "w+") as f:

    with open(loc, "wb") as f:

      f.write(data)

  def test_full_recovery(self):

    validate_target_files.ValidateInstallRecoveryScript(self._tempdir,

                                                        self._info)

    # Validate 'recovery-from-boot' with bonus argument.

    self._out_tmp_sink("etc/recovery-resource.dat", "bonus", "SYSTEM")

    self._out_tmp_sink("etc/recovery-resource.dat", b"bonus", "SYSTEM")

    common.MakeRecoveryPatch(self._tempdir, self._out_tmp_sink,

                             recovery_image, boot_image, self._info)

    validate_target_files.ValidateInstallRecoveryScript(self._tempdir,

class MockScriptWriter(object):

  """A class that mocks edify_generator.EdifyGenerator.

  """

  """A class that mocks edify_generator.EdifyGenerator."""

  def __init__(self, enable_comments=False):

    self.lines = []

    self.enable_comments = enable_comments

  def Comment(self, comment):

    if self.enable_comments:

      self.lines.append("# {}".format(comment))

      self.lines.append('# {}'.format(comment))

  def AppendExtra(self, extra):

    self.lines.append(extra)

  def __str__(self):

    return "\n".join(self.lines)

    return '\n'.join(self.lines)

class MockBlockDifference(object):

  def __init__(self, partition, tgt, src=None):

    self.partition = partition

    self.tgt = tgt

    self.src = src

  def WriteScript(self, script, _, progress=None,

                  write_verify_script=False):

    if progress:

    script.AppendExtra("patch({});".format(self.partition))

    if write_verify_script:

      self.WritePostInstallVerifyScript(script)

  def WritePostInstallVerifyScript(self, script):

    script.AppendExtra("verify({});".format(self.partition))

class FakeSparseImage(object):

  def __init__(self, size):

    self.blocksize = 4096

    self.total_blocks = size // 4096

class DynamicPartitionsDifferenceTest(test_utils.ReleaseToolsTestCase):

  @staticmethod

  def get_op_list(output_path):

    with zipfile.ZipFile(output_path) as output_zip:

      with output_zip.open("dynamic_partitions_op_list") as op_list:

        return [line.strip() for line in op_list.readlines()

                if not line.startswith("#")]

      with output_zip.open('dynamic_partitions_op_list') as op_list:

        return [line.decode().strip() for line in op_list.readlines()

                if not line.startswith(b'#')]

  def setUp(self):

    self.script = MockScriptWriter()

    }

    self.hash_algorithm = "sha256"

    self.fixed_salt = \

        "aee087a5be3b982978c923f566a94613496b417f2af592639bc80d141e34dfe7"

    self.expected_root_hash = \

        "0b7c4565e87b1026e11fbab91c0bc29e185c847a5b44d40e6e86e461e8adf80d"

    self.fixed_salt = (

        "aee087a5be3b982978c923f566a94613496b417f2af592639bc80d141e34dfe7")

    self.expected_root_hash = (

        "0b7c4565e87b1026e11fbab91c0bc29e185c847a5b44d40e6e86e461e8adf80d")

  def _create_simg(self, raw_data):

  def _CreateSimg(self, raw_data):  # pylint: disable=no-self-use

    output_file = common.MakeTempFile()

    raw_image = common.MakeTempFile()

    with open(raw_image, 'wb') as f:

      f.write(raw_data)

    cmd = ["img2simg", raw_image, output_file, '4096']

    p = common.Run(cmd)

    p.communicate()

    self.assertEqual(0, p.returncode)

    common.RunAndCheckOutput(cmd)

    return output_file

  def _generate_image(self):

  def _GenerateImage(self):

    partition_size = 1024 * 1024

    prop_dict = {

        'partition_size': str(partition_size),

    self.assertIsNotNone(verity_image_builder)

    adjusted_size = verity_image_builder.CalculateMaxImageSize()

    raw_image = ""

    raw_image = bytearray(adjusted_size)

    for i in range(adjusted_size):

      raw_image += str(i % 10)

      raw_image[i] = ord('0') + i % 10

    output_file = self._create_simg(raw_image)

    output_file = self._CreateSimg(raw_image)

    # Append the verity metadata.

    verity_image_builder.Build(output_file)

  @SkipIfExternalToolsUnavailable()

  def test_CreateHashtreeInfoGenerator(self):

    image_file = sparse_img.SparseImage(self._generate_image())

    image_file = sparse_img.SparseImage(self._GenerateImage())

    generator = CreateHashtreeInfoGenerator(

        'system', image_file, self.prop_dict)

  @SkipIfExternalToolsUnavailable()

  def test_DecomposeSparseImage(self):

    image_file = sparse_img.SparseImage(self._generate_image())

    image_file = sparse_img.SparseImage(self._GenerateImage())

    generator = VerifiedBootVersion1HashtreeInfoGenerator(

        self.partition_size, 4096, True)

  @SkipIfExternalToolsUnavailable()

  def test_ParseHashtreeMetadata(self):

    image_file = sparse_img.SparseImage(self._generate_image())

    image_file = sparse_img.SparseImage(self._GenerateImage())

    generator = VerifiedBootVersion1HashtreeInfoGenerator(

        self.partition_size, 4096, True)

    generator.DecomposeSparseImage(image_file)

  def test_ValidateHashtree_smoke(self):

    generator = VerifiedBootVersion1HashtreeInfoGenerator(

        self.partition_size, 4096, True)

    generator.image = sparse_img.SparseImage(self._generate_image())

    generator.image = sparse_img.SparseImage(self._GenerateImage())

    generator.hashtree_info = info = HashtreeInfo()

    info.filesystem_range = RangeSet(data=[0, 991232 / 4096])

    info.filesystem_range = RangeSet(data=[0, 991232 // 4096])

    info.hashtree_range = RangeSet(

        data=[991232 / 4096, (991232 + 12288) / 4096])

        data=[991232 // 4096, (991232 + 12288) // 4096])

    info.hash_algorithm = self.hash_algorithm

    info.salt = self.fixed_salt

    info.root_hash = self.expected_root_hash

  def test_ValidateHashtree_failure(self):

    generator = VerifiedBootVersion1HashtreeInfoGenerator(

        self.partition_size, 4096, True)

    generator.image = sparse_img.SparseImage(self._generate_image())

    generator.image = sparse_img.SparseImage(self._GenerateImage())

    generator.hashtree_info = info = HashtreeInfo()

    info.filesystem_range = RangeSet(data=[0, 991232 / 4096])

    info.filesystem_range = RangeSet(data=[0, 991232 // 4096])

    info.hashtree_range = RangeSet(

        data=[991232 / 4096, (991232 + 12288) / 4096])

        data=[991232 // 4096, (991232 + 12288) // 4096])

    info.hash_algorithm = self.hash_algorithm

    info.salt = self.fixed_salt

    info.root_hash = "a" + self.expected_root_hash[1:]

  @SkipIfExternalToolsUnavailable()

  def test_Generate(self):

    image_file = sparse_img.SparseImage(self._generate_image())

    image_file = sparse_img.SparseImage(self._GenerateImage())

    generator = CreateHashtreeInfoGenerator('system', 4096, self.prop_dict)

    info = generator.Generate(image_file)

    self.assertEqual(RangeSet(data=[0, 991232 / 4096]), info.filesystem_range)

    self.assertEqual(RangeSet(data=[991232 / 4096, (991232 + 12288) / 4096]),

    self.assertEqual(RangeSet(data=[0, 991232 // 4096]), info.filesystem_range)

    self.assertEqual(RangeSet(data=[991232 // 4096, (991232 + 12288) // 4096]),

                     info.hashtree_range)

    self.assertEqual(self.hash_algorithm, info.hash_algorithm)

    self.assertEqual(self.fixed_salt, info.salt)

  """Constructs and returns a File object. Rounds up its size if needed."""

  assert os.path.exists(unpacked_name)

  with open(unpacked_name, 'r') as f:

  with open(unpacked_name, 'rb') as f:

    file_data = f.read()

  file_size = len(file_data)

  if round_up:

    BuildVerityImageError: On error.

  """

  try:

    with open(target, "a") as out_file, open(file_to_append, "r") as input_file:

    with open(target, 'ab') as out_file, \

        open(file_to_append, 'rb') as input_file:

      for line in input_file:

        out_file.write(line)

  except IOError:

    # key_path and algorithm are only available when chain partition is used.

    key_path = prop_dict.get("avb_key_path")

    algorithm = prop_dict.get("avb_algorithm")

    # Image uses hash footer.

    if prop_dict.get("avb_hash_enable") == "true":

      return VerifiedBootVersion2VerityImageBuilder(

          prop_dict["partition_name"],

          algorithm,

          prop_dict.get("avb_salt"),

          prop_dict["avb_add_hash_footer_args"])

    else:

    # Image uses hashtree footer.

    return VerifiedBootVersion2VerityImageBuilder(

        prop_dict["partition_name"],

        partition_size,

    self.metadata_size = metadata_size

    self.hashtree_info.filesystem_range = RangeSet(

        data=[0, adjusted_size / self.block_size])

        data=[0, adjusted_size // self.block_size])

    self.hashtree_info.hashtree_range = RangeSet(

        data=[adjusted_size / self.block_size,

              (adjusted_size + verity_tree_size) / self.block_size])

        data=[adjusted_size // self.block_size,

              (adjusted_size + verity_tree_size) // self.block_size])

  def _ParseHashtreeMetadata(self):

    """Parses the hash_algorithm, root_hash, salt from the metadata block."""

    metadata_start = self.filesystem_size + self.hashtree_size

    metadata_range = RangeSet(

        data=[metadata_start / self.block_size,

              (metadata_start + self.metadata_size) / self.block_size])

    meta_data = ''.join(self.image.ReadRangeSet(metadata_range))

        data=[metadata_start // self.block_size,

              (metadata_start + self.metadata_size) // self.block_size])

    meta_data = b''.join(self.image.ReadRangeSet(metadata_range))

    # More info about the metadata structure available in:

    # system/extras/verity/build_verity_metadata.py

    assert (int(table_entries[5]) * self.block_size == self.filesystem_size and

            int(table_entries[6]) * self.block_size == self.filesystem_size)

    self.hashtree_info.hash_algorithm = table_entries[7]

    self.hashtree_info.root_hash = table_entries[8]

    self.hashtree_info.salt = table_entries[9]

    self.hashtree_info.hash_algorithm = table_entries[7].decode()

    self.hashtree_info.root_hash = table_entries[8].decode()

    self.hashtree_info.salt = table_entries[9].decode()

  def ValidateHashtree(self):

    """Checks that we can reconstruct the verity hash tree."""

    # Reads the generated hash tree and checks if it has the exact same bytes

    # as the one in the sparse image.

    with open(generated_verity_tree, "rb") as fd:

      return fd.read() == ''.join(self.image.ReadRangeSet(

    with open(generated_verity_tree, 'rb') as fd:

      return fd.read() == b''.join(self.image.ReadRangeSet(

          self.hashtree_info.hashtree_range))

  def Generate(self, image):