#!/usr/bin/env python # # Copyright (C) 2020 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. # """Unit tests for apexer.""" import hashlib import logging import os import re import shutil import subprocess import tempfile import unittest from zipfile import ZipFile from apex_manifest import ValidateApexManifest logger = logging.getLogger(__name__) TEST_APEX = "com.android.example.apex" TEST_APEX_LEGACY = "com.android.example-legacy.apex" TEST_APEX_WITH_LOGGING_PARENT = "com.android.example-logging_parent.apex" TEST_APEX_WITH_OVERRIDDEN_PACKAGE_NAME = "com.android.example-overridden_package_name.apex" TEST_PRIVATE_KEY = os.path.join("testdata", "com.android.example.apex.pem") TEST_X509_KEY = os.path.join("testdata", "com.android.example.apex.x509.pem") TEST_PK8_KEY = os.path.join("testdata", "com.android.example.apex.pk8") TEST_AVB_PUBLIC_KEY = os.path.join("testdata", "com.android.example.apex.avbpubkey") def run(args, verbose=None, **kwargs): """Creates and returns a subprocess.Popen object. Args: args: The command represented as a list of strings. verbose: Whether the commands should be shown. Default to the global verbosity if unspecified. kwargs: Any additional args to be passed to subprocess.Popen(), such as env, stdin, etc. stdout and stderr will default to subprocess.PIPE and subprocess.STDOUT respectively unless caller specifies any of them. universal_newlines will default to True, as most of the users in releasetools expect string output. Returns: A subprocess.Popen object. """ if 'stdout' not in kwargs and 'stderr' not in kwargs: kwargs['stdout'] = subprocess.PIPE kwargs['stderr'] = subprocess.STDOUT if 'universal_newlines' not in kwargs: kwargs['universal_newlines'] = True # Don't log any if caller explicitly says so. if DEBUG_TEST: print("\nRunning: \n%s\n" % " ".join(args)) if verbose: logger.info(" Running: \"%s\"", " ".join(args)) return subprocess.Popen(args, **kwargs) def run_host_command(args, verbose=None, **kwargs): host_build_top = os.environ.get("ANDROID_BUILD_TOP") if host_build_top: host_command_dir = os.path.join(host_build_top, "out/soong/host/linux-x86/bin") args[0] = os.path.join(host_command_dir, args[0]) return run_and_check_output(args, verbose, **kwargs) def run_and_check_output(args, verbose=None, **kwargs): """Runs the given command and returns the output. Args: args: The command represented as a list of strings. verbose: Whether the commands should be shown. Default to the global verbosity if unspecified. kwargs: Any additional args to be passed to subprocess.Popen(), such as env, stdin, etc. stdout and stderr will default to subprocess.PIPE and subprocess.STDOUT respectively unless caller specifies any of them. Returns: The output string. Raises: ExternalError: On non-zero exit from the command. """ proc = run(args, verbose=verbose, **kwargs) output, _ = proc.communicate() if output is None: output = "" # Don't log any if caller explicitly says so. if verbose: logger.info("%s", output.rstrip()) if proc.returncode != 0: raise RuntimeError( "Failed to run command '{}' (exit code {}):\n{}".format( args, proc.returncode, output)) return output def get_sha1sum(file_path): h = hashlib.sha256() with open(file_path, 'rb') as file: while True: # Reading is buffered, so we can read smaller chunks. chunk = file.read(h.block_size) if not chunk: break h.update(chunk) return h.hexdigest() def get_current_dir(): """Returns the current dir, relative to the script dir.""" # The script dir is the one we want, which could be different from pwd. current_dir = os.path.dirname(os.path.realpath(__file__)) return current_dir def round_up(size, unit): assert unit & (unit - 1) == 0 return (size + unit - 1) & (~(unit - 1)) # In order to debug test failures, set DEBUG_TEST to True and run the test from # local workstation bypassing atest, e.g.: # $ m apexer_test && out/host/linux-x86/nativetest64/apexer_test/apexer_test # # the test will print out the command used, and the temporary files used by the # test. You need to compare e.g. /tmp/test_simple_apex_input_XXXXXXXX.apex with # /tmp/test_simple_apex_repacked_YYYYYYYY.apex to check where they are # different. # A simple script to analyze the differences: # # FILE_INPUT=/tmp/test_simple_apex_input_XXXXXXXX.apex # FILE_OUTPUT=/tmp/test_simple_apex_repacked_YYYYYYYY.apex # # cd ~/tmp/ # rm -rf input output # mkdir input output # unzip ${FILE_INPUT} -d input/ # unzip ${FILE_OUTPUT} -d output/ # # diff -r input/ output/ # # For analyzing binary diffs I had mild success using the vbindiff utility. DEBUG_TEST = False class ApexerRebuildTest(unittest.TestCase): def setUp(self): self._to_cleanup = [] def tearDown(self): if not DEBUG_TEST: for i in self._to_cleanup: if os.path.isdir(i): shutil.rmtree(i, ignore_errors=True) else: os.remove(i) del self._to_cleanup[:] else: print(self._to_cleanup) def _get_container_files(self, apex_file_path): dir_name = tempfile.mkdtemp(prefix=self._testMethodName+"_container_files_") self._to_cleanup.append(dir_name) with ZipFile(apex_file_path, 'r') as zip_obj: zip_obj.extractall(path=dir_name) files = {} for i in ["apex_manifest.json", "apex_manifest.pb", "apex_build_info.pb", "assets", "apex_payload.img", "apex_payload.zip"]: file_path = os.path.join(dir_name, i) if os.path.exists(file_path): files[i] = file_path self.assertIn("apex_manifest.pb", files) self.assertIn("apex_build_info.pb", files) image_file = None if "apex_payload.img" in files: image_file = files["apex_payload.img"] elif "apex_payload.zip" in files: image_file = files["apex_payload.zip"] self.assertIsNotNone(image_file) files["apex_payload"] = image_file return files def _extract_payload_from_img(self, img_file_path): dir_name = tempfile.mkdtemp(prefix=self._testMethodName+"_extracted_payload_") self._to_cleanup.append(dir_name) cmd = ["debugfs_static", '-R', 'rdump ./ %s' % dir_name, img_file_path] run_host_command(cmd) # Remove payload files added by apexer and e2fs tools. for i in ["apex_manifest.json", "apex_manifest.pb"]: if os.path.exists(os.path.join(dir_name, i)): os.remove(os.path.join(dir_name, i)) if os.path.isdir(os.path.join(dir_name, "lost+found")): shutil.rmtree(os.path.join(dir_name, "lost+found")) return dir_name def _extract_payload(self, apex_file_path): dir_name = tempfile.mkdtemp(prefix=self._testMethodName+"_extracted_payload_") self._to_cleanup.append(dir_name) cmd = ["deapexer", "extract", apex_file_path, dir_name] run_host_command(cmd) # Remove payload files added by apexer and e2fs tools. for i in ["apex_manifest.json", "apex_manifest.pb"]: if os.path.exists(os.path.join(dir_name, i)): os.remove(os.path.join(dir_name, i)) if os.path.isdir(os.path.join(dir_name, "lost+found")): shutil.rmtree(os.path.join(dir_name, "lost+found")) return dir_name def _run_apexer(self, container_files, payload_dir, args=[]): unsigned_payload_only = False payload_only = False if "--unsigned_payload_only" in args: unsigned_payload_only = True if unsigned_payload_only or "--payload_only" in args: payload_only = True os.environ["APEXER_TOOL_PATH"] = ( "out/soong/host/linux-x86/bin:prebuilts/sdk/tools/linux/bin") cmd = ["apexer", "--force", "--include_build_info", "--do_not_check_keyname"] cmd.extend(["--manifest", container_files["apex_manifest.pb"]]) if "apex_manifest.json" in container_files: cmd.extend(["--manifest_json", container_files["apex_manifest.json"]]) cmd.extend(["--build_info", container_files["apex_build_info.pb"]]) if not payload_only and "assets" in container_files: cmd.extend(["--assets_dir", "assets"]) if not unsigned_payload_only: cmd.extend(["--key", os.path.join(get_current_dir(), TEST_PRIVATE_KEY)]) cmd.extend(["--pubkey", os.path.join(get_current_dir(), TEST_AVB_PUBLIC_KEY)]) cmd.extend(args) # Decide on output file name apex_suffix = ".apex.unsigned" if payload_only: apex_suffix = ".payload" fd, fn = tempfile.mkstemp(prefix=self._testMethodName+"_repacked_", suffix=apex_suffix) os.close(fd) self._to_cleanup.append(fn) cmd.extend([payload_dir, fn]) run_host_command(cmd) return fn def _sign_apk_container(self, unsigned_apex): fd, fn = tempfile.mkstemp(prefix=self._testMethodName+"_repacked_", suffix=".apex") os.close(fd) self._to_cleanup.append(fn) cmd = [ "prebuilts/jdk/jdk11/linux-x86/bin/java", "-Djava.library.path=out/soong/host/linux-x86/lib64", "-jar", "out/soong/host/linux-x86/framework/signapk.jar", "-a", "4096", os.path.join(get_current_dir(), TEST_X509_KEY), os.path.join(get_current_dir(), TEST_PK8_KEY), unsigned_apex, fn] run_and_check_output(cmd) return fn def _sign_payload(self, container_files, unsigned_payload): fd, signed_payload = \ tempfile.mkstemp(prefix=self._testMethodName+"_repacked_", suffix=".payload") os.close(fd) self._to_cleanup.append(signed_payload) shutil.copyfile(unsigned_payload, signed_payload) cmd = ['avbtool'] cmd.append('add_hashtree_footer') cmd.append('--do_not_generate_fec') cmd.extend(['--algorithm', 'SHA256_RSA4096']) cmd.extend(['--hash_algorithm', 'sha256']) cmd.extend(['--key', os.path.join(get_current_dir(), TEST_PRIVATE_KEY)]) manifest_apex = ValidateApexManifest(container_files["apex_manifest.pb"]) cmd.extend(['--prop', 'apex.key:' + manifest_apex.name]) # Set up the salt based on manifest content which includes name # and version salt = hashlib.sha256(manifest_apex.SerializeToString()).hexdigest() cmd.extend(['--salt', salt]) cmd.extend(['--image', signed_payload]) cmd.append('--no_hashtree') run_and_check_output(cmd) return signed_payload def _verify_payload(self, payload): """Verifies that the payload is properly signed by avbtool""" cmd = ["avbtool", "verify_image", "--image", payload, "--accept_zeroed_hashtree"] run_and_check_output(cmd) def _run_build_test(self, apex_name): apex_file_path = os.path.join(get_current_dir(), apex_name + ".apex") if DEBUG_TEST: fd, fn = tempfile.mkstemp(prefix=self._testMethodName+"_input_", suffix=".apex") os.close(fd) shutil.copyfile(apex_file_path, fn) self._to_cleanup.append(fn) container_files = self._get_container_files(apex_file_path) payload_dir = self._extract_payload(apex_file_path) repack_apex_file_path = self._run_apexer(container_files, payload_dir) resigned_apex_file_path = self._sign_apk_container(repack_apex_file_path) self.assertEqual(get_sha1sum(apex_file_path), get_sha1sum(resigned_apex_file_path)) def test_simple_apex(self): self._run_build_test(TEST_APEX) def test_legacy_apex(self): self._run_build_test(TEST_APEX_LEGACY) def test_output_payload_only(self): """Assert that payload-only output from apexer is same as the payload we get by unzipping apex. """ apex_file_path = os.path.join(get_current_dir(), TEST_APEX + ".apex") container_files = self._get_container_files(apex_file_path) payload_dir = self._extract_payload(apex_file_path) payload_only_file_path = self._run_apexer(container_files, payload_dir, ["--payload_only"]) self._verify_payload(payload_only_file_path) self.assertEqual(get_sha1sum(payload_only_file_path), get_sha1sum(container_files["apex_payload"])) def test_output_unsigned_payload_only(self): """Assert that when unsigned-payload-only output from apexer is signed by the avb key, it is same as the payload we get by unzipping apex. """ apex_file_path = os.path.join(get_current_dir(), TEST_APEX + ".apex") container_files = self._get_container_files(apex_file_path) payload_dir = self._extract_payload(apex_file_path) unsigned_payload_only_file_path = self._run_apexer(container_files, payload_dir, ["--unsigned_payload_only"]) with self.assertRaises(RuntimeError) as error: self._verify_payload(unsigned_payload_only_file_path) self.assertIn("Given image does not look like a vbmeta image", str(error.exception)) signed_payload = self._sign_payload(container_files, unsigned_payload_only_file_path) self.assertEqual(get_sha1sum(signed_payload), get_sha1sum(container_files["apex_payload"])) # Now assert that given an unsigned image and the original container # files, we can produce an identical unsigned image. unsigned_payload_dir = self._extract_payload_from_img(unsigned_payload_only_file_path) unsigned_payload_only_2_file_path = self._run_apexer(container_files, unsigned_payload_dir, ["--unsigned_payload_only"]) self.assertEqual(get_sha1sum(unsigned_payload_only_file_path), get_sha1sum(unsigned_payload_only_2_file_path)) def test_apex_with_logging_parent(self): self._run_build_test(TEST_APEX_WITH_LOGGING_PARENT) def test_apex_with_overridden_package_name(self): self._run_build_test(TEST_APEX_WITH_OVERRIDDEN_PACKAGE_NAME) if __name__ == '__main__': unittest.main(verbosity=2)