tools/releasetools/test_validate_target_files.py

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

"""Unittests for validate_target_files.py."""

import os
import os.path
import shutil
import zipfile

import common
import test_utils
from rangelib import RangeSet
from validate_target_files import (ValidateVerifiedBootImages,
                                   ValidateFileConsistency, CheckBuildPropDuplicity)
from verity_utils import CreateVerityImageBuilder

class ValidateTargetFilesTest(test_utils.ReleaseToolsTestCase):

  def setUp(self):
    self.testdata_dir = test_utils.get_testdata_dir()

  def _generate_boot_image(self, output_file):
    kernel = common.MakeTempFile(prefix='kernel-')
    with open(kernel, 'wb') as kernel_fp:
      kernel_fp.write(os.urandom(10))

    cmd = ['mkbootimg', '--kernel', kernel, '-o', output_file]
    proc = common.Run(cmd)
    stdoutdata, _ = proc.communicate()
    self.assertEqual(
        0, proc.returncode,
        "Failed to run mkbootimg: {}".format(stdoutdata))

    cmd = ['boot_signer', '/boot', output_file,
           os.path.join(self.testdata_dir, 'testkey.pk8'),
           os.path.join(self.testdata_dir, 'testkey.x509.pem'), output_file]
    proc = common.Run(cmd)
    stdoutdata, _ = proc.communicate()
    self.assertEqual(
        0, proc.returncode,
        "Failed to sign boot image with boot_signer: {}".format(stdoutdata))

  @test_utils.SkipIfExternalToolsUnavailable()
  def test_ValidateVerifiedBootImages_bootImage(self):
    input_tmp = common.MakeTempDir()
    os.mkdir(os.path.join(input_tmp, 'IMAGES'))
    boot_image = os.path.join(input_tmp, 'IMAGES', 'boot.img')
    self._generate_boot_image(boot_image)

    info_dict = {
        'boot_signer' : 'true',
    }
    options = {
        'verity_key' : os.path.join(self.testdata_dir, 'testkey.x509.pem'),
    }
    ValidateVerifiedBootImages(input_tmp, info_dict, options)

  @test_utils.SkipIfExternalToolsUnavailable()
  def test_ValidateVerifiedBootImages_bootImage_wrongKey(self):
    input_tmp = common.MakeTempDir()
    os.mkdir(os.path.join(input_tmp, 'IMAGES'))
    boot_image = os.path.join(input_tmp, 'IMAGES', 'boot.img')
    self._generate_boot_image(boot_image)

    info_dict = {
        'boot_signer' : 'true',
    }
    options = {
        'verity_key' : os.path.join(self.testdata_dir, 'verity.x509.pem'),
    }
    self.assertRaises(
        AssertionError, ValidateVerifiedBootImages, input_tmp, info_dict,
        options)

  @test_utils.SkipIfExternalToolsUnavailable()
  def test_ValidateVerifiedBootImages_bootImage_corrupted(self):
    input_tmp = common.MakeTempDir()
    os.mkdir(os.path.join(input_tmp, 'IMAGES'))
    boot_image = os.path.join(input_tmp, 'IMAGES', 'boot.img')
    self._generate_boot_image(boot_image)

    # Corrupt the late byte of the image.
    with open(boot_image, 'r+b') as boot_fp:
      boot_fp.seek(-1, os.SEEK_END)
      last_byte = boot_fp.read(1)
      last_byte = bytes([255 - ord(last_byte)])
      boot_fp.seek(-1, os.SEEK_END)
      boot_fp.write(last_byte)

    info_dict = {
        'boot_signer' : 'true',
    }
    options = {
        'verity_key' : os.path.join(self.testdata_dir, 'testkey.x509.pem'),
    }
    self.assertRaises(
        AssertionError, ValidateVerifiedBootImages, input_tmp, info_dict,
        options)

  def _generate_system_image(self, output_file, system_root=None,
                             file_map=None):
    prop_dict = {
        'partition_size': str(1024 * 1024),
        'verity': 'true',
        'verity_block_device': '/dev/block/system',
        'verity_key' : os.path.join(self.testdata_dir, 'testkey'),
        'verity_fec': "true",
        'verity_signer_cmd': 'verity_signer',
    }
    verity_image_builder = CreateVerityImageBuilder(prop_dict)
    image_size = verity_image_builder.CalculateMaxImageSize()

    # Use an empty root directory.
    if not system_root:
      system_root = common.MakeTempDir()
    cmd = ['mkuserimg_mke2fs', '-s', system_root, output_file, 'ext4',
           '/system', str(image_size), '-j', '0']
    if file_map:
      cmd.extend(['-B', file_map])
    proc = common.Run(cmd)
    stdoutdata, _ = proc.communicate()
    self.assertEqual(
        0, proc.returncode,
        "Failed to create system image with mkuserimg_mke2fs: {}".format(
            stdoutdata))

    # Append the verity metadata.
    verity_image_builder.Build(output_file)

  @test_utils.SkipIfExternalToolsUnavailable()
  def test_ValidateVerifiedBootImages_systemRootImage(self):
    input_tmp = common.MakeTempDir()
    os.mkdir(os.path.join(input_tmp, 'IMAGES'))
    system_image = os.path.join(input_tmp, 'IMAGES', 'system.img')
    self._generate_system_image(system_image)

    # Pack the verity key.
    verity_key_mincrypt = os.path.join(input_tmp, 'ROOT', 'verity_key')
    os.makedirs(os.path.dirname(verity_key_mincrypt))
    shutil.copyfile(
        os.path.join(self.testdata_dir, 'testkey_mincrypt'),
        verity_key_mincrypt)

    info_dict = {
        'verity' : 'true',
    }
    options = {
        'verity_key' : os.path.join(self.testdata_dir, 'testkey.x509.pem'),
        'verity_key_mincrypt' : verity_key_mincrypt,
    }
    ValidateVerifiedBootImages(input_tmp, info_dict, options)

  @test_utils.SkipIfExternalToolsUnavailable()
  def test_ValidateVerifiedBootImages_nonSystemRootImage(self):
    input_tmp = common.MakeTempDir()
    os.mkdir(os.path.join(input_tmp, 'IMAGES'))
    system_image = os.path.join(input_tmp, 'IMAGES', 'system.img')
    self._generate_system_image(system_image)

    # Pack the verity key into the root dir in system.img.
    verity_key_mincrypt = os.path.join(input_tmp, 'ROOT', 'verity_key')
    os.makedirs(os.path.dirname(verity_key_mincrypt))
    shutil.copyfile(
        os.path.join(self.testdata_dir, 'testkey_mincrypt'),
        verity_key_mincrypt)

    # And a copy in ramdisk.
    verity_key_ramdisk = os.path.join(
        input_tmp, 'BOOT', 'RAMDISK', 'verity_key')
    os.makedirs(os.path.dirname(verity_key_ramdisk))
    shutil.copyfile(
        os.path.join(self.testdata_dir, 'testkey_mincrypt'),
        verity_key_ramdisk)

    info_dict = {
        'verity' : 'true',
    }
    options = {
        'verity_key' : os.path.join(self.testdata_dir, 'testkey.x509.pem'),
        'verity_key_mincrypt' : verity_key_mincrypt,
    }
    ValidateVerifiedBootImages(input_tmp, info_dict, options)

  @test_utils.SkipIfExternalToolsUnavailable()
  def test_ValidateVerifiedBootImages_nonSystemRootImage_mismatchingKeys(self):
    input_tmp = common.MakeTempDir()
    os.mkdir(os.path.join(input_tmp, 'IMAGES'))
    system_image = os.path.join(input_tmp, 'IMAGES', 'system.img')
    self._generate_system_image(system_image)

    # Pack the verity key into the root dir in system.img.
    verity_key_mincrypt = os.path.join(input_tmp, 'ROOT', 'verity_key')
    os.makedirs(os.path.dirname(verity_key_mincrypt))
    shutil.copyfile(
        os.path.join(self.testdata_dir, 'testkey_mincrypt'),
        verity_key_mincrypt)

    # And an invalid copy in ramdisk.
    verity_key_ramdisk = os.path.join(
        input_tmp, 'BOOT', 'RAMDISK', 'verity_key')
    os.makedirs(os.path.dirname(verity_key_ramdisk))
    shutil.copyfile(
        os.path.join(self.testdata_dir, 'verity_mincrypt'),
        verity_key_ramdisk)

    info_dict = {
        'verity' : 'true',
    }
    options = {
        'verity_key' : os.path.join(self.testdata_dir, 'testkey.x509.pem'),
        'verity_key_mincrypt' : verity_key_mincrypt,
    }
    self.assertRaises(
        AssertionError, ValidateVerifiedBootImages, input_tmp, info_dict,
        options)

  @test_utils.SkipIfExternalToolsUnavailable()
  def test_ValidateFileConsistency_incompleteRange(self):
    input_tmp = common.MakeTempDir()
    os.mkdir(os.path.join(input_tmp, 'IMAGES'))
    system_image = os.path.join(input_tmp, 'IMAGES', 'system.img')
    system_root = os.path.join(input_tmp, "SYSTEM")
    os.mkdir(system_root)

    # Write test files that contain multiple blocks of zeros, and these zero
    # blocks will be omitted by kernel. Each test file will occupy one block in
    # the final system image.
    with open(os.path.join(system_root, 'a'), 'w') as f:
      f.write('aaa')
      f.write('\0' * 4096 * 3)
    with open(os.path.join(system_root, 'b'), 'w') as f:
      f.write('bbb')
      f.write('\0' * 4096 * 3)

    raw_file_map = os.path.join(input_tmp, 'IMAGES', 'raw_system.map')
    self._generate_system_image(system_image, system_root, raw_file_map)

    # Parse the generated file map and update the block ranges for each file.
    file_map_list = {}
    image_ranges = RangeSet()
    with open(raw_file_map) as f:
      for line in f.readlines():
        info = line.split()
        self.assertEqual(2, len(info))
        image_ranges = image_ranges.union(RangeSet(info[1]))
        file_map_list[info[0]] = RangeSet(info[1])

    # Add one unoccupied block as the shared block for all test files.
    mock_shared_block = RangeSet("10-20").subtract(image_ranges).first(1)
    with open(os.path.join(input_tmp, 'IMAGES', 'system.map'), 'w') as f:
      for key in sorted(file_map_list.keys()):
        line = '{} {}\n'.format(
            key, file_map_list[key].union(mock_shared_block))
        f.write(line)

    # Prepare for the target zip file
    input_file = common.MakeTempFile()
    all_entries = ['SYSTEM/', 'SYSTEM/b', 'SYSTEM/a', 'IMAGES/',
                   'IMAGES/system.map', 'IMAGES/system.img']
    with zipfile.ZipFile(input_file, 'w', allowZip64=True) as input_zip:
      for name in all_entries:
        input_zip.write(os.path.join(input_tmp, name), arcname=name)

    # Expect the validation to pass and both files are skipped due to
    # 'incomplete' block range.
    with zipfile.ZipFile(input_file) as input_zip:
      info_dict = {'extfs_sparse_flag': '-s'}
      ValidateFileConsistency(input_zip, input_tmp, info_dict)

  @test_utils.SkipIfExternalToolsUnavailable()
  def test_ValidateFileConsistency_nonMonotonicRanges(self):
    input_tmp = common.MakeTempDir()
    os.mkdir(os.path.join(input_tmp, 'IMAGES'))
    system_image = os.path.join(input_tmp, 'IMAGES', 'system.img')
    system_root = os.path.join(input_tmp, "SYSTEM")
    os.mkdir(system_root)

    # Write the test file that contain three blocks of 'a', 'b', 'c'.
    with open(os.path.join(system_root, 'abc'), 'w') as f:
      f.write('a' * 4096 + 'b' * 4096 + 'c' * 4096)
    raw_file_map = os.path.join(input_tmp, 'IMAGES', 'raw_system.map')
    self._generate_system_image(system_image, system_root, raw_file_map)

    # Parse the generated file map and manipulate the block ranges of 'abc' to
    # be 'cba'.
    file_map_list = {}
    with open(raw_file_map) as f:
      for line in f.readlines():
        info = line.split()
        self.assertEqual(2, len(info))
        ranges = RangeSet(info[1])
        self.assertTrue(ranges.monotonic)
        blocks = reversed(list(ranges.next_item()))
        file_map_list[info[0]] = ' '.join([str(block) for block in blocks])

    # Update the contents of 'abc' to be 'cba'.
    with open(os.path.join(system_root, 'abc'), 'w') as f:
      f.write('c' * 4096 + 'b' * 4096 + 'a' * 4096)

    # Update the system.map.
    with open(os.path.join(input_tmp, 'IMAGES', 'system.map'), 'w') as f:
      for key in sorted(file_map_list.keys()):
        f.write('{} {}\n'.format(key, file_map_list[key]))

    # Get the target zip file.
    input_file = common.MakeTempFile()
    all_entries = ['SYSTEM/', 'SYSTEM/abc', 'IMAGES/',
                   'IMAGES/system.map', 'IMAGES/system.img']
    with zipfile.ZipFile(input_file, 'w', allowZip64=True) as input_zip:
      for name in all_entries:
        input_zip.write(os.path.join(input_tmp, name), arcname=name)

    with zipfile.ZipFile(input_file) as input_zip:
      info_dict = {'extfs_sparse_flag': '-s'}
      ValidateFileConsistency(input_zip, input_tmp, info_dict)

  @staticmethod
  def make_build_prop(build_prop):
    input_tmp = common.MakeTempDir()
    system_dir = os.path.join(input_tmp, 'SYSTEM')
    os.makedirs(system_dir)
    prop_file = os.path.join(system_dir, 'build.prop')
    with open(prop_file, 'w') as output_file:
      output_file.write("\n".join(build_prop))
    return input_tmp

  def test_checkDuplicateProps_noDuplicate(self):
    build_prop = [
        'ro.odm.build.date.utc=1578430045',
        'ro.odm.build.fingerprint='
        'google/coral/coral:10/RP1A.200325.001/6337676:user/dev-keys',
        'ro.product.odm.device=coral',
    ]
    input_tmp = ValidateTargetFilesTest.make_build_prop(build_prop)
    CheckBuildPropDuplicity(input_tmp)

  def test_checkDuplicateProps_withDuplicate(self):
    build_prop = [
        'ro.odm.build.date.utc=1578430045',
        'ro.odm.build.date.utc=1578430049',
        'ro.odm.build.fingerprint='
        'google/coral/coral:10/RP1A.200325.001/6337676:user/dev-keys',
        'ro.product.odm.device=coral',
    ]
    input_tmp = ValidateTargetFilesTest.make_build_prop(build_prop)

    self.assertRaises(ValueError, CheckBuildPropDuplicity, input_tmp)