Device-generic CLI for the torvalds-based hardware layer

NOTICE

@@ -1,7 +1,8 @@
 Default project license: GPL-2.0-or-later (LICENSE).
 Pongo/libusb uploader: checkra1n team, MIT (LICENSES/Pongo.txt).
-Touch conversion (j71b/touch.py): GPL-2.0-or-later. Packet layouts and Z2FW
+Touch conversion (apple/z2fw.py): GPL-2.0-or-later. Packet layouts and Z2FW
 framing are informed by Corellium's Project Sandcastle and Paul Praschl's
-(Pauli1Go) makez2fw:
+(Pauli1Go) makez2fw. convert_packets() and convert_mtfw() port makez2fw's
+Gen2 mtprops and MTFW HBPP item conversions (mtfw.c, makez2fw.c):
 https://github.com/Pauli1Go/HoolockLinux-linux-firmware
 Reference revision: 944541e67b4c603df328c77d30093551de802c17.

pyproject.toml

@@ -4,13 +4,13 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "orchard-cli"
-version = "0.1.10"
+version = "0.1.11"
 description = "Apple device Linux boot and installation tooling"
 readme = "README.md"
 license = "GPL-2.0-or-later AND MIT"
 license-files = ["LICENSE", "NOTICE", "LICENSES/*"]
 requires-python = ">=3.9"
-dependencies = ["orchard-hardware>=0.1.10,<0.2", "jsonschema>=4.25,<5", "tomli>=2; python_version < '3.11'"]
+dependencies = ["orchard-hardware>=0.1.11,<0.2", "jsonschema>=4.25,<5", "tomli>=2; python_version < '3.11'"]
 
 [project.urls]
 Repository = "https://github.com/OrchardLinux/cli"
@@ -23,7 +23,7 @@ orch = "orchard_cli.command:main"
 
 [tool.setuptools]
 include-package-data = false
-packages = ["orchard_cli", "orchard_cli.j71b", "orchard_cli.apple", "orchard_cli.transports", "orchard_cli_schemas"]
+packages = ["orchard_cli", "orchard_cli.apple", "orchard_cli.transports", "orchard_cli_schemas"]
 package-dir = {"" = "src", orchard_cli_schemas = "schemas"}
 
 [tool.setuptools.package-data]
@@ -36,12 +36,12 @@ dev = ["mypy==1.18.2"]
 [tool.mypy]
 python_version = "3.9"
 mypy_path = "src"
-files = ["src/orchard_cli/contracts.py", "src/orchard_cli/devices.py", "src/orchard_cli/j71b/adapter.py", "src/orchard_cli/j71b/inputs.py", "src/orchard_cli/j71b/mixin.py", "src/orchard_cli/service.py", "src/orchard_cli/command.py", "tests/typecheck_adapter.py"]
+files = ["src/orchard_cli/contracts.py", "src/orchard_cli/devices.py", "src/orchard_cli/apple/device.py", "src/orchard_cli/service.py", "src/orchard_cli/command.py", "tests/typecheck_adapter.py"]
 check_untyped_defs = true
 follow_imports = "silent"
 ignore_missing_imports = true
 warn_unused_ignores = true
 
 [[tool.mypy.overrides]]
-module = ["orchard_cli.contracts", "orchard_cli.devices", "orchard_cli.j71b.adapter", "orchard_cli.j71b.inputs", "orchard_cli.j71b.mixin"]
+module = ["orchard_cli.contracts", "orchard_cli.devices", "orchard_cli.apple.device"]
 disallow_untyped_defs = true

src/orchard_cli/apple/device.py

@@ -0,0 +1,90 @@
+"""Generic Apple device adapter: identity from hardware's device.toml, shared transports.
+
+Supporting a new device needs only its hardware descriptor and firmware
+recipe, not CLI code.
+"""
+from contextlib import AbstractContextManager
+from functools import partial
+from typing import Any, Dict, List, Optional, cast
+
+from orchard_cli.contracts import BootAssets, BootRequest, Connection, SelectedDevice
+from orchard_cli.state import BootState, DeviceState
+from orchard_cli.progress import Progress
+from orchard_cli.apple import usb, boot as apple_boot, adt_capture
+from orchard_cli.apple.adt_tree import parse
+from orchard_cli.transports import ssh
+from orchard_cli import firmware
+from orchard_hardware.apple.adt import validate as validate_adt
+
+
+class Device:
+    def __init__(self, descriptor: Dict[str, Any]):
+        identity = descriptor['device']
+        self.PROFILE = identity['profile']
+        self.NAME = identity['name']
+        self.APPLE_PRODUCT = identity['apple_products'][0]
+        # USB product of the device's recovery system; absent until it has one.
+        self.PRODUCT = descriptor['match'].get('linux', {}).get('product')
+        self.CHIP = descriptor['match']['dfu']['chip']
+        self.BOARD = descriptor['match']['dfu']['board']
+        support = descriptor.get('support', {})
+        self.INSTALL = bool(support.get('install', False))
+        self.STATUS = support.get('status', 'experimental')
+        self.CAVEATS = list(support.get('caveats', []))
+        self.FEATURES = dict(descriptor.get('features', {}))
+        self.REBOOT_MESSAGE = ('A fresh recovery boot is required.\n'
+                               f'Put the {self.NAME} in DFU mode, then run: orchard install')
+        self.firmware_inputs = partial(firmware.ensure, self.PROFILE)
+        self.prepare_mixin = partial(firmware.mixin, self.PROFILE)
+
+    def matches_dfu(self, device: Dict[str, Any]) -> bool:
+        return (device.get('chip'), device.get('board')) == (self.CHIP, self.BOARD)
+
+    def recovery_endpoint(self) -> Dict[str, Any]:
+        if self.PRODUCT is None:
+            raise ValueError(f'The {self.NAME} has no Orchard recovery system yet')
+        return usb.recovery(self.PRODUCT)
+
+    def connect(self, device: SelectedDevice) -> AbstractContextManager[Connection]:
+        return ssh.session(DeviceState(self.PROFILE, device['ecid']), self.recovery_endpoint,
+                           expected_boot=device.get('boot_id'))
+
+    def connect_saved(self, device: SelectedDevice) -> AbstractContextManager[Connection]:
+        if device['mode'] != 'linux' or not device.get('boot_id'):
+            raise ValueError('Diagnostic access requires the selected running Linux boot')
+        return ssh.saved_session(DeviceState(self.PROFILE, device['ecid']), self.recovery_endpoint,
+                                 expected_boot=device['boot_id'])
+
+    def forget_access(self, scope: BootState) -> None:
+        if scope.device.profile != self.PROFILE:
+            raise ValueError('Credential scope differs from adapter profile')
+        return ssh.forget_access(scope)
+
+    def capture_adt(self, device: Optional[SelectedDevice] = None) -> bytes:
+        # Bring-up and diagnostics only; installs do not need the ADT.
+        # DFU/Pongo and Linux location IDs differ: bind the fresh ADT to the ECID.
+        data = adt_capture.capture(self.recovery_endpoint)
+        if parse(data).properties.get('model', b'').rstrip(b'\0') != self.APPLE_PRODUCT.encode():
+            raise ValueError('Captured ADT is not for ' + self.APPLE_PRODUCT)
+        if device is not None and validate_adt(data) != device['ecid']:
+            raise ValueError('Fresh ADT identity differs from the confirmed device')
+        return data
+
+    def list_devices(self) -> List[Dict[str, Any]]:
+        return usb.devices()
+
+    def discover(self) -> SelectedDevice:
+        device = usb.sole_target()
+        if device['mode'] == 'linux':
+            endpoint = self.recovery_endpoint()
+            BootState(DeviceState(self.PROFILE, endpoint['ecid']), endpoint['boot_id'])
+            return cast(SelectedDevice, {**endpoint, 'resumed': True})
+        if not self.matches_dfu(device) or not device['ecid']:
+            raise ValueError(f'The connected device is not a {self.NAME}')
+        DeviceState(self.PROFILE, device['ecid'])
+        return cast(SelectedDevice, {**device, 'resumed': False})
+
+    def boot(self, device: SelectedDevice, assets: BootAssets,
+             manifest: BootRequest, progress: Progress) -> None:
+        return apple_boot.boot(device, assets, manifest, progress,
+                               profile=self.PROFILE, chip=self.CHIP, board=self.BOARD, product=self.PRODUCT)

src/orchard_cli/apple/ipsw.py

@@ -11,7 +11,6 @@
 
 from orchard_cli.cache import cache_root
 from orchard_cli.integrity import digest
-from orchard_cli.releases import materialize
 
 TOOL = {'url': 'https://github.com/blacktop/ipsw/releases/download/v3.1.739/ipsw_3.1.739_macOS_arm64.tar.gz',
         'sha256': 'c09e8eba02c332f5b50d15f94383e59dd35646d70f6ea9e78d185e9afda6abc2',
@@ -46,6 +45,8 @@ def hcd_from_binary(data, expected):
 
 
 def tool(clock, progress):
+    # Keep the pure firmware selector usable on-device without download tooling.
+    from orchard_cli.releases import materialize
     supplied = os.environ.get('ORCHARD_IPSW_TOOL') or shutil.which('ipsw')
     if supplied:
         return supplied

src/orchard_cli/apple/kernelcache.py

@@ -0,0 +1,83 @@
+"""Bounded installed IMG4/LZFSE kernel decoding using Apple's system library."""
+import ctypes
+import struct
+
+MAX_INPUT = 64 * 1024 * 1024
+MAX_OUTPUT = 128 * 1024 * 1024
+
+
+def der_items(blob):
+    pos = 0
+    while pos < len(blob):
+        if pos + 2 > len(blob):
+            raise ValueError('Truncated DER header')
+        tag, length = blob[pos:pos+2]
+        pos += 2
+        if length & 128:
+            count = length & 127
+            if not 1 <= count <= 4 or pos+count > len(blob) or blob[pos] == 0:
+                raise ValueError('Invalid DER length')
+            length = int.from_bytes(blob[pos:pos+count], 'big')
+            pos += count
+            if length < 128:
+                raise ValueError('Noncanonical DER length')
+        if pos+length > len(blob):
+            raise ValueError('Truncated DER value')
+        yield tag, blob[pos:pos+length]
+        pos += length
+
+
+def kernel_payload(blob):
+    if not 1 <= len(blob) <= MAX_INPUT:
+        raise ValueError('Kernel input exceeds bounds')
+    outer = list(der_items(blob))
+    if len(outer) != 1 or outer[0][0] != 0x30:
+        raise ValueError('Expected one IMG4/IM4P sequence')
+    nodes = list(der_items(outer[0][1]))
+    if nodes and nodes[0] == (0x16, b'IMG4'):
+        if len(nodes) < 2 or nodes[1][0] != 0x30:
+            raise ValueError('Missing IM4P')
+        nodes = list(der_items(nodes[1][1]))
+    if (len(nodes) < 4 or nodes[0] != (0x16, b'IM4P') or
+            nodes[1] != (0x16, b'krnl') or nodes[2][0] != 0x16 or nodes[3][0] != 4):
+        raise ValueError('Not a kernel IM4P payload')
+    return nodes[3][1]
+
+
+def arm64_slice(blob):
+    if blob[:4] == b'\xca\xfe\xba\xbe':
+        if len(blob) < 8:
+            raise ValueError('Truncated fat kernel')
+        count = struct.unpack_from('>I', blob, 4)[0]
+        if not 1 <= count <= 16 or 8+20*count > len(blob):
+            raise ValueError('Invalid fat architecture table')
+        selected = []
+        for i in range(count):
+            cpu, subtype, offset, size, alignment = struct.unpack_from('>5I', blob, 8+20*i)
+            if offset < 8+20*count or offset+size > len(blob) or alignment > 31 or offset % (1 << alignment):
+                raise ValueError('Invalid fat slice bounds/alignment')
+            if cpu == 0x100000c:
+                selected.append(blob[offset:offset+size])
+        if len(selected) != 1:
+            raise ValueError('No unique ARM64 kernel slice')
+        blob = selected[0]
+    if len(blob) < 32 or struct.unpack_from('<II', blob) != (0xfeedfacf, 0x100000c):
+        raise ValueError('Not an ARM64 Mach-O kernel')
+    return blob
+
+
+def decode(blob):
+    payload = kernel_payload(blob)
+    if payload[:4] != b'bvx2':
+        raise ValueError('Only installed LZFSE kernel payloads are qualified')
+    lib = ctypes.CDLL('/usr/lib/libcompression.dylib')
+    call = lib.compression_decode_buffer
+    call.argtypes = [ctypes.c_void_p, ctypes.c_size_t, ctypes.c_void_p,
+                     ctypes.c_size_t, ctypes.c_void_p, ctypes.c_int]
+    call.restype = ctypes.c_size_t
+    source = ctypes.create_string_buffer(payload)
+    target = ctypes.create_string_buffer(MAX_OUTPUT)
+    size = call(target, MAX_OUTPUT, source, len(payload), None, 0x801)
+    if not 0 < size < MAX_OUTPUT:
+        raise ValueError('LZFSE decode failed or output exceeds bound')
+    return arm64_slice(target.raw[:size])