1#
2# Copyright (C) 2018 The Android Open Source Project
3#
4# Licensed under the Apache License, Version 2.0 (the "License");
5# you may not use this file except in compliance with the License.
6# You may obtain a copy of the License at
7#
8#      http://www.apache.org/licenses/LICENSE-2.0
9#
10# Unless required by applicable law or agreed to in writing, software
11# distributed under the License is distributed on an "AS IS" BASIS,
12# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13# See the License for the specific language governing permissions and
14# limitations under the License.
15#
16
17import copy
18import os
19import os.path
20import zipfile
21
22import common
23import test_utils
24from ota_from_target_files import (
25    _LoadOemDicts, AbOtaPropertyFiles, FinalizeMetadata,
26    GetPackageMetadata, GetTargetFilesZipForSecondaryImages,
27    GetTargetFilesZipWithoutPostinstallConfig, NonAbOtaPropertyFiles,
28    Payload, PayloadSigner, POSTINSTALL_CONFIG, PropertyFiles,
29    StreamingPropertyFiles, WriteFingerprintAssertion,
30    CalculateRuntimeDevicesAndFingerprints)
31
32
33def construct_target_files(secondary=False):
34  """Returns a target-files.zip file for generating OTA packages."""
35  target_files = common.MakeTempFile(prefix='target_files-', suffix='.zip')
36  with zipfile.ZipFile(target_files, 'w') as target_files_zip:
37    # META/update_engine_config.txt
38    target_files_zip.writestr(
39        'META/update_engine_config.txt',
40        "PAYLOAD_MAJOR_VERSION=2\nPAYLOAD_MINOR_VERSION=4\n")
41
42    # META/postinstall_config.txt
43    target_files_zip.writestr(
44        POSTINSTALL_CONFIG,
45        '\n'.join([
46            "RUN_POSTINSTALL_system=true",
47            "POSTINSTALL_PATH_system=system/bin/otapreopt_script",
48            "FILESYSTEM_TYPE_system=ext4",
49            "POSTINSTALL_OPTIONAL_system=true",
50        ]))
51
52    ab_partitions = [
53        ('IMAGES', 'boot'),
54        ('IMAGES', 'system'),
55        ('IMAGES', 'vendor'),
56        ('RADIO', 'bootloader'),
57        ('RADIO', 'modem'),
58    ]
59    # META/ab_partitions.txt
60    target_files_zip.writestr(
61        'META/ab_partitions.txt',
62        '\n'.join([partition[1] for partition in ab_partitions]))
63
64    # Create fake images for each of them.
65    for path, partition in ab_partitions:
66      target_files_zip.writestr(
67          '{}/{}.img'.format(path, partition),
68          os.urandom(len(partition)))
69
70    # system_other shouldn't appear in META/ab_partitions.txt.
71    if secondary:
72      target_files_zip.writestr('IMAGES/system_other.img',
73                                os.urandom(len("system_other")))
74
75  return target_files
76
77
78class LoadOemDictsTest(test_utils.ReleaseToolsTestCase):
79
80  def test_NoneDict(self):
81    self.assertIsNone(_LoadOemDicts(None))
82
83  def test_SingleDict(self):
84    dict_file = common.MakeTempFile()
85    with open(dict_file, 'w') as dict_fp:
86      dict_fp.write('abc=1\ndef=2\nxyz=foo\na.b.c=bar\n')
87
88    oem_dicts = _LoadOemDicts([dict_file])
89    self.assertEqual(1, len(oem_dicts))
90    self.assertEqual('foo', oem_dicts[0]['xyz'])
91    self.assertEqual('bar', oem_dicts[0]['a.b.c'])
92
93  def test_MultipleDicts(self):
94    oem_source = []
95    for i in range(3):
96      dict_file = common.MakeTempFile()
97      with open(dict_file, 'w') as dict_fp:
98        dict_fp.write(
99            'ro.build.index={}\ndef=2\nxyz=foo\na.b.c=bar\n'.format(i))
100      oem_source.append(dict_file)
101
102    oem_dicts = _LoadOemDicts(oem_source)
103    self.assertEqual(3, len(oem_dicts))
104    for i, oem_dict in enumerate(oem_dicts):
105      self.assertEqual('2', oem_dict['def'])
106      self.assertEqual('foo', oem_dict['xyz'])
107      self.assertEqual('bar', oem_dict['a.b.c'])
108      self.assertEqual('{}'.format(i), oem_dict['ro.build.index'])
109
110
111class OtaFromTargetFilesTest(test_utils.ReleaseToolsTestCase):
112  TEST_TARGET_INFO_DICT = {
113      'build.prop': common.PartitionBuildProps.FromDictionary(
114          'system', {
115              'ro.product.device': 'product-device',
116              'ro.build.fingerprint': 'build-fingerprint-target',
117              'ro.build.version.incremental': 'build-version-incremental-target',
118              'ro.build.version.sdk': '27',
119              'ro.build.version.security_patch': '2017-12-01',
120              'ro.build.date.utc': '1500000000'}
121      )
122  }
123
124  TEST_SOURCE_INFO_DICT = {
125      'build.prop': common.PartitionBuildProps.FromDictionary(
126          'system', {
127              'ro.product.device': 'product-device',
128              'ro.build.fingerprint': 'build-fingerprint-source',
129              'ro.build.version.incremental': 'build-version-incremental-source',
130              'ro.build.version.sdk': '25',
131              'ro.build.version.security_patch': '2016-12-01',
132              'ro.build.date.utc': '1400000000'}
133      )
134  }
135
136  TEST_INFO_DICT_USES_OEM_PROPS = {
137      'build.prop': common.PartitionBuildProps.FromDictionary(
138          'system', {
139              'ro.product.name': 'product-name',
140              'ro.build.thumbprint': 'build-thumbprint',
141              'ro.build.bar': 'build-bar'}
142      ),
143      'vendor.build.prop': common.PartitionBuildProps.FromDictionary(
144          'vendor', {
145               'ro.vendor.build.fingerprint': 'vendor-build-fingerprint'}
146      ),
147      'property1': 'value1',
148      'property2': 4096,
149      'oem_fingerprint_properties': 'ro.product.device ro.product.brand',
150  }
151
152  TEST_OEM_DICTS = [
153      {
154          'ro.product.brand': 'brand1',
155          'ro.product.device': 'device1',
156      },
157      {
158          'ro.product.brand': 'brand2',
159          'ro.product.device': 'device2',
160      },
161      {
162          'ro.product.brand': 'brand3',
163          'ro.product.device': 'device3',
164      },
165  ]
166
167  def setUp(self):
168    self.testdata_dir = test_utils.get_testdata_dir()
169    self.assertTrue(os.path.exists(self.testdata_dir))
170
171    # Reset the global options as in ota_from_target_files.py.
172    common.OPTIONS.incremental_source = None
173    common.OPTIONS.downgrade = False
174    common.OPTIONS.retrofit_dynamic_partitions = False
175    common.OPTIONS.timestamp = False
176    common.OPTIONS.wipe_user_data = False
177    common.OPTIONS.no_signing = False
178    common.OPTIONS.package_key = os.path.join(self.testdata_dir, 'testkey')
179    common.OPTIONS.key_passwords = {
180        common.OPTIONS.package_key : None,
181    }
182
183    common.OPTIONS.search_path = test_utils.get_search_path()
184
185  def test_GetPackageMetadata_abOta_full(self):
186    target_info_dict = copy.deepcopy(self.TEST_TARGET_INFO_DICT)
187    target_info_dict['ab_update'] = 'true'
188    target_info = common.BuildInfo(target_info_dict, None)
189    metadata = GetPackageMetadata(target_info)
190    self.assertDictEqual(
191        {
192            'ota-type' : 'AB',
193            'ota-required-cache' : '0',
194            'post-build' : 'build-fingerprint-target',
195            'post-build-incremental' : 'build-version-incremental-target',
196            'post-sdk-level' : '27',
197            'post-security-patch-level' : '2017-12-01',
198            'post-timestamp' : '1500000000',
199            'pre-device' : 'product-device',
200        },
201        metadata)
202
203  def test_GetPackageMetadata_abOta_incremental(self):
204    target_info_dict = copy.deepcopy(self.TEST_TARGET_INFO_DICT)
205    target_info_dict['ab_update'] = 'true'
206    target_info = common.BuildInfo(target_info_dict, None)
207    source_info = common.BuildInfo(self.TEST_SOURCE_INFO_DICT, None)
208    common.OPTIONS.incremental_source = ''
209    metadata = GetPackageMetadata(target_info, source_info)
210    self.assertDictEqual(
211        {
212            'ota-type' : 'AB',
213            'ota-required-cache' : '0',
214            'post-build' : 'build-fingerprint-target',
215            'post-build-incremental' : 'build-version-incremental-target',
216            'post-sdk-level' : '27',
217            'post-security-patch-level' : '2017-12-01',
218            'post-timestamp' : '1500000000',
219            'pre-device' : 'product-device',
220            'pre-build' : 'build-fingerprint-source',
221            'pre-build-incremental' : 'build-version-incremental-source',
222        },
223        metadata)
224
225  def test_GetPackageMetadata_nonAbOta_full(self):
226    target_info = common.BuildInfo(self.TEST_TARGET_INFO_DICT, None)
227    metadata = GetPackageMetadata(target_info)
228    self.assertDictEqual(
229        {
230            'ota-type' : 'BLOCK',
231            'post-build' : 'build-fingerprint-target',
232            'post-build-incremental' : 'build-version-incremental-target',
233            'post-sdk-level' : '27',
234            'post-security-patch-level' : '2017-12-01',
235            'post-timestamp' : '1500000000',
236            'pre-device' : 'product-device',
237        },
238        metadata)
239
240  def test_GetPackageMetadata_nonAbOta_incremental(self):
241    target_info = common.BuildInfo(self.TEST_TARGET_INFO_DICT, None)
242    source_info = common.BuildInfo(self.TEST_SOURCE_INFO_DICT, None)
243    common.OPTIONS.incremental_source = ''
244    metadata = GetPackageMetadata(target_info, source_info)
245    self.assertDictEqual(
246        {
247            'ota-type' : 'BLOCK',
248            'post-build' : 'build-fingerprint-target',
249            'post-build-incremental' : 'build-version-incremental-target',
250            'post-sdk-level' : '27',
251            'post-security-patch-level' : '2017-12-01',
252            'post-timestamp' : '1500000000',
253            'pre-device' : 'product-device',
254            'pre-build' : 'build-fingerprint-source',
255            'pre-build-incremental' : 'build-version-incremental-source',
256        },
257        metadata)
258
259  def test_GetPackageMetadata_wipe(self):
260    target_info = common.BuildInfo(self.TEST_TARGET_INFO_DICT, None)
261    common.OPTIONS.wipe_user_data = True
262    metadata = GetPackageMetadata(target_info)
263    self.assertDictEqual(
264        {
265            'ota-type' : 'BLOCK',
266            'ota-wipe' : 'yes',
267            'post-build' : 'build-fingerprint-target',
268            'post-build-incremental' : 'build-version-incremental-target',
269            'post-sdk-level' : '27',
270            'post-security-patch-level' : '2017-12-01',
271            'post-timestamp' : '1500000000',
272            'pre-device' : 'product-device',
273        },
274        metadata)
275
276  def test_GetPackageMetadata_retrofitDynamicPartitions(self):
277    target_info = common.BuildInfo(self.TEST_TARGET_INFO_DICT, None)
278    common.OPTIONS.retrofit_dynamic_partitions = True
279    metadata = GetPackageMetadata(target_info)
280    self.assertDictEqual(
281        {
282            'ota-retrofit-dynamic-partitions' : 'yes',
283            'ota-type' : 'BLOCK',
284            'post-build' : 'build-fingerprint-target',
285            'post-build-incremental' : 'build-version-incremental-target',
286            'post-sdk-level' : '27',
287            'post-security-patch-level' : '2017-12-01',
288            'post-timestamp' : '1500000000',
289            'pre-device' : 'product-device',
290        },
291        metadata)
292
293  @staticmethod
294  def _test_GetPackageMetadata_swapBuildTimestamps(target_info, source_info):
295    (target_info['build.prop'].build_props['ro.build.date.utc'],
296     source_info['build.prop'].build_props['ro.build.date.utc']) = (
297         source_info['build.prop'].build_props['ro.build.date.utc'],
298         target_info['build.prop'].build_props['ro.build.date.utc'])
299
300  def test_GetPackageMetadata_unintentionalDowngradeDetected(self):
301    target_info_dict = copy.deepcopy(self.TEST_TARGET_INFO_DICT)
302    source_info_dict = copy.deepcopy(self.TEST_SOURCE_INFO_DICT)
303    self._test_GetPackageMetadata_swapBuildTimestamps(
304        target_info_dict, source_info_dict)
305
306    target_info = common.BuildInfo(target_info_dict, None)
307    source_info = common.BuildInfo(source_info_dict, None)
308    common.OPTIONS.incremental_source = ''
309    self.assertRaises(RuntimeError, GetPackageMetadata, target_info,
310                      source_info)
311
312  def test_GetPackageMetadata_downgrade(self):
313    target_info_dict = copy.deepcopy(self.TEST_TARGET_INFO_DICT)
314    source_info_dict = copy.deepcopy(self.TEST_SOURCE_INFO_DICT)
315    self._test_GetPackageMetadata_swapBuildTimestamps(
316        target_info_dict, source_info_dict)
317
318    target_info = common.BuildInfo(target_info_dict, None)
319    source_info = common.BuildInfo(source_info_dict, None)
320    common.OPTIONS.incremental_source = ''
321    common.OPTIONS.downgrade = True
322    common.OPTIONS.wipe_user_data = True
323    metadata = GetPackageMetadata(target_info, source_info)
324    self.assertDictEqual(
325        {
326            'ota-downgrade' : 'yes',
327            'ota-type' : 'BLOCK',
328            'ota-wipe' : 'yes',
329            'post-build' : 'build-fingerprint-target',
330            'post-build-incremental' : 'build-version-incremental-target',
331            'post-sdk-level' : '27',
332            'post-security-patch-level' : '2017-12-01',
333            'post-timestamp' : '1400000000',
334            'pre-device' : 'product-device',
335            'pre-build' : 'build-fingerprint-source',
336            'pre-build-incremental' : 'build-version-incremental-source',
337        },
338        metadata)
339
340  @test_utils.SkipIfExternalToolsUnavailable()
341  def test_GetTargetFilesZipForSecondaryImages(self):
342    input_file = construct_target_files(secondary=True)
343    target_file = GetTargetFilesZipForSecondaryImages(input_file)
344
345    with zipfile.ZipFile(target_file) as verify_zip:
346      namelist = verify_zip.namelist()
347      ab_partitions = verify_zip.read('META/ab_partitions.txt').decode()
348
349    self.assertIn('META/ab_partitions.txt', namelist)
350    self.assertIn('IMAGES/system.img', namelist)
351    self.assertIn('RADIO/bootloader.img', namelist)
352    self.assertIn(POSTINSTALL_CONFIG, namelist)
353
354    self.assertNotIn('IMAGES/boot.img', namelist)
355    self.assertNotIn('IMAGES/system_other.img', namelist)
356    self.assertNotIn('IMAGES/system.map', namelist)
357    self.assertNotIn('RADIO/modem.img', namelist)
358
359    expected_ab_partitions = ['system', 'bootloader']
360    self.assertEqual('\n'.join(expected_ab_partitions), ab_partitions)
361
362  @test_utils.SkipIfExternalToolsUnavailable()
363  def test_GetTargetFilesZipForSecondaryImages_skipPostinstall(self):
364    input_file = construct_target_files(secondary=True)
365    target_file = GetTargetFilesZipForSecondaryImages(
366        input_file, skip_postinstall=True)
367
368    with zipfile.ZipFile(target_file) as verify_zip:
369      namelist = verify_zip.namelist()
370
371    self.assertIn('META/ab_partitions.txt', namelist)
372    self.assertIn('IMAGES/system.img', namelist)
373    self.assertIn('RADIO/bootloader.img', namelist)
374
375    self.assertNotIn('IMAGES/boot.img', namelist)
376    self.assertNotIn('IMAGES/system_other.img', namelist)
377    self.assertNotIn('IMAGES/system.map', namelist)
378    self.assertNotIn('RADIO/modem.img', namelist)
379    self.assertNotIn(POSTINSTALL_CONFIG, namelist)
380
381  @test_utils.SkipIfExternalToolsUnavailable()
382  def test_GetTargetFilesZipForSecondaryImages_withoutRadioImages(self):
383    input_file = construct_target_files(secondary=True)
384    common.ZipDelete(input_file, 'RADIO/bootloader.img')
385    common.ZipDelete(input_file, 'RADIO/modem.img')
386    target_file = GetTargetFilesZipForSecondaryImages(input_file)
387
388    with zipfile.ZipFile(target_file) as verify_zip:
389      namelist = verify_zip.namelist()
390
391    self.assertIn('META/ab_partitions.txt', namelist)
392    self.assertIn('IMAGES/system.img', namelist)
393    self.assertIn(POSTINSTALL_CONFIG, namelist)
394
395    self.assertNotIn('IMAGES/boot.img', namelist)
396    self.assertNotIn('IMAGES/system_other.img', namelist)
397    self.assertNotIn('IMAGES/system.map', namelist)
398    self.assertNotIn('RADIO/bootloader.img', namelist)
399    self.assertNotIn('RADIO/modem.img', namelist)
400
401  @test_utils.SkipIfExternalToolsUnavailable()
402  def test_GetTargetFilesZipForSecondaryImages_dynamicPartitions(self):
403    input_file = construct_target_files(secondary=True)
404    misc_info = '\n'.join([
405        'use_dynamic_partition_size=true',
406        'use_dynamic_partitions=true',
407        'dynamic_partition_list=system vendor product',
408        'super_partition_groups=google_dynamic_partitions',
409        'super_google_dynamic_partitions_group_size=4873781248',
410        'super_google_dynamic_partitions_partition_list=system vendor product',
411    ])
412    dynamic_partitions_info = '\n'.join([
413        'super_partition_groups=google_dynamic_partitions',
414        'super_google_dynamic_partitions_group_size=4873781248',
415        'super_google_dynamic_partitions_partition_list=system vendor product',
416    ])
417
418    with zipfile.ZipFile(input_file, 'a') as append_zip:
419      common.ZipWriteStr(append_zip, 'META/misc_info.txt', misc_info)
420      common.ZipWriteStr(append_zip, 'META/dynamic_partitions_info.txt',
421                         dynamic_partitions_info)
422
423    target_file = GetTargetFilesZipForSecondaryImages(input_file)
424
425    with zipfile.ZipFile(target_file) as verify_zip:
426      namelist = verify_zip.namelist()
427      updated_misc_info = verify_zip.read('META/misc_info.txt').decode()
428      updated_dynamic_partitions_info = verify_zip.read(
429          'META/dynamic_partitions_info.txt').decode()
430
431    self.assertIn('META/ab_partitions.txt', namelist)
432    self.assertIn('IMAGES/system.img', namelist)
433    self.assertIn(POSTINSTALL_CONFIG, namelist)
434    self.assertIn('META/misc_info.txt', namelist)
435    self.assertIn('META/dynamic_partitions_info.txt', namelist)
436
437    self.assertNotIn('IMAGES/boot.img', namelist)
438    self.assertNotIn('IMAGES/system_other.img', namelist)
439    self.assertNotIn('IMAGES/system.map', namelist)
440
441    # Check the vendor & product are removed from the partitions list.
442    expected_misc_info = misc_info.replace('system vendor product',
443                                           'system')
444    expected_dynamic_partitions_info = dynamic_partitions_info.replace(
445        'system vendor product', 'system')
446    self.assertEqual(expected_misc_info, updated_misc_info)
447    self.assertEqual(expected_dynamic_partitions_info,
448                     updated_dynamic_partitions_info)
449
450  @test_utils.SkipIfExternalToolsUnavailable()
451  def test_GetTargetFilesZipWithoutPostinstallConfig(self):
452    input_file = construct_target_files()
453    target_file = GetTargetFilesZipWithoutPostinstallConfig(input_file)
454    with zipfile.ZipFile(target_file) as verify_zip:
455      self.assertNotIn(POSTINSTALL_CONFIG, verify_zip.namelist())
456
457  @test_utils.SkipIfExternalToolsUnavailable()
458  def test_GetTargetFilesZipWithoutPostinstallConfig_missingEntry(self):
459    input_file = construct_target_files()
460    common.ZipDelete(input_file, POSTINSTALL_CONFIG)
461    target_file = GetTargetFilesZipWithoutPostinstallConfig(input_file)
462    with zipfile.ZipFile(target_file) as verify_zip:
463      self.assertNotIn(POSTINSTALL_CONFIG, verify_zip.namelist())
464
465  def _test_FinalizeMetadata(self, large_entry=False):
466    entries = [
467        'required-entry1',
468        'required-entry2',
469    ]
470    zip_file = PropertyFilesTest.construct_zip_package(entries)
471    # Add a large entry of 1 GiB if requested.
472    if large_entry:
473      with zipfile.ZipFile(zip_file, 'a') as zip_fp:
474        zip_fp.writestr(
475            # Using 'zoo' so that the entry stays behind others after signing.
476            'zoo',
477            'A' * 1024 * 1024 * 1024,
478            zipfile.ZIP_STORED)
479
480    metadata = {}
481    output_file = common.MakeTempFile(suffix='.zip')
482    needed_property_files = (
483        TestPropertyFiles(),
484    )
485    FinalizeMetadata(metadata, zip_file, output_file, needed_property_files)
486    self.assertIn('ota-test-property-files', metadata)
487
488  @test_utils.SkipIfExternalToolsUnavailable()
489  def test_FinalizeMetadata(self):
490    self._test_FinalizeMetadata()
491
492  @test_utils.SkipIfExternalToolsUnavailable()
493  def test_FinalizeMetadata_withNoSigning(self):
494    common.OPTIONS.no_signing = True
495    self._test_FinalizeMetadata()
496
497  @test_utils.SkipIfExternalToolsUnavailable()
498  def test_FinalizeMetadata_largeEntry(self):
499    self._test_FinalizeMetadata(large_entry=True)
500
501  @test_utils.SkipIfExternalToolsUnavailable()
502  def test_FinalizeMetadata_largeEntry_withNoSigning(self):
503    common.OPTIONS.no_signing = True
504    self._test_FinalizeMetadata(large_entry=True)
505
506  @test_utils.SkipIfExternalToolsUnavailable()
507  def test_FinalizeMetadata_insufficientSpace(self):
508    entries = [
509        'required-entry1',
510        'required-entry2',
511        'optional-entry1',
512        'optional-entry2',
513    ]
514    zip_file = PropertyFilesTest.construct_zip_package(entries)
515    with zipfile.ZipFile(zip_file, 'a') as zip_fp:
516      zip_fp.writestr(
517          # 'foo-entry1' will appear ahead of all other entries (in alphabetical
518          # order) after the signing, which will in turn trigger the
519          # InsufficientSpaceException and an automatic retry.
520          'foo-entry1',
521          'A' * 1024 * 1024,
522          zipfile.ZIP_STORED)
523
524    metadata = {}
525    needed_property_files = (
526        TestPropertyFiles(),
527    )
528    output_file = common.MakeTempFile(suffix='.zip')
529    FinalizeMetadata(metadata, zip_file, output_file, needed_property_files)
530    self.assertIn('ota-test-property-files', metadata)
531
532  def test_WriteFingerprintAssertion_without_oem_props(self):
533    target_info = common.BuildInfo(self.TEST_TARGET_INFO_DICT, None)
534    source_info_dict = copy.deepcopy(self.TEST_TARGET_INFO_DICT)
535    source_info_dict['build.prop'].build_props['ro.build.fingerprint'] = (
536        'source-build-fingerprint')
537    source_info = common.BuildInfo(source_info_dict, None)
538
539    script_writer = test_utils.MockScriptWriter()
540    WriteFingerprintAssertion(script_writer, target_info, source_info)
541    self.assertEqual(
542        [('AssertSomeFingerprint', 'source-build-fingerprint',
543          'build-fingerprint-target')],
544        script_writer.lines)
545
546  def test_WriteFingerprintAssertion_with_source_oem_props(self):
547    target_info = common.BuildInfo(self.TEST_TARGET_INFO_DICT, None)
548    source_info = common.BuildInfo(self.TEST_INFO_DICT_USES_OEM_PROPS,
549                                   self.TEST_OEM_DICTS)
550
551    script_writer = test_utils.MockScriptWriter()
552    WriteFingerprintAssertion(script_writer, target_info, source_info)
553    self.assertEqual(
554        [('AssertFingerprintOrThumbprint', 'build-fingerprint-target',
555          'build-thumbprint')],
556        script_writer.lines)
557
558  def test_WriteFingerprintAssertion_with_target_oem_props(self):
559    target_info = common.BuildInfo(self.TEST_INFO_DICT_USES_OEM_PROPS,
560                                   self.TEST_OEM_DICTS)
561    source_info = common.BuildInfo(self.TEST_TARGET_INFO_DICT, None)
562
563    script_writer = test_utils.MockScriptWriter()
564    WriteFingerprintAssertion(script_writer, target_info, source_info)
565    self.assertEqual(
566        [('AssertFingerprintOrThumbprint', 'build-fingerprint-target',
567          'build-thumbprint')],
568        script_writer.lines)
569
570  def test_WriteFingerprintAssertion_with_both_oem_props(self):
571    target_info = common.BuildInfo(self.TEST_INFO_DICT_USES_OEM_PROPS,
572                                   self.TEST_OEM_DICTS)
573    source_info_dict = copy.deepcopy(self.TEST_INFO_DICT_USES_OEM_PROPS)
574    source_info_dict['build.prop'].build_props['ro.build.thumbprint'] = (
575        'source-build-thumbprint')
576    source_info = common.BuildInfo(source_info_dict, self.TEST_OEM_DICTS)
577
578    script_writer = test_utils.MockScriptWriter()
579    WriteFingerprintAssertion(script_writer, target_info, source_info)
580    self.assertEqual(
581        [('AssertSomeThumbprint', 'build-thumbprint',
582          'source-build-thumbprint')],
583        script_writer.lines)
584
585
586class TestPropertyFiles(PropertyFiles):
587  """A class that extends PropertyFiles for testing purpose."""
588
589  def __init__(self):
590    super(TestPropertyFiles, self).__init__()
591    self.name = 'ota-test-property-files'
592    self.required = (
593        'required-entry1',
594        'required-entry2',
595    )
596    self.optional = (
597        'optional-entry1',
598        'optional-entry2',
599    )
600
601
602class PropertyFilesTest(test_utils.ReleaseToolsTestCase):
603
604  def setUp(self):
605    common.OPTIONS.no_signing = False
606
607  @staticmethod
608  def construct_zip_package(entries):
609    zip_file = common.MakeTempFile(suffix='.zip')
610    with zipfile.ZipFile(zip_file, 'w') as zip_fp:
611      for entry in entries:
612        zip_fp.writestr(
613            entry,
614            entry.replace('.', '-').upper(),
615            zipfile.ZIP_STORED)
616    return zip_file
617
618  @staticmethod
619  def _parse_property_files_string(data):
620    result = {}
621    for token in data.split(','):
622      name, info = token.split(':', 1)
623      result[name] = info
624    return result
625
626  def _verify_entries(self, input_file, tokens, entries):
627    for entry in entries:
628      offset, size = map(int, tokens[entry].split(':'))
629      with open(input_file, 'rb') as input_fp:
630        input_fp.seek(offset)
631        if entry == 'metadata':
632          expected = b'META-INF/COM/ANDROID/METADATA'
633        else:
634          expected = entry.replace('.', '-').upper().encode()
635        self.assertEqual(expected, input_fp.read(size))
636
637  @test_utils.SkipIfExternalToolsUnavailable()
638  def test_Compute(self):
639    entries = (
640        'required-entry1',
641        'required-entry2',
642    )
643    zip_file = self.construct_zip_package(entries)
644    property_files = TestPropertyFiles()
645    with zipfile.ZipFile(zip_file, 'r') as zip_fp:
646      property_files_string = property_files.Compute(zip_fp)
647
648    tokens = self._parse_property_files_string(property_files_string)
649    self.assertEqual(3, len(tokens))
650    self._verify_entries(zip_file, tokens, entries)
651
652  def test_Compute_withOptionalEntries(self):
653    entries = (
654        'required-entry1',
655        'required-entry2',
656        'optional-entry1',
657        'optional-entry2',
658    )
659    zip_file = self.construct_zip_package(entries)
660    property_files = TestPropertyFiles()
661    with zipfile.ZipFile(zip_file, 'r') as zip_fp:
662      property_files_string = property_files.Compute(zip_fp)
663
664    tokens = self._parse_property_files_string(property_files_string)
665    self.assertEqual(5, len(tokens))
666    self._verify_entries(zip_file, tokens, entries)
667
668  def test_Compute_missingRequiredEntry(self):
669    entries = (
670        'required-entry2',
671    )
672    zip_file = self.construct_zip_package(entries)
673    property_files = TestPropertyFiles()
674    with zipfile.ZipFile(zip_file, 'r') as zip_fp:
675      self.assertRaises(KeyError, property_files.Compute, zip_fp)
676
677  @test_utils.SkipIfExternalToolsUnavailable()
678  def test_Finalize(self):
679    entries = [
680        'required-entry1',
681        'required-entry2',
682        'META-INF/com/android/metadata',
683    ]
684    zip_file = self.construct_zip_package(entries)
685    property_files = TestPropertyFiles()
686    with zipfile.ZipFile(zip_file, 'r') as zip_fp:
687      raw_metadata = property_files.GetPropertyFilesString(
688          zip_fp, reserve_space=False)
689      streaming_metadata = property_files.Finalize(zip_fp, len(raw_metadata))
690    tokens = self._parse_property_files_string(streaming_metadata)
691
692    self.assertEqual(3, len(tokens))
693    # 'META-INF/com/android/metadata' will be key'd as 'metadata' in the
694    # streaming metadata.
695    entries[2] = 'metadata'
696    self._verify_entries(zip_file, tokens, entries)
697
698  @test_utils.SkipIfExternalToolsUnavailable()
699  def test_Finalize_assertReservedLength(self):
700    entries = (
701        'required-entry1',
702        'required-entry2',
703        'optional-entry1',
704        'optional-entry2',
705        'META-INF/com/android/metadata',
706    )
707    zip_file = self.construct_zip_package(entries)
708    property_files = TestPropertyFiles()
709    with zipfile.ZipFile(zip_file, 'r') as zip_fp:
710      # First get the raw metadata string (i.e. without padding space).
711      raw_metadata = property_files.GetPropertyFilesString(
712          zip_fp, reserve_space=False)
713      raw_length = len(raw_metadata)
714
715      # Now pass in the exact expected length.
716      streaming_metadata = property_files.Finalize(zip_fp, raw_length)
717      self.assertEqual(raw_length, len(streaming_metadata))
718
719      # Or pass in insufficient length.
720      self.assertRaises(
721          PropertyFiles.InsufficientSpaceException,
722          property_files.Finalize,
723          zip_fp,
724          raw_length - 1)
725
726      # Or pass in a much larger size.
727      streaming_metadata = property_files.Finalize(
728          zip_fp,
729          raw_length + 20)
730      self.assertEqual(raw_length + 20, len(streaming_metadata))
731      self.assertEqual(' ' * 20, streaming_metadata[raw_length:])
732
733  def test_Verify(self):
734    entries = (
735        'required-entry1',
736        'required-entry2',
737        'optional-entry1',
738        'optional-entry2',
739        'META-INF/com/android/metadata',
740    )
741    zip_file = self.construct_zip_package(entries)
742    property_files = TestPropertyFiles()
743    with zipfile.ZipFile(zip_file, 'r') as zip_fp:
744      # First get the raw metadata string (i.e. without padding space).
745      raw_metadata = property_files.GetPropertyFilesString(
746          zip_fp, reserve_space=False)
747
748      # Should pass the test if verification passes.
749      property_files.Verify(zip_fp, raw_metadata)
750
751      # Or raise on verification failure.
752      self.assertRaises(
753          AssertionError, property_files.Verify, zip_fp, raw_metadata + 'x')
754
755
756class StreamingPropertyFilesTest(PropertyFilesTest):
757  """Additional validity checks specialized for StreamingPropertyFiles."""
758
759  def test_init(self):
760    property_files = StreamingPropertyFiles()
761    self.assertEqual('ota-streaming-property-files', property_files.name)
762    self.assertEqual(
763        (
764            'payload.bin',
765            'payload_properties.txt',
766        ),
767        property_files.required)
768    self.assertEqual(
769        (
770            'care_map.pb',
771            'care_map.txt',
772            'compatibility.zip',
773        ),
774        property_files.optional)
775
776  def test_Compute(self):
777    entries = (
778        'payload.bin',
779        'payload_properties.txt',
780        'care_map.txt',
781        'compatibility.zip',
782    )
783    zip_file = self.construct_zip_package(entries)
784    property_files = StreamingPropertyFiles()
785    with zipfile.ZipFile(zip_file, 'r') as zip_fp:
786      property_files_string = property_files.Compute(zip_fp)
787
788    tokens = self._parse_property_files_string(property_files_string)
789    self.assertEqual(5, len(tokens))
790    self._verify_entries(zip_file, tokens, entries)
791
792  def test_Finalize(self):
793    entries = [
794        'payload.bin',
795        'payload_properties.txt',
796        'care_map.txt',
797        'compatibility.zip',
798        'META-INF/com/android/metadata',
799    ]
800    zip_file = self.construct_zip_package(entries)
801    property_files = StreamingPropertyFiles()
802    with zipfile.ZipFile(zip_file, 'r') as zip_fp:
803      raw_metadata = property_files.GetPropertyFilesString(
804          zip_fp, reserve_space=False)
805      streaming_metadata = property_files.Finalize(zip_fp, len(raw_metadata))
806    tokens = self._parse_property_files_string(streaming_metadata)
807
808    self.assertEqual(5, len(tokens))
809    # 'META-INF/com/android/metadata' will be key'd as 'metadata' in the
810    # streaming metadata.
811    entries[4] = 'metadata'
812    self._verify_entries(zip_file, tokens, entries)
813
814  def test_Verify(self):
815    entries = (
816        'payload.bin',
817        'payload_properties.txt',
818        'care_map.txt',
819        'compatibility.zip',
820        'META-INF/com/android/metadata',
821    )
822    zip_file = self.construct_zip_package(entries)
823    property_files = StreamingPropertyFiles()
824    with zipfile.ZipFile(zip_file, 'r') as zip_fp:
825      # First get the raw metadata string (i.e. without padding space).
826      raw_metadata = property_files.GetPropertyFilesString(
827          zip_fp, reserve_space=False)
828
829      # Should pass the test if verification passes.
830      property_files.Verify(zip_fp, raw_metadata)
831
832      # Or raise on verification failure.
833      self.assertRaises(
834          AssertionError, property_files.Verify, zip_fp, raw_metadata + 'x')
835
836
837class AbOtaPropertyFilesTest(PropertyFilesTest):
838  """Additional validity checks specialized for AbOtaPropertyFiles."""
839
840  # The size for payload and metadata signature size.
841  SIGNATURE_SIZE = 256
842
843  def setUp(self):
844    self.testdata_dir = test_utils.get_testdata_dir()
845    self.assertTrue(os.path.exists(self.testdata_dir))
846
847    common.OPTIONS.wipe_user_data = False
848    common.OPTIONS.payload_signer = None
849    common.OPTIONS.payload_signer_args = None
850    common.OPTIONS.package_key = os.path.join(self.testdata_dir, 'testkey')
851    common.OPTIONS.key_passwords = {
852        common.OPTIONS.package_key : None,
853    }
854
855  def test_init(self):
856    property_files = AbOtaPropertyFiles()
857    self.assertEqual('ota-property-files', property_files.name)
858    self.assertEqual(
859        (
860            'payload.bin',
861            'payload_properties.txt',
862        ),
863        property_files.required)
864    self.assertEqual(
865        (
866            'care_map.pb',
867            'care_map.txt',
868            'compatibility.zip',
869        ),
870        property_files.optional)
871
872  @test_utils.SkipIfExternalToolsUnavailable()
873  def test_GetPayloadMetadataOffsetAndSize(self):
874    target_file = construct_target_files()
875    payload = Payload()
876    payload.Generate(target_file)
877
878    payload_signer = PayloadSigner()
879    payload.Sign(payload_signer)
880
881    output_file = common.MakeTempFile(suffix='.zip')
882    with zipfile.ZipFile(output_file, 'w') as output_zip:
883      payload.WriteToZip(output_zip)
884
885    # Find out the payload metadata offset and size.
886    property_files = AbOtaPropertyFiles()
887    with zipfile.ZipFile(output_file) as input_zip:
888      # pylint: disable=protected-access
889      payload_offset, metadata_total = (
890          property_files._GetPayloadMetadataOffsetAndSize(input_zip))
891
892    # The signature proto has the following format (details in
893    #  /platform/system/update_engine/update_metadata.proto):
894    #  message Signature {
895    #    optional uint32 version = 1;
896    #    optional bytes data = 2;
897    #    optional fixed32 unpadded_signature_size = 3;
898    #  }
899    #
900    # According to the protobuf encoding, the tail of the signature message will
901    # be [signature string(256 bytes) + encoding of the fixed32 number 256]. And
902    # 256 is encoded as 'x1d\x00\x01\x00\x00':
903    # [3 (field number) << 3 | 5 (type) + byte reverse of 0x100 (256)].
904    # Details in (https://developers.google.com/protocol-buffers/docs/encoding)
905    signature_tail_length = self.SIGNATURE_SIZE + 5
906    self.assertGreater(metadata_total, signature_tail_length)
907    with open(output_file, 'rb') as verify_fp:
908      verify_fp.seek(payload_offset + metadata_total - signature_tail_length)
909      metadata_signature_proto_tail = verify_fp.read(signature_tail_length)
910
911    self.assertEqual(b'\x1d\x00\x01\x00\x00',
912                     metadata_signature_proto_tail[-5:])
913    metadata_signature = metadata_signature_proto_tail[:-5]
914
915    # Now we extract the metadata hash via brillo_update_payload script, which
916    # will serve as the oracle result.
917    payload_sig_file = common.MakeTempFile(prefix="sig-", suffix=".bin")
918    metadata_sig_file = common.MakeTempFile(prefix="sig-", suffix=".bin")
919    cmd = ['brillo_update_payload', 'hash',
920           '--unsigned_payload', payload.payload_file,
921           '--signature_size', str(self.SIGNATURE_SIZE),
922           '--metadata_hash_file', metadata_sig_file,
923           '--payload_hash_file', payload_sig_file]
924    proc = common.Run(cmd)
925    stdoutdata, _ = proc.communicate()
926    self.assertEqual(
927        0, proc.returncode,
928        'Failed to run brillo_update_payload:\n{}'.format(stdoutdata))
929
930    signed_metadata_sig_file = payload_signer.Sign(metadata_sig_file)
931
932    # Finally we can compare the two signatures.
933    with open(signed_metadata_sig_file, 'rb') as verify_fp:
934      self.assertEqual(verify_fp.read(), metadata_signature)
935
936  @staticmethod
937  def construct_zip_package_withValidPayload(with_metadata=False):
938    # Cannot use construct_zip_package() since we need a "valid" payload.bin.
939    target_file = construct_target_files()
940    payload = Payload()
941    payload.Generate(target_file)
942
943    payload_signer = PayloadSigner()
944    payload.Sign(payload_signer)
945
946    zip_file = common.MakeTempFile(suffix='.zip')
947    with zipfile.ZipFile(zip_file, 'w') as zip_fp:
948      # 'payload.bin',
949      payload.WriteToZip(zip_fp)
950
951      # Other entries.
952      entries = ['care_map.txt', 'compatibility.zip']
953
954      # Put META-INF/com/android/metadata if needed.
955      if with_metadata:
956        entries.append('META-INF/com/android/metadata')
957
958      for entry in entries:
959        zip_fp.writestr(
960            entry, entry.replace('.', '-').upper(), zipfile.ZIP_STORED)
961
962    return zip_file
963
964  @test_utils.SkipIfExternalToolsUnavailable()
965  def test_Compute(self):
966    zip_file = self.construct_zip_package_withValidPayload()
967    property_files = AbOtaPropertyFiles()
968    with zipfile.ZipFile(zip_file, 'r') as zip_fp:
969      property_files_string = property_files.Compute(zip_fp)
970
971    tokens = self._parse_property_files_string(property_files_string)
972    # "6" indcludes the four entries above, one metadata entry, and one entry
973    # for payload-metadata.bin.
974    self.assertEqual(6, len(tokens))
975    self._verify_entries(
976        zip_file, tokens, ('care_map.txt', 'compatibility.zip'))
977
978  @test_utils.SkipIfExternalToolsUnavailable()
979  def test_Finalize(self):
980    zip_file = self.construct_zip_package_withValidPayload(with_metadata=True)
981    property_files = AbOtaPropertyFiles()
982    with zipfile.ZipFile(zip_file, 'r') as zip_fp:
983      raw_metadata = property_files.GetPropertyFilesString(
984          zip_fp, reserve_space=False)
985      property_files_string = property_files.Finalize(zip_fp, len(raw_metadata))
986
987    tokens = self._parse_property_files_string(property_files_string)
988    # "6" indcludes the four entries above, one metadata entry, and one entry
989    # for payload-metadata.bin.
990    self.assertEqual(6, len(tokens))
991    self._verify_entries(
992        zip_file, tokens, ('care_map.txt', 'compatibility.zip'))
993
994  @test_utils.SkipIfExternalToolsUnavailable()
995  def test_Verify(self):
996    zip_file = self.construct_zip_package_withValidPayload(with_metadata=True)
997    property_files = AbOtaPropertyFiles()
998    with zipfile.ZipFile(zip_file, 'r') as zip_fp:
999      raw_metadata = property_files.GetPropertyFilesString(
1000          zip_fp, reserve_space=False)
1001
1002      property_files.Verify(zip_fp, raw_metadata)
1003
1004
1005class NonAbOtaPropertyFilesTest(PropertyFilesTest):
1006  """Additional validity checks specialized for NonAbOtaPropertyFiles."""
1007
1008  def test_init(self):
1009    property_files = NonAbOtaPropertyFiles()
1010    self.assertEqual('ota-property-files', property_files.name)
1011    self.assertEqual((), property_files.required)
1012    self.assertEqual((), property_files.optional)
1013
1014  def test_Compute(self):
1015    entries = ()
1016    zip_file = self.construct_zip_package(entries)
1017    property_files = NonAbOtaPropertyFiles()
1018    with zipfile.ZipFile(zip_file) as zip_fp:
1019      property_files_string = property_files.Compute(zip_fp)
1020
1021    tokens = self._parse_property_files_string(property_files_string)
1022    self.assertEqual(1, len(tokens))
1023    self._verify_entries(zip_file, tokens, entries)
1024
1025  def test_Finalize(self):
1026    entries = [
1027        'META-INF/com/android/metadata',
1028    ]
1029    zip_file = self.construct_zip_package(entries)
1030    property_files = NonAbOtaPropertyFiles()
1031    with zipfile.ZipFile(zip_file) as zip_fp:
1032      raw_metadata = property_files.GetPropertyFilesString(
1033          zip_fp, reserve_space=False)
1034      property_files_string = property_files.Finalize(zip_fp, len(raw_metadata))
1035    tokens = self._parse_property_files_string(property_files_string)
1036
1037    self.assertEqual(1, len(tokens))
1038    # 'META-INF/com/android/metadata' will be key'd as 'metadata'.
1039    entries[0] = 'metadata'
1040    self._verify_entries(zip_file, tokens, entries)
1041
1042  def test_Verify(self):
1043    entries = (
1044        'META-INF/com/android/metadata',
1045    )
1046    zip_file = self.construct_zip_package(entries)
1047    property_files = NonAbOtaPropertyFiles()
1048    with zipfile.ZipFile(zip_file) as zip_fp:
1049      raw_metadata = property_files.GetPropertyFilesString(
1050          zip_fp, reserve_space=False)
1051
1052      property_files.Verify(zip_fp, raw_metadata)
1053
1054
1055class PayloadSignerTest(test_utils.ReleaseToolsTestCase):
1056
1057  SIGFILE = 'sigfile.bin'
1058  SIGNED_SIGFILE = 'signed-sigfile.bin'
1059
1060  def setUp(self):
1061    self.testdata_dir = test_utils.get_testdata_dir()
1062    self.assertTrue(os.path.exists(self.testdata_dir))
1063
1064    common.OPTIONS.payload_signer = None
1065    common.OPTIONS.payload_signer_args = []
1066    common.OPTIONS.package_key = os.path.join(self.testdata_dir, 'testkey')
1067    common.OPTIONS.key_passwords = {
1068        common.OPTIONS.package_key : None,
1069    }
1070
1071  def _assertFilesEqual(self, file1, file2):
1072    with open(file1, 'rb') as fp1, open(file2, 'rb') as fp2:
1073      self.assertEqual(fp1.read(), fp2.read())
1074
1075  @test_utils.SkipIfExternalToolsUnavailable()
1076  def test_init(self):
1077    payload_signer = PayloadSigner()
1078    self.assertEqual('openssl', payload_signer.signer)
1079    self.assertEqual(256, payload_signer.maximum_signature_size)
1080
1081  @test_utils.SkipIfExternalToolsUnavailable()
1082  def test_init_withPassword(self):
1083    common.OPTIONS.package_key = os.path.join(
1084        self.testdata_dir, 'testkey_with_passwd')
1085    common.OPTIONS.key_passwords = {
1086        common.OPTIONS.package_key : 'foo',
1087    }
1088    payload_signer = PayloadSigner()
1089    self.assertEqual('openssl', payload_signer.signer)
1090
1091  def test_init_withExternalSigner(self):
1092    common.OPTIONS.payload_signer = 'abc'
1093    common.OPTIONS.payload_signer_args = ['arg1', 'arg2']
1094    common.OPTIONS.payload_signer_maximum_signature_size = '512'
1095    payload_signer = PayloadSigner()
1096    self.assertEqual('abc', payload_signer.signer)
1097    self.assertEqual(['arg1', 'arg2'], payload_signer.signer_args)
1098    self.assertEqual(512, payload_signer.maximum_signature_size)
1099
1100  @test_utils.SkipIfExternalToolsUnavailable()
1101  def test_GetMaximumSignatureSizeInBytes_512Bytes(self):
1102    signing_key = os.path.join(self.testdata_dir, 'testkey_RSA4096.key')
1103    # pylint: disable=protected-access
1104    signature_size = PayloadSigner._GetMaximumSignatureSizeInBytes(signing_key)
1105    self.assertEqual(512, signature_size)
1106
1107  @test_utils.SkipIfExternalToolsUnavailable()
1108  def test_GetMaximumSignatureSizeInBytes_ECKey(self):
1109    signing_key = os.path.join(self.testdata_dir, 'testkey_EC.key')
1110    # pylint: disable=protected-access
1111    signature_size = PayloadSigner._GetMaximumSignatureSizeInBytes(signing_key)
1112    self.assertEqual(72, signature_size)
1113
1114  @test_utils.SkipIfExternalToolsUnavailable()
1115  def test_Sign(self):
1116    payload_signer = PayloadSigner()
1117    input_file = os.path.join(self.testdata_dir, self.SIGFILE)
1118    signed_file = payload_signer.Sign(input_file)
1119
1120    verify_file = os.path.join(self.testdata_dir, self.SIGNED_SIGFILE)
1121    self._assertFilesEqual(verify_file, signed_file)
1122
1123  def test_Sign_withExternalSigner_openssl(self):
1124    """Uses openssl as the external payload signer."""
1125    common.OPTIONS.payload_signer = 'openssl'
1126    common.OPTIONS.payload_signer_args = [
1127        'pkeyutl', '-sign', '-keyform', 'DER', '-inkey',
1128        os.path.join(self.testdata_dir, 'testkey.pk8'),
1129        '-pkeyopt', 'digest:sha256']
1130    payload_signer = PayloadSigner()
1131    input_file = os.path.join(self.testdata_dir, self.SIGFILE)
1132    signed_file = payload_signer.Sign(input_file)
1133
1134    verify_file = os.path.join(self.testdata_dir, self.SIGNED_SIGFILE)
1135    self._assertFilesEqual(verify_file, signed_file)
1136
1137  def test_Sign_withExternalSigner_script(self):
1138    """Uses testdata/payload_signer.sh as the external payload signer."""
1139    common.OPTIONS.payload_signer = os.path.join(
1140        self.testdata_dir, 'payload_signer.sh')
1141    os.chmod(common.OPTIONS.payload_signer, 0o700)
1142    common.OPTIONS.payload_signer_args = [
1143        os.path.join(self.testdata_dir, 'testkey.pk8')]
1144    payload_signer = PayloadSigner()
1145    input_file = os.path.join(self.testdata_dir, self.SIGFILE)
1146    signed_file = payload_signer.Sign(input_file)
1147
1148    verify_file = os.path.join(self.testdata_dir, self.SIGNED_SIGFILE)
1149    self._assertFilesEqual(verify_file, signed_file)
1150
1151
1152class PayloadTest(test_utils.ReleaseToolsTestCase):
1153
1154  def setUp(self):
1155    self.testdata_dir = test_utils.get_testdata_dir()
1156    self.assertTrue(os.path.exists(self.testdata_dir))
1157
1158    common.OPTIONS.wipe_user_data = False
1159    common.OPTIONS.payload_signer = None
1160    common.OPTIONS.payload_signer_args = None
1161    common.OPTIONS.package_key = os.path.join(self.testdata_dir, 'testkey')
1162    common.OPTIONS.key_passwords = {
1163        common.OPTIONS.package_key : None,
1164    }
1165
1166  @staticmethod
1167  def _create_payload_full(secondary=False):
1168    target_file = construct_target_files(secondary)
1169    payload = Payload(secondary)
1170    payload.Generate(target_file)
1171    return payload
1172
1173  @staticmethod
1174  def _create_payload_incremental():
1175    target_file = construct_target_files()
1176    source_file = construct_target_files()
1177    payload = Payload()
1178    payload.Generate(target_file, source_file)
1179    return payload
1180
1181  @test_utils.SkipIfExternalToolsUnavailable()
1182  def test_Generate_full(self):
1183    payload = self._create_payload_full()
1184    self.assertTrue(os.path.exists(payload.payload_file))
1185
1186  @test_utils.SkipIfExternalToolsUnavailable()
1187  def test_Generate_incremental(self):
1188    payload = self._create_payload_incremental()
1189    self.assertTrue(os.path.exists(payload.payload_file))
1190
1191  @test_utils.SkipIfExternalToolsUnavailable()
1192  def test_Generate_additionalArgs(self):
1193    target_file = construct_target_files()
1194    source_file = construct_target_files()
1195    payload = Payload()
1196    # This should work the same as calling payload.Generate(target_file,
1197    # source_file).
1198    payload.Generate(
1199        target_file, additional_args=["--source_image", source_file])
1200    self.assertTrue(os.path.exists(payload.payload_file))
1201
1202  @test_utils.SkipIfExternalToolsUnavailable()
1203  def test_Generate_invalidInput(self):
1204    target_file = construct_target_files()
1205    common.ZipDelete(target_file, 'IMAGES/vendor.img')
1206    payload = Payload()
1207    self.assertRaises(common.ExternalError, payload.Generate, target_file)
1208
1209  @test_utils.SkipIfExternalToolsUnavailable()
1210  def test_Sign_full(self):
1211    payload = self._create_payload_full()
1212    payload.Sign(PayloadSigner())
1213
1214    output_file = common.MakeTempFile(suffix='.zip')
1215    with zipfile.ZipFile(output_file, 'w') as output_zip:
1216      payload.WriteToZip(output_zip)
1217
1218    import check_ota_package_signature
1219    check_ota_package_signature.VerifyAbOtaPayload(
1220        os.path.join(self.testdata_dir, 'testkey.x509.pem'),
1221        output_file)
1222
1223  @test_utils.SkipIfExternalToolsUnavailable()
1224  def test_Sign_incremental(self):
1225    payload = self._create_payload_incremental()
1226    payload.Sign(PayloadSigner())
1227
1228    output_file = common.MakeTempFile(suffix='.zip')
1229    with zipfile.ZipFile(output_file, 'w') as output_zip:
1230      payload.WriteToZip(output_zip)
1231
1232    import check_ota_package_signature
1233    check_ota_package_signature.VerifyAbOtaPayload(
1234        os.path.join(self.testdata_dir, 'testkey.x509.pem'),
1235        output_file)
1236
1237  @test_utils.SkipIfExternalToolsUnavailable()
1238  def test_Sign_withDataWipe(self):
1239    common.OPTIONS.wipe_user_data = True
1240    payload = self._create_payload_full()
1241    payload.Sign(PayloadSigner())
1242
1243    with open(payload.payload_properties) as properties_fp:
1244      self.assertIn("POWERWASH=1", properties_fp.read())
1245
1246  @test_utils.SkipIfExternalToolsUnavailable()
1247  def test_Sign_secondary(self):
1248    payload = self._create_payload_full(secondary=True)
1249    payload.Sign(PayloadSigner())
1250
1251    with open(payload.payload_properties) as properties_fp:
1252      self.assertIn("SWITCH_SLOT_ON_REBOOT=0", properties_fp.read())
1253
1254  @test_utils.SkipIfExternalToolsUnavailable()
1255  def test_Sign_badSigner(self):
1256    """Tests that signing failure can be captured."""
1257    payload = self._create_payload_full()
1258    payload_signer = PayloadSigner()
1259    payload_signer.signer_args.append('bad-option')
1260    self.assertRaises(common.ExternalError, payload.Sign, payload_signer)
1261
1262  @test_utils.SkipIfExternalToolsUnavailable()
1263  def test_WriteToZip(self):
1264    payload = self._create_payload_full()
1265    payload.Sign(PayloadSigner())
1266
1267    output_file = common.MakeTempFile(suffix='.zip')
1268    with zipfile.ZipFile(output_file, 'w') as output_zip:
1269      payload.WriteToZip(output_zip)
1270
1271    with zipfile.ZipFile(output_file) as verify_zip:
1272      # First make sure we have the essential entries.
1273      namelist = verify_zip.namelist()
1274      self.assertIn(Payload.PAYLOAD_BIN, namelist)
1275      self.assertIn(Payload.PAYLOAD_PROPERTIES_TXT, namelist)
1276
1277      # Then assert these entries are stored.
1278      for entry_info in verify_zip.infolist():
1279        if entry_info.filename not in (Payload.PAYLOAD_BIN,
1280                                       Payload.PAYLOAD_PROPERTIES_TXT):
1281          continue
1282        self.assertEqual(zipfile.ZIP_STORED, entry_info.compress_type)
1283
1284  @test_utils.SkipIfExternalToolsUnavailable()
1285  def test_WriteToZip_unsignedPayload(self):
1286    """Unsigned payloads should not be allowed to be written to zip."""
1287    payload = self._create_payload_full()
1288
1289    output_file = common.MakeTempFile(suffix='.zip')
1290    with zipfile.ZipFile(output_file, 'w') as output_zip:
1291      self.assertRaises(AssertionError, payload.WriteToZip, output_zip)
1292
1293    # Also test with incremental payload.
1294    payload = self._create_payload_incremental()
1295
1296    output_file = common.MakeTempFile(suffix='.zip')
1297    with zipfile.ZipFile(output_file, 'w') as output_zip:
1298      self.assertRaises(AssertionError, payload.WriteToZip, output_zip)
1299
1300  @test_utils.SkipIfExternalToolsUnavailable()
1301  def test_WriteToZip_secondary(self):
1302    payload = self._create_payload_full(secondary=True)
1303    payload.Sign(PayloadSigner())
1304
1305    output_file = common.MakeTempFile(suffix='.zip')
1306    with zipfile.ZipFile(output_file, 'w') as output_zip:
1307      payload.WriteToZip(output_zip)
1308
1309    with zipfile.ZipFile(output_file) as verify_zip:
1310      # First make sure we have the essential entries.
1311      namelist = verify_zip.namelist()
1312      self.assertIn(Payload.SECONDARY_PAYLOAD_BIN, namelist)
1313      self.assertIn(Payload.SECONDARY_PAYLOAD_PROPERTIES_TXT, namelist)
1314
1315      # Then assert these entries are stored.
1316      for entry_info in verify_zip.infolist():
1317        if entry_info.filename not in (
1318            Payload.SECONDARY_PAYLOAD_BIN,
1319            Payload.SECONDARY_PAYLOAD_PROPERTIES_TXT):
1320          continue
1321        self.assertEqual(zipfile.ZIP_STORED, entry_info.compress_type)
1322
1323
1324class RuntimeFingerprintTest(test_utils.ReleaseToolsTestCase):
1325  MISC_INFO = [
1326      'recovery_api_version=3',
1327      'fstab_version=2',
1328      'recovery_as_boot=true',
1329  ]
1330
1331  BUILD_PROP = [
1332      'ro.build.version.release=version-release',
1333      'ro.build.id=build-id',
1334      'ro.build.version.incremental=version-incremental',
1335      'ro.build.type=build-type',
1336      'ro.build.tags=build-tags',
1337      'ro.build.version.sdk=30',
1338      'ro.build.version.security_patch=2020',
1339      'ro.build.date.utc=12345678'
1340  ]
1341
1342  VENDOR_BUILD_PROP = [
1343      'ro.product.vendor.brand=vendor-product-brand',
1344      'ro.product.vendor.name=vendor-product-name',
1345      'ro.product.vendor.device=vendor-product-device'
1346  ]
1347
1348  def setUp(self):
1349    common.OPTIONS.oem_dicts = None
1350    self.test_dir = common.MakeTempDir()
1351    self.writeFiles({'META/misc_info.txt': '\n'.join(self.MISC_INFO)},
1352                    self.test_dir)
1353
1354  def writeFiles(self, contents_dict, out_dir):
1355    for path, content in contents_dict.items():
1356      abs_path = os.path.join(out_dir, path)
1357      dir_name = os.path.dirname(abs_path)
1358      if not os.path.exists(dir_name):
1359        os.makedirs(dir_name)
1360      with open(abs_path, 'w') as f:
1361        f.write(content)
1362
1363  @staticmethod
1364  def constructFingerprint(prefix):
1365    return '{}:version-release/build-id/version-incremental:' \
1366           'build-type/build-tags'.format(prefix)
1367
1368  def test_CalculatePossibleFingerprints_no_dynamic_fingerprint(self):
1369    build_prop = copy.deepcopy(self.BUILD_PROP)
1370    build_prop.extend([
1371        'ro.product.brand=product-brand',
1372        'ro.product.name=product-name',
1373        'ro.product.device=product-device',
1374    ])
1375    self.writeFiles({
1376        'SYSTEM/build.prop': '\n'.join(build_prop),
1377        'VENDOR/build.prop': '\n'.join(self.VENDOR_BUILD_PROP),
1378    }, self.test_dir)
1379
1380    build_info = common.BuildInfo(common.LoadInfoDict(self.test_dir))
1381    expected = ({'product-device'},
1382                {self.constructFingerprint(
1383                    'product-brand/product-name/product-device')})
1384    self.assertEqual(expected,
1385                     CalculateRuntimeDevicesAndFingerprints(build_info, {}))
1386
1387  def test_CalculatePossibleFingerprints_single_override(self):
1388    vendor_build_prop = copy.deepcopy(self.VENDOR_BUILD_PROP)
1389    vendor_build_prop.extend([
1390        'import /vendor/etc/build_${ro.boot.sku_name}.prop',
1391    ])
1392    self.writeFiles({
1393        'SYSTEM/build.prop': '\n'.join(self.BUILD_PROP),
1394        'VENDOR/build.prop': '\n'.join(vendor_build_prop),
1395        'VENDOR/etc/build_std.prop':
1396        'ro.product.vendor.name=vendor-product-std',
1397        'VENDOR/etc/build_pro.prop':
1398        'ro.product.vendor.name=vendor-product-pro',
1399    }, self.test_dir)
1400
1401    build_info = common.BuildInfo(common.LoadInfoDict(self.test_dir))
1402    boot_variable_values = {'ro.boot.sku_name': ['std', 'pro']}
1403
1404    expected = ({'vendor-product-device'}, {
1405        self.constructFingerprint(
1406            'vendor-product-brand/vendor-product-name/vendor-product-device'),
1407        self.constructFingerprint(
1408            'vendor-product-brand/vendor-product-std/vendor-product-device'),
1409        self.constructFingerprint(
1410            'vendor-product-brand/vendor-product-pro/vendor-product-device'),
1411    })
1412    self.assertEqual(
1413        expected, CalculateRuntimeDevicesAndFingerprints(
1414            build_info, boot_variable_values))
1415
1416  def test_CalculatePossibleFingerprints_multiple_overrides(self):
1417    vendor_build_prop = copy.deepcopy(self.VENDOR_BUILD_PROP)
1418    vendor_build_prop.extend([
1419        'import /vendor/etc/build_${ro.boot.sku_name}.prop',
1420        'import /vendor/etc/build_${ro.boot.device_name}.prop',
1421    ])
1422    self.writeFiles({
1423        'SYSTEM/build.prop': '\n'.join(self.BUILD_PROP),
1424        'VENDOR/build.prop': '\n'.join(vendor_build_prop),
1425        'VENDOR/etc/build_std.prop':
1426        'ro.product.vendor.name=vendor-product-std',
1427        'VENDOR/etc/build_product1.prop':
1428        'ro.product.vendor.device=vendor-device-product1',
1429        'VENDOR/etc/build_pro.prop':
1430        'ro.product.vendor.name=vendor-product-pro',
1431        'VENDOR/etc/build_product2.prop':
1432        'ro.product.vendor.device=vendor-device-product2',
1433    }, self.test_dir)
1434
1435    build_info = common.BuildInfo(common.LoadInfoDict(self.test_dir))
1436    boot_variable_values = {
1437        'ro.boot.sku_name': ['std', 'pro'],
1438        'ro.boot.device_name': ['product1', 'product2'],
1439    }
1440
1441    expected_devices = {'vendor-product-device', 'vendor-device-product1',
1442                        'vendor-device-product2'}
1443    expected_fingerprints = {
1444        self.constructFingerprint(
1445            'vendor-product-brand/vendor-product-name/vendor-product-device'),
1446        self.constructFingerprint(
1447            'vendor-product-brand/vendor-product-std/vendor-device-product1'),
1448        self.constructFingerprint(
1449            'vendor-product-brand/vendor-product-pro/vendor-device-product1'),
1450        self.constructFingerprint(
1451            'vendor-product-brand/vendor-product-std/vendor-device-product2'),
1452        self.constructFingerprint(
1453            'vendor-product-brand/vendor-product-pro/vendor-device-product2')
1454    }
1455    self.assertEqual((expected_devices, expected_fingerprints),
1456                     CalculateRuntimeDevicesAndFingerprints(
1457                         build_info, boot_variable_values))
1458
1459  def test_GetPackageMetadata_full_package(self):
1460    vendor_build_prop = copy.deepcopy(self.VENDOR_BUILD_PROP)
1461    vendor_build_prop.extend([
1462        'import /vendor/etc/build_${ro.boot.sku_name}.prop',
1463    ])
1464    self.writeFiles({
1465        'SYSTEM/build.prop': '\n'.join(self.BUILD_PROP),
1466        'VENDOR/build.prop': '\n'.join(vendor_build_prop),
1467        'VENDOR/etc/build_std.prop':
1468        'ro.product.vendor.name=vendor-product-std',
1469        'VENDOR/etc/build_pro.prop':
1470        'ro.product.vendor.name=vendor-product-pro',
1471    }, self.test_dir)
1472
1473    common.OPTIONS.boot_variable_file = common.MakeTempFile()
1474    with open(common.OPTIONS.boot_variable_file, 'w') as f:
1475      f.write('ro.boot.sku_name=std,pro')
1476
1477    build_info = common.BuildInfo(common.LoadInfoDict(self.test_dir))
1478    metadata = GetPackageMetadata(build_info)
1479    self.assertEqual('vendor-product-device', metadata['pre-device'])
1480    fingerprints = [
1481        self.constructFingerprint(
1482            'vendor-product-brand/vendor-product-name/vendor-product-device'),
1483        self.constructFingerprint(
1484            'vendor-product-brand/vendor-product-pro/vendor-product-device'),
1485        self.constructFingerprint(
1486            'vendor-product-brand/vendor-product-std/vendor-product-device'),
1487    ]
1488    self.assertEqual('|'.join(fingerprints), metadata['post-build'])
1489
1490  def test_GetPackageMetadata_incremental_package(self):
1491    vendor_build_prop = copy.deepcopy(self.VENDOR_BUILD_PROP)
1492    vendor_build_prop.extend([
1493        'import /vendor/etc/build_${ro.boot.sku_name}.prop',
1494    ])
1495    self.writeFiles({
1496        'SYSTEM/build.prop': '\n'.join(self.BUILD_PROP),
1497        'VENDOR/build.prop': '\n'.join(vendor_build_prop),
1498        'VENDOR/etc/build_std.prop':
1499        'ro.product.vendor.device=vendor-device-std',
1500        'VENDOR/etc/build_pro.prop':
1501        'ro.product.vendor.device=vendor-device-pro',
1502    }, self.test_dir)
1503
1504    common.OPTIONS.boot_variable_file = common.MakeTempFile()
1505    with open(common.OPTIONS.boot_variable_file, 'w') as f:
1506      f.write('ro.boot.sku_name=std,pro')
1507
1508    source_dir = common.MakeTempDir()
1509    source_build_prop = [
1510        'ro.build.version.release=source-version-release',
1511        'ro.build.id=source-build-id',
1512        'ro.build.version.incremental=source-version-incremental',
1513        'ro.build.type=build-type',
1514        'ro.build.tags=build-tags',
1515        'ro.build.version.sdk=29',
1516        'ro.build.version.security_patch=2020',
1517        'ro.build.date.utc=12340000'
1518    ]
1519    self.writeFiles({
1520        'META/misc_info.txt': '\n'.join(self.MISC_INFO),
1521        'SYSTEM/build.prop': '\n'.join(source_build_prop),
1522        'VENDOR/build.prop': '\n'.join(vendor_build_prop),
1523        'VENDOR/etc/build_std.prop':
1524        'ro.product.vendor.device=vendor-device-std',
1525        'VENDOR/etc/build_pro.prop':
1526        'ro.product.vendor.device=vendor-device-pro',
1527    }, source_dir)
1528    common.OPTIONS.incremental_source = source_dir
1529
1530    target_info = common.BuildInfo(common.LoadInfoDict(self.test_dir))
1531    source_info = common.BuildInfo(common.LoadInfoDict(source_dir))
1532
1533    metadata = GetPackageMetadata(target_info, source_info)
1534    self.assertEqual(
1535        'vendor-device-pro|vendor-device-std|vendor-product-device',
1536        metadata['pre-device'])
1537    suffix = ':source-version-release/source-build-id/' \
1538             'source-version-incremental:build-type/build-tags'
1539    pre_fingerprints = [
1540        'vendor-product-brand/vendor-product-name/vendor-device-pro'
1541        '{}'.format(suffix),
1542        'vendor-product-brand/vendor-product-name/vendor-device-std'
1543        '{}'.format(suffix),
1544        'vendor-product-brand/vendor-product-name/vendor-product-device'
1545        '{}'.format(suffix),
1546    ]
1547    self.assertEqual('|'.join(pre_fingerprints), metadata['pre-build'])
1548
1549    post_fingerprints = [
1550        self.constructFingerprint(
1551            'vendor-product-brand/vendor-product-name/vendor-device-pro'),
1552        self.constructFingerprint(
1553            'vendor-product-brand/vendor-product-name/vendor-device-std'),
1554        self.constructFingerprint(
1555            'vendor-product-brand/vendor-product-name/vendor-product-device'),
1556    ]
1557    self.assertEqual('|'.join(post_fingerprints), metadata['post-build'])
1558