diff --git a/NOTICE b/NOTICE index 047a2bc..4cb2ae5 100644 --- a/NOTICE +++ b/NOTICE @@ -5,3 +5,9 @@ framing are informed by Corellium's Project Sandcastle and Paul Praschl's (Pauli1Go) makez2fw: https://github.com/Pauli1Go/HoolockLinux-linux-firmware Reference revision: 944541e67b4c603df328c77d30093551de802c17. +J71t touch packet selection and ADT calibration copies (j71t/touch.py, +j71t/calibration.py): GPL-2.0-or-later, from the omarchy-ipad5 iPad 5 port +(https://github.com/brkosebay/omarchy-ipad5), which builds on Paul Praschl's +(Pauli1Go) HoolockLinux A10 work. +VXD firmware record layout (apple/d5500.py): GPL-2.0-or-later, from the +omarchy-ipad5 video decoder work, which reads the same record. diff --git a/README.md b/README.md index 4b80aeb..fa92ae5 100644 --- a/README.md +++ b/README.md @@ -2,6 +2,15 @@ Host-side tools for installing, booting, and debugging Linux on supported Apple devices. +## Supported devices + +| Device | Identifier | Profile | +| --- | --- | --- | +| iPad 6 Wi-Fi | `iPad7,5` / J71b | `j71b` | +| iPad 5 Wi-Fi (TSMC) | `iPad6,11` / J71t (S8003) | `j71t` | + +The connected device selects its adapter. `orchard devices` lists observations from every adapter; the other iPad 5 Wi-Fi variant, J71s (S8000), is not supported. + ## Install or update Requires Linux or macOS, Python 3.9+ with venv/pip, libusb, and OpenSSH. Linux also requires sudo, udev, and [apfs-fuse](https://github.com/sgan81/apfs-fuse) with its FUSE runtime for firmware extraction. Install `apfs-fuse` separately and put it in PATH, or set `IPSW_APFS_FUSE_PATH` to its executable. `setup-host` checks for the executable before downloading tools or installing USB rules. diff --git a/docs/adapter-contract.md b/docs/adapter-contract.md index 8eeb9a3..7b18037 100644 --- a/docs/adapter-contract.md +++ b/docs/adapter-contract.md @@ -7,8 +7,8 @@ | Member | Contract and effects | | --- | --- | | `PROFILE`, `NAME`, `REBOOT_MESSAGE` | Registered profile key, human-readable name and fresh-boot instructions. The profile must match its registry entry. | -| `list_devices()` | Raw discovery observations for display; read-only, no boot/preparation or implicit adapter selection. Vendor-specific observation fields are not interpreted by the common workflow. | -| `discover()` | Read-only identification of exactly one supported device. Return a validated `SelectedDevice`; raise `ValueError` when no unique supported device is available. | +| `list_devices()` | Raw discovery observations for display; read-only, no boot/preparation or implicit adapter selection. Vendor-specific observation fields are not interpreted by the common workflow. `orchard devices` merges every registered adapter's observations without duplicates. | +| `discover()` | Read-only identification of exactly one supported device. Return a validated `SelectedDevice`; raise `UnsupportedDevice` (a `ValueError`) when the connected target is another model, and `ValueError` when no unique supported device is available. | | `boot(device, assets, manifest, progress)` | Enter recovery using already-verified boot artifacts. A resumed recovery is not rebooted/uploaded. Keep identity/port checks and timeout budgets; never retry an uncertain mutation. | | `connect(device)` | Context-managed serial bootstrap and pinned SSH access bound to the selected ECID/boot. No host-side device storage/service recipes. | | `connect_saved(device)` | Context-managed use of the selected device/boot's retained pinned credentials. Never reprovision credentials or drive a login prompt. | @@ -17,6 +17,8 @@ | `firmware_inputs(clock, progress)` | Return verified cached/acquired host firmware inputs; account download time separately. No device hardware or storage mutation. | | `prepare_mixin(adt, firmware_dir, outdir)` | Return `(Path, MixinReceipt)` for a private data-only archive bound to the selected device/recipe. Preserve firmware hash, containment, calibration and output guards. | +`devices.choose()` tries every registered adapter. Exactly one adapter must return a selection; `UnsupportedDevice` refusals are not errors, identical shared USB errors are reported once, and a target that no adapter supports is reported with the list of supported devices. `orchard mixin` selects the profile whose hardware descriptor lists the ADT's `model`, or takes an explicit `--profile`; the adapter still validates the ADT itself. + Arguments and return types are defined in `contracts.py`, not repeated in a second runtime schema. Image-facing protocol validation, manifest validation and hardware identity verification remain in their existing owners. ## Records @@ -31,7 +33,7 @@ TypedDict records describe the boundary. They do not replace validation of untru ## Enforcement -The registry checks required member presence, profile identity and presentation metadata without invoking discovery. Python runtime Protocol checks do not verify function signatures; static checking does that. The real J71b module is assigned directly to `Adapter` in `tests/typecheck_adapter.py`, so signature drift is rejected without creating a wrapper implementation. +The registry checks required member presence, profile identity and presentation metadata without invoking discovery. Python runtime Protocol checks do not verify function signatures; static checking does that. The real J71b and J71t modules are assigned directly to `Adapter` in `tests/typecheck_adapter.py`, so signature drift is rejected without creating a wrapper implementation. From `cli/`: @@ -39,6 +41,6 @@ From `cli/`: uv tool run --from 'mypy==1.18.2' mypy --config-file pyproject.toml ``` -The development-only checker is declared in the dev dependency group, not runtime dependencies. It targets Python 3.9 and checks the contract, registry, registered implementation and host command/workflow call sites. The wheel includes `py.typed` for consumers of this interface. +The development-only checker is declared in the dev dependency group, not runtime dependencies. It targets Python 3.9 and checks the contract, registry, registered implementations and host command/workflow call sites. The wheel includes `py.typed` for consumers of this interface. Unit tests cover conformance/member/profile checks, loader non-mutation and explicit resume state. Existing CLI tests continue covering identity binding, uncertain-outcome/no-retry behavior, private archives, offline imports, source-denied wheel operation and SSH retirement. These contract checks require no physical device. diff --git a/pyproject.toml b/pyproject.toml index 5d21340..2363a65 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -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.j71b", "orchard_cli.j71t", "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/j71b/adapter.py", "src/orchard_cli/j71b/inputs.py", "src/orchard_cli/j71b/mixin.py", "src/orchard_cli/j71t/adapter.py", "src/orchard_cli/j71t/inputs.py", "src/orchard_cli/j71t/mixin.py", "src/orchard_cli/j71t/calibration.py", "src/orchard_cli/j71t/touch.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.j71b.adapter", "orchard_cli.j71b.inputs", "orchard_cli.j71b.mixin", "orchard_cli.j71t.adapter", "orchard_cli.j71t.inputs", "orchard_cli.j71t.mixin", "orchard_cli.j71t.calibration", "orchard_cli.j71t.touch"] disallow_untyped_defs = true diff --git a/src/orchard_cli/apple/d5500.py b/src/orchard_cli/apple/d5500.py new file mode 100644 index 0000000..276a8b1 --- /dev/null +++ b/src/orchard_cli/apple/d5500.py @@ -0,0 +1,68 @@ +#!/usr/bin/env python3 +"""Extract the VXD (DEVA) video-decoder firmware from a legacy prelinked kernelcache. + +AppleD5500 has no __firmware section. Its __DATA.__data holds a record + + +0x00 char *name -> __TEXT.__cstring ("REL_812.0_14_Deva") + +0x08 char *build -> __TEXT.__cstring (build date) + +0x10 u32 mtx_size (memory the decoder core needs; <= image size) + +0x14 u32 size (image size) + +0x18 u8 *image -> __TEXT.__const + +The record is found by that pointer pattern, not by fixed offsets, and the +image is copied unchanged. The caller pins its SHA-256. +""" +import argparse +import hashlib +import json +from pathlib import Path +import struct +from typing import Any, Dict, Tuple + +from orchard_cli.apple.prelink import Kernelcache, named + +D5500 = 'com.apple.driver.AppleD5500' +DATA, CSTRING, CONST = ('__DATA', '__data'), ('__TEXT', '__cstring'), ('__TEXT', '__const') +RECORD = struct.Struct(' Tuple[bytes, Dict[str, Any]]: + kernel = Kernelcache(blob) + sections = named(kernel.kext_sections(D5500, 'AppleD5500'), DATA, CSTRING, CONST) + + def inside(addr: int, key: Tuple[str, str], size: int = 1) -> bool: + s = sections[key] + return bool(s['addr'] <= addr and addr + size <= s['addr'] + s['size']) + + def cstring(addr: int) -> str: + s = sections[CSTRING] + text = kernel.read(addr, min(64, s['addr'] + s['size'] - addr)).split(b'\0')[0] + value = text.decode('ascii', 'replace') + return value if value.isprintable() and '�' not in value else '' + + data = kernel.read(sections[DATA]['addr'], sections[DATA]['size']) + records = [] + for offset in range(0, len(data) - RECORD.size + 1, 8): + name, build, mtx_size, size, image = RECORD.unpack_from(data, offset) + if (inside(name, CSTRING) and inside(build, CSTRING) and 4096 < size <= 4 * 1024 * 1024 + and mtx_size <= size and inside(image, CONST, size) and cstring(name).endswith('_Deva')): + records.append((cstring(name), cstring(build), image, size)) + if len(records) != 1: + raise ValueError('No unique VXD firmware record') + name, build, image, size = records[0] + firmware = kernel.read(image, size) + report = dict(kext=D5500, firmware=name, build=build, kernel_file_offset=kernel.offset(image, size), + firmware_size=size, sha256=hashlib.sha256(firmware).hexdigest()) + return firmware, report + + +if __name__ == '__main__': + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument('kernel', type=Path) + parser.add_argument('output', type=Path) + args = parser.parse_args() + data, report = extract(args.kernel.read_bytes()) + # Never silently replace a firmware artifact. + with args.output.open('xb') as out: + out.write(data) + print(json.dumps(report, indent=2)) diff --git a/src/orchard_cli/apple/prelink.py b/src/orchard_cli/apple/prelink.py new file mode 100644 index 0000000..7b27eed --- /dev/null +++ b/src/orchard_cli/apple/prelink.py @@ -0,0 +1,57 @@ +"""Locate one kext in a legacy prelinked ARM64 kernelcache. + +No hardcoded offsets: the kext is selected by its prelink metadata, and its +virtual addresses are mapped through the outer kernel's file-backed segments. +For MH_FILESET kernels use the upstream ipsw kext extractor instead. +""" +from typing import Any, Dict, List, Tuple + +from orchard_cli.apple.macho import macho +from orchard_cli.apple.osserialize import apple_xml + +Section = Dict[str, Any] + + +class Kernelcache: + def __init__(self, blob: bytes): + kind, self.segments, self.sections = macho(blob) + if kind != 2: + raise ValueError('Expected legacy MH_EXECUTE kernelcache') + self.blob = blob + + def offset(self, addr: int, size: int) -> int: + offsets = {s['offset'] + addr - s['vm'] for s in self.segments + if s['vm'] <= addr and addr + size <= s['vm'] + s['size']} + if len(offsets) != 1: + raise ValueError('Section has no unique file-backed mapping') + offset = offsets.pop() + if offset + size > len(self.blob): + raise ValueError('Section exceeds file') + return offset + + def read(self, addr: int, size: int) -> bytes: + offset = self.offset(addr, size) + return self.blob[offset:offset + size] + + def kext_sections(self, bundle: str, label: str) -> List[Section]: + """The sections of the one prelinked kext with this bundle identifier.""" + info = [s for s in self.sections if (s['segment'], s['name']) == ('__PRELINK_INFO', '__info')] + if len(info) != 1: + raise ValueError('No unique prelink metadata section') + metadata = apple_xml(self.read(info[0]['addr'], info[0]['size'])) + kexts = [k for k in metadata['_PrelinkInfoDictionary'] if k.get('CFBundleIdentifier') == bundle] + if len(kexts) != 1: + raise ValueError(f'No unique {label} kext') + _, _, sections = macho(self.blob, self.offset(kexts[0]['_PrelinkExecutableLoadAddr'], 32)) + return sections + + +def named(sections: List[Section], *keys: Tuple[str, str]) -> Dict[Tuple[str, str], Section]: + """Exactly one section per (segment, section) key.""" + result = {} + for key in keys: + found = [s for s in sections if (s['segment'], s['name']) == key] + if len(found) != 1: + raise ValueError('No unique section %s.%s' % key) + result[key] = found[0] + return result diff --git a/src/orchard_cli/apple/smartio.py b/src/orchard_cli/apple/smartio.py new file mode 100644 index 0000000..0d4f1d6 --- /dev/null +++ b/src/orchard_cli/apple/smartio.py @@ -0,0 +1,43 @@ +#!/usr/bin/env python3 +"""Extract SmartIO __DATA.__firmware from a legacy prelinked ARM64 kernelcache. + +No hardcoded offsets; select the exact kext via its prelink metadata, then map +its section virtual address through the outer kernel's file-backed segments. +For MH_FILESET kernels use the upstream ipsw kext extractor instead. The kext +bundle identifier is the caller's board policy (T8010 by default; S8000 for A9). +""" +import argparse +import hashlib +import json +from pathlib import Path + +from orchard_cli.apple.prelink import Kernelcache + +T8010 = 'com.apple.driver.AppleT8010SmartIO' +S8000 = 'com.apple.driver.AppleS8000SmartIO' + + +def extract(blob, bundle=T8010): + kernel = Kernelcache(blob) + ksections = kernel.kext_sections(bundle, 'SmartIO') + firmware = [s for s in ksections if (s['segment'], s['name']) == ('__DATA', '__firmware')] + if len(firmware) != 1 or not 4096 <= firmware[0]['size'] <= 4*1024*1024: + raise ValueError('No unique bounded firmware section') + f = firmware[0]; offset = kernel.offset(f['addr'], f['size']) + data = blob[offset:offset+f['size']] + report = dict(kext=bundle, kernel_file_offset=offset, + firmware_size=len(data), sha256=hashlib.sha256(data).hexdigest()) + return data, report + + +if __name__ == '__main__': + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument('kernel', type=Path) + parser.add_argument('output', type=Path) + parser.add_argument('--bundle', choices=(T8010, S8000), default=T8010) + args = parser.parse_args() + data, report = extract(args.kernel.read_bytes(), args.bundle) + # Never silently replace a firmware artifact. + with args.output.open('xb') as out: + out.write(data) + print(json.dumps(report, indent=2)) diff --git a/src/orchard_cli/command.py b/src/orchard_cli/command.py index e63f3ef..975ad8e 100644 --- a/src/orchard_cli/command.py +++ b/src/orchard_cli/command.py @@ -9,7 +9,7 @@ import sys from importlib.metadata import version -from orchard_cli.devices import load, choose +from orchard_cli.devices import SUPPORTED, load, choose, list_devices, profile_for_adt from orchard_cli.progress import Progress from orchard_cli.protocol import Client from orchard_cli.releases import resolve, catalogue, register @@ -36,7 +36,7 @@ def main(): distros.add_argument('source', nargs='?') check = commands.add_parser('resolve', help='verify inputs without accessing a device') check.add_argument('source') - check.add_argument('--profile', default='j71b') + check.add_argument('--profile', default='j71b', choices=SUPPORTED) check.add_argument('--boot', action='store_true', help='resolve only boot assets') commands.add_parser('devices', help='list USB devices without modifying them') commands.add_parser('setup-host', help='install missing boot tools and Linux USB permissions; no device access') @@ -51,6 +51,7 @@ def main(): mixin = commands.add_parser('mixin', help='prepare a data-only private mixin') for name in ('adt', 'firmware-dir', 'output'): mixin.add_argument('--' + name, type=Path, required=True) + mixin.add_argument('--profile', choices=SUPPORTED, help='device profile (default: selected from the ADT model)') for action in ('status', 'diagnostics', 'targets', 'destination', 'prepare', 'install-root', 'handoff', 'abort'): command = commands.add_parser(action, help='invoke the device-owned ' + action + ' operation') command.add_argument('--recovery', action='store_true', help='provision access through the recovery serial shell') @@ -90,7 +91,7 @@ def main(): record, assets = resolve(args.source, args.profile, progress, purpose='boot' if args.boot else 'install') result = {'manifest': record, 'files': {key: str(path) for key, path in assets.items()}} elif args.action == 'devices': - result = load().list_devices() + result = list_devices() elif args.action == 'setup-host': from orchard_cli.host_setup import setup result = setup(progress) @@ -124,7 +125,7 @@ def main(): stream.write(data) result = {'saved': str(args.output), 'bytes': len(data)} elif args.action == 'mixin': - adapter = load() + adapter = load(args.profile or profile_for_adt(args.adt.read_bytes())) archive, result = adapter.prepare_mixin(args.adt, args.firmware_dir, args.output) result = {**result, 'path': str(archive)} else: diff --git a/src/orchard_cli/contracts.py b/src/orchard_cli/contracts.py index 0a32e53..8090b8e 100644 --- a/src/orchard_cli/contracts.py +++ b/src/orchard_cli/contracts.py @@ -59,6 +59,14 @@ class MixinReceipt(TypedDict): sha256: str +class UnsupportedDevice(ValueError): + """discover() identified a connected target that this adapter does not support. + + The registry tries every adapter; this refusal lets another adapter claim the + target instead of reporting an error. Other ValueErrors remain errors. + """ + + @dataclass(frozen=True) class Connection: argv: List[str] @@ -71,7 +79,8 @@ class Adapter(Protocol): """Required interface of a registered adapter module. Discovery is read-only and raises ValueError when no unique supported device - is present. Boot must never retry uncertain mutations. connect(device) binds + is present; UnsupportedDevice when the target is another model. Boot must + never retry uncertain mutations. connect(device) binds the selected ECID/boot; connect_saved(device) must never reprovision. Exceptions propagate: the host owns confirmation/reporting, not recovery recipes. There are no optional members or legacy fallbacks. diff --git a/src/orchard_cli/devices.py b/src/orchard_cli/devices.py index aa8a55f..f3c515e 100644 --- a/src/orchard_cli/devices.py +++ b/src/orchard_cli/devices.py @@ -1,14 +1,15 @@ """Device adapter registry. Shared workflow never interprets hardware identities.""" import importlib -from typing import List, Optional, Tuple +from typing import Any, Dict, List, Tuple -from .contracts import Adapter, SelectedDevice +from orchard_hardware import device as hardware_device +from .apple.adt_tree import parse +from .contracts import Adapter, SelectedDevice, UnsupportedDevice -SUPPORTED = ('j71b',) +SUPPORTED = ('j71b', 'j71t') -def load(profile: Optional[str] = None) -> Adapter: - profile = profile or SUPPORTED[0] +def load(profile: str) -> Adapter: if profile not in SUPPORTED: raise ValueError('Unsupported device profile') adapter = importlib.import_module('orchard_cli.' + profile + '.adapter') @@ -21,15 +22,43 @@ def load(profile: Optional[str] = None) -> Adapter: return adapter +def list_devices() -> List[Dict[str, Any]]: + """Raw observations from every registered adapter, without selecting one.""" + found: List[Dict[str, Any]] = [] + for profile in SUPPORTED: + for observation in load(profile).list_devices(): + if observation not in found: + found.append(observation) + return found + + def choose() -> Tuple[Adapter, SelectedDevice]: matches: List[Tuple[Adapter, SelectedDevice]] = [] errors: List[str] = [] + names: List[str] = [] for profile in SUPPORTED: adapter = load(profile) + names.append(adapter.NAME + ', ' + profile[0].upper() + profile[1:]) try: matches.append((adapter, adapter.discover())) + except UnsupportedDevice: + continue except ValueError as error: - errors.append(str(error)) - if len(matches) != 1: - raise ValueError('; '.join(errors) or 'Connect exactly one supported device') - return matches[0] + # Shared USB checks fail identically for every adapter; report once. + if str(error) not in errors: + errors.append(str(error)) + if len(matches) == 1: + return matches[0] + if matches: + raise ValueError('The connected device matches several adapters; refusing to choose') + raise ValueError('; '.join(errors) or 'Unsupported device; supported: ' + '; '.join(names)) + + +def profile_for_adt(blob: bytes) -> str: + """Select the registered profile whose public Apple product names this ADT's model.""" + model = parse(blob).properties.get('model', b'').split(b'\0', 1)[0].decode('ascii', errors='replace') + profiles = [profile for profile in SUPPORTED + if model in hardware_device(profile)['device']['apple_products']] + if len(profiles) != 1: + raise ValueError('Cannot select a device profile from this ADT; specify --profile') + return profiles[0] diff --git a/src/orchard_cli/j71b/adapter.py b/src/orchard_cli/j71b/adapter.py index 51c7da9..1f9c0ec 100644 --- a/src/orchard_cli/j71b/adapter.py +++ b/src/orchard_cli/j71b/adapter.py @@ -2,7 +2,7 @@ from contextlib import AbstractContextManager from typing import Any, Dict, List, Optional, cast -from orchard_cli.contracts import BootAssets, BootRequest, Connection, SelectedDevice +from orchard_cli.contracts import BootAssets, BootRequest, Connection, SelectedDevice, UnsupportedDevice from orchard_cli.state import BootState, DeviceState, EndpointUnavailable from orchard_cli.progress import Progress from orchard_cli.apple import usb, boot as apple_boot, adt_capture @@ -63,10 +63,14 @@ def discover() -> SelectedDevice: except EndpointUnavailable as error: raise ValueError(str(error)) from error if device['mode'] == 'linux': + if device.get('product') != PRODUCT: + raise UnsupportedDevice('Connected recovery is not for iPad 6 (Wi-Fi), J71b') endpoint = recovery_endpoint() BootState(DeviceState(PROFILE, endpoint['ecid']), endpoint['boot_id']) return cast(SelectedDevice, {**endpoint, 'resumed': True}) - if (device['chip'], device['board']) != (CHIP, BOARD) or not device['ecid']: + if (device['chip'], device['board']) != (CHIP, BOARD): + raise UnsupportedDevice('Unsupported device; only iPad 6 (Wi-Fi), J71b is supported') + if not device['ecid']: raise ValueError('Unsupported device; only iPad 6 (Wi-Fi), J71b is supported') DeviceState(PROFILE, device['ecid']) return cast(SelectedDevice, {**device, 'resumed': False}) diff --git a/src/orchard_cli/j71b/extract_smartio.py b/src/orchard_cli/j71b/extract_smartio.py deleted file mode 100644 index d0ea4bf..0000000 --- a/src/orchard_cli/j71b/extract_smartio.py +++ /dev/null @@ -1,63 +0,0 @@ -#!/usr/bin/env python3 -"""Extract SmartIO __DATA.__firmware from a legacy prelinked ARM64 kernelcache. - -No hardcoded offsets; select the exact kext via its prelink metadata, then map -its section virtual address through the outer kernel's file-backed segments. -For MH_FILESET kernels use the upstream ipsw kext extractor instead. -""" -import argparse -import hashlib -import json -from pathlib import Path - -from orchard_cli.apple.macho import macho -from orchard_cli.apple.osserialize import apple_xml - - -def extract(blob): - kind, segments, sections = macho(blob) - if kind != 2: - raise ValueError('Expected legacy MH_EXECUTE kernelcache') - - def mapped(addr, size): - offsets = {s['offset']+addr-s['vm'] for s in segments - if s['vm'] <= addr and addr+size <= s['vm']+s['size']} - if len(offsets) != 1: - raise ValueError('Section has no unique file-backed mapping') - off = offsets.pop() - if off+size > len(blob): - raise ValueError('Section exceeds file') - return off - - info = [s for s in sections if (s['segment'], s['name']) == ('__PRELINK_INFO', '__info')] - if len(info) != 1: - raise ValueError('No unique prelink metadata section') - s = info[0]; off = mapped(s['addr'], s['size']) - metadata = apple_xml(blob[off:off+s['size']]) - kexts = [k for k in metadata['_PrelinkInfoDictionary'] - if k.get('CFBundleIdentifier') == 'com.apple.driver.AppleT8010SmartIO'] - if len(kexts) != 1: - raise ValueError('No unique SmartIO kext') - addr = kexts[0]['_PrelinkExecutableLoadAddr'] - offset = mapped(addr, 32) - _, _, ksections = macho(blob, offset) - firmware = [s for s in ksections if (s['segment'], s['name']) == ('__DATA', '__firmware')] - if len(firmware) != 1 or not 4096 <= firmware[0]['size'] <= 4*1024*1024: - raise ValueError('No unique bounded firmware section') - f = firmware[0]; offset = mapped(f['addr'], f['size']) - data = blob[offset:offset+f['size']] - report = dict(kext='com.apple.driver.AppleT8010SmartIO', kernel_file_offset=offset, - firmware_size=len(data), sha256=hashlib.sha256(data).hexdigest()) - return data, report - - -if __name__ == '__main__': - parser = argparse.ArgumentParser(description=__doc__) - parser.add_argument('kernel', type=Path) - parser.add_argument('output', type=Path) - args = parser.parse_args() - data, report = extract(args.kernel.read_bytes()) - # Never silently replace a firmware artifact. - with args.output.open('xb') as out: - out.write(data) - print(json.dumps(report, indent=2)) diff --git a/src/orchard_cli/j71b/inputs.py b/src/orchard_cli/j71b/inputs.py index c596d5f..2cf3706 100644 --- a/src/orchard_cli/j71b/inputs.py +++ b/src/orchard_cli/j71b/inputs.py @@ -9,7 +9,7 @@ from orchard_cli.firmware import ensure_archive, missing_inputs, store_input from orchard_cli.integrity import digest from .config import MANIFEST, PROFILE -from .extract_smartio import extract as smartio +from orchard_cli.apple.smartio import extract as smartio def ensure(clock: Clock, progress: Progress, *, allow_download: bool = True) -> Path: diff --git a/src/orchard_cli/j71t/__init__.py b/src/orchard_cli/j71t/__init__.py new file mode 100644 index 0000000..837ab62 --- /dev/null +++ b/src/orchard_cli/j71t/__init__.py @@ -0,0 +1 @@ +"""J71t-specific installer inputs and validation.""" diff --git a/src/orchard_cli/j71t/adapter.py b/src/orchard_cli/j71t/adapter.py new file mode 100644 index 0000000..362f085 --- /dev/null +++ b/src/orchard_cli/j71t/adapter.py @@ -0,0 +1,87 @@ +"""J71t identity/recipe policy binding the shared Apple boot and transport code.""" +from contextlib import AbstractContextManager +from typing import Any, Dict, List, Optional, cast + +from orchard_cli.contracts import BootAssets, BootRequest, Connection, SelectedDevice, UnsupportedDevice +from orchard_cli.state import BootState, DeviceState, EndpointUnavailable +from orchard_cli.progress import Progress +from orchard_cli.apple import usb, boot as apple_boot, adt_capture +from orchard_hardware.apple.adt import validate as validate_adt +from orchard_cli.apple.adt_tree import parse +from orchard_cli.transports import ssh +from .calibration import identify +from .mixin import build as prepare_mixin +from .inputs import ensure as firmware_inputs +from .config import PROFILE, NAME, PRODUCT, CHIP, BOARD, APPLE_PRODUCT, BOARD_CONFIG + +REBOOT_MESSAGE = 'A fresh recovery boot is required.\nPut the iPad in DFU mode, then run: orchard install' +UNSUPPORTED = 'Unsupported device; only iPad 5 (Wi-Fi), J71t is supported' + + +def recovery_endpoint() -> Dict[str, Any]: + return usb.recovery(PRODUCT) + + +def connect(device: SelectedDevice) -> AbstractContextManager[Connection]: + return ssh.session(DeviceState(PROFILE, device['ecid']), recovery_endpoint, + expected_boot=device.get('boot_id')) + + +def connect_saved(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(PROFILE, device['ecid']), recovery_endpoint, + expected_boot=device['boot_id']) + + +def forget_access(scope: BootState) -> None: + if scope.device.profile != PROFILE: + raise ValueError('Credential scope differs from adapter profile') + return ssh.forget_access(scope) + + +def read_adt() -> bytes: + data = adt_capture.capture(recovery_endpoint) + try: + identify(parse(data), APPLE_PRODUCT, BOARD_CONFIG) + except ValueError as error: + raise ValueError('Captured ADT is not for ' + APPLE_PRODUCT + ' (' + BOARD_CONFIG + ')') from error + return data + + +def capture_adt(device: Optional[SelectedDevice] = None) -> bytes: + # DFU/Pongo and Linux location IDs differ: bind the fresh ADT to the ECID. + data = read_adt() + 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() -> List[Dict[str, Any]]: + return usb.devices() + + +def discover() -> SelectedDevice: + try: + device = usb.sole_target() + except EndpointUnavailable as error: + raise ValueError(str(error)) from error + if device['mode'] == 'linux': + if device.get('product') != PRODUCT: + raise UnsupportedDevice('Connected recovery is not for iPad 5 (Wi-Fi), J71t') + endpoint = recovery_endpoint() + BootState(DeviceState(PROFILE, endpoint['ecid']), endpoint['boot_id']) + return cast(SelectedDevice, {**endpoint, 'resumed': True}) + # J71s (S8000) shares the board ID and product name; the chip differs. + if (device['chip'], device['board']) != (CHIP, BOARD): + raise UnsupportedDevice(UNSUPPORTED) + if not device['ecid']: + raise ValueError(UNSUPPORTED) + DeviceState(PROFILE, device['ecid']) + return cast(SelectedDevice, {**device, 'resumed': False}) + + +def boot(device: SelectedDevice, assets: BootAssets, + manifest: BootRequest, progress: Progress) -> None: + return apple_boot.boot(device, assets, manifest, progress, + profile=PROFILE, chip=CHIP, board=BOARD, product=PRODUCT) diff --git a/src/orchard_cli/j71t/calibration.py b/src/orchard_cli/j71t/calibration.py new file mode 100644 index 0000000..94309e9 --- /dev/null +++ b/src/orchard_cli/j71t/calibration.py @@ -0,0 +1,108 @@ +#!/usr/bin/env python3 +# SPDX-License-Identifier: GPL-2.0-or-later +"""Copy J71t calibration and identity records from the device's boot ADT. + +iBoot resolves these syscfg-backed properties before m1n1 starts, so the fresh +recovery ADT carries the unit's own bytes. They are copied unmodified into +private firmware files and are never logged. +""" +import argparse +import hashlib +import json +import os +from pathlib import Path +from typing import Dict, List, Optional, Tuple + +from orchard_cli.apple.adt_tree import Node, parse + +GENERATOR = 'j71t-adt' +# Linux firmware name, ADT node, property, exact size (None: bounded, versioned +# record validated by its kernel consumer). +RECORDS: Tuple[Tuple[str, str, str, Optional[int]], ...] = ( + ('j71t-n1-calibration.bin', '/device-tree/arm-io/spi3/multi-touch', 'multi-touch-calibration', 1024), + ('j71t-LSCI.bin', '/device-tree/arm-io/i2c2/als1', 'alsCalibration', None), + ('j71t-CBAT.bin', '/device-tree/charger', 'usb-input-limit-calibration', 24), + ('j71t-BMac.bin', '/device-tree/arm-io/uart3/bluetooth', 'local-mac-address', 6), +) +UNRESOLVED = (b'syscfg/', b'macaddr/', b'zeroes/') + + +def identify(tree: Node, model: str, board_config: str) -> None: + if tree.properties.get('model', b'').split(b'\0', 1)[0] != model.encode(): + raise ValueError('ADT does not describe ' + model) + compatible = tree.properties.get('compatible', b'').split(b'\0') + if board_config.encode() not in compatible: + raise ValueError('ADT is not for board ' + board_config) + + +def checked(name: str, value: Optional[bytes], size: Optional[int]) -> bytes: + if not isinstance(value, bytes) or not value: + raise ValueError('Missing ADT record: ' + name) + if value.startswith(UNRESOLVED): + raise ValueError('Unresolved ADT record: ' + name) + if (size is not None and len(value) != size) or (size is None and not 8 <= len(value) <= 1024): + raise ValueError('Unexpected ADT record size: ' + name) + if value == b'\0' * len(value) or value == b'\xff' * len(value): + raise ValueError('Blank ADT record: ' + name) + return value + + +def records(blob: bytes, model: str, board_config: str) -> Dict[str, bytes]: + tree = parse(blob) + identify(tree, model, board_config) + paths = {path: [node for walked, node in tree.walk() if walked == path] + for _, path, _, _ in RECORDS} + result = {} + for name, path, prop, size in RECORDS: + if len(paths[path]) != 1: + raise ValueError('No unique ADT node for ' + name) + result[name] = checked(name, paths[path][0].properties.get(prop), size) + address = result['j71t-BMac.bin'] + if address[0] & 1: + raise ValueError('Bluetooth address is not unicast') + chosen = [node for walked, node in tree.walk() if walked == '/device-tree/chosen'] + published = chosen[0].properties.get('mac-address-bluetooth0') if len(chosen) == 1 else None + if published is not None and published != address: + raise ValueError('Bluetooth address records disagree') + return result + + +def prepare(adt_path: Path, outdir: Path, device: str, model: str, board_config: str, + expected: List[str]) -> List[Path]: + """Write each record and a staging receipt; return the receipt paths.""" + if sorted(expected) != sorted(name for name, _, _, _ in RECORDS): + raise ValueError('Firmware recipe and ADT generator records differ') + if outdir.is_symlink() or (outdir.exists() and any(outdir.iterdir())): + raise ValueError('Calibration output must be absent or an empty directory') + blob = adt_path.read_bytes() + values = records(blob, model, board_config) + outdir.mkdir(parents=True, exist_ok=True, mode=0o700) + receipts = [] + for name, data in sorted(values.items()): + fd = os.open(outdir / name, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600) + with os.fdopen(fd, 'wb') as stream: + stream.write(data) + receipt = outdir / (name + '.json') + fd = os.open(receipt, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600) + with os.fdopen(fd, 'w') as stream: + json.dump({'device': device, 'name': name, 'generator': GENERATOR, 'bytes': len(data), + 'sha256': hashlib.sha256(data).hexdigest(), + 'adt_sha256': hashlib.sha256(blob).hexdigest()}, stream, indent=2) + stream.write('\n') + receipts.append(receipt) + return receipts + + +def main() -> None: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument('--adt', type=Path, required=True) + parser.add_argument('--output', type=Path, required=True) + args = parser.parse_args() + receipts = prepare(args.adt, args.output, 'iPad6,11', 'iPad6,11', 'J71tAP', + [name for name, _, _, _ in RECORDS]) + # Sizes only: receipts stay private beside the generated files. + print(json.dumps({'output': str(args.output), 'files': len(receipts)}, indent=2)) + + +if __name__ == '__main__': + main() diff --git a/src/orchard_cli/j71t/config.py b/src/orchard_cli/j71t/config.py new file mode 100644 index 0000000..f1fda4a --- /dev/null +++ b/src/orchard_cli/j71t/config.py @@ -0,0 +1,14 @@ +"""J71t facts from the independently installed hardware data package.""" +from orchard_hardware import device, firmware_path + +DEVICE = device('j71t') +PROFILE = DEVICE['device']['profile'] +NAME = DEVICE['device']['name'] +PRODUCT = DEVICE['match']['linux']['product'] +CHIP = DEVICE['match']['dfu']['chip'] +BOARD = DEVICE['match']['dfu']['board'] +APPLE_PRODUCT = DEVICE['device']['apple_products'][0] +MANIFEST = firmware_path(PROFILE) +# iPad6,11 also ships as J71s (S8000). The ADT model is shared; the board +# configuration in the root compatible list is not. +BOARD_CONFIG = 'J71tAP' diff --git a/src/orchard_cli/j71t/inputs.py b/src/orchard_cli/j71t/inputs.py new file mode 100644 index 0000000..1b1f322 --- /dev/null +++ b/src/orchard_cli/j71t/inputs.py @@ -0,0 +1,73 @@ +"""J71t firmware recipe and board-specific selections; shared code owns acquisition.""" +import json +from pathlib import Path +import tempfile +from typing import Dict, Optional + +from orchard_cli.apple import ipsw +from orchard_cli.apple.d5500 import extract as video_decoder +from orchard_cli.apple.smartio import S8000, extract as smartio +from orchard_cli.cache import cache_root +from orchard_cli.progress import Clock, Progress +from orchard_cli.firmware import ensure_archive, missing_inputs, store_input +from orchard_cli.integrity import digest +from .config import MANIFEST, PROFILE +from .touch import firmware as touch_firmware + +PROFILE_INPUT = 'J71t.mtprops' +BLUETOOTH = 'BCM4355C0-FearA-USI.hcd' +SMARTIO = 's8000-smartio.bin' +TOUCH = 'j71t-n1.bin' +VIDEO_DECODER = 'vxd-deva.bin' +DERIVED = (BLUETOOTH, SMARTIO, TOUCH, VIDEO_DECODER) + + +def verified(raw: Path, name: str, checksum: str) -> bytes: + candidates = [p for p in raw.rglob(name) if p.is_file() and digest(p) == checksum] + if len(candidates) != 1: + raise ValueError('Expected one verified extracted firmware input: ' + name) + return candidates[0].read_bytes() + + +def ensure(clock: Clock, progress: Progress, *, allow_download: bool = True) -> Path: + spec = json.loads(MANIFEST.read_text()) + root = cache_root() / 'firmware' / PROFILE + root.mkdir(parents=True, exist_ok=True, mode=0o700) + root.chmod(0o700) + expected: Dict[str, str] = {row['name']: row['sha256'] for row in spec['files'] + spec['extracted_inputs']} + locations = {name: root / ('extra' if name in DERIVED else 'extracted') / name for name in expected} + if not missing_inputs(locations, expected): + progress.phase('Verified cached firmware') + return root + if not allow_download: + raise ValueError('Verified private firmware cache is missing. Launch does not download an IPSW by default; ' + 'explicitly use --allow-firmware-download to acquire it.') + archive = ensure_archive(spec['ipsw_url'], spec['ipsw_bytes'], spec['ipsw_sha256'], + root / 'archive' / (spec['ipsw_sha256'] + '.ipsw'), + clock, progress, label='Apple IPSW') + executable = ipsw.tool(clock, progress) + progress.phase('Extracting firmware…') + with tempfile.TemporaryDirectory(prefix='extract-', dir=root.parent) as directory: + raw = Path(directory) + ipsw.extract_files(executable, archive, (*spec['extract_patterns'], 'usr/sbin/BlueTool$'), raw) + kernel: Optional[bytes] = None + for name in missing_inputs(locations, expected): + if name == BLUETOOTH: + candidates = list(raw.rglob('BlueTool')) + if len(candidates) != 1: + raise ValueError('Expected one extracted BlueTool') + data = ipsw.hcd_from_binary(candidates[0].read_bytes(), expected[name]) + elif name == SMARTIO: + kernel = kernel or ipsw.extract_kernel(executable, archive, spec['device'], raw) + data, _ = smartio(kernel, S8000) + elif name == VIDEO_DECODER: + kernel = kernel or ipsw.extract_kernel(executable, archive, spec['device'], raw) + data, _ = video_decoder(kernel) + elif name == TOUCH: + data = touch_firmware(verified(raw, PROFILE_INPUT, expected[PROFILE_INPUT])) + else: + data = verified(raw, name, expected[name]) + store_input(locations[name], data, expected[name]) + if missing_inputs(locations, expected): + raise ValueError('Incomplete firmware input cache') + return root diff --git a/src/orchard_cli/j71t/mixin.py b/src/orchard_cli/j71t/mixin.py new file mode 100644 index 0000000..a9f1ee6 --- /dev/null +++ b/src/orchard_cli/j71t/mixin.py @@ -0,0 +1,19 @@ +"""Prepare J71t's private inputs, then use the shared data-only mixin packer.""" +from pathlib import Path +from typing import Tuple + +from orchard_cli.contracts import MixinReceipt +from orchard_cli.firmware import pack_mixin +from .prepare import prepare, verify +from .config import MANIFEST, PROFILE + + +def build(adt: Path, firmware_dir: Path, outdir: Path) -> Tuple[Path, MixinReceipt]: + if outdir.exists(): + raise ValueError('Choose a new private output directory') + outdir.mkdir(parents=True, mode=0o700) + prepared = prepare(adt, firmware_dir, outdir / 'prepared') + receipt = verify(prepared, MANIFEST, adt) + metadata = {'schema': 1, 'profile': PROFILE, 'adt_sha256': receipt['adt_sha256'], + 'files': receipt['files']} + return pack_mixin(prepared, metadata, outdir) diff --git a/src/orchard_cli/j71t/prepare.py b/src/orchard_cli/j71t/prepare.py new file mode 100644 index 0000000..75e0aaf --- /dev/null +++ b/src/orchard_cli/j71t/prepare.py @@ -0,0 +1,87 @@ +"""Prepare verified J71t firmware from explicit inputs into an explicit output.""" +from __future__ import annotations + +import json +import os +from pathlib import Path +import tempfile + +from orchard_cli.staging import stage as stage_firmware +from orchard_hardware.apple.adt import validate as validate_adt +from orchard_cli.integrity import digest +from .calibration import GENERATOR, prepare as prepare_calibration + +from .config import MANIFEST, APPLE_PRODUCT, BOARD_CONFIG, PROFILE + + +def verify(directory: Path, manifest: Path, adt: Path) -> dict: + record = json.loads((directory / 'firmware.json').read_text()) + device_id = f'{validate_adt(adt.read_bytes()):016x}' + if (record.get('schema') != 1 or record.get('profile') != PROFILE or + record.get('device_id') != device_id or record.get('adt_sha256') != digest(adt) or + record.get('manifest_sha256') != digest(manifest)): + raise ValueError('Prepared firmware identity differs') + files = record['files'] + actual = set() + for path in (directory / 'firmware').rglob('*'): + if path.is_symlink(): + raise ValueError('Prepared firmware contains a symlink') + if path.is_file(): + actual.add(path.relative_to(directory / 'firmware').as_posix()) + if actual != set(files): + raise ValueError('Prepared firmware file set differs') + for name, expected in files.items(): + path = directory / 'firmware' / name + if path.stat().st_size != expected['bytes'] or digest(path) != expected['sha256']: + raise ValueError(f'Prepared firmware differs: {name}') + spec = json.loads(manifest.read_text()) + for row in spec['generated']: + receipt = json.loads((directory / 'calibration' / (row['name'] + '.json')).read_text()) + if (receipt.get('adt_sha256') != digest(adt) or + files[row['destination']]['sha256'] != receipt.get('sha256')): + raise ValueError('Calibration differs from its ADT receipt: ' + row['name']) + return record + + +def check_outdir(outdir: Path): + if (outdir.is_symlink() or + (outdir.exists() and (not outdir.is_dir() or any(outdir.iterdir())))): + raise ValueError('Outdir must be absent or an empty directory') + + +def prepare(adt: Path, firmware_dir: Path, outdir: Path) -> Path: + manifest = MANIFEST + spec = json.loads(manifest.read_text()) + if spec.get('device') != APPLE_PRODUCT: + raise ValueError('Wrong J71t firmware manifest') + if any(row.get('generator') != GENERATOR for row in spec['generated']): + raise ValueError('Unsupported J71t firmware generator') + if not firmware_dir.is_dir() or firmware_dir.is_symlink(): + raise ValueError('Expected a regular firmware input directory') + check_outdir(outdir) + firmware_dir = firmware_dir.resolve() + outdir = outdir.resolve() + if outdir == firmware_dir or firmware_dir in outdir.parents: + raise ValueError('Output must not be inside the firmware inputs') + device_id = f'{validate_adt(adt.read_bytes()):016x}' + outdir.parent.mkdir(parents=True, exist_ok=True, mode=0o700) + with tempfile.TemporaryDirectory(prefix='.j71t-', dir=outdir.parent) as temporary: + work = Path(temporary) + calibration = work / 'calibration' + receipts = prepare_calibration(adt, calibration, spec['device'], APPLE_PRODUCT, BOARD_CONFIG, + [row['name'] for row in spec['generated']]) + count = stage_firmware(manifest, [firmware_dir, calibration], receipts, work / 'firmware') + if count != len(spec['files']) + len(spec['generated']): + raise ValueError('Incomplete J71t firmware stage') + records = {} + for path in sorted((work / 'firmware').rglob('*')): + if path.is_file(): + records[path.relative_to(work / 'firmware').as_posix()] = { + 'bytes': path.stat().st_size, 'sha256': digest(path)} + receipt = {'schema': 1, 'profile': PROFILE, 'device_id': device_id, + 'manifest_sha256': digest(manifest), 'adt_sha256': digest(adt), 'files': records} + (work / 'firmware.json').write_text(json.dumps(receipt, indent=2) + '\n') + verify(work, manifest, adt) + check_outdir(outdir) + os.replace(work, outdir) + return outdir diff --git a/src/orchard_cli/j71t/touch.py b/src/orchard_cli/j71t/touch.py new file mode 100644 index 0000000..b235c1b --- /dev/null +++ b/src/orchard_cli/j71t/touch.py @@ -0,0 +1,59 @@ +#!/usr/bin/env python3 +# SPDX-License-Identifier: GPL-2.0-or-later +"""Select J71t's N1 touch boot packet from the public multitouch profile. + +The Linux N1 driver uploads iOS's preconstructed C1F15,1 packet unmodified +(no Z2FW wrapper) and the unit's MtCl calibration separately, so this only +validates the packet framing and returns it verbatim. +""" +import argparse +import hashlib +import json +from pathlib import Path +import plistlib + +PERSONALITY = 'C1F15,1' +PREFIX = b'\x18\xe1\x30\x01' + + +def halfword32(data: bytes) -> int: + return int.from_bytes(bytes((data[1], data[0], data[3], data[2])), 'little') + + +def validate(packet: bytes) -> None: + if len(packet) < 20 or packet[:4] != PREFIX: + raise ValueError('Not a prefixed N1 boot packet') + if int.from_bytes(packet[4:6], 'big') * 4 + 16 != len(packet): + raise ValueError('N1 packet length mismatch') + if halfword32(packet[6:10]): + raise ValueError('Unexpected N1 packet load address') + if sum(packet[4:10]) != int.from_bytes(packet[10:12], 'big'): + raise ValueError('N1 packet header checksum mismatch') + if sum(packet[12:-4]) & 0xffffffff != halfword32(packet[-4:]): + raise ValueError('N1 packet payload checksum mismatch') + + +def firmware(profile: bytes) -> bytes: + profiles = plistlib.loads(profile) + if not isinstance(profiles, dict) or set(profiles) != {PERSONALITY}: + raise ValueError('Unexpected J71t multitouch profile set') + packet = profiles[PERSONALITY].get('Constructed Firmware') + if not isinstance(packet, bytes): + raise ValueError('Expected data-form constructed firmware') + validate(packet) + return packet + + +def main() -> None: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument('profile', type=Path) + parser.add_argument('output', type=Path) + args = parser.parse_args() + packet = firmware(args.profile.read_bytes()) + with args.output.open('xb') as stream: + stream.write(packet) + print(json.dumps({'bytes': len(packet), 'sha256': hashlib.sha256(packet).hexdigest()}, indent=2)) + + +if __name__ == '__main__': + main() diff --git a/tests/test_adapter_contract.py b/tests/test_adapter_contract.py index 85507b3..aa6601a 100644 --- a/tests/test_adapter_contract.py +++ b/tests/test_adapter_contract.py @@ -1,9 +1,10 @@ """Registry/adapter conformance checks, with no USB access or boot actions.""" +import struct from types import SimpleNamespace import unittest from unittest.mock import Mock, patch -from orchard_cli.contracts import Adapter +from orchard_cli.contracts import Adapter, UnsupportedDevice from orchard_cli import devices from orchard_cli.j71b import adapter @@ -53,11 +54,61 @@ def test_dfu_discovery_returns_explicit_resume_state_without_mutating_observatio def test_choose_passes_the_validated_selection_through_unchanged(self): selection = {'mode':'dfu','location':'usb','ecid':1,'resumed':False} implementation = module(discover=Mock(return_value=selection)) - with patch.object(devices, 'load', return_value=implementation): + other = module(discover=Mock(side_effect=UnsupportedDevice('another model'))) + with patch.object(devices, 'SUPPORTED', ('first', 'second')), \ + patch.object(devices, 'load', side_effect=lambda profile: other if profile == 'first' else implementation): loaded, selected = devices.choose() self.assertIs(loaded, implementation) self.assertIs(selected, selection) implementation.discover.assert_called_once_with() + def test_choose_reports_shared_errors_once_and_names_supported_devices(self): + failing = module(discover=Mock(side_effect=ValueError('Connect exactly one target'))) + with patch.object(devices, 'SUPPORTED', ('first', 'second')), \ + patch.object(devices, 'load', return_value=failing): + with self.assertRaisesRegex(ValueError, '^Connect exactly one target$'): + devices.choose() + refusing = module(NAME='Tablet', discover=Mock(side_effect=UnsupportedDevice('another model'))) + with patch.object(devices, 'SUPPORTED', ('first', 'second')), \ + patch.object(devices, 'load', return_value=refusing): + with self.assertRaisesRegex(ValueError, 'supported: Tablet, First; Tablet, Second$'): + devices.choose() + accepting = module(discover=Mock(return_value={'mode':'dfu','location':'usb','ecid':1,'resumed':False})) + with patch.object(devices, 'SUPPORTED', ('first', 'second')), \ + patch.object(devices, 'load', return_value=accepting): + with self.assertRaisesRegex(ValueError, 'several adapters'): + devices.choose() + + def test_j71b_refuses_other_models_without_claiming_an_error(self): + for observed in ({'mode':'dfu', 'location':'usb', 'ecid':1, 'chip':0x8003, 'board':0x10}, + {'mode':'linux', 'location':'usb', 'ecid':1, 'product':'Other Recovery', + 'serial_port':'/dev/other', 'boot_id':'11111111-1111-4111-8111-111111111111'}): + with self.subTest(mode=observed['mode']), patch.object(adapter.usb, 'devices', return_value=[observed]): + with self.assertRaises(UnsupportedDevice): + adapter.discover() + missing = {'mode':'dfu', 'location':'usb', 'ecid':None, 'chip':adapter.CHIP, 'board':adapter.BOARD} + with patch.object(adapter.usb, 'devices', return_value=[missing]): + with self.assertRaises(ValueError) as raised: + adapter.discover() + self.assertNotIsInstance(raised.exception, UnsupportedDevice) + + def test_device_listing_merges_observations_without_selection(self): + observed = [{'mode':'dfu','location':'usb'}] + implementation = module(list_devices=Mock(return_value=observed), + discover=Mock(side_effect=AssertionError('Unexpected discovery'))) + with patch.object(devices, 'SUPPORTED', ('first', 'second')), \ + patch.object(devices, 'load', return_value=implementation): + self.assertEqual(devices.list_devices(), observed) + + def test_profile_selection_from_adt_model(self): + def tree(model): + name = b'name'.ljust(32, b'\0') + struct.pack('CFBundleIdentifier{bundle}' + f'_PrelinkExecutableLoadAddr{hex(base + header)}') + firmware[bundle] = data + info = ('_PrelinkInfoDictionary' + ''.join(records) + '').encode() + info_offset = len(body) + body += info + segment = struct.pack('_PrelinkInfoDictionaryCFBundleIdentifier' + f'{d5500.D5500}_PrelinkExecutableLoadAddr' + f'{hex(base + header)}').encode() + info_offset = len(body) + body += info + segment = struct.pack('H', len(payload) // 4) + encode32(0) + return (touch.PREFIX + header + struct.pack('>H', sum(header)) + payload + encode32(sum(payload))) + + +class Registry(unittest.TestCase): + def test_registered_module_satisfies_the_contract_without_discovery(self): + with patch.object(adapter, 'discover', side_effect=AssertionError('Unexpected USB access')): + self.assertIs(devices.load('j71t'), adapter) + self.assertIsInstance(adapter, Adapter) + record = hardware_device('j71t') + self.assertEqual((adapter.CHIP, adapter.BOARD, adapter.PRODUCT, adapter.APPLE_PRODUCT), + (0x8003, 0x10, 'J71t Recovery', 'iPad6,11')) + self.assertEqual(adapter.NAME, record['device']['name']) + + def choose(self, observed): + with patch.object(usb, 'devices', return_value=observed): + return devices.choose() + + def test_each_model_selects_only_its_adapter(self): + dfu = {'mode': 'dfu', 'location': 'usb', 'ecid': ECID, 'product': None, 'serial_port': None} + for chip, board, expected in ((0x8010, 0x18, j71b), (0x8003, 0x10, adapter)): + with self.subTest(chip=chip): + loaded, selected = self.choose([{**dfu, 'chip': chip, 'board': board}]) + self.assertIs(loaded, expected) + self.assertIs(selected['resumed'], False) + linux = {'mode': 'linux', 'location': 'usb', 'ecid': ECID, 'boot_id': BOOT, + 'serial_port': '/dev/recovery', 'chip': None, 'board': None} + for product, expected in (('J71b Recovery', j71b), ('J71t Recovery', adapter)): + with self.subTest(product=product): + loaded, selected = self.choose([{**linux, 'product': product}]) + self.assertIs(loaded, expected) + self.assertIs(selected['resumed'], True) + + def test_unsupported_and_ambiguous_targets_report_one_error(self): + j71s = {'mode': 'dfu', 'location': 'usb', 'ecid': ECID, 'chip': 0x8000, 'board': 0x10} + with self.assertRaisesRegex(ValueError, '^Unsupported device; supported: ' + 'iPad 6 \\(Wi-Fi\\), J71b; iPad 5 \\(Wi-Fi\\), J71t$'): + self.choose([j71s]) + with self.assertRaisesRegex(ValueError, '^Unsupported device; supported:'): + self.choose([{'mode': 'linux', 'location': 'usb', 'product': 'Other Recovery'}]) + with self.assertRaisesRegex(ValueError, '^Selected USB target is not connected$'): + self.choose([]) + with self.assertRaises(ValueError) as raised: + self.choose([j71s, {**j71s, 'location': 'other'}]) + self.assertTrue(str(raised.exception).startswith('Connect exactly one target')) + self.assertEqual(str(raised.exception).count('Connect exactly one target'), 1) + + def test_profile_selection_from_adt_and_explicit_mixin_profile(self): + self.assertEqual(devices.profile_for_adt(adt()), 'j71t') + + +class Calibration(unittest.TestCase): + def test_records_are_copied_verbatim_and_staged_with_receipts(self): + names = [row['name'] for row in json.loads(prepare.MANIFEST.read_text())['generated']] + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + (root / 'adt').write_bytes(adt(chosen_mac=RECORDS['local-mac-address'])) + receipts = calibration.prepare(root / 'adt', root / 'out', 'iPad6,11', 'iPad6,11', 'J71tAP', names) + self.assertEqual(len(receipts), 4) + expected = {'j71t-n1-calibration.bin': 'multi-touch-calibration', 'j71t-LSCI.bin': 'alsCalibration', + 'j71t-CBAT.bin': 'usb-input-limit-calibration', 'j71t-BMac.bin': 'local-mac-address'} + for name, prop in expected.items(): + self.assertEqual((root / 'out' / name).read_bytes(), RECORDS[prop]) + self.assertEqual((root / 'out' / name).stat().st_mode & 0o777, 0o600) + manifest = root / 'manifest.json' + manifest.write_text(json.dumps({'device': 'iPad6,11', 'files': [], 'generated': [ + {'name': name, 'destination': 'lib/firmware/apple/' + name, 'generator': 'j71t-adt'} + for name in names]})) + self.assertEqual(stage(manifest, [root / 'out'], receipts, root / 'staged'), 4) + with self.assertRaisesRegex(ValueError, 'empty directory'): + calibration.prepare(root / 'adt', root / 'out', 'iPad6,11', 'iPad6,11', 'J71tAP', names) + with self.assertRaisesRegex(ValueError, 'recipe and ADT generator'): + calibration.prepare(root / 'adt', root / 'other', 'iPad6,11', 'iPad6,11', 'J71tAP', names[1:]) + + def test_wrong_board_and_unusable_records_are_refused(self): + cases = [(adt(board=b'J71sAP'), 'not for board'), (adt(model=b'iPad7,5'), 'does not describe'), + (adt(**{'multi-touch-calibration': b'syscfg/MtCl/0x400,zeroes/0x400'}), 'Unresolved'), + (adt(**{'multi-touch-calibration': b'\0' * 1024}), 'Blank'), + (adt(**{'multi-touch-calibration': b'\xff' * 1024}), 'Blank'), + (adt(**{'usb-input-limit-calibration': b'\x01' * 23}), 'size'), + (adt(**{'alsCalibration': b'\x01' * 4}), 'size'), + (adt(**{'local-mac-address': b'\x01\x11\x22\x33\x44\x55'}), 'unicast'), + (adt(chosen_mac=b'\x02\x11\x22\x33\x44\x66'), 'disagree')] + for blob, message in cases: + with self.subTest(message=message), self.assertRaisesRegex(ValueError, message): + calibration.records(blob, 'iPad6,11', 'J71tAP') + + def test_capture_requires_the_j71t_board(self): + with patch.object(adapter.adt_capture, 'capture', return_value=adt(board=b'J71sAP')): + with self.assertRaisesRegex(ValueError, 'Captured ADT is not for iPad6,11 \\(J71tAP\\)'): + adapter.read_adt() + with patch.object(adapter.adt_capture, 'capture', return_value=adt()): + self.assertEqual(adapter.capture_adt({'ecid': ECID}), adt()) + with self.assertRaisesRegex(ValueError, 'identity differs'): + adapter.capture_adt({'ecid': ECID + 1}) + + +class Touch(unittest.TestCase): + def test_boot_packet_is_returned_verbatim_after_framing_checks(self): + data = packet() + profile = plistlib.dumps({'C1F15,1': {'Constructed Firmware': data, 'Constructed Firmware Version': 'x'}}) + self.assertEqual(touch.firmware(profile), data) + broken = [data[:3] + b'\0' + data[4:], data[:-1], data[:6] + encode32(1) + data[10:], + data[:10] + b'\0\0' + data[12:], data[:20] + bytes([data[20] ^ 1]) + data[21:]] + for item in broken: + with self.subTest(item=item[:12]), self.assertRaises(ValueError): + touch.validate(item) + for document in ({'C1F15,1': {}}, {'C1F15,1': {'Constructed Firmware': data}, 'other': {}}): + with self.assertRaises(ValueError): + touch.firmware(plistlib.dumps(document)) + + +class Inputs(unittest.TestCase): + def test_derived_inputs_use_their_board_sources(self): + mtprops = plistlib.dumps({'C1F15,1': {'Constructed Firmware': packet()}}) + derived = {'BCM4355C0-FearA-USI.hcd': b'hcd', 's8000-smartio.bin': b'smartio', 'j71t-n1.bin': packet(), + 'vxd-deva.bin': b'deva', 'feara.trx': b'wifi'} + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + spec = {'device': 'iPad6,11', 'ipsw_url': 'https://example.invalid/ipsw', 'ipsw_bytes': 1, + 'ipsw_sha256': '0' * 64, 'extract_patterns': ['C-4355__s-C0/', 'J71t[.]mtprops$'], + 'extracted_inputs': [{'name': 'J71t.mtprops', 'sha256': hashlib.sha256(mtprops).hexdigest()}], + 'files': [{'name': name, 'sha256': hashlib.sha256(data).hexdigest(), + 'destination': 'lib/firmware/' + name} for name, data in derived.items()]} + (root / 'recipe.json').write_text(json.dumps(spec)) + + def extract(_tool, _archive, patterns, output): + self.assertIn('usr/sbin/BlueTool$', patterns) + (output / 'wifi').mkdir() + (output / 'wifi/feara.trx').write_bytes(b'wifi') + (output / 'J71t.mtprops').write_bytes(mtprops) + (output / 'BlueTool').write_bytes(b'tool') + with patch.object(inputs, 'MANIFEST', root / 'recipe.json'), \ + patch.object(inputs, 'cache_root', return_value=root / 'cache'), \ + patch.object(inputs, 'ensure_archive', return_value=root / 'archive'), \ + patch.object(inputs.ipsw, 'tool', return_value='ipsw'), \ + patch.object(inputs.ipsw, 'extract_files', side_effect=extract), \ + patch.object(inputs.ipsw, 'hcd_from_binary', return_value=b'hcd') as hcd, \ + patch.object(inputs.ipsw, 'extract_kernel', return_value=b'kernel') as kernel, \ + patch.object(inputs, 'smartio', return_value=(b'smartio', {})) as smartio, \ + patch.object(inputs, 'video_decoder', return_value=(b'deva', {})) as video_decoder: + firmware = inputs.ensure(Mock(), Mock()) + hcd.assert_called_once_with(b'tool', hashlib.sha256(b'hcd').hexdigest()) + kernel.assert_called_once() + smartio.assert_called_once_with(b'kernel', S8000) + video_decoder.assert_called_once_with(b'kernel') + self.assertEqual((firmware / 'extra/vxd-deva.bin').read_bytes(), b'deva') + self.assertEqual((firmware / 'extra/j71t-n1.bin').read_bytes(), packet()) + self.assertEqual((firmware / 'extracted/feara.trx').read_bytes(), b'wifi') + self.assertEqual(inputs.ensure(Mock(), Mock(), allow_download=False), firmware) + + def test_mixin_contains_exactly_the_recipe_firmware(self): + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + pinned = {'feara.trx': b'wifi', 'j71t-n1.bin': packet()} + real = json.loads(prepare.MANIFEST.read_text()) + spec = {'device': 'iPad6,11', 'files': [ + {'name': name, 'sha256': hashlib.sha256(data).hexdigest(), 'destination': 'lib/firmware/x/' + name} + for name, data in pinned.items()], 'generated': real['generated']} + (root / 'recipe.json').write_text(json.dumps(spec)) + (root / 'firmware/extracted').mkdir(parents=True) + for name, data in pinned.items(): + (root / 'firmware/extracted' / name).write_bytes(data) + (root / 'adt').write_bytes(adt()) + with patch.object(prepare, 'MANIFEST', root / 'recipe.json'), \ + patch.object(mixin, 'MANIFEST', root / 'recipe.json'): + archive, receipt = adapter.prepare_mixin(root / 'adt', root / 'firmware', root / 'mixin') + self.assertEqual(receipt['sha256'], hashlib.sha256(archive.read_bytes()).hexdigest()) + with tarfile.open(archive) as bundle: + names = sorted(bundle.getnames()) + metadata = json.loads(bundle.extractfile('mixin.json').read()) + destinations = sorted([row['destination'] for row in spec['files'] + spec['generated']]) + self.assertEqual(names, sorted(destinations + ['mixin.json'])) + self.assertEqual(metadata['profile'], 'j71t') + self.assertEqual(metadata['adt_sha256'], hashlib.sha256(adt()).hexdigest()) + + +@unittest.skipUnless(os.environ.get('ORCHARD_TEST_LOCAL_J71T'), + 'set ORCHARD_TEST_LOCAL_J71T to a directory with local 20H350 inputs') +class LocalInputs(unittest.TestCase): + """Optional: reproduce pinned recipe hashes from a developer's own extracted 20H350 files.""" + + def find(self, name): + matches = [p for p in Path(os.environ['ORCHARD_TEST_LOCAL_J71T']).rglob(name) if p.is_file()] + if len(matches) != 1: + self.skipTest('missing local input ' + name) + return matches[0].read_bytes() + + def test_pinned_derived_inputs_reproduce(self): + pinned = {row['name']: row['sha256'] for row in json.loads(prepare.MANIFEST.read_text())['files']} + checks = {'j71t-n1.bin': lambda: touch.firmware(self.find('J71t.mtprops')), + 's8000-smartio.bin': lambda: smartio_extract(self.find('kernelcache*'), S8000)[0], + 'vxd-deva.bin': lambda: video_decoder_extract(self.find('kernelcache*'))[0], + 'BCM4355C0-FearA-USI.hcd': lambda: inputs.ipsw.hcd_from_binary( + self.find('BlueTool'), pinned['BCM4355C0-FearA-USI.hcd'])} + for name, produce in checks.items(): + with self.subTest(name=name): + self.assertEqual(hashlib.sha256(produce()).hexdigest(), pinned[name]) + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/test_package_boundaries.py b/tests/test_package_boundaries.py index b6de31c..482cf6f 100644 --- a/tests/test_package_boundaries.py +++ b/tests/test_package_boundaries.py @@ -23,15 +23,16 @@ class PackageBoundaries(unittest.TestCase): def test_removed_board_local_modules_are_not_shipped(self): package=Path(orchard_cli.__file__).parent self.assertFalse((package/'apple/adt.py').exists()) - for name in ('range_download','ssh','usb','linux_usb','adt','adt_tree','adt_capture'): - self.assertFalse((package/'j71b'/(name+'.py')).exists(), name) + for name in ('range_download','ssh','usb','linux_usb','adt','adt_tree','adt_capture','extract_smartio'): + for board in ('j71b', 'j71t'): + self.assertFalse((package/board/(name+'.py')).exists(), name) - def test_shared_imports_and_offline_commands_do_not_import_j71b(self): + def test_shared_imports_and_offline_commands_do_not_import_boards(self): script = ''' import importlib.abc, os, sys class NoBoard(importlib.abc.MetaPathFinder): def find_spec(self, name, path=None, target=None): - if name == 'orchard_cli.j71b' or name.startswith('orchard_cli.j71b.'): + if name.split('.')[:2] in (['orchard_cli', 'j71b'], ['orchard_cli', 'j71t']): raise AssertionError('Offline/common code imported board: '+name) sys.meta_path.insert(0, NoBoard()) from orchard_cli import downloads, firmware, http, integrity diff --git a/tests/typecheck_adapter.py b/tests/typecheck_adapter.py index 8e0e328..18e64b4 100644 --- a/tests/typecheck_adapter.py +++ b/tests/typecheck_adapter.py @@ -1,5 +1,7 @@ -"""Static conformance check: the registered module must satisfy Adapter.""" +"""Static conformance check: the registered modules must satisfy Adapter.""" from orchard_cli.contracts import Adapter from orchard_cli.j71b import adapter +from orchard_cli.j71t import adapter as j71t_adapter J71B: Adapter = adapter +J71T: Adapter = j71t_adapter