diff --git a/README.md b/README.md index 8dfd9eb..69fba02 100644 --- a/README.md +++ b/README.md @@ -4,8 +4,6 @@ avbroot is a script for patching Android boot images with Magisk root while pres I do not recommend using this project without a deep understanding of the implementation of AVB and A/B OTAs. It is meant for use with proprietary stock firmware. For folks running open-source Android firmware, I highly recommend adding Magisk to the build process and then compiling from source instead. -**NOTE**: avbroot currently only supports running on Linux due to `magiskboot`'s requirements. - ### Patches avbroot applies two patches to the boot images: @@ -78,7 +76,7 @@ The boot-related components are signed with an AVB key and OTA-related component git submodule update --init --recursive ``` -3. Ensure that `openssl`, `python3`, and `python3-protobuf` are installed. +3. Ensure that `openssl`, `python3`, `python3-lz4`, and `python3-protobuf` are installed. 4. Follow the steps to [generate signing keys](#generating-keys). diff --git a/avbroot.py b/avbroot.py index 7973216..746e235 100644 --- a/avbroot.py +++ b/avbroot.py @@ -74,7 +74,7 @@ def patch_ota_payload(f_in, f_out, file_size, magisk, privkey_avb, ota.extract_images(f_in, manifest, blob_offset, extract_dir, images) boot_patches = [boot.MagiskRootPatch(magisk)] - otacert_patches = [boot.OtaCertPatch(magisk, cert_ota)] + otacert_patches = [boot.OtaCertPatch(cert_ota)] if otacert_image == boot_image: boot_patches.extend(otacert_patches) diff --git a/avbroot/boot.py b/avbroot/boot.py index 1082d21..ecf736e 100644 --- a/avbroot/boot.py +++ b/avbroot/boot.py @@ -1,8 +1,7 @@ -import os +import hashlib +import io +import lzma import shutil -import subprocess -import struct -import tempfile import zipfile import avbtool @@ -10,76 +9,124 @@ import avbtool from . import openssl from . import util from . import vbmeta +from .formats import bootimage +from .formats import compression +from .formats import cpio + + +def _load_ramdisk(ramdisk): + with ( + io.BytesIO(ramdisk) as f_raw, + compression.CompressedFile(f_raw, 'rb') as f, + ): + return cpio.load(f.fp), f.format + + +def _save_ramdisk(entries, format): + with io.BytesIO() as f_raw: + with compression.CompressedFile(f_raw, 'wb', format=format) as f: + cpio.save(f.fp, entries) + + return f_raw.getvalue() class BootImagePatch: - def __init__(self, magisk_apk, to_extract): - self.magisk_apk = magisk_apk - self.to_extract = to_extract - def __call__(self, image_file): - with tempfile.TemporaryDirectory() as temp_dir: - with zipfile.ZipFile(self.magisk_apk, 'r') as zip: - for source, extract_info in self.to_extract.items(): - try: - info = zip.getinfo(source) - except KeyError: - if extract_info.get('optional', False): - continue - raise + # Load the boot image + with open(image_file, 'r+b') as f: + boot_image = bootimage.load_autodetect(f) - info.filename = extract_info['dest'] - zip.extract(info, path=temp_dir) + self.patch(image_file, boot_image) - self.patch(image_file, temp_dir) + f.seek(0) + f.truncate(0) - def patch(self, image_file, temp_dir): + boot_image.generate(f) + + def patch(self, image_file, boot_image): raise NotImplementedError() class MagiskRootPatch(BootImagePatch): ''' - Root the boot image using Magisk's patch script. + Root the boot image with Magisk. ''' - EXTRACT_MAP = { - 'assets/boot_patch.sh': {'dest': 'boot_patch.sh'}, - 'assets/stub.apk': { - 'dest': 'stub.apk', - # Only exists after the Magisk commit: - # ad0e6511e11ebec65aa9b5b916e1397342850319 - 'optional': True, - }, - 'assets/util_functions.sh': {'dest': 'util_functions.sh'}, - 'lib/arm64-v8a/libmagisk64.so': {'dest': 'magisk64'}, - 'lib/arm64-v8a/libmagiskinit.so': {'dest': 'magiskinit'}, - 'lib/armeabi-v7a/libmagisk32.so': {'dest': 'magisk32'}, - # The x86 binary is used because the x86_64 binary breaks if the PID - # exceeds 65535. This is due to the pthread_mutex implementation in the - # version of bionic libc compiled into magiskboot, which uses the - # thread ID as a 16-bit integer. On systems with a large PID limit, - # like Fedora 37, which sets `kernel.pid_max = 4194304`, the x86_64 - # binary is almost never able to run successfully. In the future, if - # there's a need to work around this, we can unshare() a new PID - # namespace to get small PIDs. - 'lib/x86/libmagiskboot.so': {'dest': 'magiskboot'}, - } - def __init__(self, magisk_apk): - super().__init__(magisk_apk, self.EXTRACT_MAP) + self.magisk_apk = magisk_apk - def patch(self, image_file, temp_dir): - subprocess.check_call( - ['sh', './boot_patch.sh', image_file], - cwd=temp_dir, - env={ - 'BOOTMODE': 'true', - 'KEEPVERITY': 'true', - 'KEEPFORCEENCRYPT': 'true', - }, - ) + def patch(self, image_file, boot_image): + with zipfile.ZipFile(self.magisk_apk, 'r') as zip: + self._patch(image_file, boot_image, zip) - shutil.copyfile(os.path.join(temp_dir, 'new-boot.img'), image_file) + def _patch(self, image_file, boot_image, zip): + # Magisk saves the original SHA1 digest in its config file + with open(image_file, 'rb') as f: + hasher = util.hash_file(f, hashlib.sha1()) + + # Load the ramdisk + entries, ramdisk_format = _load_ramdisk(boot_image.ramdisks[0]) + + # Create magisk directory structure + for path, perms in ( + (b'.backup', 0o000), + (b'overlay.d', 0o750), + (b'overlay.d/sbin', 0o750), + ): + entries.append(cpio.CpioEntryNew.new_directory(path, perms=perms)) + + # Move original init to backup path + orig_init = next(e for e in entries if e.name == b'init') + orig_init.name = b'.backup/init' + + # Add magiskinit + with zip.open('lib/arm64-v8a/libmagiskinit.so', 'r') as f: + entries.append(cpio.CpioEntryNew.new_file( + b'init', perms=0o750, data=f.read())) + + # Add xz-compressed magisk32 and magisk64 + xz_files = { + 'lib/armeabi-v7a/libmagisk32.so': b'magisk32.xz', + 'lib/arm64-v8a/libmagisk64.so': b'magisk64.xz', + } + + # Add stub apk, which only exists after the Magisk commit: + # ad0e6511e11ebec65aa9b5b916e1397342850319 + if 'assets/stub.apk' in zip.namelist(): + xz_files['assets/stub.apk'] = b'stub.xz' + + for source, target in xz_files.items(): + with ( + zip.open(source, 'r') as f_in, + io.BytesIO() as f_out_raw, + ): + with lzma.open(f_out_raw, 'wb', preset=9, + check=lzma.CHECK_CRC32) as f_out: + shutil.copyfileobj(f_in, f_out) + + entries.append(cpio.CpioEntryNew.new_file( + b'overlay.d/sbin/' + target, perms=0o644, + data=f_out_raw.getvalue())) + + # Create magisk boot-time removal list file + rmlist_files = sorted(e.name for e in entries + if e.name.startswith(b'overlay.d')) + rmlist_data = b'\0'.join(rmlist_files) + b'\0' + entries.append(cpio.CpioEntryNew.new_file( + b'.backup/.rmlist', perms=0o000, data=rmlist_data)) + + # Create magisk config + magisk_config = \ + b'KEEPVERITY=true\n' \ + b'KEEPFORCEENCRYPT=true\n' \ + b'PATCHVBMETAFLAG=false\n' \ + b'RECOVERYMODE=false\n' \ + b'SHA1=%s\n' % hasher.hexdigest().encode('ascii') + entries.append(cpio.CpioEntryNew.new_file( + b'.backup/.magisk', perms=0o000, data=magisk_config)) + + # Repack ramdisk + boot_image.ramdisks[0] = _save_ramdisk(entries, ramdisk_format) class OtaCertPatch(BootImagePatch): @@ -88,86 +135,48 @@ class OtaCertPatch(BootImagePatch): signing certificate. ''' - EXTRACT_MAP = { - 'lib/x86/libmagiskboot.so': {'dest': 'magiskboot'}, - } + OTACERTS_PATH = b'system/etc/security/otacerts.zip' - def __init__(self, magisk_apk, cert_ota): - super().__init__(magisk_apk, self.EXTRACT_MAP) + def __init__(self, cert_ota): self.cert_ota = cert_ota - def _read_header_version(self, image_file): - with open(image_file, 'rb') as f: - magic = f.read(8) + def patch(self, image_file, boot_image): + found_otacerts = False - if magic == b'ANDROID!': - f.seek(0x28) - is_vendor = False - elif magic == b'VNDRBOOT': - # Version is immediately after magic - is_vendor = True + # Check each ramdisk + for i, ramdisk in enumerate(boot_image.ramdisks): + entries, ramdisk_format = _load_ramdisk(ramdisk) + + # Fail hard if otacerts does not exist. We don't want to lock the + # user out of future updates if the OTA certificate mechanism has + # changed. + otacerts = next((e for e in entries if e.name == + self.OTACERTS_PATH), None) + if otacerts: + found_otacerts = True else: - raise Exception(b'Invalid magic: {magic}') + continue - return struct.unpack('I', f.read(4))[0], is_vendor + # Create new otacerts archive. The old certs are ignored since + # flashing a stock OTA will render the device unbootable. + with io.BytesIO() as f_zip: + with zipfile.ZipFile(f_zip, 'w') as z: + # Use zeroed-out metadata to ensure the archive is bit for + # bit reproducible across runs. + info = zipfile.ZipInfo('ota.x509.pem') + with ( + z.open(info, 'w') as f_out, + open(self.cert_ota, 'rb') as f_in, + ): + shutil.copyfileobj(f_in, f_out) - def _is_uncompressed(self, ramdisk_file): - with open(ramdisk_file, 'rb') as f: - magic = f.read(6) + otacerts.content = f_zip.getvalue() - return magic == b'070701' or magic == b'070702' + # Repack ramdisk + boot_image.ramdisks[i] = _save_ramdisk(entries, ramdisk_format) - def patch(self, image_file, temp_dir): - def run(*args): - subprocess.check_call(['./magiskboot', *args], cwd=temp_dir) - - os.chmod(os.path.join(temp_dir, 'magiskboot'), 0o755) - - # Unpack the boot image - run('unpack', image_file) - - # magiskboot currently does not automatically decompress v4 vendor boot - # ramdisks as there may be more than one. This is not the case for - # Android 13 on the Pixel 6 Pro. - ramdisk_file = 'ramdisk.cpio' - header_version, is_vendor = self._read_header_version(image_file) - need_decompress = header_version == 4 and is_vendor - - if need_decompress: - ramdisk_file = 'decompressed.cpio' - run('decompress', 'ramdisk.cpio', ramdisk_file) - - # Fail hard if otacerts does not exist. We don't want to lock the user - # out of future updates if the OTA certificate mechanism has changed. - run( - 'cpio', ramdisk_file, - 'exists system/etc/security/otacerts.zip', - ) - - # Create new otacerts archive. The old certs are ignored since flashing - # a stock OTA will render the device unbootable. - with zipfile.ZipFile(os.path.join(temp_dir, 'otacerts.zip'), 'w') as z: - # Construct our own timestamp so the archive is reproducible - info = zipfile.ZipInfo('ota.x509.pem') - with z.open(info, 'w') as f_out: - with open(self.cert_ota, 'rb') as f_in: - shutil.copyfileobj(f_in, f_out) - - # Repack ramdisk - run( - 'cpio', ramdisk_file, - 'rm system/etc/security/otacerts.zip', - 'add 644 system/etc/security/otacerts.zip otacerts.zip', - ) - - # Recompress ramdisk - if need_decompress: - run('compress=lz4_legacy', ramdisk_file, 'ramdisk.cpio') - - # Repack image - run('repack', image_file) - - shutil.copyfile(os.path.join(temp_dir, 'new-boot.img'), image_file) + if not found_otacerts: + raise Exception(f'{self.OTACERTS_PATH} not found in ramdisk') def patch_boot(avb, input_path, output_path, key, passphrase, diff --git a/avbroot/formats/bootimage.py b/avbroot/formats/bootimage.py new file mode 100644 index 0000000..2fbe23d --- /dev/null +++ b/avbroot/formats/bootimage.py @@ -0,0 +1,612 @@ +import collections +import os +import struct +import typing + +from . import padding +from .. import util + + +BOOT_MAGIC = b'ANDROID!' +BOOT_NAME_SIZE = 16 +BOOT_ARGS_SIZE = 512 +BOOT_EXTRA_ARGS_SIZE = 1024 + +VENDOR_BOOT_MAGIC = b'VNDRBOOT' +VENDOR_BOOT_ARGS_SIZE = 2048 +VENDOR_BOOT_NAME_SIZE = 16 + +VENDOR_RAMDISK_TYPE_NONE = 0 +VENDOR_RAMDISK_TYPE_PLATFORM = 1 +VENDOR_RAMDISK_TYPE_RECOVERY = 2 +VENDOR_RAMDISK_TYPE_DLKM = 3 +VENDOR_RAMDISK_NAME_SIZE = 32 +VENDOR_RAMDISK_TABLE_ENTRY_BOARD_ID_SIZE = 16 + +PAGE_SIZE = 4096 + +BOOT_IMG_HDR_V0 = struct.Struct( + '<' + f'{len(BOOT_MAGIC)}s' # magic + 'I' # kernel_size + 'I' # kernel_addr + 'I' # ramdisk_size + 'I' # ramdisk_addr + 'I' # second_size + 'I' # second_addr + 'I' # tags_addr + 'I' # page_size + 'I' # header_version + 'I' # os_version + f'{BOOT_NAME_SIZE}s' # name + f'{BOOT_ARGS_SIZE}s' # cmdline + f'{8 * 4}s' # id (uint32_t[8]) + f'{BOOT_EXTRA_ARGS_SIZE}s' # extra_cmdline +) + +BOOT_IMG_HDR_V1_EXTRA = struct.Struct( + '<' + 'I' # recovery_dtbo_size + 'Q' # recovery_dtbo_offset + 'I' # header_size +) + +BOOT_IMG_HDR_V2_EXTRA = struct.Struct( + '<' + 'I' # dtb_size + 'Q' # dtb_addr +) + +BOOT_IMG_HDR_V3 = struct.Struct( + '<' + f'{len(BOOT_MAGIC)}s' # magic + 'I' # kernel_size + 'I' # ramdisk_size + 'I' # os_version + 'I' # header_size + '16s' # reserved (uint32_t[4]) + 'I' # header_version + f'{BOOT_ARGS_SIZE + BOOT_EXTRA_ARGS_SIZE}s' # cmdline +) + +VENDOR_BOOT_IMG_HDR_V3 = struct.Struct( + '<' + f'{len(VENDOR_BOOT_MAGIC)}s' # magic + 'I' # header_version + 'I' # page_size + 'I' # kernel_addr + 'I' # ramdisk_addr + 'I' # vendor_ramdisk_size + f'{VENDOR_BOOT_ARGS_SIZE}s' # cmdline + 'I' # tags_addr + f'{VENDOR_BOOT_NAME_SIZE}s' # name + 'I' # header_size + 'I' # dtb_size + 'Q' # dtb_addr +) + +BOOT_IMG_HDR_V4_EXTRA = struct.Struct( + '<' + 'I' # signature_size +) + +VENDOR_BOOT_IMG_HDR_V4_EXTRA = struct.Struct( + '<' + 'I' # vendor_ramdisk_table_size + 'I' # vendor_ramdisk_table_entry_num + 'I' # vendor_ramdisk_table_entry_size + 'I' # bootconfig_size +) + +VENDOR_RAMDISK_TABLE_ENTRY_V4 = struct.Struct( + '<' + 'I' # ramdisk_size + 'I' # ramdisk_offset + 'I' # ramdisk_type + f'{VENDOR_RAMDISK_NAME_SIZE}s' # ramdisk_name + f'{VENDOR_RAMDISK_TABLE_ENTRY_BOARD_ID_SIZE * 4}s' # board_id (uint32_t[]) +) + + +class WrongFormat(ValueError): + pass + + +class BootImage: + def __init__(self) -> None: + self.kernel: None | bytes = None + self.ramdisks: list[bytes] = [] + self.second: None | bytes = None + self.recovery_dtbo: None | bytes = None + self.dtb: None | bytes = None + self.bootconfig: None | bytes = None + + def generate(self, f: typing.BinaryIO) -> None: + raise NotImplementedError() + + +class _BootImageV0Through2(BootImage): + def __init__(self, f: typing.BinaryIO) -> None: + super().__init__() + + # Common fields for v0 through v2 + magic, kernel_size, kernel_addr, ramdisk_size, ramdisk_addr, \ + second_size, second_addr, tags_addr, page_size, header_version, \ + os_version, name, cmdline, id, extra_cmdline = \ + BOOT_IMG_HDR_V0.unpack(util.read_exact(f, BOOT_IMG_HDR_V0.size)) + + if magic != BOOT_MAGIC: + raise WrongFormat(f'Unknown magic: {magic}') + elif header_version not in (0, 1, 2): + raise WrongFormat(f'Unknown header version: {header_version}') + + self.kernel_addr = kernel_addr + self.ramdisk_addr = ramdisk_addr + self.second_addr = second_addr + self.tags_addr = tags_addr + self.page_size = page_size + self.header_version = header_version + self.os_version = os_version + self.name = name.rstrip(b'\0') + self.cmdline = cmdline.rstrip(b'\0') + self.id = id + self.extra_cmdline = extra_cmdline.rstrip(b'\0') + + # Parse v1 fields + if header_version >= 1: + recovery_dtbo_size, recovery_dtbo_offset, header_size = \ + BOOT_IMG_HDR_V1_EXTRA.unpack( + util.read_exact(f, BOOT_IMG_HDR_V1_EXTRA.size)) + + self.recovery_dtbo_offset = recovery_dtbo_offset + + # Parse v2 fields + if header_version == 2: + dtb_size, dtb_addr = BOOT_IMG_HDR_V2_EXTRA.unpack( + util.read_exact(f, BOOT_IMG_HDR_V2_EXTRA.size)) + + self.dtb_addr = dtb_addr + + if header_version >= 1 and f.tell() != header_size: + raise ValueError(f'Invalid header size: {header_size}') + + padding.read_skip(f, page_size) + + if kernel_size > 0: + self.kernel = util.read_exact(f, kernel_size) + padding.read_skip(f, page_size) + + if ramdisk_size > 0: + self.ramdisks.append(util.read_exact(f, ramdisk_size)) + padding.read_skip(f, page_size) + + if second_size > 0: + self.second = util.read_exact(f, second_size) + padding.read_skip(f, page_size) + + if header_version >= 1 and recovery_dtbo_size > 0: + self.recovery_dtbo = util.read_exact(f, recovery_dtbo_size) + padding.read_skip(f, page_size) + + if header_version == 2 and dtb_size > 0: + self.dtb = util.read_exact(f, dtb_size) + padding.read_skip(f, page_size) + + def generate(self, f: typing.BinaryIO) -> None: + if len(self.ramdisks) > 1: + raise ValueError('Only one ramdisk is supported') + elif self.bootconfig is not None: + raise ValueError('Boot config is not supported') + elif self.header_version < 1 and self.recovery_dtbo is not None: + raise ValueError('Recovery dtbo/acpio is not supported') + elif self.header_version < 2 and self.dtb is not None: + raise ValueError('Device tree is not supported') + + f.write(BOOT_IMG_HDR_V0.pack( + BOOT_MAGIC, + len(self.kernel) if self.kernel else 0, + self.kernel_addr, + len(self.ramdisks[0]) if self.ramdisks else 0, + self.ramdisk_addr, + len(self.second) if self.second else 0, + self.second_addr, + self.tags_addr, + self.page_size, + self.header_version, + self.os_version, + self.name, + self.cmdline, + self.id, + self.extra_cmdline, + )) + + if self.header_version >= 1: + header_size = BOOT_IMG_HDR_V0.size + if self.header_version >= 1: + header_size += BOOT_IMG_HDR_V1_EXTRA.size + if self.header_version == 2: + header_size += BOOT_IMG_HDR_V2_EXTRA.size + + f.write(BOOT_IMG_HDR_V1_EXTRA.pack( + len(self.recovery_dtbo) if self.recovery_dtbo else 0, + self.recovery_dtbo_offset, + header_size, + )) + + if self.header_version == 2: + f.write(BOOT_IMG_HDR_V2_EXTRA.pack( + len(self.dtb) if self.dtb else 0, + self.dtb_addr, + )) + + padding.write(f, self.page_size) + + if self.kernel: + f.write(self.kernel) + padding.write(f, self.page_size) + + if self.ramdisks: + f.write(self.ramdisks[0]) + padding.write(f, self.page_size) + + if self.second: + f.write(self.second) + padding.write(f, self.page_size) + + if self.header_version >= 1 and self.recovery_dtbo: + f.write(self.recovery_dtbo) + padding.write(f, self.page_size) + + if self.header_version == 2 and self.dtb: + f.write(self.dtb) + padding.write(f, self.page_size) + + def __str__(self) -> str: + kernel_size = len(self.kernel) if self.kernel else 0 + ramdisk_size = len(self.ramdisks[0]) if self.ramdisks else 0 + second_size = len(self.second) if self.second else 0 + + result = \ + f'Boot image v{self.header_version} header:\n' \ + f'- Kernel size: {kernel_size}\n' \ + f'- Kernel address: 0x{self.kernel_addr:x}\n' \ + f'- Ramdisk size: {ramdisk_size}\n' \ + f'- Ramdisk address: 0x{self.ramdisk_addr:x}\n' \ + f'- Second stage size: {second_size}\n' \ + f'- Second stage address: 0x{self.second_addr:x}\n' \ + f'- Kernel tags address: 0x{self.tags_addr:x}\n' \ + f'- Page size: {self.page_size}\n' \ + f'- OS version: 0x{self.os_version:x}\n' \ + f'- Name: {self.name!r}\n' \ + f'- Kernel cmdline: {self.cmdline!r}\n' \ + f'- ID: {self.id.hex()}\n' \ + f'- Extra kernel cmdline: {self.extra_cmdline!r}\n' + + if self.header_version >= 1: + recovery_dtbo_size = len(self.recovery_dtbo) \ + if self.recovery_dtbo else 0 + + result += \ + f'- Recovery dtbo size: {recovery_dtbo_size}\n' \ + f'- Recovery dtbo offset: {self.recovery_dtbo_offset}\n' + + if self.header_version == 2: + dtb_size = len(self.dtb) if self.dtb else 0 + + result += \ + f'- Device tree size: {dtb_size}\n' \ + f'- Device tree address: {self.dtb_addr}\n' + + return result + + +class _BootImageV3Through4(BootImage): + def __init__(self, f: typing.BinaryIO) -> None: + super().__init__() + + # Common fields for both v3 and v4 + magic, kernel_size, ramdisk_size, os_version, header_size, reserved, \ + header_version, cmdline = BOOT_IMG_HDR_V3.unpack( + util.read_exact(f, BOOT_IMG_HDR_V3.size)) + + if magic != BOOT_MAGIC: + raise WrongFormat(f'Unknown magic: {magic}') + elif header_version not in (3, 4): + raise WrongFormat(f'Unknown header version: {header_version}') + + # Parse v4 fields + if header_version == 4: + signature_size, = BOOT_IMG_HDR_V4_EXTRA.unpack( + util.read_exact(f, BOOT_IMG_HDR_V4_EXTRA.size)) + + if f.tell() != header_size: + raise ValueError(f'Invalid header size: {header_size}') + + self.header_version = header_version + self.os_version = os_version + self.reserved = reserved + self.cmdline = cmdline.rstrip(b'\0') + + padding.read_skip(f, PAGE_SIZE) + + if kernel_size > 0: + self.kernel = util.read_exact(f, kernel_size) + padding.read_skip(f, PAGE_SIZE) + + if ramdisk_size > 0: + self.ramdisks.append(util.read_exact(f, ramdisk_size)) + padding.read_skip(f, PAGE_SIZE) + + if header_version == 4: + # Don't preserve the signature. It is only used for VTS tests and + # is not relevant for booting + f.seek(signature_size, os.SEEK_CUR) + padding.read_skip(f, PAGE_SIZE) + + def generate(self, f: typing.BinaryIO) -> None: + if len(self.ramdisks) > 1: + raise ValueError('Only one ramdisk is supported') + elif self.second is not None: + raise ValueError('Second stage bootloader is not supported') + elif self.recovery_dtbo is not None: + raise ValueError('Recovery dtbo/acpio is not supported') + elif self.dtb is not None: + raise ValueError('Device tree is not supported') + elif self.bootconfig is not None: + raise ValueError('Boot config is not supported') + + f.write(BOOT_IMG_HDR_V3.pack( + BOOT_MAGIC, + len(self.kernel) if self.kernel else 0, + len(self.ramdisks[0]) if self.ramdisks else 0, + self.os_version, + BOOT_IMG_HDR_V3.size + (BOOT_IMG_HDR_V4_EXTRA.size + if self.header_version == 4 else 0), + self.reserved, + self.header_version, + self.cmdline, + )) + + if self.header_version == 4: + f.write(BOOT_IMG_HDR_V4_EXTRA.pack( + # We don't care about the VTS signature + 0 + )) + + padding.write(f, PAGE_SIZE) + + if self.kernel: + f.write(self.kernel) + padding.write(f, PAGE_SIZE) + + if self.ramdisks: + f.write(self.ramdisks[0]) + padding.write(f, PAGE_SIZE) + + def __str__(self) -> str: + kernel_size = len(self.kernel) if self.kernel else 0 + ramdisk_size = len(self.ramdisks[0]) if self.ramdisks else 0 + + return \ + f'Boot image v{self.header_version} header:\n' \ + f'- Kernel size: {kernel_size}\n' \ + f'- Ramdisk size: {ramdisk_size}\n' \ + f'- OS version: 0x{self.os_version:x}\n' \ + f'- Reserved: {self.reserved.hex()}\n' \ + f'- Kernel cmdline: {self.cmdline!r}\n' + + +_RamdiskMeta = collections.namedtuple( + '_RamdiskMeta', ['type', 'name', 'board_id']) + + +class _VendorBootImageV3Through4(BootImage): + def __init__(self, f: typing.BinaryIO) -> None: + super().__init__() + + # Common fields for both v3 and v4 + magic, header_version, page_size, kernel_addr, ramdisk_addr, \ + vendor_ramdisk_size, cmdline, tags_addr, name, header_size, \ + dtb_size, dtb_addr = VENDOR_BOOT_IMG_HDR_V3.unpack( + util.read_exact(f, VENDOR_BOOT_IMG_HDR_V3.size)) + + if magic != VENDOR_BOOT_MAGIC: + raise WrongFormat(f'Unknown magic: {magic}') + elif header_version not in (3, 4): + raise WrongFormat(f'Unknown header version: {header_version}') + + # Parse v4 fields + if header_version == 4: + vendor_ramdisk_table_size, vendor_ramdisk_table_entry_num, \ + vendor_ramdisk_table_entry_size, bootconfig_size = \ + VENDOR_BOOT_IMG_HDR_V4_EXTRA.unpack( + util.read_exact(f, VENDOR_BOOT_IMG_HDR_V4_EXTRA.size)) + + if vendor_ramdisk_table_entry_size != \ + VENDOR_RAMDISK_TABLE_ENTRY_V4.size: + raise ValueError('Invalid ramdisk table entry size: ' + f'{vendor_ramdisk_table_entry_size}') + elif vendor_ramdisk_table_size != vendor_ramdisk_table_entry_num \ + * vendor_ramdisk_table_entry_size: + raise ValueError('Invalid ramdisk table size: ' + f'{vendor_ramdisk_table_size}') + + if f.tell() != header_size: + raise ValueError(f'Invalid header size: {header_size}') + + self.page_size = page_size + self.header_version = header_version + self.kernel_addr = kernel_addr + self.ramdisk_addr = ramdisk_addr + self.cmdline = cmdline.rstrip(b'\0') + self.tags_addr = tags_addr + self.name = name.rstrip(b'\0') + self.dtb_addr = dtb_addr + + padding.read_skip(f, page_size) + + vendor_ramdisk_offset = f.tell() + + if header_version == 3: + # v3 has one big ramdisk + self.ramdisks.append(util.read_exact(f, vendor_ramdisk_size)) + else: + # v4 has multiple ramdisks, processed later + f.seek(vendor_ramdisk_size, os.SEEK_CUR) + + padding.read_skip(f, page_size) + + if dtb_size > 0: + self.dtb = util.read_exact(f, dtb_size) + padding.read_skip(f, page_size) + + if header_version == 4: + self.ramdisks_meta = [] + + total_ramdisk_size = 0 + + for _ in range(0, vendor_ramdisk_table_entry_num): + ramdisk_size, ramdisk_offset, ramdisk_type, ramdisk_name, \ + board_id = VENDOR_RAMDISK_TABLE_ENTRY_V4.unpack( + util.read_exact(f, VENDOR_RAMDISK_TABLE_ENTRY_V4.size)) + + table_offset = f.tell() + f.seek(vendor_ramdisk_offset + ramdisk_offset) + + self.ramdisks.append(util.read_exact(f, ramdisk_size)) + self.ramdisks_meta.append(_RamdiskMeta( + ramdisk_type, + ramdisk_name.rstrip(b'\0'), + board_id, + )) + + f.seek(table_offset) + + total_ramdisk_size += ramdisk_size + + if total_ramdisk_size != vendor_ramdisk_size: + raise ValueError('Invalid vendor ramdisk size: ' + f'{vendor_ramdisk_size}') + + padding.read_skip(f, page_size) + + if bootconfig_size > 0: + self.bootconfig = util.read_exact(f, bootconfig_size) + padding.read_skip(f, page_size) + + def generate(self, f: typing.BinaryIO) -> None: + if self.header_version == 3: + if len(self.ramdisks) > 1: + raise ValueError('Only one ramdisk is supported') + elif self.bootconfig is not None: + raise ValueError('Boot config is not supported') + else: + if len(self.ramdisks) != len(self.ramdisks_meta): + raise ValueError('Mismatched ramdisk and ramdisk_meta') + + if self.second is not None: + raise ValueError('Second stage bootloader is not supported') + elif self.recovery_dtbo is not None: + raise ValueError('Recovery dtbo/acpio is not supported') + + vendor_ramdisk_size = sum(len(r) for r in self.ramdisks) + + f.write(VENDOR_BOOT_IMG_HDR_V3.pack( + VENDOR_BOOT_MAGIC, + self.header_version, + self.page_size, + self.kernel_addr, + self.ramdisk_addr, + vendor_ramdisk_size, + self.cmdline, + self.tags_addr, + self.name, + VENDOR_BOOT_IMG_HDR_V3.size + ( + VENDOR_BOOT_IMG_HDR_V4_EXTRA.size + if self.header_version == 4 else 0), + len(self.dtb) if self.dtb else 0, + self.dtb_addr, + )) + + if self.header_version == 4: + f.write(VENDOR_BOOT_IMG_HDR_V4_EXTRA.pack( + len(self.ramdisks) * VENDOR_RAMDISK_TABLE_ENTRY_V4.size, + len(self.ramdisks), + VENDOR_RAMDISK_TABLE_ENTRY_V4.size, + len(self.bootconfig) if self.bootconfig else 0, + )) + + padding.write(f, self.page_size) + + for ramdisk in self.ramdisks: + f.write(ramdisk) + + padding.write(f, self.page_size) + + if self.dtb: + f.write(self.dtb) + padding.write(f, self.page_size) + + if self.header_version == 4: + ramdisk_offset = 0 + + for ramdisk, meta in zip(self.ramdisks, self.ramdisks_meta): + f.write(VENDOR_RAMDISK_TABLE_ENTRY_V4.pack( + len(ramdisk), + ramdisk_offset, + meta.type, + meta.name, + meta.board_id, + )) + + ramdisk_offset += len(ramdisk) + + padding.write(f, self.page_size) + + if self.bootconfig: + f.write(self.bootconfig) + padding.write(f, self.page_size) + + def __str__(self) -> str: + dtb_size = len(self.dtb) if self.dtb else 0 + + result = \ + f'Vendor boot image v{self.header_version} header:\n' \ + f'- Page size: {self.page_size}\n' \ + f'- Kernel address: 0x{self.kernel_addr:x}\n' \ + f'- Ramdisk address: 0x{self.ramdisk_addr:x}\n' \ + f'- Kernel cmdline: {self.cmdline!r}\n' \ + f'- Kernel tags address: 0x{self.tags_addr:x}\n' \ + f'- Name: {self.name!r}\n' \ + f'- Device tree size: {dtb_size}\n' \ + f'- Device tree address: {self.dtb_addr}\n' + + if self.header_version == 4: + for ramdisk, meta in zip(self.ramdisks, self.ramdisks_meta): + result += \ + '- Ramdisk:\n' \ + f' - Size: {len(ramdisk)}\n' \ + f' - Type: {meta.type}\n' \ + f' - Name: {meta.name}\n' \ + f' - Board ID: {meta.board_id.hex()}\n' + + bootconfig_size = len(self.bootconfig) if self.bootconfig else 0 + + result += f'- Bootconfig size: {bootconfig_size}\n' + + return result + + +def load_autodetect(f: typing.BinaryIO) -> BootImage: + for cls in ( + _BootImageV0Through2, + _BootImageV3Through4, + _VendorBootImageV3Through4, + ): + try: + f.seek(0) + return cls(f) + except WrongFormat: + continue + + raise ValueError('Unknown boot image format') diff --git a/avbroot/formats/compression.py b/avbroot/formats/compression.py new file mode 100644 index 0000000..98d679b --- /dev/null +++ b/avbroot/formats/compression.py @@ -0,0 +1,184 @@ +import enum +import gzip +import typing + +import lz4.block + +from .. import util + + +GZIP_MAGIC = b'\x1f\x8b' + + +class Lz4Legacy: + MAGIC = b'\x02\x21\x4c\x18' + MAX_BLOCK_SIZE = 8 * 1024 * 1024 + + def __init__(self, fp: typing.BinaryIO, + mode: typing.Literal['rb', 'wb'] = 'rb'): + if mode not in ('rb', 'wb'): + raise ValueError(f'Invalid mode: {mode}') + + self.fp = fp + self.mode = mode + + if mode == 'rb': + magic = util.read_exact(self.fp, len(self.MAGIC)) + if magic != self.MAGIC: + raise ValueError(f'Invalid magic: {magic!r}') + + self.rblock = b'' + self.rblock_offset = 0 + else: + self.fp.write(self.MAGIC) + + self.wblock = bytearray() + + self.file_offset = 0 + + def __enter__(self) -> 'Lz4Legacy': + return self + + def __exit__(self, *exc_args) -> None: + self.close() + + def _read_block(self) -> None: + if self.rblock_offset < len(self.rblock): + # Haven't finished reading block yet + return + + size_raw = self.fp.read(4) + if not size_raw or size_raw == self.MAGIC: + self.rblock = b'' + self.rblock_offset = 0 + return + elif len(size_raw) != 4: + raise EOFError('Failed to read block size') + + size_compressed = int.from_bytes(size_raw, 'little') + + compressed = util.read_exact(self.fp, size_compressed) + self.rblock = lz4.block.decompress(compressed, self.MAX_BLOCK_SIZE) + self.rblock_offset = 0 + + def _write_block(self, force=False) -> None: + if not force and len(self.wblock) < self.MAX_BLOCK_SIZE: + # Block not fully filled yet + return + + compressed = lz4.block.compress( + self.wblock, + mode='high_compression', + compression=12, + store_size=False, + ) + + self.fp.write(len(compressed).to_bytes(4, 'little')) + self.fp.write(compressed) + + self.wblock.clear() + + def read(self, size=None) -> bytes: + assert self.mode == 'rb' + + result = bytearray() + + while size is None or size > 0: + self._read_block() + + to_read = len(self.rblock) - self.rblock_offset + if to_read == 0: + # EOF + break + elif size is not None: + to_read = min(to_read, size) + + result.extend(self.rblock[self.rblock_offset: + self.rblock_offset + to_read]) + + self.rblock_offset += to_read + self.file_offset += to_read + + if size is not None: + size -= to_read + + return result + + def write(self, data: bytes) -> int: + assert self.mode == 'wb' + + offset = 0 + + while offset < len(data): + self._write_block() + + to_write = min( + self.MAX_BLOCK_SIZE - len(self.wblock), + len(data) - offset, + ) + + self.wblock.extend(data[offset:offset + to_write]) + + self.file_offset += to_write + offset += to_write + + return len(data) + + def flush(self) -> None: + assert self.mode == 'wb' + + self._write_block(force=True) + + def close(self) -> None: + try: + if self.mode == 'wb': + self.flush() + finally: + self.mode = 'closed' + + def tell(self) -> int: + return self.file_offset + + +Format = enum.Enum('Format', ['GZIP', 'LZ4_LEGACY']) + + +_MAGIC_TO_FORMAT = { + GZIP_MAGIC: Format.GZIP, + Lz4Legacy.MAGIC: Format.LZ4_LEGACY, +} +_MAGIC_MAX_SIZE = max(len(m) for m in _MAGIC_TO_FORMAT) + + +class CompressedFile: + def __init__( + self, + fp: typing.BinaryIO, + mode: typing.Literal['rb', 'wb'] = 'rb', + format: None | Format = None, + ): + if mode == 'rb' and not format: + magic = fp.read(_MAGIC_MAX_SIZE) + fp.seek(0) + + for m, f in _MAGIC_TO_FORMAT.items(): + if magic.startswith(m): + format = f + break + + if format == Format.GZIP: + format_fp = gzip.open(fp, mode) + elif format == Format.LZ4_LEGACY: + format_fp = Lz4Legacy(fp, mode) + else: + raise ValueError('Unknown compression format') + + self.fp = format_fp + self.format = format + + def __enter__(self): + self.fp.__enter__() + return self + + def __exit__(self, *exc_args): + self.fp.__exit__(*exc_args) diff --git a/avbroot/formats/cpio.py b/avbroot/formats/cpio.py new file mode 100644 index 0000000..c60437e --- /dev/null +++ b/avbroot/formats/cpio.py @@ -0,0 +1,280 @@ +# This is a miniature implementation of cpio, originally written for +# DualBootPatcher, supporting only enough of the file format for messing with +# boot image ramdisks. Only the "new format" for cpio entries are supported. + +import stat +import typing + +from . import padding +from .. import util + +MAGIC_NEW = b'070701' # new format +MAGIC_NEW_CRC = b'070702' # new format w/crc + +# Constants from cpio.h + +# A header with a filename "TRAILER!!!" indicates the end of the archive. +CPIO_TRAILER = b'TRAILER!!!' + +C_ISCTG = 0o0110000 + +IO_BLOCK_SIZE = 512 + + +def _read_int(f: typing.BinaryIO) -> int: + return int(util.read_exact(f, 8), 16) + + +def _write_int(f: typing.BinaryIO, value: int) -> int: + if value < 0 or value > 0xffffffff: + raise ValueError(f'{value} out of range for 32-bit integer') + + return f.write(b'%08x' % value) + + +class CpioEntryNew: + # c_magic - "070701" for "new" portable format + # "070702" for CRC format + # c_ino + # c_mode + # c_uid + # c_gid + # c_nlink + # c_mtime + # c_filesize - must be 0 for FIFOs and directories + # c_dev_maj + # c_dev_min + # c_rdev_maj - only valid for chr and blk special files + # c_rdev_min - only valid for chr and blk special files + # c_namesize - count includes terminating NUL in pathname + # c_chksum - 0 for "new" portable format; for CRC format + # the sum of all the bytes in the file + + @staticmethod + def new_trailer() -> 'CpioEntryNew': + entry = CpioEntryNew() + entry.nlink = 1 # Must be 1 for crc format + entry.name = CPIO_TRAILER + + return entry + + @staticmethod + def new_symlink(link_target: bytes, name: bytes) -> 'CpioEntryNew': + if not link_target: + raise ValueError('Symlink target is empty') + elif not name: + raise ValueError('Symlink name is empty') + + entry = CpioEntryNew() + entry.mode = stat.S_IFLNK | 0o777 + entry.nlink = 1 + entry.name = name + entry.content = link_target + + return entry + + @staticmethod + def new_directory(name: bytes, perms: int = 0o755) -> 'CpioEntryNew': + if not name: + raise ValueError('Directory name is empty') + + entry = CpioEntryNew() + entry.mode = stat.S_IFDIR | stat.S_IMODE(perms) + entry.nlink = 1 + entry.name = name + + return entry + + @staticmethod + def new_file(name: bytes, perms: int = 0o644, + data: bytes = b'') -> 'CpioEntryNew': + if not name: + raise ValueError('File name is empty') + + entry = CpioEntryNew() + entry.mode = stat.S_IFREG | stat.S_IMODE(perms) + entry.nlink = 1 + entry.name = name + entry.content = data + + return entry + + def __init__(self, f: typing.Optional[typing.BinaryIO] = None) -> None: + super(CpioEntryNew, self).__init__() + + if f is None: + self.magic = MAGIC_NEW + self.ino = 0 + self.mode = 0 + self.uid = 0 + self.gid = 0 + self.nlink = 0 + self.mtime = 0 + self.filesize = 0 + self.dev_maj = 0 + self.dev_min = 0 + self.rdev_maj = 0 + self.rdev_min = 0 + self.namesize = 0 + self.chksum = 0 + + self._name = b'' + self._content = b'' + else: + self.magic = util.read_exact(f, 6) + if self.magic != MAGIC_NEW and self.magic != MAGIC_NEW_CRC: + raise Exception(f'Unknown magic: {self.magic!r}') + + self.ino = _read_int(f) + self.mode = _read_int(f) + self.uid = _read_int(f) + self.gid = _read_int(f) + self.nlink = _read_int(f) + self.mtime = _read_int(f) + self.filesize = _read_int(f) + self.dev_maj = _read_int(f) + self.dev_min = _read_int(f) + self.rdev_maj = _read_int(f) + self.rdev_min = _read_int(f) + self.namesize = _read_int(f) + self.chksum = _read_int(f) + + # Filename + self._name = util.read_exact(f, self.namesize - 1) + # Discard NULL terminator + util.read_exact(f, 1) + padding.read_skip(f, 4) + + # File contents + self._content = util.read_exact(f, self.filesize) + padding.read_skip(f, 4) + + def write(self, f: typing.BinaryIO): + if len(self.magic) != 6: + raise ValueError(f'Magic is not 6 bytes: {self.magic!r}') + + f.write(self.magic) + + _write_int(f, self.ino) + _write_int(f, self.mode) + _write_int(f, self.uid) + _write_int(f, self.gid) + _write_int(f, self.nlink) + _write_int(f, self.mtime) + _write_int(f, self.filesize) + _write_int(f, self.dev_maj) + _write_int(f, self.dev_min) + _write_int(f, self.rdev_maj) + _write_int(f, self.rdev_min) + _write_int(f, self.namesize) + _write_int(f, self.chksum) + + # Filename + f.write(self._name) + f.write(b'\x00') + padding.write(f, 4) + + # File contents + f.write(self._content) + padding.write(f, 4) + + @property + def name(self) -> bytes: + return self._name + + @name.setter + def name(self, value: bytes): + self._name = value + self.namesize = len(value) + 1 + + @property + def content(self) -> bytes: + return self._content + + @content.setter + def content(self, value: bytes): + self._content = value + self.filesize = len(value) + + def __str__(self) -> str: + filetype = stat.S_IFMT(self.mode) + + if stat.S_ISDIR(self.mode): + ftypestr = 'directory' + elif stat.S_ISLNK(self.mode): + ftypestr = 'symbolic link' + elif stat.S_ISREG(self.mode): + ftypestr = 'regular file' + elif stat.S_ISFIFO(self.mode): + ftypestr = 'pipe' + elif stat.S_ISCHR(self.mode): + ftypestr = 'character device' + elif stat.S_ISBLK(self.mode): + ftypestr = 'block device' + elif stat.S_ISSOCK(self.mode): + ftypestr = 'socket' + elif filetype == C_ISCTG: + ftypestr = 'reserved' + else: + ftypestr = 'unknown (%o)' % filetype + + return \ + f'Filename: {self.name!r}\n' \ + f'Filetype: {ftypestr}\n' \ + f'Magic: {self.magic!r}\n' \ + f'Inode: {self.ino}\n' \ + f'Mode: {self.mode:o}\n' \ + f'Permissions: {self.mode - filetype:o}\n' \ + f'UID: {self.uid}\n' \ + f'GID: {self.gid}\n' \ + f'Links: {self.nlink}\n' \ + f'Modified: {self.mtime}\n' \ + f'File size: {self.filesize}\n' \ + f'dev_maj: {self.dev_maj:x}\n' \ + f'dev_min: {self.dev_min:x}\n' \ + f'rdev_maj: {self.rdev_maj:x}\n' \ + f'rdev_min: {self.rdev_min:x}\n' \ + f'Filename length: {self.namesize}\n' \ + f'Checksum: {self.chksum:x}\n' + + +def load(f: typing.BinaryIO) -> list[CpioEntryNew]: + entries = [] + + while True: + entry = CpioEntryNew(f) + + if entry.name == CPIO_TRAILER: + break + + if stat.S_IFMT(entry.mode) != stat.S_IFDIR and entry.nlink > 1: + raise ValueError(f'Hard links are not supported: {entry.name!r}') + + # Inodes are reassigned on save + entry.ino = 0 + + entries.append(entry) + + return entries + + +def save(f: typing.BinaryIO, entries: list[CpioEntryNew], sort=True, + pad_to_block_size=False): + inode = 300000 + + if sort: + entries = sorted(entries, key=lambda e: e.name) + + for entry in entries: + entry.ino = inode + inode += 1 + + entry.write(f) + + trailer = CpioEntryNew.new_trailer() + trailer.ino = inode + trailer.write(f) + + # Pad until end of block + if pad_to_block_size: + padding.write(f, IO_BLOCK_SIZE) diff --git a/avbroot/formats/padding.py b/avbroot/formats/padding.py new file mode 100644 index 0000000..2e9855b --- /dev/null +++ b/avbroot/formats/padding.py @@ -0,0 +1,47 @@ +import os +import typing + + +def _is_power_of_2(n: int) -> bool: + if hasattr(n, 'bit_count'): + return n.bit_count() == 1 + else: + return bin(n).count('1') == 1 + + +def calc(offset: int, page_size: int) -> int: + ''' + Calculate the amount of padding that needs to be added to align the + specified offset to a page boundary. The page size must be a power of 2. + ''' + + if not _is_power_of_2(page_size): + raise ValueError(f'{page_size} is not a power of 2') + + return (page_size - (offset & (page_size - 1))) & (page_size - 1) + + +def read_skip(f: typing.BinaryIO, page_size: int) -> int: + ''' + Seek file to the next page boundary if it is not already at a page + boundary. If the file does not support seeking, then data is read and + discarded. + ''' + + padding = calc(f.tell(), page_size) + + if hasattr(f, 'seek'): + f.seek(padding, os.SEEK_CUR) + else: + f.read(padding) + + return padding + + +def write(f: typing.BinaryIO, page_size: int) -> int: + ''' + Write null bytes to pad the file to the next page boundary if it is not + already at a page boundary. + ''' + + return f.write(calc(f.tell(), page_size) * b'\x00') diff --git a/avbroot/util.py b/avbroot/util.py index 12dd460..e285988 100644 --- a/avbroot/util.py +++ b/avbroot/util.py @@ -21,6 +21,24 @@ def open_output_file(path): raise +def hash_file(f, hasher, buf_size=16384): + ''' + Update when the data from until EOF. + ''' + + buf = bytearray(buf_size) + buf_view = memoryview(buf) + + while True: + n = f.readinto(buf_view) + if not n: + break + + hasher.update(buf_view[:n]) + + return hasher + + def copyfileobj_n(f_in, f_out, size, buf_size=16384, hasher=None): ''' Copy bytes from to . @@ -89,3 +107,16 @@ def zero_n(f_out, size, buf_size=16384): to_write = min(len(buf_view), size) f_out.write(buf_view[:to_write]) size -= to_write + + +def read_exact(f, size: int) -> bytes: + ''' + Read exactly bytes from or raise an EOFError. + ''' + + data = f.read(size) + if len(data) != size: + raise EOFError(f'Unexpected EOF: expected {size} bytes, ' + f'but only read {len(data)} bytes') + + return data