Merge pull request #82 from chenxiaolong/magisk-25207

Magisk >=25207 require a rules device ID

README.md

@@ -141,14 +141,15 @@ To install the Python dependencies:
 
     If `--output` is not specified, then the output file is written to `<input>.patched`.
 
-    If you already have a prepatched boot image and don't want avbroot to apply the Magisk patches itself, see the [advanced usage section](#advanced-usage).
+    **NOTE:** If you are using Magisk version >=25207, you need to know the rules device ID (`--magisk-rules-device <ID>`). For details, see the [Magisk rules device section](#magisk-rules-device).
+
+    If you prefer to use an existing boot image patched by the Magisk app or you want to use KernelSU, see the [advanced usage section](#advanced-usage).
 
 6. **[Initial setup only]** Unlock the bootloader. This will trigger a data wipe.
 
 7. **[Initial setup only]** Extract the patched images from the patched OTA.
 
     ```bash
-    mkdir extracted
     python avbroot.py \
         extract \
         --input /path/to/ota.zip.patched \
@@ -203,6 +204,34 @@ python clearotacerts/build.py
 
 and flash the `clearotacerts/dist/clearotacerts-<version>.zip` file in Magisk. The module simply overrides `/system/etc/security/otacerts.zip` at runtime with an empty zip so that even if an OTA is downloaded, signature verification will fail.
 
+## Magisk rules device
+
+Magisk versions 25207 and newer require the device ID for a writable partition that is used to store custom SELinux rules. This can only be determined on a real device, so avbroot requires the device ID to be specified via `--magisk-rules-device <ID>`. To find the device ID:
+
+1. Extract the boot image from the original/unpatched OTA:
+
+    ```bash
+    python avbroot.py \
+        extract \
+        --input /path/to/ota.zip \
+        --directory . \
+        --boot-only
+    ```
+
+2. Patch the boot image via the Magisk app. This **MUST** be done on the target device! The device ID will be incorrect if patched from Magisk on a different device.
+
+3. Find the device ID from the patched boot image:
+
+    ```bash
+    python avbroot.py \
+        magisk-info \
+        --image magisk_patched-*.img
+    ```
+
+    The device ID shown in the output (`RULESDEVICE=<ID>`) can be passed to avbroot via `--magisk-rules-device <ID>`. The ID should be saved somewhere for future reference since it does not change across updates.
+
+If it's not possible to run the Magisk app on the target device (eg. device is currently unbootable), patch and flash the OTA once using `--ignore-magisk-warnings`, follow these steps, and the repatch and reflash the OTA with `--magisk-rules-device <ID>`.
+
 ## Advanced Usage
 
 ### Using a prepatched boot image

avbroot/boot.py

@@ -52,8 +52,35 @@ class MagiskRootPatch(BootImagePatch):
     Root the boot image with Magisk.
     '''
 
-    def __init__(self, magisk_apk):
+    # Half-open intervals
+    VER_SUPPORTED = util.Range(22000, 25211)
+    VER_RULES_DEVICE = util.Range(25207, VER_SUPPORTED.end)
+
+    def __init__(self, magisk_apk, rules_device):
         self.magisk_apk = magisk_apk
+        self.rules_device = rules_device
+
+    def _get_version(self):
+        with zipfile.ZipFile(self.magisk_apk, '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('Failed to determine Magisk version from: '
+                        f'{self.magisk_apk}')
+
+    def validate(self):
+        version = self._get_version()
+
+        if version not in self.VER_SUPPORTED:
+            raise ValueError(f'Unsupported Magisk version {version} '
+                             f'(supported: >={self.VER_SUPPORTED})')
+
+        if self.rules_device is None and version in self.VER_RULES_DEVICE:
+            raise ValueError(f'Magisk version {version} '
+                             f'({self.VER_RULES_DEVICE}) requires a rules '
+                             f'device to be specified')
 
     def patch(self, image_file, boot_image):
         with zipfile.ZipFile(self.magisk_apk, 'r') as zip:
@@ -125,8 +152,10 @@ def _patch(self, image_file, boot_image, zip):
             b'KEEPVERITY=true\n' \
             b'KEEPFORCEENCRYPT=true\n' \
             b'PATCHVBMETAFLAG=false\n' \
-            b'RECOVERYMODE=false\n' \
-            b'SHA1=%s\n' % hasher.hexdigest().encode('ascii')
+            b'RECOVERYMODE=false\n'
+        if self.rules_device is not None:
+            magisk_config += b'RULESDEVICE=%d\n' % self.rules_device
+        magisk_config += b'SHA1=%s\n' % hasher.hexdigest().encode('ascii')
         entries.append(cpio.CpioEntryNew.new_file(
             b'.backup/.magisk', perms=0o000, data=magisk_config))
 

avbroot/formats/compression.py

@@ -156,6 +156,7 @@ def __init__(
         fp: typing.BinaryIO,
         mode: typing.Literal['rb', 'wb'] = 'rb',
         format: typing.Optional[Format] = None,
+        raw_if_unknown = False,
     ):
         if mode == 'rb' and not format:
             magic = fp.read(_MAGIC_MAX_SIZE)
@@ -170,6 +171,8 @@ def __init__(
             format_fp = gzip.GzipFile(fileobj=fp, mode=mode, mtime=0)
         elif format == Format.LZ4_LEGACY:
             format_fp = Lz4Legacy(fp, mode)
+        elif raw_if_unknown:
+            format_fp = fp
         else:
             raise ValueError('Unknown compression format')
 

avbroot/main.py

@@ -1,6 +1,7 @@
 import argparse
 import concurrent.futures
 import copy
+import io
 import os
 import shutil
 import tempfile
@@ -14,6 +15,9 @@
 from . import ota
 from . import util
 from . import vbmeta
+from .formats import bootimage
+from .formats import compression
+from .formats import cpio
 
 
 PATH_METADATA = 'META-INF/com/android/metadata'
@@ -22,10 +26,6 @@
 PATH_PAYLOAD = 'payload.bin'
 PATH_PROPERTIES = 'payload_properties.txt'
 
-# Half-open range
-MAGISK_MIN_VERSION = 22000
-MAGISK_MAX_VERSION = 25300
-
 PARTITION_PRIORITIES = {
     '@vbmeta': ('vbmeta',),
     # The kernel is always in boot
@@ -76,7 +76,7 @@ def get_required_images(manifest, boot_partition):
 
 
 def patch_ota_payload(f_in, open_more_f_in, f_out, file_size, boot_partition,
-                      magisk, prepatched, clear_vbmeta_flags, privkey_avb,
+                      root_patch, clear_vbmeta_flags, privkey_avb,
                       passphrase_avb, privkey_ota, passphrase_ota, cert_ota):
     with tempfile.TemporaryDirectory() as temp_dir:
         extract_dir = os.path.join(temp_dir, 'extract')
@@ -95,15 +95,7 @@ def patch_ota_payload(f_in, open_more_f_in, f_out, file_size, boot_partition,
         ota.extract_images(open_more_f_in, manifest, blob_offset,
                            extract_dir, unique_images)
 
-        image_patches = {}
-
-        if magisk is not None:
-            image_patches[images['@rootpatch']] = \
-                [boot.MagiskRootPatch(magisk)]
-        else:
-            image_patches[images['@rootpatch']] = \
-                [boot.PrepatchedImage(prepatched)]
-
+        image_patches = {images['@rootpatch']: [root_patch]}
         image_patches.setdefault(images['@otacerts'], []).append(
             boot.OtaCertPatch(cert_ota))
 
@@ -158,7 +150,7 @@ def apply_patches(image, patches):
         )
 
 
-def patch_ota_zip(f_zip_in, f_zip_out, boot_partition, magisk, prepatched,
+def patch_ota_zip(f_zip_in, f_zip_out, boot_partition, root_patch,
                   clear_vbmeta_flags, privkey_avb, passphrase_avb, privkey_ota,
                   passphrase_ota, cert_ota):
     with (
@@ -249,8 +241,7 @@ def patch_ota_zip(f_zip_in, f_zip_out, boot_partition, magisk, prepatched,
                         f_out,
                         info.file_size,
                         boot_partition,
-                        magisk,
-                        prepatched,
+                        root_patch,
                         clear_vbmeta_flags,
                         privkey_avb,
                         passphrase_avb,
@@ -287,33 +278,24 @@ def patch_ota_zip(f_zip_in, f_zip_out, boot_partition, magisk, prepatched,
         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'
 
     if args.magisk is not None:
-        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}, ' \
-                    f'<{MAGISK_MAX_VERSION})'
-
-            if args.ignore_magisk_version:
-                print_warning(message)
+        root_patch = boot.MagiskRootPatch(args.magisk,
+                                          args.magisk_rules_device)
+
+        try:
+            root_patch.validate()
+        except ValueError as e:
+            if args.ignore_magisk_warnings:
+                print_warning(e)
             else:
-                raise Exception(message)
+                raise e
+    else:
+        root_patch = boot.PrepatchedImage(args.prepatched)
 
     # Get passphrases for keys
     passphrase_avb = openssl.prompt_passphrase(args.privkey_avb)
@@ -336,8 +318,7 @@ def patch_subcommand(args):
                 args.input,
                 temp,
                 args.boot_partition,
-                args.magisk,
-                args.prepatched,
+                root_patch,
                 args.clear_vbmeta_flags,
                 args.privkey_avb,
                 passphrase_avb,
@@ -368,7 +349,10 @@ def extract_subcommand(args):
                                 for p in manifest.partitions)
         else:
             images = get_required_images(manifest, args.boot_partition)
-            unique_images = set(images.values())
+            if args.boot_only:
+                unique_images = {images['@rootpatch']}
+            else:
+                unique_images = set(images.values())
 
         print_status('Extracting', ', '.join(sorted(unique_images)),
                      'from the payload')
@@ -383,6 +367,26 @@ def extract_subcommand(args):
                            unique_images)
 
 
+def magisk_info_subcommand(args):
+    with open(args.image, 'rb') as f:
+        img = bootimage.load_autodetect(f)
+
+    if not img.ramdisks:
+        raise ValueError('Boot image does not have a ramdisk')
+
+    with (
+        io.BytesIO(img.ramdisks[0]) as f_raw,
+        compression.CompressedFile(f_raw, 'rb', raw_if_unknown=True) as f,
+    ):
+        entries = cpio.load(f.fp)
+        config = next((e for e in entries if e.name == b'.backup/.magisk'),
+                      None)
+        if config is None:
+            raise ValueError('Not a Magisk-patched boot image')
+
+        print(config.content.decode('ascii'), end='')
+
+
 def parse_args(argv=None):
     parser = argparse.ArgumentParser()
     subparsers = parser.add_subparsers(dest='subcommand', required=True,
@@ -409,8 +413,10 @@ def parse_args(argv=None):
     boot_group.add_argument('--prepatched',
                             help='Path to prepatched boot image')
 
-    patch.add_argument('--ignore-magisk-version', action='store_true',
-                       help='Allow patching with unsupported Magisk versions')
+    patch.add_argument('--magisk-rules-device', type=int,
+                       help='Magisk rules device ID')
+    patch.add_argument('--ignore-magisk-warnings', action='store_true',
+                       help='Ignore Magisk compatibility/version warnings')
 
     patch.add_argument('--clear-vbmeta-flags', action='store_true',
                        help='Forcibly clear vbmeta flags if they disable AVB')
@@ -422,18 +428,28 @@ def parse_args(argv=None):
                          help='Path to patched OTA zip')
     extract.add_argument('--directory', default='.',
                          help='Output directory for extracted images')
-    extract.add_argument('--all', action='store_true',
-                         help='Extract all images from payload')
+    extract_group = extract.add_mutually_exclusive_group()
+    extract_group.add_argument('--all', action='store_true',
+                               help='Extract all images from the payload')
+    extract_group.add_argument('--boot-only', action='store_true',
+                               help='Extract only the boot image')
 
     for subcmd in (patch, extract):
         subcmd.add_argument('--boot-partition', default='@gki_ramdisk',
                             help='Boot partition name')
 
+    magisk_info = subparsers.add_parser(
+        'magisk-info', help='Print Magisk config from a patched boot image')
+    magisk_info.add_argument('--image', required=True,
+                             help='Patch to Magisk-patched boot image')
+
     args = parser.parse_args(args=argv)
 
-    if args.subcommand == 'patch' and \
-            args.ignore_magisk_version and args.magisk is None:
-        parser.error('--ignore-magisk-version requires --magisk')
+    if args.subcommand == 'patch' and args.magisk is None:
+        if args.magisk_rules_device:
+            parser.error('--magisk-rules-device requires --magisk')
+        elif args.ignore_magisk_warnings:
+            parser.error('--ignore-magisk-warnings requires --magisk')
 
     return args
 
@@ -445,5 +461,7 @@ def main(argv=None):
         patch_subcommand(args)
     elif args.subcommand == 'extract':
         extract_subcommand(args)
+    elif args.subcommand == 'magisk-info':
+        magisk_info_subcommand(args)
     else:
         raise NotImplementedError()

avbroot/util.py

@@ -1,11 +1,45 @@
 import contextlib
+import dataclasses
+import functools
 import os
 import tempfile
 
 
 _ZERO_BLOCK = memoryview(b'\0' * 16384)
 
 
+@dataclasses.dataclass
+@functools.total_ordering
+class Range:
+    '''
+    Simple class to represent a half-open interval.
+    '''
+
+    start: int
+    end: int
+
+    def __repr__(self) -> str:
+        return f'[{self.start}, {self.end})'
+
+    def __str__(self) -> str:
+        return f'>={self.start}, <{self.end}'
+
+    def __lt__(self, other) -> bool:
+        return (self.start, self.end) < (other.start, other.end)
+
+    def __eq__(self, other) -> bool:
+        return (self.start, self.end) == (other.start, other.end)
+
+    def __contains__(self, item) -> bool:
+        return item >= self.start and item < self.end
+
+    def __bool__(self) -> bool:
+        return self.start < self.end
+
+    def size(self) -> int:
+        return self.end - self.start
+
+
 @contextlib.contextmanager
 def open_output_file(path):
     '''