1318 lines
49 KiB
Python
1318 lines
49 KiB
Python
#
|
|
# 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.
|
|
#
|
|
|
|
import copy
|
|
import os
|
|
import os.path
|
|
import zipfile
|
|
|
|
import common
|
|
import test_utils
|
|
from ota_from_target_files import (
|
|
_LoadOemDicts, AbOtaPropertyFiles, FinalizeMetadata,
|
|
GetPackageMetadata, GetTargetFilesZipForSecondaryImages,
|
|
GetTargetFilesZipWithoutPostinstallConfig, NonAbOtaPropertyFiles,
|
|
Payload, PayloadSigner, POSTINSTALL_CONFIG, PropertyFiles,
|
|
StreamingPropertyFiles, WriteFingerprintAssertion)
|
|
|
|
|
|
def construct_target_files(secondary=False):
|
|
"""Returns a target-files.zip file for generating OTA packages."""
|
|
target_files = common.MakeTempFile(prefix='target_files-', suffix='.zip')
|
|
with zipfile.ZipFile(target_files, 'w') as target_files_zip:
|
|
# META/update_engine_config.txt
|
|
target_files_zip.writestr(
|
|
'META/update_engine_config.txt',
|
|
"PAYLOAD_MAJOR_VERSION=2\nPAYLOAD_MINOR_VERSION=4\n")
|
|
|
|
# META/postinstall_config.txt
|
|
target_files_zip.writestr(
|
|
POSTINSTALL_CONFIG,
|
|
'\n'.join([
|
|
"RUN_POSTINSTALL_system=true",
|
|
"POSTINSTALL_PATH_system=system/bin/otapreopt_script",
|
|
"FILESYSTEM_TYPE_system=ext4",
|
|
"POSTINSTALL_OPTIONAL_system=true",
|
|
]))
|
|
|
|
ab_partitions = [
|
|
('IMAGES', 'boot'),
|
|
('IMAGES', 'system'),
|
|
('IMAGES', 'vendor'),
|
|
('RADIO', 'bootloader'),
|
|
('RADIO', 'modem'),
|
|
]
|
|
# META/ab_partitions.txt
|
|
target_files_zip.writestr(
|
|
'META/ab_partitions.txt',
|
|
'\n'.join([partition[1] for partition in ab_partitions]))
|
|
|
|
# Create dummy images for each of them.
|
|
for path, partition in ab_partitions:
|
|
target_files_zip.writestr(
|
|
'{}/{}.img'.format(path, partition),
|
|
os.urandom(len(partition)))
|
|
|
|
# system_other shouldn't appear in META/ab_partitions.txt.
|
|
if secondary:
|
|
target_files_zip.writestr('IMAGES/system_other.img',
|
|
os.urandom(len("system_other")))
|
|
|
|
return target_files
|
|
|
|
|
|
class LoadOemDictsTest(test_utils.ReleaseToolsTestCase):
|
|
|
|
def test_NoneDict(self):
|
|
self.assertIsNone(_LoadOemDicts(None))
|
|
|
|
def test_SingleDict(self):
|
|
dict_file = common.MakeTempFile()
|
|
with open(dict_file, 'w') as dict_fp:
|
|
dict_fp.write('abc=1\ndef=2\nxyz=foo\na.b.c=bar\n')
|
|
|
|
oem_dicts = _LoadOemDicts([dict_file])
|
|
self.assertEqual(1, len(oem_dicts))
|
|
self.assertEqual('foo', oem_dicts[0]['xyz'])
|
|
self.assertEqual('bar', oem_dicts[0]['a.b.c'])
|
|
|
|
def test_MultipleDicts(self):
|
|
oem_source = []
|
|
for i in range(3):
|
|
dict_file = common.MakeTempFile()
|
|
with open(dict_file, 'w') as dict_fp:
|
|
dict_fp.write(
|
|
'ro.build.index={}\ndef=2\nxyz=foo\na.b.c=bar\n'.format(i))
|
|
oem_source.append(dict_file)
|
|
|
|
oem_dicts = _LoadOemDicts(oem_source)
|
|
self.assertEqual(3, len(oem_dicts))
|
|
for i, oem_dict in enumerate(oem_dicts):
|
|
self.assertEqual('2', oem_dict['def'])
|
|
self.assertEqual('foo', oem_dict['xyz'])
|
|
self.assertEqual('bar', oem_dict['a.b.c'])
|
|
self.assertEqual('{}'.format(i), oem_dict['ro.build.index'])
|
|
|
|
|
|
class OtaFromTargetFilesTest(test_utils.ReleaseToolsTestCase):
|
|
|
|
TEST_TARGET_INFO_DICT = {
|
|
'build.prop' : {
|
|
'ro.product.device' : 'product-device',
|
|
'ro.build.fingerprint' : 'build-fingerprint-target',
|
|
'ro.build.version.incremental' : 'build-version-incremental-target',
|
|
'ro.build.version.sdk' : '27',
|
|
'ro.build.version.security_patch' : '2017-12-01',
|
|
'ro.build.date.utc' : '1500000000',
|
|
},
|
|
}
|
|
|
|
TEST_SOURCE_INFO_DICT = {
|
|
'build.prop' : {
|
|
'ro.product.device' : 'product-device',
|
|
'ro.build.fingerprint' : 'build-fingerprint-source',
|
|
'ro.build.version.incremental' : 'build-version-incremental-source',
|
|
'ro.build.version.sdk' : '25',
|
|
'ro.build.version.security_patch' : '2016-12-01',
|
|
'ro.build.date.utc' : '1400000000',
|
|
},
|
|
}
|
|
|
|
TEST_INFO_DICT_USES_OEM_PROPS = {
|
|
'build.prop' : {
|
|
'ro.product.name' : 'product-name',
|
|
'ro.build.thumbprint' : 'build-thumbprint',
|
|
'ro.build.bar' : 'build-bar',
|
|
},
|
|
'vendor.build.prop' : {
|
|
'ro.vendor.build.fingerprint' : 'vendor-build-fingerprint',
|
|
},
|
|
'property1' : 'value1',
|
|
'property2' : 4096,
|
|
'oem_fingerprint_properties' : 'ro.product.device ro.product.brand',
|
|
}
|
|
|
|
TEST_OEM_DICTS = [
|
|
{
|
|
'ro.product.brand' : 'brand1',
|
|
'ro.product.device' : 'device1',
|
|
},
|
|
{
|
|
'ro.product.brand' : 'brand2',
|
|
'ro.product.device' : 'device2',
|
|
},
|
|
{
|
|
'ro.product.brand' : 'brand3',
|
|
'ro.product.device' : 'device3',
|
|
},
|
|
]
|
|
|
|
def setUp(self):
|
|
self.testdata_dir = test_utils.get_testdata_dir()
|
|
self.assertTrue(os.path.exists(self.testdata_dir))
|
|
|
|
# Reset the global options as in ota_from_target_files.py.
|
|
common.OPTIONS.incremental_source = None
|
|
common.OPTIONS.downgrade = False
|
|
common.OPTIONS.retrofit_dynamic_partitions = False
|
|
common.OPTIONS.timestamp = False
|
|
common.OPTIONS.wipe_user_data = False
|
|
common.OPTIONS.no_signing = False
|
|
common.OPTIONS.package_key = os.path.join(self.testdata_dir, 'testkey')
|
|
common.OPTIONS.key_passwords = {
|
|
common.OPTIONS.package_key : None,
|
|
}
|
|
|
|
common.OPTIONS.search_path = test_utils.get_search_path()
|
|
|
|
def test_GetPackageMetadata_abOta_full(self):
|
|
target_info_dict = copy.deepcopy(self.TEST_TARGET_INFO_DICT)
|
|
target_info_dict['ab_update'] = 'true'
|
|
target_info = common.BuildInfo(target_info_dict, None)
|
|
metadata = GetPackageMetadata(target_info)
|
|
self.assertDictEqual(
|
|
{
|
|
'ota-type' : 'AB',
|
|
'ota-required-cache' : '0',
|
|
'post-build' : 'build-fingerprint-target',
|
|
'post-build-incremental' : 'build-version-incremental-target',
|
|
'post-sdk-level' : '27',
|
|
'post-security-patch-level' : '2017-12-01',
|
|
'post-timestamp' : '1500000000',
|
|
'pre-device' : 'product-device',
|
|
},
|
|
metadata)
|
|
|
|
def test_GetPackageMetadata_abOta_incremental(self):
|
|
target_info_dict = copy.deepcopy(self.TEST_TARGET_INFO_DICT)
|
|
target_info_dict['ab_update'] = 'true'
|
|
target_info = common.BuildInfo(target_info_dict, None)
|
|
source_info = common.BuildInfo(self.TEST_SOURCE_INFO_DICT, None)
|
|
common.OPTIONS.incremental_source = ''
|
|
metadata = GetPackageMetadata(target_info, source_info)
|
|
self.assertDictEqual(
|
|
{
|
|
'ota-type' : 'AB',
|
|
'ota-required-cache' : '0',
|
|
'post-build' : 'build-fingerprint-target',
|
|
'post-build-incremental' : 'build-version-incremental-target',
|
|
'post-sdk-level' : '27',
|
|
'post-security-patch-level' : '2017-12-01',
|
|
'post-timestamp' : '1500000000',
|
|
'pre-device' : 'product-device',
|
|
'pre-build' : 'build-fingerprint-source',
|
|
'pre-build-incremental' : 'build-version-incremental-source',
|
|
},
|
|
metadata)
|
|
|
|
def test_GetPackageMetadata_nonAbOta_full(self):
|
|
target_info = common.BuildInfo(self.TEST_TARGET_INFO_DICT, None)
|
|
metadata = GetPackageMetadata(target_info)
|
|
self.assertDictEqual(
|
|
{
|
|
'ota-type' : 'BLOCK',
|
|
'post-build' : 'build-fingerprint-target',
|
|
'post-build-incremental' : 'build-version-incremental-target',
|
|
'post-sdk-level' : '27',
|
|
'post-security-patch-level' : '2017-12-01',
|
|
'post-timestamp' : '1500000000',
|
|
'pre-device' : 'product-device',
|
|
},
|
|
metadata)
|
|
|
|
def test_GetPackageMetadata_nonAbOta_incremental(self):
|
|
target_info = common.BuildInfo(self.TEST_TARGET_INFO_DICT, None)
|
|
source_info = common.BuildInfo(self.TEST_SOURCE_INFO_DICT, None)
|
|
common.OPTIONS.incremental_source = ''
|
|
metadata = GetPackageMetadata(target_info, source_info)
|
|
self.assertDictEqual(
|
|
{
|
|
'ota-type' : 'BLOCK',
|
|
'post-build' : 'build-fingerprint-target',
|
|
'post-build-incremental' : 'build-version-incremental-target',
|
|
'post-sdk-level' : '27',
|
|
'post-security-patch-level' : '2017-12-01',
|
|
'post-timestamp' : '1500000000',
|
|
'pre-device' : 'product-device',
|
|
'pre-build' : 'build-fingerprint-source',
|
|
'pre-build-incremental' : 'build-version-incremental-source',
|
|
},
|
|
metadata)
|
|
|
|
def test_GetPackageMetadata_wipe(self):
|
|
target_info = common.BuildInfo(self.TEST_TARGET_INFO_DICT, None)
|
|
common.OPTIONS.wipe_user_data = True
|
|
metadata = GetPackageMetadata(target_info)
|
|
self.assertDictEqual(
|
|
{
|
|
'ota-type' : 'BLOCK',
|
|
'ota-wipe' : 'yes',
|
|
'post-build' : 'build-fingerprint-target',
|
|
'post-build-incremental' : 'build-version-incremental-target',
|
|
'post-sdk-level' : '27',
|
|
'post-security-patch-level' : '2017-12-01',
|
|
'post-timestamp' : '1500000000',
|
|
'pre-device' : 'product-device',
|
|
},
|
|
metadata)
|
|
|
|
def test_GetPackageMetadata_retrofitDynamicPartitions(self):
|
|
target_info = common.BuildInfo(self.TEST_TARGET_INFO_DICT, None)
|
|
common.OPTIONS.retrofit_dynamic_partitions = True
|
|
metadata = GetPackageMetadata(target_info)
|
|
self.assertDictEqual(
|
|
{
|
|
'ota-retrofit-dynamic-partitions' : 'yes',
|
|
'ota-type' : 'BLOCK',
|
|
'post-build' : 'build-fingerprint-target',
|
|
'post-build-incremental' : 'build-version-incremental-target',
|
|
'post-sdk-level' : '27',
|
|
'post-security-patch-level' : '2017-12-01',
|
|
'post-timestamp' : '1500000000',
|
|
'pre-device' : 'product-device',
|
|
},
|
|
metadata)
|
|
|
|
@staticmethod
|
|
def _test_GetPackageMetadata_swapBuildTimestamps(target_info, source_info):
|
|
(target_info['build.prop']['ro.build.date.utc'],
|
|
source_info['build.prop']['ro.build.date.utc']) = (
|
|
source_info['build.prop']['ro.build.date.utc'],
|
|
target_info['build.prop']['ro.build.date.utc'])
|
|
|
|
def test_GetPackageMetadata_unintentionalDowngradeDetected(self):
|
|
target_info_dict = copy.deepcopy(self.TEST_TARGET_INFO_DICT)
|
|
source_info_dict = copy.deepcopy(self.TEST_SOURCE_INFO_DICT)
|
|
self._test_GetPackageMetadata_swapBuildTimestamps(
|
|
target_info_dict, source_info_dict)
|
|
|
|
target_info = common.BuildInfo(target_info_dict, None)
|
|
source_info = common.BuildInfo(source_info_dict, None)
|
|
common.OPTIONS.incremental_source = ''
|
|
self.assertRaises(RuntimeError, GetPackageMetadata, target_info,
|
|
source_info)
|
|
|
|
def test_GetPackageMetadata_downgrade(self):
|
|
target_info_dict = copy.deepcopy(self.TEST_TARGET_INFO_DICT)
|
|
source_info_dict = copy.deepcopy(self.TEST_SOURCE_INFO_DICT)
|
|
self._test_GetPackageMetadata_swapBuildTimestamps(
|
|
target_info_dict, source_info_dict)
|
|
|
|
target_info = common.BuildInfo(target_info_dict, None)
|
|
source_info = common.BuildInfo(source_info_dict, None)
|
|
common.OPTIONS.incremental_source = ''
|
|
common.OPTIONS.downgrade = True
|
|
common.OPTIONS.wipe_user_data = True
|
|
metadata = GetPackageMetadata(target_info, source_info)
|
|
self.assertDictEqual(
|
|
{
|
|
'ota-downgrade' : 'yes',
|
|
'ota-type' : 'BLOCK',
|
|
'ota-wipe' : 'yes',
|
|
'post-build' : 'build-fingerprint-target',
|
|
'post-build-incremental' : 'build-version-incremental-target',
|
|
'post-sdk-level' : '27',
|
|
'post-security-patch-level' : '2017-12-01',
|
|
'post-timestamp' : '1400000000',
|
|
'pre-device' : 'product-device',
|
|
'pre-build' : 'build-fingerprint-source',
|
|
'pre-build-incremental' : 'build-version-incremental-source',
|
|
},
|
|
metadata)
|
|
|
|
@test_utils.SkipIfExternalToolsUnavailable()
|
|
def test_GetTargetFilesZipForSecondaryImages(self):
|
|
input_file = construct_target_files(secondary=True)
|
|
target_file = GetTargetFilesZipForSecondaryImages(input_file)
|
|
|
|
with zipfile.ZipFile(target_file) as verify_zip:
|
|
namelist = verify_zip.namelist()
|
|
ab_partitions = verify_zip.read('META/ab_partitions.txt').decode()
|
|
|
|
self.assertIn('META/ab_partitions.txt', namelist)
|
|
self.assertIn('IMAGES/system.img', namelist)
|
|
self.assertIn('RADIO/bootloader.img', namelist)
|
|
self.assertIn(POSTINSTALL_CONFIG, namelist)
|
|
|
|
self.assertNotIn('IMAGES/boot.img', namelist)
|
|
self.assertNotIn('IMAGES/system_other.img', namelist)
|
|
self.assertNotIn('IMAGES/system.map', namelist)
|
|
self.assertNotIn('RADIO/modem.img', namelist)
|
|
|
|
expected_ab_partitions = ['system', 'bootloader']
|
|
self.assertEqual('\n'.join(expected_ab_partitions), ab_partitions)
|
|
|
|
@test_utils.SkipIfExternalToolsUnavailable()
|
|
def test_GetTargetFilesZipForSecondaryImages_skipPostinstall(self):
|
|
input_file = construct_target_files(secondary=True)
|
|
target_file = GetTargetFilesZipForSecondaryImages(
|
|
input_file, skip_postinstall=True)
|
|
|
|
with zipfile.ZipFile(target_file) as verify_zip:
|
|
namelist = verify_zip.namelist()
|
|
|
|
self.assertIn('META/ab_partitions.txt', namelist)
|
|
self.assertIn('IMAGES/system.img', namelist)
|
|
self.assertIn('RADIO/bootloader.img', namelist)
|
|
|
|
self.assertNotIn('IMAGES/boot.img', namelist)
|
|
self.assertNotIn('IMAGES/system_other.img', namelist)
|
|
self.assertNotIn('IMAGES/system.map', namelist)
|
|
self.assertNotIn('RADIO/modem.img', namelist)
|
|
self.assertNotIn(POSTINSTALL_CONFIG, namelist)
|
|
|
|
@test_utils.SkipIfExternalToolsUnavailable()
|
|
def test_GetTargetFilesZipForSecondaryImages_withoutRadioImages(self):
|
|
input_file = construct_target_files(secondary=True)
|
|
common.ZipDelete(input_file, 'RADIO/bootloader.img')
|
|
common.ZipDelete(input_file, 'RADIO/modem.img')
|
|
target_file = GetTargetFilesZipForSecondaryImages(input_file)
|
|
|
|
with zipfile.ZipFile(target_file) as verify_zip:
|
|
namelist = verify_zip.namelist()
|
|
|
|
self.assertIn('META/ab_partitions.txt', namelist)
|
|
self.assertIn('IMAGES/system.img', namelist)
|
|
self.assertIn(POSTINSTALL_CONFIG, namelist)
|
|
|
|
self.assertNotIn('IMAGES/boot.img', namelist)
|
|
self.assertNotIn('IMAGES/system_other.img', namelist)
|
|
self.assertNotIn('IMAGES/system.map', namelist)
|
|
self.assertNotIn('RADIO/bootloader.img', namelist)
|
|
self.assertNotIn('RADIO/modem.img', namelist)
|
|
|
|
@test_utils.SkipIfExternalToolsUnavailable()
|
|
def test_GetTargetFilesZipForSecondaryImages_dynamicPartitions(self):
|
|
input_file = construct_target_files(secondary=True)
|
|
misc_info = '\n'.join([
|
|
'use_dynamic_partition_size=true',
|
|
'use_dynamic_partitions=true',
|
|
'dynamic_partition_list=system vendor product',
|
|
'super_partition_groups=google_dynamic_partitions',
|
|
'super_google_dynamic_partitions_group_size=4873781248',
|
|
'super_google_dynamic_partitions_partition_list=system vendor product',
|
|
])
|
|
dynamic_partitions_info = '\n'.join([
|
|
'super_partition_groups=google_dynamic_partitions',
|
|
'super_google_dynamic_partitions_group_size=4873781248',
|
|
'super_google_dynamic_partitions_partition_list=system vendor product',
|
|
])
|
|
|
|
with zipfile.ZipFile(input_file, 'a') as append_zip:
|
|
common.ZipWriteStr(append_zip, 'META/misc_info.txt', misc_info)
|
|
common.ZipWriteStr(append_zip, 'META/dynamic_partitions_info.txt',
|
|
dynamic_partitions_info)
|
|
|
|
target_file = GetTargetFilesZipForSecondaryImages(input_file)
|
|
|
|
with zipfile.ZipFile(target_file) as verify_zip:
|
|
namelist = verify_zip.namelist()
|
|
updated_misc_info = verify_zip.read('META/misc_info.txt').decode()
|
|
updated_dynamic_partitions_info = verify_zip.read(
|
|
'META/dynamic_partitions_info.txt').decode()
|
|
|
|
self.assertIn('META/ab_partitions.txt', namelist)
|
|
self.assertIn('IMAGES/system.img', namelist)
|
|
self.assertIn(POSTINSTALL_CONFIG, namelist)
|
|
self.assertIn('META/misc_info.txt', namelist)
|
|
self.assertIn('META/dynamic_partitions_info.txt', namelist)
|
|
|
|
self.assertNotIn('IMAGES/boot.img', namelist)
|
|
self.assertNotIn('IMAGES/system_other.img', namelist)
|
|
self.assertNotIn('IMAGES/system.map', namelist)
|
|
|
|
# Check the vendor & product are removed from the partitions list.
|
|
expected_misc_info = misc_info.replace('system vendor product',
|
|
'system')
|
|
expected_dynamic_partitions_info = dynamic_partitions_info.replace(
|
|
'system vendor product', 'system')
|
|
self.assertEqual(expected_misc_info, updated_misc_info)
|
|
self.assertEqual(expected_dynamic_partitions_info,
|
|
updated_dynamic_partitions_info)
|
|
|
|
@test_utils.SkipIfExternalToolsUnavailable()
|
|
def test_GetTargetFilesZipWithoutPostinstallConfig(self):
|
|
input_file = construct_target_files()
|
|
target_file = GetTargetFilesZipWithoutPostinstallConfig(input_file)
|
|
with zipfile.ZipFile(target_file) as verify_zip:
|
|
self.assertNotIn(POSTINSTALL_CONFIG, verify_zip.namelist())
|
|
|
|
@test_utils.SkipIfExternalToolsUnavailable()
|
|
def test_GetTargetFilesZipWithoutPostinstallConfig_missingEntry(self):
|
|
input_file = construct_target_files()
|
|
common.ZipDelete(input_file, POSTINSTALL_CONFIG)
|
|
target_file = GetTargetFilesZipWithoutPostinstallConfig(input_file)
|
|
with zipfile.ZipFile(target_file) as verify_zip:
|
|
self.assertNotIn(POSTINSTALL_CONFIG, verify_zip.namelist())
|
|
|
|
def _test_FinalizeMetadata(self, large_entry=False):
|
|
entries = [
|
|
'required-entry1',
|
|
'required-entry2',
|
|
]
|
|
zip_file = PropertyFilesTest.construct_zip_package(entries)
|
|
# Add a large entry of 1 GiB if requested.
|
|
if large_entry:
|
|
with zipfile.ZipFile(zip_file, 'a') as zip_fp:
|
|
zip_fp.writestr(
|
|
# Using 'zoo' so that the entry stays behind others after signing.
|
|
'zoo',
|
|
'A' * 1024 * 1024 * 1024,
|
|
zipfile.ZIP_STORED)
|
|
|
|
metadata = {}
|
|
output_file = common.MakeTempFile(suffix='.zip')
|
|
needed_property_files = (
|
|
TestPropertyFiles(),
|
|
)
|
|
FinalizeMetadata(metadata, zip_file, output_file, needed_property_files)
|
|
self.assertIn('ota-test-property-files', metadata)
|
|
|
|
@test_utils.SkipIfExternalToolsUnavailable()
|
|
def test_FinalizeMetadata(self):
|
|
self._test_FinalizeMetadata()
|
|
|
|
@test_utils.SkipIfExternalToolsUnavailable()
|
|
def test_FinalizeMetadata_withNoSigning(self):
|
|
common.OPTIONS.no_signing = True
|
|
self._test_FinalizeMetadata()
|
|
|
|
@test_utils.SkipIfExternalToolsUnavailable()
|
|
def test_FinalizeMetadata_largeEntry(self):
|
|
self._test_FinalizeMetadata(large_entry=True)
|
|
|
|
@test_utils.SkipIfExternalToolsUnavailable()
|
|
def test_FinalizeMetadata_largeEntry_withNoSigning(self):
|
|
common.OPTIONS.no_signing = True
|
|
self._test_FinalizeMetadata(large_entry=True)
|
|
|
|
@test_utils.SkipIfExternalToolsUnavailable()
|
|
def test_FinalizeMetadata_insufficientSpace(self):
|
|
entries = [
|
|
'required-entry1',
|
|
'required-entry2',
|
|
'optional-entry1',
|
|
'optional-entry2',
|
|
]
|
|
zip_file = PropertyFilesTest.construct_zip_package(entries)
|
|
with zipfile.ZipFile(zip_file, 'a') as zip_fp:
|
|
zip_fp.writestr(
|
|
# 'foo-entry1' will appear ahead of all other entries (in alphabetical
|
|
# order) after the signing, which will in turn trigger the
|
|
# InsufficientSpaceException and an automatic retry.
|
|
'foo-entry1',
|
|
'A' * 1024 * 1024,
|
|
zipfile.ZIP_STORED)
|
|
|
|
metadata = {}
|
|
needed_property_files = (
|
|
TestPropertyFiles(),
|
|
)
|
|
output_file = common.MakeTempFile(suffix='.zip')
|
|
FinalizeMetadata(metadata, zip_file, output_file, needed_property_files)
|
|
self.assertIn('ota-test-property-files', metadata)
|
|
|
|
def test_WriteFingerprintAssertion_without_oem_props(self):
|
|
target_info = common.BuildInfo(self.TEST_TARGET_INFO_DICT, None)
|
|
source_info_dict = copy.deepcopy(self.TEST_TARGET_INFO_DICT)
|
|
source_info_dict['build.prop']['ro.build.fingerprint'] = (
|
|
'source-build-fingerprint')
|
|
source_info = common.BuildInfo(source_info_dict, None)
|
|
|
|
script_writer = test_utils.MockScriptWriter()
|
|
WriteFingerprintAssertion(script_writer, target_info, source_info)
|
|
self.assertEqual(
|
|
[('AssertSomeFingerprint', 'source-build-fingerprint',
|
|
'build-fingerprint-target')],
|
|
script_writer.lines)
|
|
|
|
def test_WriteFingerprintAssertion_with_source_oem_props(self):
|
|
target_info = common.BuildInfo(self.TEST_TARGET_INFO_DICT, None)
|
|
source_info = common.BuildInfo(self.TEST_INFO_DICT_USES_OEM_PROPS,
|
|
self.TEST_OEM_DICTS)
|
|
|
|
script_writer = test_utils.MockScriptWriter()
|
|
WriteFingerprintAssertion(script_writer, target_info, source_info)
|
|
self.assertEqual(
|
|
[('AssertFingerprintOrThumbprint', 'build-fingerprint-target',
|
|
'build-thumbprint')],
|
|
script_writer.lines)
|
|
|
|
def test_WriteFingerprintAssertion_with_target_oem_props(self):
|
|
target_info = common.BuildInfo(self.TEST_INFO_DICT_USES_OEM_PROPS,
|
|
self.TEST_OEM_DICTS)
|
|
source_info = common.BuildInfo(self.TEST_TARGET_INFO_DICT, None)
|
|
|
|
script_writer = test_utils.MockScriptWriter()
|
|
WriteFingerprintAssertion(script_writer, target_info, source_info)
|
|
self.assertEqual(
|
|
[('AssertFingerprintOrThumbprint', 'build-fingerprint-target',
|
|
'build-thumbprint')],
|
|
script_writer.lines)
|
|
|
|
def test_WriteFingerprintAssertion_with_both_oem_props(self):
|
|
target_info = common.BuildInfo(self.TEST_INFO_DICT_USES_OEM_PROPS,
|
|
self.TEST_OEM_DICTS)
|
|
source_info_dict = copy.deepcopy(self.TEST_INFO_DICT_USES_OEM_PROPS)
|
|
source_info_dict['build.prop']['ro.build.thumbprint'] = (
|
|
'source-build-thumbprint')
|
|
source_info = common.BuildInfo(source_info_dict, self.TEST_OEM_DICTS)
|
|
|
|
script_writer = test_utils.MockScriptWriter()
|
|
WriteFingerprintAssertion(script_writer, target_info, source_info)
|
|
self.assertEqual(
|
|
[('AssertSomeThumbprint', 'build-thumbprint',
|
|
'source-build-thumbprint')],
|
|
script_writer.lines)
|
|
|
|
|
|
class TestPropertyFiles(PropertyFiles):
|
|
"""A class that extends PropertyFiles for testing purpose."""
|
|
|
|
def __init__(self):
|
|
super(TestPropertyFiles, self).__init__()
|
|
self.name = 'ota-test-property-files'
|
|
self.required = (
|
|
'required-entry1',
|
|
'required-entry2',
|
|
)
|
|
self.optional = (
|
|
'optional-entry1',
|
|
'optional-entry2',
|
|
)
|
|
|
|
|
|
class PropertyFilesTest(test_utils.ReleaseToolsTestCase):
|
|
|
|
def setUp(self):
|
|
common.OPTIONS.no_signing = False
|
|
|
|
@staticmethod
|
|
def construct_zip_package(entries):
|
|
zip_file = common.MakeTempFile(suffix='.zip')
|
|
with zipfile.ZipFile(zip_file, 'w') as zip_fp:
|
|
for entry in entries:
|
|
zip_fp.writestr(
|
|
entry,
|
|
entry.replace('.', '-').upper(),
|
|
zipfile.ZIP_STORED)
|
|
return zip_file
|
|
|
|
@staticmethod
|
|
def _parse_property_files_string(data):
|
|
result = {}
|
|
for token in data.split(','):
|
|
name, info = token.split(':', 1)
|
|
result[name] = info
|
|
return result
|
|
|
|
def _verify_entries(self, input_file, tokens, entries):
|
|
for entry in entries:
|
|
offset, size = map(int, tokens[entry].split(':'))
|
|
with open(input_file, 'rb') as input_fp:
|
|
input_fp.seek(offset)
|
|
if entry == 'metadata':
|
|
expected = b'META-INF/COM/ANDROID/METADATA'
|
|
else:
|
|
expected = entry.replace('.', '-').upper().encode()
|
|
self.assertEqual(expected, input_fp.read(size))
|
|
|
|
@test_utils.SkipIfExternalToolsUnavailable()
|
|
def test_Compute(self):
|
|
entries = (
|
|
'required-entry1',
|
|
'required-entry2',
|
|
)
|
|
zip_file = self.construct_zip_package(entries)
|
|
property_files = TestPropertyFiles()
|
|
with zipfile.ZipFile(zip_file, 'r') as zip_fp:
|
|
property_files_string = property_files.Compute(zip_fp)
|
|
|
|
tokens = self._parse_property_files_string(property_files_string)
|
|
self.assertEqual(3, len(tokens))
|
|
self._verify_entries(zip_file, tokens, entries)
|
|
|
|
def test_Compute_withOptionalEntries(self):
|
|
entries = (
|
|
'required-entry1',
|
|
'required-entry2',
|
|
'optional-entry1',
|
|
'optional-entry2',
|
|
)
|
|
zip_file = self.construct_zip_package(entries)
|
|
property_files = TestPropertyFiles()
|
|
with zipfile.ZipFile(zip_file, 'r') as zip_fp:
|
|
property_files_string = property_files.Compute(zip_fp)
|
|
|
|
tokens = self._parse_property_files_string(property_files_string)
|
|
self.assertEqual(5, len(tokens))
|
|
self._verify_entries(zip_file, tokens, entries)
|
|
|
|
def test_Compute_missingRequiredEntry(self):
|
|
entries = (
|
|
'required-entry2',
|
|
)
|
|
zip_file = self.construct_zip_package(entries)
|
|
property_files = TestPropertyFiles()
|
|
with zipfile.ZipFile(zip_file, 'r') as zip_fp:
|
|
self.assertRaises(KeyError, property_files.Compute, zip_fp)
|
|
|
|
@test_utils.SkipIfExternalToolsUnavailable()
|
|
def test_Finalize(self):
|
|
entries = [
|
|
'required-entry1',
|
|
'required-entry2',
|
|
'META-INF/com/android/metadata',
|
|
]
|
|
zip_file = self.construct_zip_package(entries)
|
|
property_files = TestPropertyFiles()
|
|
with zipfile.ZipFile(zip_file, 'r') as zip_fp:
|
|
raw_metadata = property_files.GetPropertyFilesString(
|
|
zip_fp, reserve_space=False)
|
|
streaming_metadata = property_files.Finalize(zip_fp, len(raw_metadata))
|
|
tokens = self._parse_property_files_string(streaming_metadata)
|
|
|
|
self.assertEqual(3, len(tokens))
|
|
# 'META-INF/com/android/metadata' will be key'd as 'metadata' in the
|
|
# streaming metadata.
|
|
entries[2] = 'metadata'
|
|
self._verify_entries(zip_file, tokens, entries)
|
|
|
|
@test_utils.SkipIfExternalToolsUnavailable()
|
|
def test_Finalize_assertReservedLength(self):
|
|
entries = (
|
|
'required-entry1',
|
|
'required-entry2',
|
|
'optional-entry1',
|
|
'optional-entry2',
|
|
'META-INF/com/android/metadata',
|
|
)
|
|
zip_file = self.construct_zip_package(entries)
|
|
property_files = TestPropertyFiles()
|
|
with zipfile.ZipFile(zip_file, 'r') as zip_fp:
|
|
# First get the raw metadata string (i.e. without padding space).
|
|
raw_metadata = property_files.GetPropertyFilesString(
|
|
zip_fp, reserve_space=False)
|
|
raw_length = len(raw_metadata)
|
|
|
|
# Now pass in the exact expected length.
|
|
streaming_metadata = property_files.Finalize(zip_fp, raw_length)
|
|
self.assertEqual(raw_length, len(streaming_metadata))
|
|
|
|
# Or pass in insufficient length.
|
|
self.assertRaises(
|
|
PropertyFiles.InsufficientSpaceException,
|
|
property_files.Finalize,
|
|
zip_fp,
|
|
raw_length - 1)
|
|
|
|
# Or pass in a much larger size.
|
|
streaming_metadata = property_files.Finalize(
|
|
zip_fp,
|
|
raw_length + 20)
|
|
self.assertEqual(raw_length + 20, len(streaming_metadata))
|
|
self.assertEqual(' ' * 20, streaming_metadata[raw_length:])
|
|
|
|
def test_Verify(self):
|
|
entries = (
|
|
'required-entry1',
|
|
'required-entry2',
|
|
'optional-entry1',
|
|
'optional-entry2',
|
|
'META-INF/com/android/metadata',
|
|
)
|
|
zip_file = self.construct_zip_package(entries)
|
|
property_files = TestPropertyFiles()
|
|
with zipfile.ZipFile(zip_file, 'r') as zip_fp:
|
|
# First get the raw metadata string (i.e. without padding space).
|
|
raw_metadata = property_files.GetPropertyFilesString(
|
|
zip_fp, reserve_space=False)
|
|
|
|
# Should pass the test if verification passes.
|
|
property_files.Verify(zip_fp, raw_metadata)
|
|
|
|
# Or raise on verification failure.
|
|
self.assertRaises(
|
|
AssertionError, property_files.Verify, zip_fp, raw_metadata + 'x')
|
|
|
|
|
|
class StreamingPropertyFilesTest(PropertyFilesTest):
|
|
"""Additional sanity checks specialized for StreamingPropertyFiles."""
|
|
|
|
def test_init(self):
|
|
property_files = StreamingPropertyFiles()
|
|
self.assertEqual('ota-streaming-property-files', property_files.name)
|
|
self.assertEqual(
|
|
(
|
|
'payload.bin',
|
|
'payload_properties.txt',
|
|
),
|
|
property_files.required)
|
|
self.assertEqual(
|
|
(
|
|
'care_map.pb',
|
|
'care_map.txt',
|
|
'compatibility.zip',
|
|
),
|
|
property_files.optional)
|
|
|
|
def test_Compute(self):
|
|
entries = (
|
|
'payload.bin',
|
|
'payload_properties.txt',
|
|
'care_map.txt',
|
|
'compatibility.zip',
|
|
)
|
|
zip_file = self.construct_zip_package(entries)
|
|
property_files = StreamingPropertyFiles()
|
|
with zipfile.ZipFile(zip_file, 'r') as zip_fp:
|
|
property_files_string = property_files.Compute(zip_fp)
|
|
|
|
tokens = self._parse_property_files_string(property_files_string)
|
|
self.assertEqual(5, len(tokens))
|
|
self._verify_entries(zip_file, tokens, entries)
|
|
|
|
def test_Finalize(self):
|
|
entries = [
|
|
'payload.bin',
|
|
'payload_properties.txt',
|
|
'care_map.txt',
|
|
'compatibility.zip',
|
|
'META-INF/com/android/metadata',
|
|
]
|
|
zip_file = self.construct_zip_package(entries)
|
|
property_files = StreamingPropertyFiles()
|
|
with zipfile.ZipFile(zip_file, 'r') as zip_fp:
|
|
raw_metadata = property_files.GetPropertyFilesString(
|
|
zip_fp, reserve_space=False)
|
|
streaming_metadata = property_files.Finalize(zip_fp, len(raw_metadata))
|
|
tokens = self._parse_property_files_string(streaming_metadata)
|
|
|
|
self.assertEqual(5, len(tokens))
|
|
# 'META-INF/com/android/metadata' will be key'd as 'metadata' in the
|
|
# streaming metadata.
|
|
entries[4] = 'metadata'
|
|
self._verify_entries(zip_file, tokens, entries)
|
|
|
|
def test_Verify(self):
|
|
entries = (
|
|
'payload.bin',
|
|
'payload_properties.txt',
|
|
'care_map.txt',
|
|
'compatibility.zip',
|
|
'META-INF/com/android/metadata',
|
|
)
|
|
zip_file = self.construct_zip_package(entries)
|
|
property_files = StreamingPropertyFiles()
|
|
with zipfile.ZipFile(zip_file, 'r') as zip_fp:
|
|
# First get the raw metadata string (i.e. without padding space).
|
|
raw_metadata = property_files.GetPropertyFilesString(
|
|
zip_fp, reserve_space=False)
|
|
|
|
# Should pass the test if verification passes.
|
|
property_files.Verify(zip_fp, raw_metadata)
|
|
|
|
# Or raise on verification failure.
|
|
self.assertRaises(
|
|
AssertionError, property_files.Verify, zip_fp, raw_metadata + 'x')
|
|
|
|
|
|
class AbOtaPropertyFilesTest(PropertyFilesTest):
|
|
"""Additional sanity checks specialized for AbOtaPropertyFiles."""
|
|
|
|
# The size for payload and metadata signature size.
|
|
SIGNATURE_SIZE = 256
|
|
|
|
def setUp(self):
|
|
self.testdata_dir = test_utils.get_testdata_dir()
|
|
self.assertTrue(os.path.exists(self.testdata_dir))
|
|
|
|
common.OPTIONS.wipe_user_data = False
|
|
common.OPTIONS.payload_signer = None
|
|
common.OPTIONS.payload_signer_args = None
|
|
common.OPTIONS.package_key = os.path.join(self.testdata_dir, 'testkey')
|
|
common.OPTIONS.key_passwords = {
|
|
common.OPTIONS.package_key : None,
|
|
}
|
|
|
|
def test_init(self):
|
|
property_files = AbOtaPropertyFiles()
|
|
self.assertEqual('ota-property-files', property_files.name)
|
|
self.assertEqual(
|
|
(
|
|
'payload.bin',
|
|
'payload_properties.txt',
|
|
),
|
|
property_files.required)
|
|
self.assertEqual(
|
|
(
|
|
'care_map.pb',
|
|
'care_map.txt',
|
|
'compatibility.zip',
|
|
),
|
|
property_files.optional)
|
|
|
|
@test_utils.SkipIfExternalToolsUnavailable()
|
|
def test_GetPayloadMetadataOffsetAndSize(self):
|
|
target_file = construct_target_files()
|
|
payload = Payload()
|
|
payload.Generate(target_file)
|
|
|
|
payload_signer = PayloadSigner()
|
|
payload.Sign(payload_signer)
|
|
|
|
output_file = common.MakeTempFile(suffix='.zip')
|
|
with zipfile.ZipFile(output_file, 'w') as output_zip:
|
|
payload.WriteToZip(output_zip)
|
|
|
|
# Find out the payload metadata offset and size.
|
|
property_files = AbOtaPropertyFiles()
|
|
with zipfile.ZipFile(output_file) as input_zip:
|
|
# pylint: disable=protected-access
|
|
payload_offset, metadata_total = (
|
|
property_files._GetPayloadMetadataOffsetAndSize(input_zip))
|
|
|
|
# The signature proto has the following format (details in
|
|
# /platform/system/update_engine/update_metadata.proto):
|
|
# message Signature {
|
|
# optional uint32 version = 1;
|
|
# optional bytes data = 2;
|
|
# optional fixed32 unpadded_signature_size = 3;
|
|
# }
|
|
#
|
|
# According to the protobuf encoding, the tail of the signature message will
|
|
# be [signature string(256 bytes) + encoding of the fixed32 number 256]. And
|
|
# 256 is encoded as 'x1d\x00\x01\x00\x00':
|
|
# [3 (field number) << 3 | 5 (type) + byte reverse of 0x100 (256)].
|
|
# Details in (https://developers.google.com/protocol-buffers/docs/encoding)
|
|
signature_tail_length = self.SIGNATURE_SIZE + 5
|
|
self.assertGreater(metadata_total, signature_tail_length)
|
|
with open(output_file, 'rb') as verify_fp:
|
|
verify_fp.seek(payload_offset + metadata_total - signature_tail_length)
|
|
metadata_signature_proto_tail = verify_fp.read(signature_tail_length)
|
|
|
|
self.assertEqual(b'\x1d\x00\x01\x00\x00',
|
|
metadata_signature_proto_tail[-5:])
|
|
metadata_signature = metadata_signature_proto_tail[:-5]
|
|
|
|
# Now we extract the metadata hash via brillo_update_payload script, which
|
|
# will serve as the oracle result.
|
|
payload_sig_file = common.MakeTempFile(prefix="sig-", suffix=".bin")
|
|
metadata_sig_file = common.MakeTempFile(prefix="sig-", suffix=".bin")
|
|
cmd = ['brillo_update_payload', 'hash',
|
|
'--unsigned_payload', payload.payload_file,
|
|
'--signature_size', str(self.SIGNATURE_SIZE),
|
|
'--metadata_hash_file', metadata_sig_file,
|
|
'--payload_hash_file', payload_sig_file]
|
|
proc = common.Run(cmd)
|
|
stdoutdata, _ = proc.communicate()
|
|
self.assertEqual(
|
|
0, proc.returncode,
|
|
'Failed to run brillo_update_payload:\n{}'.format(stdoutdata))
|
|
|
|
signed_metadata_sig_file = payload_signer.Sign(metadata_sig_file)
|
|
|
|
# Finally we can compare the two signatures.
|
|
with open(signed_metadata_sig_file, 'rb') as verify_fp:
|
|
self.assertEqual(verify_fp.read(), metadata_signature)
|
|
|
|
@staticmethod
|
|
def construct_zip_package_withValidPayload(with_metadata=False):
|
|
# Cannot use construct_zip_package() since we need a "valid" payload.bin.
|
|
target_file = construct_target_files()
|
|
payload = Payload()
|
|
payload.Generate(target_file)
|
|
|
|
payload_signer = PayloadSigner()
|
|
payload.Sign(payload_signer)
|
|
|
|
zip_file = common.MakeTempFile(suffix='.zip')
|
|
with zipfile.ZipFile(zip_file, 'w') as zip_fp:
|
|
# 'payload.bin',
|
|
payload.WriteToZip(zip_fp)
|
|
|
|
# Other entries.
|
|
entries = ['care_map.txt', 'compatibility.zip']
|
|
|
|
# Put META-INF/com/android/metadata if needed.
|
|
if with_metadata:
|
|
entries.append('META-INF/com/android/metadata')
|
|
|
|
for entry in entries:
|
|
zip_fp.writestr(
|
|
entry, entry.replace('.', '-').upper(), zipfile.ZIP_STORED)
|
|
|
|
return zip_file
|
|
|
|
@test_utils.SkipIfExternalToolsUnavailable()
|
|
def test_Compute(self):
|
|
zip_file = self.construct_zip_package_withValidPayload()
|
|
property_files = AbOtaPropertyFiles()
|
|
with zipfile.ZipFile(zip_file, 'r') as zip_fp:
|
|
property_files_string = property_files.Compute(zip_fp)
|
|
|
|
tokens = self._parse_property_files_string(property_files_string)
|
|
# "6" indcludes the four entries above, one metadata entry, and one entry
|
|
# for payload-metadata.bin.
|
|
self.assertEqual(6, len(tokens))
|
|
self._verify_entries(
|
|
zip_file, tokens, ('care_map.txt', 'compatibility.zip'))
|
|
|
|
@test_utils.SkipIfExternalToolsUnavailable()
|
|
def test_Finalize(self):
|
|
zip_file = self.construct_zip_package_withValidPayload(with_metadata=True)
|
|
property_files = AbOtaPropertyFiles()
|
|
with zipfile.ZipFile(zip_file, 'r') as zip_fp:
|
|
raw_metadata = property_files.GetPropertyFilesString(
|
|
zip_fp, reserve_space=False)
|
|
property_files_string = property_files.Finalize(zip_fp, len(raw_metadata))
|
|
|
|
tokens = self._parse_property_files_string(property_files_string)
|
|
# "6" indcludes the four entries above, one metadata entry, and one entry
|
|
# for payload-metadata.bin.
|
|
self.assertEqual(6, len(tokens))
|
|
self._verify_entries(
|
|
zip_file, tokens, ('care_map.txt', 'compatibility.zip'))
|
|
|
|
@test_utils.SkipIfExternalToolsUnavailable()
|
|
def test_Verify(self):
|
|
zip_file = self.construct_zip_package_withValidPayload(with_metadata=True)
|
|
property_files = AbOtaPropertyFiles()
|
|
with zipfile.ZipFile(zip_file, 'r') as zip_fp:
|
|
raw_metadata = property_files.GetPropertyFilesString(
|
|
zip_fp, reserve_space=False)
|
|
|
|
property_files.Verify(zip_fp, raw_metadata)
|
|
|
|
|
|
class NonAbOtaPropertyFilesTest(PropertyFilesTest):
|
|
"""Additional sanity checks specialized for NonAbOtaPropertyFiles."""
|
|
|
|
def test_init(self):
|
|
property_files = NonAbOtaPropertyFiles()
|
|
self.assertEqual('ota-property-files', property_files.name)
|
|
self.assertEqual((), property_files.required)
|
|
self.assertEqual((), property_files.optional)
|
|
|
|
def test_Compute(self):
|
|
entries = ()
|
|
zip_file = self.construct_zip_package(entries)
|
|
property_files = NonAbOtaPropertyFiles()
|
|
with zipfile.ZipFile(zip_file) as zip_fp:
|
|
property_files_string = property_files.Compute(zip_fp)
|
|
|
|
tokens = self._parse_property_files_string(property_files_string)
|
|
self.assertEqual(1, len(tokens))
|
|
self._verify_entries(zip_file, tokens, entries)
|
|
|
|
def test_Finalize(self):
|
|
entries = [
|
|
'META-INF/com/android/metadata',
|
|
]
|
|
zip_file = self.construct_zip_package(entries)
|
|
property_files = NonAbOtaPropertyFiles()
|
|
with zipfile.ZipFile(zip_file) as zip_fp:
|
|
raw_metadata = property_files.GetPropertyFilesString(
|
|
zip_fp, reserve_space=False)
|
|
property_files_string = property_files.Finalize(zip_fp, len(raw_metadata))
|
|
tokens = self._parse_property_files_string(property_files_string)
|
|
|
|
self.assertEqual(1, len(tokens))
|
|
# 'META-INF/com/android/metadata' will be key'd as 'metadata'.
|
|
entries[0] = 'metadata'
|
|
self._verify_entries(zip_file, tokens, entries)
|
|
|
|
def test_Verify(self):
|
|
entries = (
|
|
'META-INF/com/android/metadata',
|
|
)
|
|
zip_file = self.construct_zip_package(entries)
|
|
property_files = NonAbOtaPropertyFiles()
|
|
with zipfile.ZipFile(zip_file) as zip_fp:
|
|
raw_metadata = property_files.GetPropertyFilesString(
|
|
zip_fp, reserve_space=False)
|
|
|
|
property_files.Verify(zip_fp, raw_metadata)
|
|
|
|
|
|
class PayloadSignerTest(test_utils.ReleaseToolsTestCase):
|
|
|
|
SIGFILE = 'sigfile.bin'
|
|
SIGNED_SIGFILE = 'signed-sigfile.bin'
|
|
|
|
def setUp(self):
|
|
self.testdata_dir = test_utils.get_testdata_dir()
|
|
self.assertTrue(os.path.exists(self.testdata_dir))
|
|
|
|
common.OPTIONS.payload_signer = None
|
|
common.OPTIONS.payload_signer_args = []
|
|
common.OPTIONS.package_key = os.path.join(self.testdata_dir, 'testkey')
|
|
common.OPTIONS.key_passwords = {
|
|
common.OPTIONS.package_key : None,
|
|
}
|
|
|
|
def _assertFilesEqual(self, file1, file2):
|
|
with open(file1, 'rb') as fp1, open(file2, 'rb') as fp2:
|
|
self.assertEqual(fp1.read(), fp2.read())
|
|
|
|
@test_utils.SkipIfExternalToolsUnavailable()
|
|
def test_init(self):
|
|
payload_signer = PayloadSigner()
|
|
self.assertEqual('openssl', payload_signer.signer)
|
|
self.assertEqual(256, payload_signer.maximum_signature_size)
|
|
|
|
@test_utils.SkipIfExternalToolsUnavailable()
|
|
def test_init_withPassword(self):
|
|
common.OPTIONS.package_key = os.path.join(
|
|
self.testdata_dir, 'testkey_with_passwd')
|
|
common.OPTIONS.key_passwords = {
|
|
common.OPTIONS.package_key : 'foo',
|
|
}
|
|
payload_signer = PayloadSigner()
|
|
self.assertEqual('openssl', payload_signer.signer)
|
|
|
|
def test_init_withExternalSigner(self):
|
|
common.OPTIONS.payload_signer = 'abc'
|
|
common.OPTIONS.payload_signer_args = ['arg1', 'arg2']
|
|
common.OPTIONS.payload_signer_maximum_signature_size = '512'
|
|
payload_signer = PayloadSigner()
|
|
self.assertEqual('abc', payload_signer.signer)
|
|
self.assertEqual(['arg1', 'arg2'], payload_signer.signer_args)
|
|
self.assertEqual(512, payload_signer.maximum_signature_size)
|
|
|
|
@test_utils.SkipIfExternalToolsUnavailable()
|
|
def test_GetMaximumSignatureSizeInBytes_512Bytes(self):
|
|
signing_key = os.path.join(self.testdata_dir, 'testkey_RSA4096.key')
|
|
# pylint: disable=protected-access
|
|
signature_size = PayloadSigner._GetMaximumSignatureSizeInBytes(signing_key)
|
|
self.assertEqual(512, signature_size)
|
|
|
|
@test_utils.SkipIfExternalToolsUnavailable()
|
|
def test_GetMaximumSignatureSizeInBytes_ECKey(self):
|
|
signing_key = os.path.join(self.testdata_dir, 'testkey_EC.key')
|
|
# pylint: disable=protected-access
|
|
signature_size = PayloadSigner._GetMaximumSignatureSizeInBytes(signing_key)
|
|
self.assertEqual(72, signature_size)
|
|
|
|
@test_utils.SkipIfExternalToolsUnavailable()
|
|
def test_Sign(self):
|
|
payload_signer = PayloadSigner()
|
|
input_file = os.path.join(self.testdata_dir, self.SIGFILE)
|
|
signed_file = payload_signer.Sign(input_file)
|
|
|
|
verify_file = os.path.join(self.testdata_dir, self.SIGNED_SIGFILE)
|
|
self._assertFilesEqual(verify_file, signed_file)
|
|
|
|
def test_Sign_withExternalSigner_openssl(self):
|
|
"""Uses openssl as the external payload signer."""
|
|
common.OPTIONS.payload_signer = 'openssl'
|
|
common.OPTIONS.payload_signer_args = [
|
|
'pkeyutl', '-sign', '-keyform', 'DER', '-inkey',
|
|
os.path.join(self.testdata_dir, 'testkey.pk8'),
|
|
'-pkeyopt', 'digest:sha256']
|
|
payload_signer = PayloadSigner()
|
|
input_file = os.path.join(self.testdata_dir, self.SIGFILE)
|
|
signed_file = payload_signer.Sign(input_file)
|
|
|
|
verify_file = os.path.join(self.testdata_dir, self.SIGNED_SIGFILE)
|
|
self._assertFilesEqual(verify_file, signed_file)
|
|
|
|
def test_Sign_withExternalSigner_script(self):
|
|
"""Uses testdata/payload_signer.sh as the external payload signer."""
|
|
common.OPTIONS.payload_signer = os.path.join(
|
|
self.testdata_dir, 'payload_signer.sh')
|
|
os.chmod(common.OPTIONS.payload_signer, 0o700)
|
|
common.OPTIONS.payload_signer_args = [
|
|
os.path.join(self.testdata_dir, 'testkey.pk8')]
|
|
payload_signer = PayloadSigner()
|
|
input_file = os.path.join(self.testdata_dir, self.SIGFILE)
|
|
signed_file = payload_signer.Sign(input_file)
|
|
|
|
verify_file = os.path.join(self.testdata_dir, self.SIGNED_SIGFILE)
|
|
self._assertFilesEqual(verify_file, signed_file)
|
|
|
|
|
|
class PayloadTest(test_utils.ReleaseToolsTestCase):
|
|
|
|
def setUp(self):
|
|
self.testdata_dir = test_utils.get_testdata_dir()
|
|
self.assertTrue(os.path.exists(self.testdata_dir))
|
|
|
|
common.OPTIONS.wipe_user_data = False
|
|
common.OPTIONS.payload_signer = None
|
|
common.OPTIONS.payload_signer_args = None
|
|
common.OPTIONS.package_key = os.path.join(self.testdata_dir, 'testkey')
|
|
common.OPTIONS.key_passwords = {
|
|
common.OPTIONS.package_key : None,
|
|
}
|
|
|
|
@staticmethod
|
|
def _create_payload_full(secondary=False):
|
|
target_file = construct_target_files(secondary)
|
|
payload = Payload(secondary)
|
|
payload.Generate(target_file)
|
|
return payload
|
|
|
|
@staticmethod
|
|
def _create_payload_incremental():
|
|
target_file = construct_target_files()
|
|
source_file = construct_target_files()
|
|
payload = Payload()
|
|
payload.Generate(target_file, source_file)
|
|
return payload
|
|
|
|
@test_utils.SkipIfExternalToolsUnavailable()
|
|
def test_Generate_full(self):
|
|
payload = self._create_payload_full()
|
|
self.assertTrue(os.path.exists(payload.payload_file))
|
|
|
|
@test_utils.SkipIfExternalToolsUnavailable()
|
|
def test_Generate_incremental(self):
|
|
payload = self._create_payload_incremental()
|
|
self.assertTrue(os.path.exists(payload.payload_file))
|
|
|
|
@test_utils.SkipIfExternalToolsUnavailable()
|
|
def test_Generate_additionalArgs(self):
|
|
target_file = construct_target_files()
|
|
source_file = construct_target_files()
|
|
payload = Payload()
|
|
# This should work the same as calling payload.Generate(target_file,
|
|
# source_file).
|
|
payload.Generate(
|
|
target_file, additional_args=["--source_image", source_file])
|
|
self.assertTrue(os.path.exists(payload.payload_file))
|
|
|
|
@test_utils.SkipIfExternalToolsUnavailable()
|
|
def test_Generate_invalidInput(self):
|
|
target_file = construct_target_files()
|
|
common.ZipDelete(target_file, 'IMAGES/vendor.img')
|
|
payload = Payload()
|
|
self.assertRaises(common.ExternalError, payload.Generate, target_file)
|
|
|
|
@test_utils.SkipIfExternalToolsUnavailable()
|
|
def test_Sign_full(self):
|
|
payload = self._create_payload_full()
|
|
payload.Sign(PayloadSigner())
|
|
|
|
output_file = common.MakeTempFile(suffix='.zip')
|
|
with zipfile.ZipFile(output_file, 'w') as output_zip:
|
|
payload.WriteToZip(output_zip)
|
|
|
|
import check_ota_package_signature
|
|
check_ota_package_signature.VerifyAbOtaPayload(
|
|
os.path.join(self.testdata_dir, 'testkey.x509.pem'),
|
|
output_file)
|
|
|
|
@test_utils.SkipIfExternalToolsUnavailable()
|
|
def test_Sign_incremental(self):
|
|
payload = self._create_payload_incremental()
|
|
payload.Sign(PayloadSigner())
|
|
|
|
output_file = common.MakeTempFile(suffix='.zip')
|
|
with zipfile.ZipFile(output_file, 'w') as output_zip:
|
|
payload.WriteToZip(output_zip)
|
|
|
|
import check_ota_package_signature
|
|
check_ota_package_signature.VerifyAbOtaPayload(
|
|
os.path.join(self.testdata_dir, 'testkey.x509.pem'),
|
|
output_file)
|
|
|
|
@test_utils.SkipIfExternalToolsUnavailable()
|
|
def test_Sign_withDataWipe(self):
|
|
common.OPTIONS.wipe_user_data = True
|
|
payload = self._create_payload_full()
|
|
payload.Sign(PayloadSigner())
|
|
|
|
with open(payload.payload_properties) as properties_fp:
|
|
self.assertIn("POWERWASH=1", properties_fp.read())
|
|
|
|
@test_utils.SkipIfExternalToolsUnavailable()
|
|
def test_Sign_secondary(self):
|
|
payload = self._create_payload_full(secondary=True)
|
|
payload.Sign(PayloadSigner())
|
|
|
|
with open(payload.payload_properties) as properties_fp:
|
|
self.assertIn("SWITCH_SLOT_ON_REBOOT=0", properties_fp.read())
|
|
|
|
@test_utils.SkipIfExternalToolsUnavailable()
|
|
def test_Sign_badSigner(self):
|
|
"""Tests that signing failure can be captured."""
|
|
payload = self._create_payload_full()
|
|
payload_signer = PayloadSigner()
|
|
payload_signer.signer_args.append('bad-option')
|
|
self.assertRaises(common.ExternalError, payload.Sign, payload_signer)
|
|
|
|
@test_utils.SkipIfExternalToolsUnavailable()
|
|
def test_WriteToZip(self):
|
|
payload = self._create_payload_full()
|
|
payload.Sign(PayloadSigner())
|
|
|
|
output_file = common.MakeTempFile(suffix='.zip')
|
|
with zipfile.ZipFile(output_file, 'w') as output_zip:
|
|
payload.WriteToZip(output_zip)
|
|
|
|
with zipfile.ZipFile(output_file) as verify_zip:
|
|
# First make sure we have the essential entries.
|
|
namelist = verify_zip.namelist()
|
|
self.assertIn(Payload.PAYLOAD_BIN, namelist)
|
|
self.assertIn(Payload.PAYLOAD_PROPERTIES_TXT, namelist)
|
|
|
|
# Then assert these entries are stored.
|
|
for entry_info in verify_zip.infolist():
|
|
if entry_info.filename not in (Payload.PAYLOAD_BIN,
|
|
Payload.PAYLOAD_PROPERTIES_TXT):
|
|
continue
|
|
self.assertEqual(zipfile.ZIP_STORED, entry_info.compress_type)
|
|
|
|
@test_utils.SkipIfExternalToolsUnavailable()
|
|
def test_WriteToZip_unsignedPayload(self):
|
|
"""Unsigned payloads should not be allowed to be written to zip."""
|
|
payload = self._create_payload_full()
|
|
|
|
output_file = common.MakeTempFile(suffix='.zip')
|
|
with zipfile.ZipFile(output_file, 'w') as output_zip:
|
|
self.assertRaises(AssertionError, payload.WriteToZip, output_zip)
|
|
|
|
# Also test with incremental payload.
|
|
payload = self._create_payload_incremental()
|
|
|
|
output_file = common.MakeTempFile(suffix='.zip')
|
|
with zipfile.ZipFile(output_file, 'w') as output_zip:
|
|
self.assertRaises(AssertionError, payload.WriteToZip, output_zip)
|
|
|
|
@test_utils.SkipIfExternalToolsUnavailable()
|
|
def test_WriteToZip_secondary(self):
|
|
payload = self._create_payload_full(secondary=True)
|
|
payload.Sign(PayloadSigner())
|
|
|
|
output_file = common.MakeTempFile(suffix='.zip')
|
|
with zipfile.ZipFile(output_file, 'w') as output_zip:
|
|
payload.WriteToZip(output_zip)
|
|
|
|
with zipfile.ZipFile(output_file) as verify_zip:
|
|
# First make sure we have the essential entries.
|
|
namelist = verify_zip.namelist()
|
|
self.assertIn(Payload.SECONDARY_PAYLOAD_BIN, namelist)
|
|
self.assertIn(Payload.SECONDARY_PAYLOAD_PROPERTIES_TXT, namelist)
|
|
|
|
# Then assert these entries are stored.
|
|
for entry_info in verify_zip.infolist():
|
|
if entry_info.filename not in (
|
|
Payload.SECONDARY_PAYLOAD_BIN,
|
|
Payload.SECONDARY_PAYLOAD_PROPERTIES_TXT):
|
|
continue
|
|
self.assertEqual(zipfile.ZIP_STORED, entry_info.compress_type)
|