From 904a9f666a2d02f1ddd226b774846b4ea4e3c0c1 Mon Sep 17 00:00:00 2001 From: Andrew Gunnerson Date: Mon, 13 Feb 2023 22:07:19 -0500 Subject: [PATCH] Add end-to-end patching tests Signed-off-by: Andrew Gunnerson --- avbroot.py | 381 +------------------------ avbroot/main.py | 377 ++++++++++++++++++++++++ tests/.gitignore | 1 + tests/README.md | 24 ++ tests/keys/TEST_KEY_DO_NOT_USE_avb.key | 52 ++++ tests/keys/TEST_KEY_DO_NOT_USE_ota.crt | 30 ++ tests/keys/TEST_KEY_DO_NOT_USE_ota.key | 52 ++++ tests/tests.conf | 36 +++ tests/tests.py | 137 +++++++++ 9 files changed, 711 insertions(+), 379 deletions(-) create mode 100644 avbroot/main.py create mode 100644 tests/.gitignore create mode 100644 tests/README.md create mode 100644 tests/keys/TEST_KEY_DO_NOT_USE_avb.key create mode 100644 tests/keys/TEST_KEY_DO_NOT_USE_ota.crt create mode 100644 tests/keys/TEST_KEY_DO_NOT_USE_ota.key create mode 100644 tests/tests.conf create mode 100755 tests/tests.py diff --git a/avbroot.py b/avbroot.py index f420287..ec4f629 100755 --- a/avbroot.py +++ b/avbroot.py @@ -1,383 +1,6 @@ #!/usr/bin/env python3 -from avbroot import external - -import argparse -import concurrent.futures -import os -import shutil -import tempfile -import time -import zipfile - -import avbtool - -from avbroot import boot -from avbroot import openssl -from avbroot import ota -from avbroot import util -from avbroot import vbmeta - - -PATH_METADATA = 'META-INF/com/android/metadata' -PATH_METADATA_PB = f'{PATH_METADATA}.pb' -PATH_OTACERT = 'META-INF/com/android/otacert' -PATH_PAYLOAD = 'payload.bin' -PATH_PROPERTIES = 'payload_properties.txt' - -# Half-open range -MAGISK_MIN_VERSION = 22000 -MAGISK_MAX_VERSION = 25300 - - -def print_status(*args, **kwargs): - print('\x1b[1m*****', *args, '*****\x1b[0m', **kwargs) - - -def print_warning(*args, **kwargs): - print('\x1b[1;31m*****', '[WARNING]', *args, '*****\x1b[0m', **kwargs) - - -def get_images(manifest): - boot_image = 'boot' - otacert_image = None - - for p in manifest.partitions: - # Devices launching with Android 13 use a GKI init_boot ramdisk - if p.partition_name == 'init_boot': - boot_image = p.partition_name - # OnePlus devices have a recovery image - elif p.partition_name == 'recovery': - # If a recovery image exists, it will always contain the OTA certs, - # even if vendor_boot exists - otacert_image = p.partition_name - # Older devices may not have vendor_boot - elif p.partition_name == 'vendor_boot': - if otacert_image is None: - otacert_image = p.partition_name - - images = ['vbmeta', boot_image] - - if otacert_image is not None: - images.append(otacert_image) - else: - otacert_image = boot_image - - return images, boot_image, otacert_image - - -def patch_ota_payload(f_in, f_out, file_size, magisk, privkey_avb, - passphrase_avb, privkey_ota, passphrase_ota, cert_ota): - with tempfile.TemporaryDirectory() as temp_dir: - extract_dir = os.path.join(temp_dir, 'extract') - patch_dir = os.path.join(temp_dir, 'patch') - payload_dir = os.path.join(temp_dir, 'payload') - os.mkdir(extract_dir) - os.mkdir(patch_dir) - os.mkdir(payload_dir) - - version, manifest, blob_offset = ota.parse_payload(f_in) - images, boot_image, otacert_image = get_images(manifest) - - print_status('Extracting', ', '.join(images), 'from the payload') - ota.extract_images(f_in, manifest, blob_offset, extract_dir, images) - - image_patches = {boot_image: [boot.MagiskRootPatch(magisk)]} - image_patches.setdefault(otacert_image, []).append( - boot.OtaCertPatch(cert_ota)) - - avb = avbtool.Avb() - - print_status('Patching', ', '.join(image_patches)) - with concurrent.futures.ThreadPoolExecutor( - max_workers=len(image_patches)) as executor: - def apply_patches(image, patches): - boot.patch_boot( - avb, - os.path.join(extract_dir, f'{image}.img'), - os.path.join(patch_dir, f'{image}.img'), - privkey_avb, - passphrase_avb, - True, - patches, - ) - - futures = [executor.submit(apply_patches, i, p) - for i, p in image_patches.items()] - - for future in concurrent.futures.as_completed(futures): - future.result() - - print_status('Building new root vbmeta image') - vbmeta.patch_vbmeta_root( - avb, - [os.path.join(patch_dir, f'{i}.img') - for i in images if i != 'vbmeta'], - os.path.join(extract_dir, 'vbmeta.img'), - os.path.join(patch_dir, 'vbmeta.img'), - privkey_avb, - passphrase_avb, - manifest.block_size, - ) - - print_status('Updating OTA payload to reference patched images') - return ota.patch_payload( - f_in, - f_out, - version, - manifest, - blob_offset, - payload_dir, - {i: os.path.join(patch_dir, f'{i}.img') for i in images}, - file_size, - privkey_ota, - passphrase_ota, - ) - - -def patch_ota_zip(f_zip_in, f_zip_out, magisk, privkey_avb, passphrase_avb, - privkey_ota, passphrase_ota, cert_ota): - with ( - zipfile.ZipFile(f_zip_in, 'r') as z_in, - zipfile.ZipFile(f_zip_out, 'w') as z_out, - ): - infolist = z_in.infolist() - missing = { - PATH_METADATA, - PATH_METADATA_PB, - PATH_OTACERT, - PATH_PAYLOAD, - PATH_PROPERTIES, - } - i_payload = -1 - i_properties = -1 - - for i, info in enumerate(infolist): - if info.filename in missing: - missing.remove(info.filename) - - if info.filename == PATH_PAYLOAD: - i_payload = i - elif info.filename == PATH_PROPERTIES: - i_properties = i - - if not missing and i_payload >= 0 and i_properties >= 0: - break - - if missing: - raise Exception(f'Missing files in zip: {missing}') - - # Ensure payload is processed before properties - if i_payload > i_properties: - infolist[i_payload], infolist[i_properties] = \ - infolist[i_properties], infolist[i_payload] - - properties = None - metadata_info = None - metadata_pb_info = None - metadata_pb_raw = None - - for info in infolist: - # Ignore because the plain-text legacy metadata file is regenerated - # from the new metadata - if info.filename == PATH_METADATA: - metadata_info = info - continue - - # The existing metadata is needed to generate a new signed zip - elif info.filename == PATH_METADATA_PB: - metadata_pb_info = info - - with z_in.open(info, 'r') as f_in: - metadata_pb_raw = f_in.read() - - continue - - # Use the user's OTA certificate - elif info.filename == PATH_OTACERT: - print_status('Replacing', info.filename) - - with ( - open(cert_ota, 'rb') as f_cert, - z_out.open(info, 'w') as f_out, - ): - shutil.copyfileobj(f_cert, f_out) - - continue - - # Copy other files, patching if needed - with ( - z_in.open(info, 'r') as f_in, - z_out.open(info, 'w') as f_out, - ): - if info.filename == PATH_PAYLOAD: - print_status('Patching', info.filename) - - if info.compress_type != zipfile.ZIP_STORED: - raise Exception( - f'{info.filename} is not stored uncompressed') - - properties = patch_ota_payload( - f_in, - f_out, - info.file_size, - magisk, - privkey_avb, - passphrase_avb, - privkey_ota, - passphrase_ota, - cert_ota, - ) - - elif info.filename == PATH_PROPERTIES: - print_status('Patching', info.filename) - - if info.compress_type != zipfile.ZIP_STORED: - raise Exception( - f'{info.filename} is not stored uncompressed') - - f_out.write(properties) - - else: - print_status('Copying', info.filename) - - shutil.copyfileobj(f_in, f_out) - - print_status('Generating', PATH_METADATA, 'and', PATH_METADATA_PB) - metadata = ota.add_metadata( - z_out, - metadata_info, - metadata_pb_info, - metadata_pb_raw, - ) - - # Signing process needs to capture the zip central directory - f_zip_out.start_capture() - - return metadata - - -def get_magisk_version(magisk): - with zipfile.ZipFile(magisk, 'r') as z: - with z.open('assets/util_functions.sh', 'r') as f: - for line in f: - if line.startswith(b'MAGISK_VER_CODE='): - return int(line[16:].strip()) - - raise Exception(f'Failed to get Magisk version from: {magisk}') - - -def patch_subcommand(args): - output = args.output - if output is None: - output = args.input + '.patched' - - magisk_version = get_magisk_version(args.magisk) - if magisk_version < MAGISK_MIN_VERSION or \ - magisk_version >= MAGISK_MAX_VERSION: - message = f'Unsupported Magisk version {magisk_version} ' \ - f'(supported: >={MAGISK_MIN_VERSION}, <{MAGISK_MAX_VERSION})' - - if args.ignore_magisk_version: - print_warning(message) - else: - raise Exception(message) - - # Get passphrases for keys - passphrase_avb = openssl.prompt_passphrase(args.privkey_avb) - passphrase_ota = openssl.prompt_passphrase(args.privkey_ota) - - # Ensure that the certificate matches the private key - if not openssl.cert_matches_key(args.cert_ota, args.privkey_ota, - passphrase_ota): - raise Exception('OTA certificate does not match private key') - - start = time.perf_counter_ns() - - with util.open_output_file(output) as temp_raw: - with ( - ota.open_signing_wrapper(temp_raw, args.privkey_ota, - passphrase_ota, args.cert_ota) as temp, - ota.match_android_zip64_limit(), - ): - metadata = patch_ota_zip( - args.input, - temp, - args.magisk, - args.privkey_avb, - passphrase_avb, - args.privkey_ota, - passphrase_ota, - args.cert_ota, - ) - - # We do a lot of low-level hackery. Reopen and verify offsets - print_status('Verifying metadata offsets') - with zipfile.ZipFile(temp_raw, 'r') as z: - ota.verify_metadata(z, metadata) - - # Excluding the time it takes for the user to type in the passwords - elapsed = time.perf_counter_ns() - start - print_status(f'Completed after {elapsed / 1_000_000_000:.1f}s') - - -def extract_subcommand(args): - with zipfile.ZipFile(args.input, 'r') as z: - info = z.getinfo(PATH_PAYLOAD) - - with z.open(info, 'r') as f: - _, manifest, blob_offset = ota.parse_payload(f) - images, _, _ = get_images(manifest) - - print_status('Extracting', ', '.join(images), 'from the payload') - os.makedirs(args.directory, exist_ok=True) - ota.extract_images(f, manifest, blob_offset, args.directory, - images) - - -def parse_args(): - parser = argparse.ArgumentParser() - subparsers = parser.add_subparsers(dest='subcommand', required=True, - help='Subcommands') - - patch = subparsers.add_parser('patch', help='Patch a full OTA zip') - - patch.add_argument('--input', required=True, - help='Path to original raw payload or OTA zip') - patch.add_argument('--output', - help='Path to new raw payload or OTA zip') - patch.add_argument('--magisk', required=True, - help='Path to Magisk API') - patch.add_argument('--ignore-magisk-version', action='store_true', - help='Allow patching with unsupported Magisk versions') - patch.add_argument('--privkey-avb', required=True, - help='Private key for signing root vbmeta image') - patch.add_argument('--privkey-ota', required=True, - help='Private key for signing OTA payload') - patch.add_argument('--cert-ota', required=True, - help='Certificate for OTA payload signing key') - - extract = subparsers.add_parser( - 'extract', help='Extract patched images from a patched OTA zip') - - extract.add_argument('--input', required=True, - help='Path to patched OTA zip') - extract.add_argument('--directory', default='.', - help='Output directory for extracted images') - - return parser.parse_args() - - -def main(): - args = parse_args() - - if args.subcommand == 'patch': - patch_subcommand(args) - elif args.subcommand == 'extract': - extract_subcommand(args) - else: - raise NotImplementedError() - +from avbroot import main if __name__ == '__main__': - main() + main.main() diff --git a/avbroot/main.py b/avbroot/main.py new file mode 100644 index 0000000..d7fabd9 --- /dev/null +++ b/avbroot/main.py @@ -0,0 +1,377 @@ +from . import external + +import argparse +import concurrent.futures +import os +import shutil +import tempfile +import time +import zipfile + +import avbtool + +from . import boot +from . import openssl +from . import ota +from . import util +from . import vbmeta + + +PATH_METADATA = 'META-INF/com/android/metadata' +PATH_METADATA_PB = f'{PATH_METADATA}.pb' +PATH_OTACERT = 'META-INF/com/android/otacert' +PATH_PAYLOAD = 'payload.bin' +PATH_PROPERTIES = 'payload_properties.txt' + +# Half-open range +MAGISK_MIN_VERSION = 22000 +MAGISK_MAX_VERSION = 25300 + + +def print_status(*args, **kwargs): + print('\x1b[1m*****', *args, '*****\x1b[0m', **kwargs) + + +def print_warning(*args, **kwargs): + print('\x1b[1;31m*****', '[WARNING]', *args, '*****\x1b[0m', **kwargs) + + +def get_images(manifest): + boot_image = 'boot' + otacert_image = None + + for p in manifest.partitions: + # Devices launching with Android 13 use a GKI init_boot ramdisk + if p.partition_name == 'init_boot': + boot_image = p.partition_name + # OnePlus devices have a recovery image + elif p.partition_name == 'recovery': + # If a recovery image exists, it will always contain the OTA certs, + # even if vendor_boot exists + otacert_image = p.partition_name + # Older devices may not have vendor_boot + elif p.partition_name == 'vendor_boot': + if otacert_image is None: + otacert_image = p.partition_name + + images = ['vbmeta', boot_image] + + if otacert_image is not None: + images.append(otacert_image) + else: + otacert_image = boot_image + + return images, boot_image, otacert_image + + +def patch_ota_payload(f_in, f_out, file_size, magisk, privkey_avb, + passphrase_avb, privkey_ota, passphrase_ota, cert_ota): + with tempfile.TemporaryDirectory() as temp_dir: + extract_dir = os.path.join(temp_dir, 'extract') + patch_dir = os.path.join(temp_dir, 'patch') + payload_dir = os.path.join(temp_dir, 'payload') + os.mkdir(extract_dir) + os.mkdir(patch_dir) + os.mkdir(payload_dir) + + version, manifest, blob_offset = ota.parse_payload(f_in) + images, boot_image, otacert_image = get_images(manifest) + + print_status('Extracting', ', '.join(images), 'from the payload') + ota.extract_images(f_in, manifest, blob_offset, extract_dir, images) + + image_patches = {boot_image: [boot.MagiskRootPatch(magisk)]} + image_patches.setdefault(otacert_image, []).append( + boot.OtaCertPatch(cert_ota)) + + avb = avbtool.Avb() + + print_status('Patching', ', '.join(image_patches)) + with concurrent.futures.ThreadPoolExecutor( + max_workers=len(image_patches)) as executor: + def apply_patches(image, patches): + boot.patch_boot( + avb, + os.path.join(extract_dir, f'{image}.img'), + os.path.join(patch_dir, f'{image}.img'), + privkey_avb, + passphrase_avb, + True, + patches, + ) + + futures = [executor.submit(apply_patches, i, p) + for i, p in image_patches.items()] + + for future in concurrent.futures.as_completed(futures): + future.result() + + print_status('Building new root vbmeta image') + vbmeta.patch_vbmeta_root( + avb, + [os.path.join(patch_dir, f'{i}.img') + for i in images if i != 'vbmeta'], + os.path.join(extract_dir, 'vbmeta.img'), + os.path.join(patch_dir, 'vbmeta.img'), + privkey_avb, + passphrase_avb, + manifest.block_size, + ) + + print_status('Updating OTA payload to reference patched images') + return ota.patch_payload( + f_in, + f_out, + version, + manifest, + blob_offset, + payload_dir, + {i: os.path.join(patch_dir, f'{i}.img') for i in images}, + file_size, + privkey_ota, + passphrase_ota, + ) + + +def patch_ota_zip(f_zip_in, f_zip_out, magisk, privkey_avb, passphrase_avb, + privkey_ota, passphrase_ota, cert_ota): + with ( + zipfile.ZipFile(f_zip_in, 'r') as z_in, + zipfile.ZipFile(f_zip_out, 'w') as z_out, + ): + infolist = z_in.infolist() + missing = { + PATH_METADATA, + PATH_METADATA_PB, + PATH_OTACERT, + PATH_PAYLOAD, + PATH_PROPERTIES, + } + i_payload = -1 + i_properties = -1 + + for i, info in enumerate(infolist): + if info.filename in missing: + missing.remove(info.filename) + + if info.filename == PATH_PAYLOAD: + i_payload = i + elif info.filename == PATH_PROPERTIES: + i_properties = i + + if not missing and i_payload >= 0 and i_properties >= 0: + break + + if missing: + raise Exception(f'Missing files in zip: {missing}') + + # Ensure payload is processed before properties + if i_payload > i_properties: + infolist[i_payload], infolist[i_properties] = \ + infolist[i_properties], infolist[i_payload] + + properties = None + metadata_info = None + metadata_pb_info = None + metadata_pb_raw = None + + for info in infolist: + # Ignore because the plain-text legacy metadata file is regenerated + # from the new metadata + if info.filename == PATH_METADATA: + metadata_info = info + continue + + # The existing metadata is needed to generate a new signed zip + elif info.filename == PATH_METADATA_PB: + metadata_pb_info = info + + with z_in.open(info, 'r') as f_in: + metadata_pb_raw = f_in.read() + + continue + + # Use the user's OTA certificate + elif info.filename == PATH_OTACERT: + print_status('Replacing', info.filename) + + with ( + open(cert_ota, 'rb') as f_cert, + z_out.open(info, 'w') as f_out, + ): + shutil.copyfileobj(f_cert, f_out) + + continue + + # Copy other files, patching if needed + with ( + z_in.open(info, 'r') as f_in, + z_out.open(info, 'w') as f_out, + ): + if info.filename == PATH_PAYLOAD: + print_status('Patching', info.filename) + + if info.compress_type != zipfile.ZIP_STORED: + raise Exception( + f'{info.filename} is not stored uncompressed') + + properties = patch_ota_payload( + f_in, + f_out, + info.file_size, + magisk, + privkey_avb, + passphrase_avb, + privkey_ota, + passphrase_ota, + cert_ota, + ) + + elif info.filename == PATH_PROPERTIES: + print_status('Patching', info.filename) + + if info.compress_type != zipfile.ZIP_STORED: + raise Exception( + f'{info.filename} is not stored uncompressed') + + f_out.write(properties) + + else: + print_status('Copying', info.filename) + + shutil.copyfileobj(f_in, f_out) + + print_status('Generating', PATH_METADATA, 'and', PATH_METADATA_PB) + metadata = ota.add_metadata( + z_out, + metadata_info, + metadata_pb_info, + metadata_pb_raw, + ) + + # Signing process needs to capture the zip central directory + f_zip_out.start_capture() + + return metadata + + +def get_magisk_version(magisk): + with zipfile.ZipFile(magisk, 'r') as z: + with z.open('assets/util_functions.sh', 'r') as f: + for line in f: + if line.startswith(b'MAGISK_VER_CODE='): + return int(line[16:].strip()) + + raise Exception(f'Failed to get Magisk version from: {magisk}') + + +def patch_subcommand(args): + output = args.output + if output is None: + output = args.input + '.patched' + + magisk_version = get_magisk_version(args.magisk) + if magisk_version < MAGISK_MIN_VERSION or \ + magisk_version >= MAGISK_MAX_VERSION: + message = f'Unsupported Magisk version {magisk_version} ' \ + f'(supported: >={MAGISK_MIN_VERSION}, <{MAGISK_MAX_VERSION})' + + if args.ignore_magisk_version: + print_warning(message) + else: + raise Exception(message) + + # Get passphrases for keys + passphrase_avb = openssl.prompt_passphrase(args.privkey_avb) + passphrase_ota = openssl.prompt_passphrase(args.privkey_ota) + + # Ensure that the certificate matches the private key + if not openssl.cert_matches_key(args.cert_ota, args.privkey_ota, + passphrase_ota): + raise Exception('OTA certificate does not match private key') + + start = time.perf_counter_ns() + + with util.open_output_file(output) as temp_raw: + with ( + ota.open_signing_wrapper(temp_raw, args.privkey_ota, + passphrase_ota, args.cert_ota) as temp, + ota.match_android_zip64_limit(), + ): + metadata = patch_ota_zip( + args.input, + temp, + args.magisk, + args.privkey_avb, + passphrase_avb, + args.privkey_ota, + passphrase_ota, + args.cert_ota, + ) + + # We do a lot of low-level hackery. Reopen and verify offsets + print_status('Verifying metadata offsets') + with zipfile.ZipFile(temp_raw, 'r') as z: + ota.verify_metadata(z, metadata) + + # Excluding the time it takes for the user to type in the passwords + elapsed = time.perf_counter_ns() - start + print_status(f'Completed after {elapsed / 1_000_000_000:.1f}s') + + +def extract_subcommand(args): + with zipfile.ZipFile(args.input, 'r') as z: + info = z.getinfo(PATH_PAYLOAD) + + with z.open(info, 'r') as f: + _, manifest, blob_offset = ota.parse_payload(f) + images, _, _ = get_images(manifest) + + print_status('Extracting', ', '.join(images), 'from the payload') + os.makedirs(args.directory, exist_ok=True) + ota.extract_images(f, manifest, blob_offset, args.directory, + images) + + +def parse_args(args=None): + parser = argparse.ArgumentParser() + subparsers = parser.add_subparsers(dest='subcommand', required=True, + help='Subcommands') + + patch = subparsers.add_parser('patch', help='Patch a full OTA zip') + + patch.add_argument('--input', required=True, + help='Path to original raw payload or OTA zip') + patch.add_argument('--output', + help='Path to new raw payload or OTA zip') + patch.add_argument('--magisk', required=True, + help='Path to Magisk API') + patch.add_argument('--ignore-magisk-version', action='store_true', + help='Allow patching with unsupported Magisk versions') + patch.add_argument('--privkey-avb', required=True, + help='Private key for signing root vbmeta image') + patch.add_argument('--privkey-ota', required=True, + help='Private key for signing OTA payload') + patch.add_argument('--cert-ota', required=True, + help='Certificate for OTA payload signing key') + + extract = subparsers.add_parser( + 'extract', help='Extract patched images from a patched OTA zip') + + extract.add_argument('--input', required=True, + help='Path to patched OTA zip') + extract.add_argument('--directory', default='.', + help='Output directory for extracted images') + + return parser.parse_args(args=args) + + +def main(args=None): + args = parse_args(args=args) + + if args.subcommand == 'patch': + patch_subcommand(args) + elif args.subcommand == 'extract': + extract_subcommand(args) + else: + raise NotImplementedError() diff --git a/tests/.gitignore b/tests/.gitignore new file mode 100644 index 0000000..bb97907 --- /dev/null +++ b/tests/.gitignore @@ -0,0 +1 @@ +/files/ diff --git a/tests/README.md b/tests/README.md new file mode 100644 index 0000000..b9d0d98 --- /dev/null +++ b/tests/README.md @@ -0,0 +1,24 @@ +# avbroot tests + +avbroot's output files are reproducible for the given input files. [`tests.conf`](./tests.conf) lists some OTA images with unique properties and the expected checksums before and after patching. These tests use pregenerated, hardcoded test keys for signing. **These keys should NEVER be used for any other purpose.** + +For each image listed in the config, the test process will: + +1. Download the OTA if it doesn't already exist in `tests/files//` +2. Verify the OTA checksum +3. Run avbroot against the OTA +4. Verify the patched OTA checksum + +## Running the tests + +aria2 must be installed before running the tests. It is used for downloading the OTA images. + +To test against all of the device OTA images listed in [`tests.conf`](./tests.conf), run: + +```bash +python tests/tests.py +``` + +To only run the tests for a specific device, pass in `-d `. This argument can be specified multiple times (eg. `-d GooglePixel7Pro -d GooglePixel6a`). + +To pass in extra arguments to `aria2c`, use `-a=`. This can be used to enable concurrent downloads (eg. `-a=-x4 -a=-s4`). diff --git a/tests/keys/TEST_KEY_DO_NOT_USE_avb.key b/tests/keys/TEST_KEY_DO_NOT_USE_avb.key new file mode 100644 index 0000000..ea9c2b2 --- /dev/null +++ b/tests/keys/TEST_KEY_DO_NOT_USE_avb.key @@ -0,0 +1,52 @@ +-----BEGIN PRIVATE KEY----- +MIIJQQIBADANBgkqhkiG9w0BAQEFAASCCSswggknAgEAAoICAQCs7h/Ap5XZFnSq +609V9NNrpviZ3bVDvl0cyHjUXXdv5iEaMwwIQzIjNLaQ27udwT+kdkKrx1iJ27p5 +yCqduhnGTUwJnzYkB3WDTp2GAT9DOhqqsL7R8iEtxTzy2Wfe1TIJweXrW1Y9ANMH +wCG50tCE6AzdIKaVIlzjHn+0B8zr3sXNoutE7Qed8+3zF1HnfOQ66Wgbx/bMTliU +U2a7ZVVQ4H3SFetRaBXpRX/u+TAbdy+SKR/VdYeC8gc0uax150afQv/UBVrkOoMB +o35W39pO3l5jn+xXq3Jo2HhxkCaJUqJ+a44//FA3Wa8xfaSgdv02/mSZ8V3ifYHc +pWb1AIONE6RCsB7NH/22Veou93m6sTzY6rZJQCzBo/gURO0kcXp8P6ZawrjTYlRQ +quFbddPcN+li9RZ5nLVVBj7rJiu1H5r0ZfyZLgfzqmaRfOCGq9mmBxbwT+FV7+eQ +HkEypNzc3PfbI+Tq3ms2ZL7oen/UxrMke+6wVTJjhF1lR+/acF9YyrkR2amPt83Y +B7/b89vx+BrD6De6depVEr5yRr/naKoe3yPEaduH35gLu5Of0fKJSDQIgjUOTFjw +RQMp+AiNprkItMTHhonrGLPAa2g7EuUYveXNV+f7Q3kt+Uw49kNnXlySscCpX0Ul +aWmkvG6h22aCdF1CPMZ2+X9Xzy1JPQIDAQABAoICAAD5CJ4G01BN+14IG4F6R8OF +RES+pd8PBRW9CEukMvXNhPDRdLHfNDr/zZRxsqkn2yR2XdhQM26jGSDHlXse/uIf +W9wkqc34v2/Q/hCGP/AwNXzcUwEkrDcbsu9oIFjKjjuBJw/0iummRYmiSmOcjKZo +FRpxV+k4MO7ozwMY6s2GNUqmLfih9LcDa5qVzlTPs7ZOddLM2dXluEMtV5iaRf6X +JLNigtkJuDCAaZXTOl1ihZam5eec1Pmm7uTHCUdTLmcwSKF4COjbw2eQ9IxR5LfT +hxMRupH2WZsUDKlm4YXy5w3Fq8kvBtco3qltRth4PY2/pJqRFwbHz0vfH6h6gO2Z +2i2/jGadcYyGTq0WVQK3IaWu1tqJ6nbOfrG/Wk0l1XKCkEoAOqQ3eXv8iGCyTe+3 +H8CU+XS92gAW77ouPUX/LhUQwa6VLlc+uFBDV3Ddr1KXMpPSLp//C1MDkbSkt08b +nyOMzcnlxRJUJ7cQ5t8RwEZEHH9IV1JPno2lBlCorPnCbTawtARPZNtmQhTbPCU0 +8Oc7mZ+oFXIHKcaaIgpuRvWYlouyHiiKHS5jnNBUKQXZqZZ9qF7JKgcEb6yOJK+g +bHEP4KMgMG2dHwrPq1h6ahzC5y6awMO4h3hyGrRwKkq0ABCyFapgyJqjQyWWr5Hq +K8ujyFIHW10dinxrog/pAoIBAQDtpW/0Qq43AlKyNEvrAuOxDDE3tP3Gl461sSM0 +sNoJwTltR1o/thP8f7uboUzZJmSEVoaVghKOeT6B96PDMldkSZiXKG5QNconIfdv +bHTXhv02B0tk2c167Zb/OHDbzyHAvZ/Tp1bBosEdCCrpGQDGMSOW6wuuruW77S/T +fcq6ENB1Rf+ceZaDLQ1NFL5nSfli7Dz52sSONlw2x1urZ3dUTeYvVaGvbp0Yk93y +wGaPUnK3UIcdLdLMWBLxqv+JTLlBhUm4kRKkAOMMs3PnZPxAiwYpt7tDozXTmHj+ +siMEExFn75mnSgXxBchl3QzisUKH452Qkb+YtXhTyMNBMUF/AoIBAQC6SStXw3jT +ksoL9jEL7u/meiiEzWprdiTXA6YJQfYO5BMh5yyX3APUrZuHW+e3Fonx+dXZxEgO +pTa79QcQBqCKp8GIaEz8qKple+8CFfNrkeO13PDGJfZmonBmSeHlyc1xsrSkBu3D +sMfBbxB3hWbIBe0vBBBk0jMPl5shJW9QQM4wM8zbezC0DJEDtXOqbY5Hy81o6FsG +4UL34JFYsz5TR3pGKKw4cY2KGoNWknRNQpGxOJP5B9muqFGpbp+uOA30My1nWPhc +mHHoXO6Cg9Zdvxm+WIoF5J2nEKM4TGZaxFVL/zz8yaT4ccLzfwpPcfX1RmG1/hnH +q7+GDD/Bf1tDAoIBAEKbySpcejoe45DhSZogYDNiqHTlvO+p0S0p7iU5oJN1XiQ8 +CJas8SMnpizz4/dFZI0XN3e2Un8uio+hGnx+orIQUr2jbm/686NOSO0pKAK06kn3 +d3FX/tkCuPzmE/P4NYyxWZfYYl+5v4jkniphgK8MXN5TGR4oFWo8C9iqfGT0i4kj +SobQKZiwyvac5tVKjaTldB3P7KGzqNsNu6rqQC8X+nmv9baXOV8HKwF8XgKz6EVr +Q8Ow8VGnKgAknqUcJRyxQD2EJAZl3T3VGppz2qQ4k0ykG8/UciTRqrAwu9HLimho +ZtfVzp6EsWujrpXney9dY+9+eoZR7t0E+31NUf8CggEAIUG2zzBCfeMS1xfYobzC +b28XlL3nZc/tu8UW+0gfWYE2OTNALsxEks97mKq7r2ACuIRVvU7hXlv4bY3M+Pt1 +5csVLfkQ5XpnET8S6Fhq4KVm9ay//1//oUsWrew6u/9PAiFZRn9GK02AG/RsCCp0 +svj74wzisF0maH40X7KDMNyg8tIMnJJdm4cyzGq0f5SXdhHYVgZFWfEVN5OL3z4A +vnVgSu4v8NO05Rbeo7H5uWFNCUqEgEz6e8fo7Eqnjz7KnvIUBQNJM56zlzwgCITo +3EulmOnNmke6LX025nIdQ88RO0QR6z9EiGYm27muaYh61JMAFrGKgK8UviWprC3E +aQKCAQB//1qiek8Vhe9zi/5Q2G2dY8oFS1OYOBfbFDUKOzd3S3IxAQ33kWLSalqs +dw3h8sazX1ejRlx6N2vnpxFo0Vk/JQ1lCMjiQl/RgIRuFPUyceMmwfI/pt5Y5ldF +f1yvFRaWHsVpQ9sQTzT6+LRfe9JzFFuBw2pdtmMD1d0Nhj06ZTT5K0iDY1UwRltC +yLm1bGOFNMP6NxG7vAVMKlKE3T6OGeCH7QlXt2gZ6uy/SNHB8zWcc+eCsgEGAiUL +ZJ/FtTy2m5J1ERsoSPZuEeCFPj5cWhQ8LFGm3ks4E2edEkvplF1l3lDMqxA2l9GP +C2N4cxXcIba2WaeXocqQVlnTRxhf +-----END PRIVATE KEY----- diff --git a/tests/keys/TEST_KEY_DO_NOT_USE_ota.crt b/tests/keys/TEST_KEY_DO_NOT_USE_ota.crt new file mode 100644 index 0000000..e658328 --- /dev/null +++ b/tests/keys/TEST_KEY_DO_NOT_USE_ota.crt @@ -0,0 +1,30 @@ +-----BEGIN CERTIFICATE----- +MIIFHzCCAwegAwIBAgIUE59WLtF/rqMDdNzNdj6ZJK9J2+IwDQYJKoZIhvcNAQEL +BQAwHjEcMBoGA1UEAwwTVEVTVCBLRVkgRE8gTk9UIFVTRTAgFw0yMzAyMTQwMDAw +MTJaGA8yMDUwMDcwMjAwMDAxMlowHjEcMBoGA1UEAwwTVEVTVCBLRVkgRE8gTk9U +IFVTRTCCAiIwDQYJKoZIhvcNAQEBBQADggIPADCCAgoCggIBAJV0bSK/ObnZ6gyg +O89pyyTfg1e/PcFDFQX0ODot9JwSGpC54+FevccrPHlSvdhRzX/qBS3IchuwHK0x +HcyaQMRszh8Uc9sk/C4aAlP+TZX5B7OrmO65xlrmLb6l6tL/B35ZMV+Zs4Aiq28Z +8cWbdk2xtR5gzsuLG3CqFhyQG1/p/h64IVMm4yQT5wjNIUkPkrTXPURtF3C9Q6Wu +/9a2FtXLj72gGQtMOt1prgXWgRC+kFPWGqJHBtImUzq/Ijoc4DazOv6e9WSD1G3o +lHfW8VIX4IQnqp5ODd24/5cKapj59WYbzWUiTaOyIt1zgu3Qj67JM5t8TnTCxFIr +Hfaq33NZAVpIjo+x1A44eqTqdZsBSN4CIH3npBIzw4s7D5+6Laat1pCndamehqmK +ICsuA/bF1WBzJMwgqRQLbb9iAhOARslXLmVylq4FboZxNVPcpsBjCWOtwSkVTbXB +vwoDQRpHdJCHl+myK8UzPuydsWZ8ohHLmCCXurU6T0HDaukbwxvN1qTAJCN40Wrr +VtOuzl4qsHV0e32MAKIZnnk8lCXaqkbzr5EOjlSOJtWZKFh+2mnUGb4oJt/nLWl/ +XuZau09bK2g5RTSAOb2MVRPsdOaOYc8WgrInwFDXXzgK2QPHRsysSdbThLEeJvi7 +EiXJzwNvJd+uG2nxBwdYpQt1l9mtAgMBAAGjUzBRMB0GA1UdDgQWBBTBpc5TQSIl +6odi9EMjhsZvUOVGnTAfBgNVHSMEGDAWgBTBpc5TQSIl6odi9EMjhsZvUOVGnTAP +BgNVHRMBAf8EBTADAQH/MA0GCSqGSIb3DQEBCwUAA4ICAQB5JKOzNylxSGaCR+4Q +qjt83gvIczRC08mgvbDi2Z90FBaT6OLXztSmW49BICvfubcWpgb5PtW76+ict5jh +9GlJTdjaDSCHnQQOvjkO/rNhdMT8G7tmGHn+I7Jj5QMidfBfRdEUuCOp5J10xGlR +bUJGt8eqVuqmvzbNHiH/xnrAgVe1WviZjeRIsp7wZ5nn+y6FxB/RezWZ3NKBnKEX +42qs6VSUgTuylYj1aGNC5+ULyZ1Hb6bOlgGREoIssE0JOByXTgpSskEh3wITMRZf +zs40f0Rf79lNvdhBqzOH7SZhe3TUqbGwzpWT/jLw5YelRCfkAYlSGPyDiYdLeVLx +ddikwbFPAvgtR6YGzNotAMhoMooRtlW3ApE6htz9SVG0sh4bG+8BKf/R6VtYzhxS +7P5SkP69fLeISkG3R/q7VV0ts3QGv/JxIAE30b5eaVGTdMEhW/CqZrvUUsbZwiFt +19IGM2uJNem9+RNT+1J6gT5rmpXJ9a4Fdp104CQJrdM4cAGvz5qydc/W3+e1E/bx +Zr4BIGQYeFYOSeoLOzOUn3rLOJTOh7kGFY1L8S72rJqYw/Ta8nso1QcRWLbqU5nZ +gABZu+DFdxyAF1CO24tH5Sf6Iwb2T3k5QHeK3dcjwxbd1thcdSAI+V37rhWFAxN6 +lbg+zoAbDmYZbtD54lEsbgaIuA== +-----END CERTIFICATE----- diff --git a/tests/keys/TEST_KEY_DO_NOT_USE_ota.key b/tests/keys/TEST_KEY_DO_NOT_USE_ota.key new file mode 100644 index 0000000..eabc7fe --- /dev/null +++ b/tests/keys/TEST_KEY_DO_NOT_USE_ota.key @@ -0,0 +1,52 @@ +-----BEGIN PRIVATE KEY----- +MIIJQwIBADANBgkqhkiG9w0BAQEFAASCCS0wggkpAgEAAoICAQCVdG0ivzm52eoM +oDvPacsk34NXvz3BQxUF9Dg6LfScEhqQuePhXr3HKzx5Ur3YUc1/6gUtyHIbsByt +MR3MmkDEbM4fFHPbJPwuGgJT/k2V+Qezq5juucZa5i2+perS/wd+WTFfmbOAIqtv +GfHFm3ZNsbUeYM7LixtwqhYckBtf6f4euCFTJuMkE+cIzSFJD5K01z1EbRdwvUOl +rv/WthbVy4+9oBkLTDrdaa4F1oEQvpBT1hqiRwbSJlM6vyI6HOA2szr+nvVkg9Rt +6JR31vFSF+CEJ6qeTg3duP+XCmqY+fVmG81lIk2jsiLdc4Lt0I+uyTObfE50wsRS +Kx32qt9zWQFaSI6PsdQOOHqk6nWbAUjeAiB956QSM8OLOw+fui2mrdaQp3Wpnoap +iiArLgP2xdVgcyTMIKkUC22/YgITgEbJVy5lcpauBW6GcTVT3KbAYwljrcEpFU21 +wb8KA0EaR3SQh5fpsivFMz7snbFmfKIRy5ggl7q1Ok9Bw2rpG8MbzdakwCQjeNFq +61bTrs5eKrB1dHt9jACiGZ55PJQl2qpG86+RDo5UjibVmShYftpp1Bm+KCbf5y1p +f17mWrtPWytoOUU0gDm9jFUT7HTmjmHPFoKyJ8BQ1184CtkDx0bMrEnW04SxHib4 +uxIlyc8DbyXfrhtp8QcHWKULdZfZrQIDAQABAoICAETAXkQRu7hnKmfMfjcX779y +orUG4J27AjzO796zrUbufRH+sXnHX33zwcn96h9M4j/po6BACV37UfXKFm88tnal +ptxdSD6TcP60MEX7Qi2vdX+NfLi09S5znK+LG88cSpIw7amQxyY2zK47PSEuNune +yfbDid3QjDrzw2A4Wp9wwNnY0luyE+NQ1IMT+i/l3hMawLBtjs7qAeiB7GcVNMP8 +9I7Cy0KLOrkIGGnPF2hggXxPjckA635Y6winTFN3XR1MreLbtP8cNeipiULnufON +0FZ6+N4CAhbxN4J/5DGjKuRh/cZ8VULse2Vr8dr8ilxZakgokt7bifxSoWILkOaG +1sMJzhQtxvrCnYbo+zApFriAdLF2gmg0v5uu5Rj3XE9F2JhSiCy4cq+MRSSswvBz +sIOsHn4IP++FKzeyugHIZLrbaW9oOKecBEiHoCBO2JLrY+954cE72dvcOp7rf8Jo +NR4hL/tRXMJTm6LurkCu+OjNtn5WyTfZbXgBiukUHnd9tvf8QklpDujS++9KW8yP +3fizTG1GFPzQvwPgfEwcsZn23NsRl3X5/031EMF7InmsezKrrO/nbQkuQ4v/nmkJ +f0xxxpKz57Jkef1Gd/Q96yEG4Q7xEcjnd7gmNdsHo3oJ2hDFYiWPoXGzz/TH+ZBH +T6un5mt8WJVG7uhAxM5xAoIBAQC41OamlO0DO9D1NESjb8/4M5nMOoBHl3dLKcI1 +0gNh9ZjbibNNmHdVbeNtuxq9Sn2qQ+VgtmLQUcKWBY9S3pKmOLDge74WG2O2BHlf +iQ0cPr07FhAcMb1uCK8jxdm3cVq0vImKc2SjIx/bbusE//n1Dtv2KeXRgDKxZtZd +n5UUypf/w1cEZYukOTDjhQNGFM+Yu0Q7DEMqqM2zzO/5y3YwkpoOqJ/kLxsfAGVb +1HcCa7QlRLLfWA/FpjZw3bbXJrmsvEgeGpbW0/QWzQqQHRG9gazSpOqKjAxtj5kb +ej2epJqMK8UHp+IPr7cpNne9pHC7n/2bOQDL93BT2VZDgFVTAoIBAQDPAGHUcFQ4 +6RLCERIihowySwVQHB2pZBPvkLZ0M0CZY3CcDtvUeeu7vH0pS8BvTRyqHUoN09lP +RXz1msFh1LLLh8RuX693CtS/vW6v+1loU1idPnflg17ESLsGiG31wOOcb+XrqpQP +aCBUd5sRbmo1otYhP4a1/yY6MrDDne1+KRbXXHOT6tYYYzlefmiofH7he+3+zMyD +flJ/ds7tzNMvJ+NwkYo6e/aLDyNn5c9iG/8GHehpP53tLcvf5b0xOhglCimkMVZH +UWwEjtlLlmg+0/uptkMYdbPYKRcFiGfStn47QVZyIliAOcRLF+VpjWCYJaIEmeYm +NxQNA43vGjT/AoIBAQC4bxnJWAo5k9KrG4DyZXxs+3CYrjebOLU9N2ooMmxVr7Dc +QMe+wkkx4flzYaUJBe/nmuCkZNqtbShycxHVa2uCmkdFebTwclxJIKXMgwGmEaTZ +9OYWfDu+NMQvOhpKRr5wY0IL+aGOeFotqLyzvIo70pwDQ0OkjfwHscpumfM1UAPk +n5ORO5LgSIFUR0JBCDsu+I7ZLR9IzjCVHgbIiBJj3aYEwpbqJ2c0xDxgKd6nd7nH +BopG+6ShYX4pmdP9VRMqHqcIAxhJPi7vIaNMsvUk2OUPPKkdnyo3mXb6SDx8tVvS +S0rMnOWjKX/njAZoIlKrprZ10afN4BciVFkT+lcpAoIBAQC/hxz9kMl8trqycVUU +OOWzCxLpYnpXZs/DU2Rd86YLxqRE4MKpv/1LgUVVVk/BK9of1GXWkXN6E1NhdEUF +neWoyAAKF+KkBJOArWei0K+TBbryEwCgjYK7nofdrZIAVu7Cqg5UlUnVT4TKYrhJ +0p3W7smtjbe44VGfe5NuC8vYdXA658HQ1PIvMm+8bL2tVzOWsjItFZUM/W6bnXQt +Nt0XvpKEb0U0g8pENEeQNGRD5J98K2QFLeWTtFH04f7Bc30vmE2bLEMmTcHiHcIU +XCAsMbui9Y7zLMSwdqRkbCeUIWJ+tR+lDv5P9iMXUA7mMPd1DyzvkO4P4dpdiDJl +TxxFAoIBAFbnY95HnFytOD8GO4dab/bM3FtjI+pRq6Xu2556nElRR5Cw9bG1PF3S +qlaTJ6I0zfzeqsRpR9oaa/ZNxJoKQKgt2qg6pjuM+0PucFc0T2pvx9AomuyUhVuw +oND3dz9eNpi0VE+M0LEiYctrPSwfn42BCcqkDBy8yb3QD47oYtd2b5nYMW1tki6h +dRUB5c02pFwdOPBC/i+WuHSWOjKyf5YBFd9YumPoT2b7t1LKwg4pywC4bBrWrOiQ +i/kwuXFoteZGAQ9SeHQeGzacIF8POIr1oMAZT995//gYQ3fW79VHiq0LKUCVO/JO +Gwsn7qf/tXc2TkAJ1ut/DMGB758tTVo= +-----END PRIVATE KEY----- diff --git a/tests/tests.conf b/tests/tests.conf new file mode 100644 index 0000000..06ab213 --- /dev/null +++ b/tests/tests.conf @@ -0,0 +1,36 @@ +[magisk] +url = https://github.com/topjohnwu/Magisk/releases/download/v25.2/Magisk-v25.2.apk +sha256 = 0bdc32918b6ea502dca769b1c7089200da51ea1def170824c2812925b426d509 + +# What's unique: init_boot (boot v4) + vendor_boot (vendor v4) +[device.GooglePixel7Pro] +url = https://dl.google.com/dl/android/aosp/cheetah-ota-tq1a.230205.002-ad27a6d7.zip +sha256 = ad27a6d7ca78c0d3fb7f51864cc6a9e6bc91f536f9e9bab5da06253df602525f +sha256_patched = a711b2033a6ebe39c2ba22e885bd570db33092897eda1793e508f3e8ffb8000a + +# What's unique: boot (boot v4, no ramdisk) + vendor_boot (vendor v4) +[device.GooglePixel6a] +url = https://dl.google.com/dl/android/aosp/bluejay-ota-tq1a.230205.002-1ec64b58.zip +sha256 = 1ec64b589cba8d9687ea8f85cf6c2e5d09c9e4fbdae15417aa06d5ba976fbe79 +sha256_patched = d0d59a90e0aba065f50fcd0c8c6c5c01910c08dc1caf9faee0dd3b9fbf1f4308 + +# What's unique: boot (boot v3) + vendor_boot (vendor v3) +[device.GooglePixel4a5G] +url = https://dl.google.com/dl/android/aosp/bramble-ota-tq1a.230205.002-49f2dc71.zip +sha256 = 49f2dc71f63eadec07537ce0d1b0c42995518a9447f008eb6266b6edd03a8a92 +sha256_patched = 6f5b4290fc28625e8cd98f85c31bd00df2fe73903a2562fd669d5fa037c4111e + +# What's unique: boot (boot v2) +[device.GooglePixel4a] +url = https://dl.google.com/dl/android/aosp/sunfish-ota-tq1a.230205.002-536bce5f.zip +sha256 = 536bce5f0038173c83a82072385bf56c0ceebbb6caefcf3190de45b0aff754a9 +sha256_patched = 948b9f3eaeaccf00f8a937b12682d7e466818ab3be7814d5119cfdb11c28f35a + +# What's unique: +# - boot (boot v4) + recovery (boot v4) +# - boot images have VTS signature block filled with all 0s +# - payload.bin uses ZERO blocks +[device.OnePlus10Pro] +url = https://android.googleapis.com/packages/ota-api/package/6812ccee9a8a891032af1e053e3217731ac7ad92.zip +sha256 = 09319bf66abf362dcb11a4af981900f393af87e10de10a03371c9d4e3785931f +sha256_patched = a9f06c7d2a3e1f2d0f81ac93262c516aa6f4dd79e909ce239bcf0df368295262 diff --git a/tests/tests.py b/tests/tests.py new file mode 100755 index 0000000..f339670 --- /dev/null +++ b/tests/tests.py @@ -0,0 +1,137 @@ +#!/usr/bin/env python3 + +import argparse +import configparser +import hashlib +import os +import subprocess +import sys + +sys.path.append(os.path.join(sys.path[0], '..')) +from avbroot import main +from avbroot import util + + +def url_filename(url): + return url.split('/')[-1] + + +def device_configs(args, config, filter=True): + for section in config.sections(): + assert section == os.path.basename(section) + + if section.startswith('device.'): + name = section.removeprefix('device.') + if filter and args.device and name not in args.device: + continue + + yield name, config[section] + + +def verify_checksum(path, sha256_hex): + with open(path, 'rb') as f: + hasher = util.hash_file(f, hashlib.sha256()) + + if hasher.digest() != bytes.fromhex(sha256_hex): + raise Exception(f'{path}: expected sha256 {sha256_hex}, ' + f'but have {hasher.hexdigest()}') + + +def download(directory, url, sha256_hex, revalidate=False, extra_args=[]): + path = os.path.join(directory, url_filename(url)) + + if os.path.exists(path) and not os.path.exists(path + '.aria2'): + if revalidate: + print('Verifying checksum of', path) + + verify_checksum(path, sha256_hex) + else: + print('Downloading', url, 'to', path) + + subprocess.check_call([ + 'aria2c', + *extra_args, + '--auto-file-renaming=false', + f'--checksum=SHA-256={sha256_hex}', + f'--dir={os.path.dirname(path)}', + f'--out={os.path.basename(path)}', + url, + ]) + + return path + + +def run_tests(args, config, files_dir): + magisk_file = download( + os.path.join(files_dir, 'magisk'), + config['magisk']['url'], + config['magisk']['sha256'], + revalidate=args.revalidate, + extra_args=args.aria2c_arg, + ) + + for device, image in device_configs(args, config): + image_dir = os.path.join(files_dir, device) + os.makedirs(image_dir, exist_ok=True) + + image_file = download( + image_dir, + image['url'], + image['sha256'], + revalidate=args.revalidate, + extra_args=args.aria2c_arg, + ) + patched_file = image_file + '.patched' + + print('Patching', image_file) + + test_key_prefix = os.path.join( + sys.path[0], 'keys', 'TEST_KEY_DO_NOT_USE_') + + main.main([ + 'patch', + '--privkey-avb', test_key_prefix + 'avb.key', + '--privkey-ota', test_key_prefix + 'ota.key', + '--cert-ota', test_key_prefix + 'ota.crt', + '--magisk', magisk_file, + '--input', image_file, + '--output', patched_file, + ]) + + print('Verifying checksum of', patched_file) + + verify_checksum(patched_file, image['sha256_patched']) + + +def parse_args(): + parser = argparse.ArgumentParser() + parser.add_argument('-a', '--aria2c-arg', default=[], action='append', + help='Argument to pass to aria2c') + parser.add_argument('-d', '--device', action='append', + help='Device image to test against') + parser.add_argument('--revalidate', action='store_true', + help='Revalidate checksums for downloaded OTAs') + + return parser.parse_args() + + +def test_main(): + args = parse_args() + + config_file = os.path.join(sys.path[0], 'tests.conf') + config = configparser.ConfigParser() + config.read(config_file) + + if args.device: + devices = set(d for d, _ in device_configs(args, config, filter=False)) + invalid = set(args.device) - devices + if invalid: + raise ValueError(f'Invalid devices: {sorted(invalid)}') + + files_dir = os.path.join(sys.path[0], 'files') + + run_tests(args, config, files_dir) + + +if __name__ == '__main__': + test_main()