awesome-android/avbroot
3
0
Fork 0
mirror of https://github.com/chenxiaolong/avbroot.git synced 2026-07-03 14:05:11 +02:00
Code Issues Packages Projects Releases Wiki Activity

Remove dependency on magiskboot

Browse Source 
This commit reimplements the magiskboot patching process natively in
Python. By doing so, avbroot gains cross platform support and there's no
longer a need to execute an Android binary on a non-Android Linux system,
which had pitfalls that already had to be worked around.

Boot images and cpio archives are handled by new custom parsers added to
avbroot. Legacy lz4 compression is done by a wrapper around the python
lz4 library (a new dependency). Gzip compression is handled natively by
python. Other compression methods and OEM-specific boot image formats
are not supported because devices that use the modern Android A/B OTA
scheme do not use those.

Output files are still bit-for-bit reproducible across runs, but they
are different from what prior avbroot commits produced:

* avbroot's cpio writer follows GNU cpio and libarchive's behavior of
  setting the mode field to 0 in the trailer entry. magiskboot sets the
  mode to 0o755.
* When patching a vendor boot v4 image, the ramdisk table entries are
  now updated correctly. Prior vendor boot images were only bootable
  because:

  * the image only contained a single ramdisk
  * the single ramdisk shrunk in size, stayed the same, or grew little
    enough to not exceed a page boundary
  * the bootloader is lenient in validating boot image fields

  magiskboot does not handle vendor boot v4 images correctly, but it
  doesn't need to just for the regular root patch. avbroot made use of
  it in an unsupported way to patch vendor boot ramdisks.

Signed-off-by: Andrew Gunnerson <accounts+github@chiller3.com>
This commit is contained in:
Andrew Gunnerson
2023-02-10 23:06:34 -05:00
parent 1cbfd96067
commit 6c116f0dba
8 changed files with 1293 additions and 132 deletions
Download Patch File Download Diff File Expand all files Collapse all files
README.md
+1 -3
View File
@@ -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).
avbroot.py
+1 -1
View File
@@ -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)
avbroot/boot.py
+137 -128
View File
@@ -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,
avbroot/formats/bootimage.py
+612
View File
@@ -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')
avbroot/formats/compression.py
+184
View File
@@ -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)
avbroot/formats/cpio.py
+280
View File
@@ -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)
avbroot/formats/padding.py
+47
View File
@@ -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')
avbroot/util.py
+31
View File
@@ -21,6 +21,24 @@ def open_output_file(path):
raise
def hash_file(f, hasher, buf_size=16384):
'''
Update <hasher> when the data from <f> 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 <size> bytes from <f_in> to <f_out>.
@@ -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 <size> bytes from <f> 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