mirror of
https://github.com/chenxiaolong/avbroot.git
synced 2026-07-03 14:05:11 +02:00
Merge pull request #46 from chenxiaolong/tests
Add end-to-end patching tests
This commit is contained in:
Andrew Gunnerson
+2
-379
@@ -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()
|
||||
|
||||
+377
@@ -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()
|
||||
@@ -0,0 +1 @@
|
||||
/files/
|
||||
@@ -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/<device>/`
|
||||
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 <device>`. This argument can be specified multiple times (eg. `-d GooglePixel7Pro -d GooglePixel6a`).
|
||||
|
||||
To pass in extra arguments to `aria2c`, use `-a=<arg>`. This can be used to enable concurrent downloads (eg. `-a=-x4 -a=-s4`).
|
||||
@@ -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-----
|
||||
@@ -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-----
|
||||
@@ -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-----
|
||||
@@ -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
|
||||
Executable
+137
@@ -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()
|
||||
Reference in New Issue
Block a user