diff --git a/NOTICE b/NOTICE index 047a2bc..1410926 100644 --- a/NOTICE +++ b/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. diff --git a/pyproject.toml b/pyproject.toml index 87f13ae..2506934 100644 --- a/pyproject.toml +++ b/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 diff --git a/src/orchard_cli/apple/device.py b/src/orchard_cli/apple/device.py new file mode 100644 index 0000000..96cb0f8 --- /dev/null +++ b/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) diff --git a/src/orchard_cli/apple/ipsw.py b/src/orchard_cli/apple/ipsw.py index 51c3640..ccbe4a8 100644 --- a/src/orchard_cli/apple/ipsw.py +++ b/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 diff --git a/src/orchard_cli/apple/kernelcache.py b/src/orchard_cli/apple/kernelcache.py new file mode 100644 index 0000000..831cd0f --- /dev/null +++ b/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('HH', value & 0xffff, value >> 16) + + +def from_xe32(data): + lo, hi = struct.unpack('>HH', data) + return lo | hi << 16 + + +def decode_blob(packet): + if len(packet) < 20 or packet[:4] != b'\x18\xe1\x30\x01': + raise ValueError('Not a prefixed HBPP14 blob') + words_minus_one = struct.unpack_from('>H', packet, 4)[0] + size = (words_minus_one + 1) * 4 + if len(packet) != 16 + size: + raise ValueError('HBPP14 length mismatch or extra trailing packets') + if sum(packet[4:10]) != struct.unpack_from('>H', packet, 10)[0]: + raise ValueError('HBPP14 header checksum mismatch') + payload = swapped(packet[12:-4]) + if sum(payload) & 0xffffffff != from_xe32(packet[-4:]): + raise ValueError('HBPP14 payload checksum mismatch') + address = from_xe32(packet[6:10]) + if address + size > 0x100000000: + raise ValueError('HBPP14 address range overflow') + return address, payload + + +def register_packet(address, mask, value): + fields = xe32(uint(address, 'register address')) + xe32(uint(mask, 'register mask')) + fields += xe32(uint(value, 'register value')) + return b'\x1e\x33' + fields + struct.pack('>H', sum(fields)) + + +def command(code, payload): + return struct.pack(' len(blob): + raise ValueError('Truncated Z2FW command') + code, size = struct.unpack_from(' len(blob): + raise ValueError('Unknown or truncated Z2FW command') + if any(blob[pos+size:next_pos]): + raise ValueError('Nonzero Z2FW padding') + records.append((code, blob[pos:pos+size])) + pos = next_pos + return records + + +def calibration_record(name, address, maximum, providers): + """Z2FW command 6: apple_z2 sends the DT property of this provider to address. + + providers collects the records so far and rejects duplicates and overlaps. + """ + if name not in PROVIDERS: + raise ValueError('No apple_z2 calibration provider for ' + repr(name)) + if any(item['provider'] == name for item in providers): + raise ValueError('Duplicate calibration provider: ' + name) + maximum = uint(maximum, 'calibration maximum', 0x40000) + if not maximum: + raise ValueError('Empty calibration range: ' + name) + address = uint(address, 'calibration address') + if not address: + raise ValueError('Calibration address must be nonzero: ' + name) + end = address + ((maximum+3)&~3) + if end > 0x100000000 or any(address < item['address'] + ((item['maximum']+3)&~3) and item['address'] < end + for item in providers): + raise ValueError('Overlapping or overflowing calibration ranges') + providers.append({'provider': name, 'maximum': maximum, 'address': address}) + return CALIBRATION, struct.pack(' 0x40000 or address + len(padded) > 0x100000000: + raise ValueError('HBPP payload too large') + header = struct.pack('>H', len(padded) // 4 - 1) + xe32(address) + return (b'\x18\xe1\x30\x01' + header + struct.pack('>H', sum(header)) + swapped(padded) + + xe32(sum(data) & 0xffffffff)) + + +def raw_xfer(tx, rx_size, flags=RX_BIT_REVERSE): + """Z2FW command 5 payload: tx length, rx length, flags, then the tx bytes.""" + if len(tx) > 16 or rx_size > 16 or not (tx or rx_size): + raise ValueError('Invalid raw transfer') + return struct.pack('H', sum(fields)) + return [(RAW_XFER, raw_xfer(request, 0)), (RAW_XFER, raw_xfer(HBPP_INIT + b'\x18\xe1' * 2, 8))] + + +def otp_preflight_records(address, index, size): + """makez2fw's OTP read sequence before the first load (mtfw_item_add_otp_preflight).""" + if not address or not size or size & 3 or size > 256: + raise ValueError('Invalid OTP address or selection') + + def write(offset, value): + return SEND, register_packet(address + offset, 0xffffffff, value) + + records = [write(0x00, 2), write(0x04, 0), write(0x08, 0)] + for word in range(size >> 2): + records += [write(0x10, 0xffffffff), write(0x0c, index + word), + write(0x04, 0x00e80000), write(0x04, 0x00e80001)] + records += register_read(address + 0x20) + [write(0x20, 1)] + register_read(address + 0x18) + return records + + +def hbpp_config(config, *, overrides): + """makez2fw's configuration words, valid mask and boot wait. + + Only present fields set their valid bit. overrides reads the MTFW's + "Device Tree Overrides" (makez2fw does so only for MTFW item lists). + Returns (valid, words or None, wait_ms, (otp address, index, size) or None). + """ + if not isinstance(config, dict): + raise ValueError('Unexpected firmware config shape') + valid, values = 0, [0] * 11 + + def section(key): + value = config.get(key, {}) + if not isinstance(value, dict): + raise ValueError('Unexpected config section: ' + key) + return value + + z2, spi, timeout = section('Z2 Config'), section('SPI Config'), section('Boot Timeout') + fields = [(config, 'Min DMA Transfer Size', 0, 1), (z2, 'Inter Packet Delay', 1, 2), + (spi, 'CS Delay', 2, 3), (spi, 'Clock Phase', 3, 4), (spi, 'Clock Polarity', 4, 5), + (spi, 'Word Delay', 5, 6), (spi, 'Clock Period Ms', 6, 7), (timeout, 'Normal Boot Ms', 8, 9)] + for container, key, bit, index in fields: + if key in container: + values[index] = uint(container[key], key) + valid |= 1 << bit + if 'Power Off On Reset' in config: + if not isinstance(config['Power Off On Reset'], bool): + raise ValueError('Power Off On Reset must be a boolean') + values[8] = int(config['Power Off On Reset']) + valid |= 1 << 7 + otp = None + if overrides: + dto = section('Device Tree Overrides') + + def data(key, minimum): + value = dto.get(key) + return value if isinstance(value, bytes) and len(value) >= minimum else None + + otp_address, chip_id, hbpp, otp_sn = (data('otp-address', 4), data('chip-id-address', 4), + data('hbpp-version', 4), data('otp-sn', 8)) + for value, bit in ((otp_address, 9), (chip_id, 10), (hbpp, 11), (otp_sn, 12), + (data('reset-sequence', 1), 13), (data('power-sequence', 1), 14)): + if value is not None: + valid |= 1 << bit + if hbpp is not None: + values[10] = struct.unpack_from(' bool: ... def list_devices(self) -> List[Dict[str, Any]]: ... @@ -97,5 +106,4 @@ def capture_adt(self, device: Optional[SelectedDevice] = None) -> bytes: ... def firmware_inputs(self, clock: Clock, progress: Progress, *, allow_download: bool = True) -> Path: ... - def prepare_mixin(self, adt: Path, firmware_dir: Path, - outdir: Path) -> Tuple[Path, MixinReceipt]: ... + def prepare_mixin(self, firmware_dir: Path, outdir: Path) -> Tuple[Path, MixinReceipt]: ... diff --git a/src/orchard_cli/devices.py b/src/orchard_cli/devices.py index aa8a55f..d40ab3d 100644 --- a/src/orchard_cli/devices.py +++ b/src/orchard_cli/devices.py @@ -1,35 +1,43 @@ -"""Device adapter registry. Shared workflow never interprets hardware identities.""" -import importlib +"""Device registry from the installed hardware data. Shared workflow never interprets hardware identities.""" +from pathlib import Path from typing import List, Optional, Tuple from .contracts import Adapter, SelectedDevice -SUPPORTED = ('j71b',) + +def profiles() -> List[str]: + import orchard_hardware + return orchard_hardware.profiles() def load(profile: Optional[str] = None) -> Adapter: - profile = profile or SUPPORTED[0] - if profile not in SUPPORTED: - raise ValueError('Unsupported device profile') - adapter = importlib.import_module('orchard_cli.' + profile + '.adapter') + """The adapter for a profile, or for the project's configured device.""" + import orchard_hardware + from .apple.device import Device + if profile is None: + from .build import configured_device + profile = configured_device(Path.cwd()) + if profile is None: + raise ValueError('No device selected: add device = "" to orchard.toml') + if profile not in profiles(): + raise ValueError('Hardware support is not installed for ' + profile) + adapter = Device(orchard_hardware.device(profile)) if not isinstance(adapter, Adapter): raise ValueError('Incomplete adapter contract for ' + profile) - if adapter.PROFILE != profile: - raise ValueError('Adapter profile does not match registered profile: ' + profile) - if not isinstance(adapter.NAME, str) or not adapter.NAME or not isinstance(adapter.REBOOT_MESSAGE, str): - raise ValueError('Invalid adapter presentation metadata for ' + profile) return adapter def choose() -> Tuple[Adapter, SelectedDevice]: - matches: List[Tuple[Adapter, SelectedDevice]] = [] - errors: List[str] = [] - for profile in SUPPORTED: - adapter = load(profile) - try: - matches.append((adapter, adapter.discover())) - except ValueError as error: - errors.append(str(error)) + """The one connected device, matched against every described device.""" + from .apple import usb + device = usb.sole_target() + adapters = [load(profile) for profile in profiles()] + if device['mode'] == 'linux': + matches = [a for a in adapters if a.PRODUCT and device.get('product') == a.PRODUCT] + else: + matches = [a for a in adapters if a.matches_dfu(device)] if len(matches) != 1: - raise ValueError('; '.join(errors) or 'Connect exactly one supported device') - return matches[0] + raise ValueError('The connected device is not one Orchard supports ' + f"(chip {device.get('chip')}, board {device.get('board')})" + if not matches else 'The connected device matches several device descriptions') + return matches[0], matches[0].discover() diff --git a/src/orchard_cli/firmware.py b/src/orchard_cli/firmware.py index e411e40..cd50ded 100644 --- a/src/orchard_cli/firmware.py +++ b/src/orchard_cli/firmware.py @@ -2,11 +2,17 @@ import hashlib import io import json +from pathlib import Path, PurePosixPath +import re import shutil import tarfile +import tempfile +import zipfile +from .cache import cache_root from .downloads import download from .integrity import digest +from . import recipe as recipes def missing_inputs(locations, expected): @@ -70,3 +76,84 @@ def pack_mixin(prepared, metadata, outdir): (outdir / 'mixin.json').write_text(json.dumps(record, indent=2) + '\n') (outdir / 'mixin.json').chmod(0o600) return archive, record + + +def recipe_for(profile): + """The installed hardware's firmware recipe for a device, validated and supported.""" + from orchard_hardware import firmware_path + return recipes.check(json.loads(firmware_path(profile).read_text())) + + +def ensure(profile, clock, progress, *, allow_download=True): + """Every firmware output of the device's recipe, verified and cached by hash. + + Only missing outputs are produced, from only the IPSW inputs they need. + Returns the directory holding the outputs, named by their sha256. + """ + recipe = recipe_for(profile) + root = cache_root() / 'firmware' / profile + outputs = root / 'outputs' + outputs.mkdir(parents=True, exist_ok=True, mode=0o700) + root.chmod(0o700) + expected = {o['sha256']: o['sha256'] for o in recipe['outputs']} + absent = set(missing_inputs({sha: outputs / sha for sha in expected}, expected)) + missing = [o for o in recipe['outputs'] if o['sha256'] in absent] + if not missing: + progress.phase('Verified cached firmware') + return outputs + 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.') + source = recipe['ipsw'] + archive = ensure_archive(source['url'], source['bytes'], source['sha256'], + root / 'archive' / (source['sha256'] + '.ipsw'), + clock, progress, label='Apple IPSW') + from .apple import ipsw + progress.phase('Extracting firmware…') + with tempfile.TemporaryDirectory(prefix='extract-', dir=root) as directory: + executable = [] + + def read_input(name): + entry = recipe['inputs'][name] + location = entry['ipsw'] + if 'zip' in location: + with zipfile.ZipFile(archive) as bundle: + return bundle.read(location['zip']) + if not executable: + executable.append(ipsw.tool(clock, progress)) + target = Path(directory) / name + target.mkdir() + if 'kernelcache' in location: + return ipsw.extract_kernel(executable[0], archive, recipe['device'], target) + ipsw.extract_files(executable[0], archive, (re.escape(location['rootfs']) + '$',), target) + leaf = PurePosixPath(location['rootfs']).name + matches = [p for p in target.rglob(leaf) if p.is_file() and not p.is_symlink() and + (not entry.get('sha256') or digest(p) == entry['sha256'])] + if len(matches) != 1: + raise ValueError('Expected one extracted firmware input: ' + name) + return matches[0].read_bytes() + + for destination, data in recipes.run(recipe, read_input, missing).items(): + checksum = next(o['sha256'] for o in missing if o['destination'] == destination) + store_input(outputs / checksum, data, checksum) + return outputs + + +def mixin(profile, firmware_dir, outdir): + """Stage the recipe's outputs under their destinations and pack the data-only mixin.""" + recipe = recipe_for(profile) + if outdir.exists(): + raise ValueError('Choose a new private output directory') + outdir.mkdir(parents=True, mode=0o700) + prepared = outdir / 'prepared' + files = {} + for output in recipe['outputs']: + source = Path(firmware_dir) / output['sha256'] + if source.is_symlink() or not source.is_file() or digest(source) != output['sha256']: + raise ValueError('Cached firmware output is missing or differs: ' + output['destination']) + target = prepared / 'firmware' / output['destination'] + target.parent.mkdir(parents=True, exist_ok=True, mode=0o700) + shutil.copyfile(source, target) + target.chmod(0o600) + files[output['destination']] = {'bytes': target.stat().st_size, 'sha256': output['sha256']} + return pack_mixin(prepared, {'schema': 1, 'profile': profile, 'files': files}, outdir) diff --git a/src/orchard_cli/j71b/__init__.py b/src/orchard_cli/j71b/__init__.py deleted file mode 100644 index 7c7d809..0000000 --- a/src/orchard_cli/j71b/__init__.py +++ /dev/null @@ -1 +0,0 @@ -"""J71b-specific installer inputs and validation.""" diff --git a/src/orchard_cli/j71b/adapter.py b/src/orchard_cli/j71b/adapter.py deleted file mode 100644 index 51c7da9..0000000 --- a/src/orchard_cli/j71b/adapter.py +++ /dev/null @@ -1,78 +0,0 @@ -"""J71b 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 -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 .mixin import build as prepare_mixin -from .inputs import ensure as firmware_inputs -from .config import PROFILE, NAME, PRODUCT, CHIP, BOARD, APPLE_PRODUCT - -REBOOT_MESSAGE = 'A fresh recovery boot is required.\nPut the iPad in DFU mode, then run: orchard install' - - -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) - if parse(data).properties.get('model', b'').rstrip(b'\0') != APPLE_PRODUCT.encode(): - raise ValueError('Captured ADT is not for ' + APPLE_PRODUCT) - 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': - 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']: - raise ValueError('Unsupported device; only iPad 6 (Wi-Fi), J71b is supported') - 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/j71b/config.py b/src/orchard_cli/j71b/config.py deleted file mode 100644 index 59f4470..0000000 --- a/src/orchard_cli/j71b/config.py +++ /dev/null @@ -1,11 +0,0 @@ -"""J71b facts from the independently installed hardware data package.""" -from orchard_hardware import device, firmware_path - -DEVICE = device('j71b') -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) diff --git a/src/orchard_cli/j71b/inputs.py b/src/orchard_cli/j71b/inputs.py deleted file mode 100644 index c596d5f..0000000 --- a/src/orchard_cli/j71b/inputs.py +++ /dev/null @@ -1,53 +0,0 @@ -"""J71b firmware recipe and board-specific selections; shared code owns acquisition.""" -import json -from pathlib import Path -import tempfile - -from orchard_cli.apple import ipsw -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 .extract_smartio import extract as smartio - - -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 = {row['name']: row['sha256'] for row in spec['files'] + spec['extracted_inputs']} - locations = {name: root / ('extra' if name in ('t8010-smartio.bin', 'BCM4355C1-FigaroA-USI.hcd') 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) - for name in missing_inputs(locations, expected): - if name == 'BCM4355C1-FigaroA-USI.hcd': - 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 == 't8010-smartio.bin': - data, _ = smartio(ipsw.extract_kernel(executable, archive, spec['device'], raw)) - else: - candidates = [p for p in raw.rglob(name) if p.is_file() and digest(p) == expected[name]] - if len(candidates) != 1: - raise ValueError('Expected one verified extracted firmware input: ' + name) - data = candidates[0].read_bytes() - 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/j71b/mixin.py b/src/orchard_cli/j71b/mixin.py deleted file mode 100644 index f4869bd..0000000 --- a/src/orchard_cli/j71b/mixin.py +++ /dev/null @@ -1,19 +0,0 @@ -"""Prepare J71b'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/j71b/prepare.py b/src/orchard_cli/j71b/prepare.py deleted file mode 100644 index e123c4a..0000000 --- a/src/orchard_cli/j71b/prepare.py +++ /dev/null @@ -1,90 +0,0 @@ -"""Prepare verified J71b 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 .touch import prepare as prepare_touch - -from .config import MANIFEST, APPLE_PRODUCT - - -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') != 'j71b' 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}') - touch = json.loads((directory / 'touch/touch.json').read_text()) - if (touch.get('sha256') != files['lib/firmware/apple/j71b-fwstart.z2fw']['sha256'] or - digest(directory / 'touch/j71b-fwstart.z2fw') != touch['sha256']): - raise ValueError('Touch firmware differs from its receipt') - 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 J71b firmware manifest') - 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}' - source = next(item for item in spec['extracted_inputs'] if item['name'] == 'J71b.mtprops') - matches = [path for path in firmware_dir.rglob(source['name']) - if path.is_file() and not path.is_symlink() and digest(path) == source['sha256']] - if len(matches) != 1: - raise ValueError('Expected one verified J71b touch profile in firmware directory') - outdir.parent.mkdir(parents=True, exist_ok=True, mode=0o700) - with tempfile.TemporaryDirectory(prefix='.j71b-', dir=outdir.parent) as temporary: - work = Path(temporary) - touch_dir = work / 'touch' - touch_dir.mkdir() - prepare_touch(matches[0], adt, touch_dir / 'j71b-fwstart.z2fw', touch_dir / 'touch.json') - count = stage_firmware(manifest, [firmware_dir, touch_dir], - [touch_dir / 'touch.json'], work / 'firmware') - expected_count = len(spec['files']) + len(spec['generated']) - if count != expected_count: - raise ValueError('Incomplete J71b 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': 'j71b', '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/j71b/touch.py b/src/orchard_cli/j71b/touch.py deleted file mode 100644 index 5305984..0000000 --- a/src/orchard_cli/j71b/touch.py +++ /dev/null @@ -1,234 +0,0 @@ -#!/usr/bin/env python3 -# SPDX-License-Identifier: GPL-2.0-or-later -"""Convert local J71b touch firmware and ADT calibration to Z2FW. - -Packet layouts/Z2FW framing informed by Corellium Project Sandcastle and -Pauli1Go's GPL makez2fw (944541e67b4c603df328c77d30093551de802c17). -""" -import argparse -import hashlib -import json -import os -from pathlib import Path -import plistlib -import struct - -from orchard_cli.apple.adt_tree import parse, u32s -from orchard_cli.apple.osserialize import apple_xml - -MAGIC = 0x5746325a -INIT, SEND, WAIT, CONFIG = 0, 1, 3, 4 -PROVIDERS = {'multi-touch-calibration', 'multitouch-to-display-offset', - 'firefly-calibration', 'pressure-effect-correction'} - - -def uint(value, name, maximum=0xffffffff): - if isinstance(value, bool): - raise ValueError('Boolean instead of integer: ' + name) - if isinstance(value, str): - if not value.isdecimal(): - raise ValueError('Nondecimal integer: ' + name) - value = int(value) - if not isinstance(value, int) or not 0 <= value <= maximum: - raise ValueError('Integer out of bounds: ' + name) - return value - - -def swapped(data): - if len(data) % 2: - raise ValueError('Word-swap requires even length') - result = bytearray(data) - result[0::2], result[1::2] = data[1::2], data[0::2] - return bytes(result) - - -def xe32(value): - return struct.pack('>HH', value & 0xffff, value >> 16) - - -def from_xe32(data): - lo, hi = struct.unpack('>HH', data) - return lo | hi << 16 - - -def decode_blob(packet): - if len(packet) < 20 or packet[:4] != b'\x18\xe1\x30\x01': - raise ValueError('Not a prefixed HBPP14 blob') - words_minus_one = struct.unpack_from('>H', packet, 4)[0] - size = (words_minus_one + 1) * 4 - if len(packet) != 16 + size: - raise ValueError('HBPP14 length mismatch or extra trailing packets') - if sum(packet[4:10]) != struct.unpack_from('>H', packet, 10)[0]: - raise ValueError('HBPP14 header checksum mismatch') - payload = swapped(packet[12:-4]) - if sum(payload) & 0xffffffff != from_xe32(packet[-4:]): - raise ValueError('HBPP14 payload checksum mismatch') - address = from_xe32(packet[6:10]) - if address + size > 0x100000000: - raise ValueError('HBPP14 address range overflow') - return address, payload - - -def calibration_packet(address, data): - address = uint(address, 'calibration address') - if not data or len(data) > 0x40000 or address + ((len(data)+3)&~3) > 0x100000000: - raise ValueError('Calibration size/address out of bounds') - padded = data + b'\0' * (-len(data) % 4) - fields = struct.pack('>H', len(padded)//4-1) + xe32(address) - packet = (b'\x18\xe1\x30\x01' + fields + struct.pack('>H', sum(fields)) - + swapped(padded) + xe32(sum(data) & 0xffffffff)) - parsed_address, parsed_data = decode_blob(packet) - if parsed_address != address or parsed_data != padded: - raise ValueError('Calibration encode/decode roundtrip mismatch') - return packet - - -def register_packet(address, mask, value): - fields = xe32(uint(address, 'register address')) + xe32(uint(mask, 'register mask')) - fields += xe32(uint(value, 'register value')) - return b'\x1e\x33' + fields + struct.pack('>H', sum(fields)) - - -def command(code, payload): - return struct.pack(' len(blob): - raise ValueError('Truncated Z2FW command') - code, size = struct.unpack_from(' len(blob): - raise ValueError('Unknown or truncated Z2FW command') - if any(blob[pos+size:next_pos]): - raise ValueError('Nonzero Z2FW padding') - records.append((code, blob[pos:pos+size])) - pos = next_pos - return records - - -def convert(profile, config, properties): - if profile.get('PreconstructedBootloadPacketType') != 'K1HBPP14' or config.get('Protocol') != 'Z2': - raise ValueError('Unexpected packet type or protocol') - if properties.get('hid-merge-personality', b'').rstrip(b'\0') != b'K1F19,4': - raise ValueError('Live touch profile is not K1F19,4') - if u32s(properties.get('hbpp-version', b'')) != [0x104]: - raise ValueError('Live HBPP version is not 0x104') - payload = profile.get('Constructed Firmware') - if not isinstance(payload, bytes): - raise ValueError('Expected data-form constructed firmware') - firmware_address, firmware_data = decode_blob(payload) - if firmware_address != 0x100000: - raise ValueError('Unexpected J71b firmware load address') - calibration = config.get('Calibration Sequence') - if not isinstance(calibration, list) or len(calibration) != 4: - raise ValueError('Expected exactly four calibration entries') - names = [item.get('Provider') for item in calibration] - if set(names) != PROVIDERS or len(set(names)) != len(names): - raise ValueError('Missing, duplicate, or unsupported calibration provider') - if not isinstance(config.get('Power Off On Reset'), bool): - raise ValueError('Power Off On Reset must be a boolean') - spi = config['SPI Config'] - timeout = config['Boot Timeout'] - # Preserve source scalar values. CS Delay=10000 and Clock Period Ms=41 - # are raw metadata, not independently established Linux units. - words = [(1<<9)-1 | (1<<11), uint(config['Min DMA Transfer Size'], 'DMA minimum'), - uint(config['Z2 Config']['Inter Packet Delay'], 'Z2 delay'), - uint(spi['CS Delay'], 'raw CS delay'), uint(spi['Clock Phase'], 'CPHA', 1), - uint(spi['Clock Polarity'], 'CPOL', 1), uint(spi['Word Delay'], 'word delay'), - uint(spi['Clock Period Ms'], 'raw clock period'), - int(config['Power Off On Reset'] is True), - uint(timeout['Normal Boot Ms'], 'normal timeout', 30000), 0x104] - first_boot = uint(timeout['First Boot Ms'], 'first timeout', 30000) - if not first_boot: - raise ValueError('First-boot timeout must be nonzero') - records = [(CONFIG, struct.pack('<11I', *words)), - (INIT, b'\x1a\xa1\x18\xe1'), (SEND, payload)] - providers = [] - addresses = [] - for item in calibration: - name = item['Provider'] - data = properties.get(name) - maximum = uint(item['Max Size'], 'calibration maximum', 0x40000) - if not isinstance(data, bytes) or not data or len(data) > maximum or data.startswith(b'syscfg/'): - raise ValueError('Missing, oversized, or unresolved calibration: ' + name) - address = uint(item['Address'], 'calibration address') - end = address + ((len(data)+3)&~3) - if end > 0x100000000 or any(address < b and a < end for a, b in addresses): - raise ValueError('Overlapping or overflowing calibration ranges') - addresses.append((address, end)) - records.append((SEND, calibration_packet(address, data))) - providers.append({'provider': name, 'bytes': len(data), 'maximum': maximum, - 'address': address}) - boot = config.get('Boot Sequence') - if not isinstance(boot, list) or len(boot) != 4: - raise ValueError('Unexpected boot sequence shape') - for item in boot: - if not isinstance(item, dict) or not {'Address', 'Value'} <= item.keys() or set(item)-{'Address','Value','Mask'}: - raise ValueError('Unsupported boot operation') - records.append((SEND, register_packet(item['Address'], item.get('Mask', 0xffffffff), item['Value']))) - records.append((WAIT, struct.pack('); a device +names its platform and DTB in devices//device.toml. """ import json from pathlib import Path @@ -19,16 +21,34 @@ def pin(hardware): if '=' in line) +def _series(hardware, *args): + return subprocess.check_output([sys.executable, str(Path(hardware) / 'kernel/series.py'), *args], + text=True) + + +def layout(hardware, device): + """(platform, dtb) for a device, from hardware's kernel series.""" + descriptor = Path(hardware) / 'devices' / device / 'device.toml' + if not descriptor.is_file(): + raise ValueError(f'Missing hardware recipe input: {descriptor}') + platform, dtb = _series(hardware, '--resolve', device).split() + if dtb == '-': + raise ValueError(f'{device} is a kernel platform, not a device') + return platform, dtb + + def work_directory(hardware, device): # Preserve existing editable source and compiler output; never move/reset it. # These are ignored scratch files, not changes to the hardware repository. - return Path(hardware) / 'kernel/.work' / device + return Path(hardware) / 'kernel/.work' / layout(hardware, device)[0] def patch_identity(hardware, device): - patches = sorted((Path(hardware) / 'kernel/patches' / device).glob('*.patch')) + """Ordered series patches and the marker kernel/build writes for them.""" + kernel = Path(hardware).resolve() / 'kernel' + patches = [Path(line) for line in _series(hardware, device).splitlines() if line] marker = b''.join(subprocess.check_output(['git', 'hash-object', str(p)]).strip() + - b' ' + p.name.encode() + b'\n' for p in patches) + b' ' + str(p.resolve().relative_to(kernel)).encode() + b'\n' for p in patches) return patches, marker @@ -136,6 +156,7 @@ def build_base(hardware, cache, device, output): output.mkdir(parents=True) for name in ('Image.gz', 'kernel.release'): shutil.copy2(work / 'full' / name, output / name) + shutil.copytree(work / 'full/dtbs', output / 'dtbs') shutil.copytree(work / 'in-tree/lib', output / 'lib') export_sdk(compiler_output, output / 'sdk') (output / 'inputs.json').write_text(json.dumps({'source_tree': tree}) + '\n') @@ -144,14 +165,15 @@ def build_base(hardware, cache, device, output): def build_device(hardware, cache, device, base, output): - """Build external modules and final DTB with a PRIVATE cached-SDK copy.""" + """Select the device DTB; build any external modules with a PRIVATE cached-SDK copy.""" from .build import command hardware, base, output = map(Path, (hardware, base, output)) + platform, dtb = layout(hardware, device) source, _ = prepare_source(hardware, cache, device) tree = sources.working_tree(source) if json.loads((base / 'inputs.json').read_text())['source_tree'] != tree: raise ValueError('Cached kernel source identity differs; rebuild the base') - if (base / 'sdk/.config').read_bytes() != (hardware / 'kernel/configs' / device / 'config').read_bytes(): + if (base / 'sdk/.config').read_bytes() != (hardware / 'kernel/configs' / platform / 'config').read_bytes(): raise ValueError('Cached kernel configuration differs; rebuild the base') release = (base / 'kernel.release').read_text().strip() if not release or '/' in release or len(release.split()) != 1: @@ -178,16 +200,14 @@ def build_device(hardware, cache, device, base, output): if (hardware / 'kernel/modules' / device).is_dir(): command(hardware / 'kernel/build-modules', '--source', source, '--build', sdk, '--modules', hardware / 'kernel/modules' / device, - '--localversion', '-orchard-' + device, '--output', work / 'external') + '--localversion', '-orchard-' + platform, '--output', work / 'external') external = ['--external', work / 'external'] command(sys.executable, hardware / 'kernel/stage-modules.py', '--build', sdk, *external, '--output', work / 'staged') - command(hardware / 'kernel/devicetree' / device / 'build', '--kernel-build', sdk, - '--headers', source / 'include', '--output', work / (device + '.dtb')) output.mkdir(parents=True) for name in ('Image.gz', 'kernel.release'): shutil.copy2(base / name, output / name) - shutil.copy2(work / (device + '.dtb'), output / (device + '.dtb')) + shutil.copy2(base / 'dtbs' / (dtb + '.dtb'), output / (device + '.dtb')) shutil.copytree(work / 'staged/lib', output / 'lib') if sources.working_tree(source) != tree: raise ValueError('Kernel source changed during module build; refusing result') diff --git a/src/orchard_cli/launch.py b/src/orchard_cli/launch.py index e21d672..d27051f 100644 --- a/src/orchard_cli/launch.py +++ b/src/orchard_cli/launch.py @@ -17,7 +17,7 @@ def launch(manifest=None, *, directory=None, target=None, allow_firmware_downloa adapter, device = choose() if ask(clock, f'Launch an installed system on {adapter.NAME}? [y/N] ', input_fn).lower() not in ('y','yes'): return {'cancelled':True} - manifest = manifest if manifest is not None else launch_manifest(adapter.PROFILE) + manifest = manifest if manifest is not None else launch_manifest(adapter.PROFILE, device.get('ecid')) record, assets = resolve(manifest, adapter.PROFILE, progress, clock=clock, purpose='boot') if not device['resumed']: adapter.boot(device, assets, record, progress) @@ -32,10 +32,7 @@ def launch(manifest=None, *, directory=None, target=None, allow_firmware_downloa firmware = adapter.firmware_inputs(clock, progress, allow_download=allow_firmware_download) with tempfile.TemporaryDirectory(prefix='launch-',dir=private) as temporary: work = Path(temporary) - adt = work / 'device.adt' - adt.write_bytes(adapter.capture_adt(device)) - adt.chmod(0o600) - mixin, receipt = adapter.prepare_mixin(adt,firmware,work/'mixin') + mixin, receipt = adapter.prepare_mixin(firmware,work/'mixin') with mixin.open('rb') as source: client.call('prepare',{'release':record['release'],**receipt},source=source, size=receipt['bytes'],timeout=180) diff --git a/src/orchard_cli/manifests.py b/src/orchard_cli/manifests.py index fb9686e..09529ee 100644 --- a/src/orchard_cli/manifests.py +++ b/src/orchard_cli/manifests.py @@ -25,6 +25,11 @@ } +def firmware_schema(): + """The firmware recipe schema shipped with the installed hardware data.""" + return json.loads((Path(orchard_hardware.__file__).parent / 'schemas/v1/firmware.schema.json').read_text()) + + class ManifestError(ValueError): pass diff --git a/src/orchard_cli/ram.py b/src/orchard_cli/ram.py index 89b77f9..48fca8d 100644 --- a/src/orchard_cli/ram.py +++ b/src/orchard_cli/ram.py @@ -30,7 +30,7 @@ def check_busybox(path): def assemble_ram(hardware, output, *, kernel, busybox, loader, pongo, init, - files=None, hostname='orchard-ram', device='j71b', modules=False, + files=None, hostname='orchard-ram', device, modules=False, bootargs='console=tty0 loglevel=7 rdinit=/init', sources=()): """Pack a caller-owned /init and arbitrary files. No firmware or modules are activated here. Files map relative RAM-root destinations to local files. @@ -99,6 +99,8 @@ def build_directory(directory, *, hardware=None, cache=None, kernel=None, """Build any DIRECTORY/init plus optional DIRECTORY/rootfs/; no layout convention.""" directory = Path(directory).expanduser().resolve() device = device if device is not None else build.configured_device(directory) + if device is None: + raise ValueError('No device selected: add device = "" to orchard.toml') if hardware is None: import orchard_hardware hardware = Path(orchard_hardware.__file__).resolve().parent diff --git a/src/orchard_cli/recipe.py b/src/orchard_cli/recipe.py new file mode 100644 index 0000000..878afc8 --- /dev/null +++ b/src/orchard_cli/recipe.py @@ -0,0 +1,178 @@ +"""Firmware recipes (schema 2): declarative steps from a fixed vocabulary. + +Hardware describes inputs and steps (docs/firmware-recipes.md in hardware); +this module implements the steps. It does not know where inputs come from: +the host reads them from the IPSW, the in-app installer from iPadOS. +""" +import hashlib +import json +import plistlib + +from orchard_cli.apple import ipsw, smartio, z2fw +from orchard_cli.apple.osserialize import apple_xml + + +def _copy(step, data, expected): + return data + + +def _bluetool_hcd(step, data, expected): + return ipsw.hcd_from_binary(data, expected) + + +def _smartio(step, data, expected): + return smartio.extract(data)[0] + + +def _im4p_payload(step, data, expected): + return im4p_payload(data) + + +def _touch_profile(data, name): + profile = plistlib.loads(data).get(name) + if not isinstance(profile, dict): + raise ValueError('Touch profile not found: ' + str(name)) + return profile + + +def _z2fw(step, data, expected): + profile = _touch_profile(data, step['profile']) + output, _ = z2fw.convert(profile, apple_xml(profile['Firmware Config']), + packet=step['packet'], load_address=step['load_address'], + hbpp_version=step['hbpp_version']) + return output + + +def _z2fw_packets(step, data, expected): + profile = _touch_profile(data, step['profile']) + output, _ = z2fw.convert_packets(profile, apple_xml(profile['Firmware Config']), packet=step['packet']) + return output + + +def _z2fw_mtfw(step, data, expected): + output, _ = z2fw.convert_mtfw(apple_xml(data), step['profile'], otp_preflight=step['otp_preflight']) + return output + + +# step@version -> (implementation, {parameter besides 'step' and 'input': type}) +STEPS = { + 'copy@1': (_copy, {}), + 'bluetool-hcd@1': (_bluetool_hcd, {}), + 'smartio-from-kernelcache@1': (_smartio, {}), + 'im4p-payload@1': (_im4p_payload, {}), + # mtprops with one preconstructed packet (J71b). + 'z2fw-from-mtprops@1': (_z2fw, {'profile': str, 'packet': str, 'load_address': int, 'hbpp_version': int}), + # mtprops with an array of preconstructed packets (iPhone 7 Plus), makez2fw's layout. + 'z2fw-from-mtprops@2': (_z2fw_packets, {'profile': str, 'packet': str}), + # MTFW (IM4P payload) with HBPP load items per personality (iPad 7), makez2fw's layout. + 'z2fw-from-mtfw@1': (_z2fw_mtfw, {'profile': str, 'otp_preflight': bool}), +} + + +class UnsupportedRecipe(ValueError): + """The recipe needs a step this Orchard version does not implement.""" + + +def check(recipe): + """Validate the recipe against the hardware schema, then its steps (host side).""" + from orchard_cli.manifests import firmware_schema + import jsonschema + jsonschema.validate(recipe, firmware_schema(), cls=jsonschema.Draft202012Validator) + return supported(recipe) + + +def supported(recipe): + """Confirm every step is implemented with its parameters, before any work. + + Needs no schema or hardware package, so the in-app installer can use it. + """ + if recipe.get('schema') != 2: + raise UnsupportedRecipe('This firmware recipe format needs a newer Orchard') + for output in recipe['outputs']: + # The first step reads a recipe input; each later step, the previous result. + if not output['steps'] or 'input' not in output['steps'][0] or \ + any('input' in step for step in output['steps'][1:]): + raise ValueError('Only the first step of an output names an input: ' + output['destination']) + for step in output['steps']: + if step['step'] not in STEPS: + raise UnsupportedRecipe(f"This firmware recipe needs step {step['step']}; " + 'update Orchard to a version that implements it') + _, parameters = STEPS[step['step']] + if set(step) - {'step', 'input'} != set(parameters): + raise ValueError(f"Wrong parameters for {step['step']}: {sorted(set(step) - {'step', 'input'})}") + for name, kind in parameters.items(): + # bool is an int subclass; neither stands in for the other. + if not isinstance(step[name], kind) or isinstance(step[name], bool) != (kind is bool): + raise ValueError(f"Parameter {name} of {step['step']} must be {kind.__name__}") + if 'input' in step and step['input'] not in recipe['inputs']: + raise ValueError('Unknown recipe input: ' + step['input']) + destinations = [output['destination'] for output in recipe['outputs']] + if len(set(destinations)) != len(destinations): + raise ValueError('Duplicate firmware destination') + return recipe + + +def needed_inputs(recipe, outputs=None): + outputs = recipe['outputs'] if outputs is None else outputs + return sorted({output['steps'][0]['input'] for output in outputs}) + + +def run(recipe, read_input, outputs=None): + """Produce outputs as {destination: bytes}; read_input(name) returns bytes. + + Every pinned input and every output is checked against its hash. + """ + cache = {} + + def source(name): + if name not in cache: + data = read_input(name) + pinned = recipe['inputs'][name].get('sha256') + if pinned and hashlib.sha256(data).hexdigest() != pinned: + raise ValueError('Firmware input differs from the recipe: ' + name) + cache[name] = data + return cache[name] + + result = {} + for output in recipe['outputs'] if outputs is None else outputs: + data = source(output['steps'][0]['input']) + for step in output['steps']: + implementation, _ = STEPS[step['step']] + data = implementation(step, data, output['sha256']) + if hashlib.sha256(data).hexdigest() != output['sha256']: + raise ValueError('Firmware output differs from the recipe: ' + output['destination']) + result[output['destination']] = data + return result + + +def im4p_payload(blob): + """Payload of an IM4P container: SEQUENCE { "IM4P", type, description, OCTET STRING, ... }.""" + def item(pos): + if pos + 2 > len(blob): + raise ValueError('Truncated IM4P') + tag, length = blob[pos], blob[pos + 1] + pos += 2 + if length & 0x80: + count = length & 0x7f + if not 1 <= count <= 4 or pos + count > len(blob): + raise ValueError('Invalid IM4P length') + length = int.from_bytes(blob[pos:pos + count], 'big') + pos += count + if pos + length > len(blob): + raise ValueError('Truncated IM4P') + return tag, blob[pos:pos + length], pos + length + + tag, body, _ = item(0) + if tag != 0x30: + raise ValueError('IM4P is not a DER sequence') + blob = body + fields, pos = [], 0 + while pos < len(blob) and len(fields) < 4: + tag, value, pos = item(pos) + fields.append((tag, value)) + if len(fields) < 4 or fields[0] != (0x16, b'IM4P') or fields[3][0] != 0x04: + raise ValueError('Not an IM4P container') + payload = fields[3][1] + if payload[:4] in (b'bvx2', b'bvx1', b'bvx-', b'bvxn'): + raise ValueError('Compressed IM4P payloads are not supported by im4p-payload@1') + return payload diff --git a/src/orchard_cli/releases.py b/src/orchard_cli/releases.py index b118c9e..35c5a2d 100644 --- a/src/orchard_cli/releases.py +++ b/src/orchard_cli/releases.py @@ -21,9 +21,37 @@ MAX_MANIFEST = 1024 * 1024 -def launch_manifest(profile): - """Select public launch assets from the detected device, never a distro.""" - source = hardware_device(profile)['boot'].get('launch_manifest') +def launcher_record(profile, ecid): + from .state import DeviceState + return DeviceState(profile, ecid).root / 'launcher.json' + + +def record_launcher(profile, ecid): + """Remember which launcher this device's installation needs.""" + from .state import private_directory + path = launcher_record(profile, ecid) + private_directory(path.parent) + temporary = path.with_suffix('.pending') + temporary.write_text(json.dumps({'manifest': launch_manifest(profile)}) + '\n') + temporary.chmod(0o600) + temporary.replace(path) + + +def launch_manifest(profile, ecid=None): + """Select public launch assets from the detected device, never a distro. + + An installation keeps the launcher it was installed for: the one recorded + at install time, or, for installs older than this record, the device's + pinned legacy launcher. + """ + boot = hardware_device(profile)['boot'] + source = boot.get('launch_manifest') + if ecid is not None: + path = launcher_record(profile, ecid) + if path.is_file() and not path.is_symlink(): + source = json.loads(path.read_text()).get('manifest') + elif boot.get('legacy_launch_manifest'): + source = boot['legacy_launch_manifest'] if not isinstance(source, str) or not source.startswith('https://'): raise ValueError('No launch assets published for connected device ' + profile) check_source({'url': source}, 'launch manifest') diff --git a/src/orchard_cli/service.py b/src/orchard_cli/service.py index 0e832ce..b10547f 100644 --- a/src/orchard_cli/service.py +++ b/src/orchard_cli/service.py @@ -11,7 +11,7 @@ from .contracts import Adapter from .progress import Clock, Progress from .protocol import Client, DeviceError -from .releases import default_manifest, resolve +from .releases import default_manifest, record_launcher, resolve from .devices import choose @@ -205,10 +205,32 @@ def finish_ssh_access(client, adapter: Adapter, status, report, path, clock, inp progress.phase('SSH access revoked for this boot; cached credentials removed') +def caveats(adapter): + """What to know before installing: features that are not working, free-text caveats.""" + import orchard_hardware + labels = orchard_hardware.features() + groups = {'missing': 'Not working yet', 'partial': 'Partly working', 'untested': 'Untested'} + rows = [] + for state, title in groups.items(): + names = [label for name, label in labels.items() if adapter.FEATURES.get(name) == state] + if names: + rows.append(f'- {title}: ' + ', '.join(names)) + lines = ['Caveats:', *rows, *(f'- {text}' for text in adapter.CAVEATS)] if rows or adapter.CAVEATS else [] + if adapter.STATUS != 'supported': + lines.append('This device has not been tested yet, proceed at your own risk!') + return lines + + @exclusive_operation def install(manifest=None, *, input_fn=input, progress=None): clock, progress = Clock(), progress or Progress() adapter, device = choose() # Definition of installation timing begins here. + if not adapter.INSTALL: + raise ValueError(f'Installing on the {adapter.NAME} is not available yet; ' + 'RAM boot is available for testing') + progress.phase(f'{adapter.NAME} ({adapter.STATUS})') + for line in caveats(adapter): + progress.phase(line) answer = ask(clock, f'Install Linux on {adapter.NAME}? [Y/n] ', input_fn) if answer.lower() not in ('', 'y', 'yes'): progress.phase('Cancelled; no download or device mutation performed') @@ -251,10 +273,7 @@ def install(manifest=None, *, input_fn=input, progress=None): if firmware is None: firmware = adapter.firmware_inputs(clock, progress) progress.phase('Preparing device hardware…') - adt = work / 'device.adt' - adt.write_bytes(adapter.capture_adt(device)) - adt.chmod(0o600) - mixin, mixin_record = adapter.prepare_mixin(adt, firmware, work / 'mixin') + mixin, mixin_record = adapter.prepare_mixin(firmware, work / 'mixin') with mixin.open('rb') as source: client.call('prepare', {'release': record['release'], **mixin_record}, source=source, size=mixin_record['bytes'], timeout=180) @@ -265,6 +284,8 @@ def install(manifest=None, *, input_fn=input, progress=None): with assets['root'].open('rb') as source: result = client.call('install', {**destination, **record['assets']['root']}, source=source, size=record['assets']['root']['bytes'], timeout=1200) + # This installation is launched by the current launcher from now on. + record_launcher(adapter.PROFILE, device['ecid']) snapshot = clock.result() save_report(private / 'last-install.json', { **identity, 'installed': True, 'ready': None, 'release': record['release'], diff --git a/src/orchard_cli/staging.py b/src/orchard_cli/staging.py deleted file mode 100644 index 8ffb366..0000000 --- a/src/orchard_cli/staging.py +++ /dev/null @@ -1,83 +0,0 @@ -#!/usr/bin/env python3 -"""Stage verified local firmware files using a device manifest.""" -import argparse -import json -from pathlib import Path, PurePosixPath -import re -import shutil - -from .integrity import digest as file_digest - - -def stage(manifest_path: Path, inputs: list[Path], receipt_paths: list[Path], output: Path): - manifest = json.loads(manifest_path.read_text()) - files = manifest['files'] - generated = manifest.get('generated', []) - receipts = {} - for path in receipt_paths: - receipt = json.loads(path.read_text()) - name = receipt.get('name') - digest = receipt.get('sha256') - size = receipt.get('bytes') - if (not isinstance(name, str) or name in receipts or - receipt.get('device') != manifest['device'] or - not isinstance(digest, str) or not re.fullmatch(r'[0-9a-f]{64}', digest) or - isinstance(size, bool) or not isinstance(size, int) or size <= 0): - raise ValueError(f'invalid or duplicate generated-file receipt: {path}') - receipts[name] = receipt - if set(receipts) != {item['name'] for item in generated}: - raise ValueError('supply exactly one local receipt for each generated firmware file') - selected = [] - destinations = set() - names = set() - for item in files + generated: - name = item['name'] - digest = receipts[name]['sha256'] if item in generated else item['sha256'] - size = receipts[name]['bytes'] if item in generated else None - destination = PurePosixPath(item['destination']) - if (not name or name in ('.', '..') or '/' in name or name in names or - destination.is_absolute() or '..' in destination.parts or - not destination.parts or str(destination) in destinations or - not re.fullmatch(r'[0-9a-f]{64}', digest)): - raise ValueError('unsafe or duplicate firmware manifest entry') - destinations.add(str(destination)) - names.add(name) - candidates = (p for directory in inputs - for p in ([directory] if directory.is_file() else directory.rglob(name)) - if p.is_file() and not p.is_symlink() and p.name == name) - matches = [p for p in candidates if (size is None or p.stat().st_size == size) - and file_digest(p) == digest] - if len(matches) != 1: - raise ValueError(f'expected one matching local input for {name}; found {len(matches)}') - selected.append((matches[0], output.joinpath(*destination.parts), digest)) - for _, dest, digest in selected: - if dest.exists() and file_digest(dest) != digest: - raise ValueError(f'existing output differs: {dest}') - for source, dest, digest in selected: - dest.parent.mkdir(parents=True, exist_ok=True) - if not dest.exists(): - shutil.copyfile(source, dest) - dest.chmod(0o600) - if file_digest(dest) != digest: - raise ValueError(f'copied firmware differs from verified input: {dest}') - return len(selected) - - -def main(): - parser = argparse.ArgumentParser(description=__doc__) - parser.add_argument('--manifest', type=Path, required=True) - parser.add_argument('--input', type=Path, action='append', required=True, - help='extracted firmware or local device-data directory; repeat as needed') - parser.add_argument('--receipt', type=Path, action='append', default=[], - help='local receipt from a device-specific generator; repeat as needed') - parser.add_argument('--output', type=Path, required=True) - args = parser.parse_args() - try: - count = stage(args.manifest, args.input, args.receipt, args.output) - except (ValueError, OSError, KeyError) as exc: - parser.error(str(exc)) - print(f'Staged {count} verified local firmware files in {args.output}') - - -if __name__ == '__main__': - main() diff --git a/tests/hardware_layout.py b/tests/hardware_layout.py new file mode 100644 index 0000000..223ada7 --- /dev/null +++ b/tests/hardware_layout.py @@ -0,0 +1,53 @@ +"""Minimal hardware checkout in the platform-series layout, for kernel cache tests.""" +from pathlib import Path +import shutil + +SERIES = '''#!/usr/bin/env python3 +from pathlib import Path +import re +import sys + +ROOT = Path(__file__).resolve().parent + + +def resolve(name): + descriptor = ROOT.parent / 'devices' / name / 'device.toml' + if descriptor.is_file(): + values = dict(re.findall(r'^(platform|dtb)\\s*=\\s*"([^"]+)"', descriptor.read_text(), re.M)) + return values['platform'], values['dtb'] + return name, None + + +def patches(name): + result = [] + for line in (ROOT / 'series' / resolve(name)[0]).read_text().splitlines(): + if line.strip(): + result += sorted((ROOT / line.strip()).rglob('*.patch')) + return result + + +if sys.argv[1] == '--resolve': + platform, dtb = resolve(sys.argv[2]) + print(platform, dtb or '-') +else: + for patch in patches(sys.argv[1]): + print(patch) +''' + + +def write(hardware, device='j71b', platform='apple', dtb='t8010-j71b'): + """Series, platform config and device descriptor; returns the patch layer.""" + hardware = Path(hardware) + files = { + 'kernel/series.py': SERIES, + 'kernel/series/' + platform: 'patches/' + platform + '\n', + 'kernel/configs/' + platform + '/config': 'config', + 'devices/' + device + '/device.toml': f'[kernel]\nplatform = "{platform}"\ndtb = "{dtb}"\n', + } + for name, text in files.items(): + path = hardware / name + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(text) + layer = hardware / 'kernel/patches' / platform + layer.mkdir(parents=True, exist_ok=True) + return layer diff --git a/tests/test_adapter_contract.py b/tests/test_adapter_contract.py index 85507b3..a88b5ae 100644 --- a/tests/test_adapter_contract.py +++ b/tests/test_adapter_contract.py @@ -1,63 +1,59 @@ """Registry/adapter conformance checks, with no USB access or boot actions.""" -from types import SimpleNamespace import unittest -from unittest.mock import Mock, patch +from unittest.mock import patch from orchard_cli.contracts import Adapter from orchard_cli import devices -from orchard_cli.j71b import adapter - -MEMBERS = ('PROFILE', 'NAME', 'REBOOT_MESSAGE', 'list_devices', 'discover', 'boot', - 'connect', 'connect_saved', 'forget_access', 'capture_adt', 'firmware_inputs', 'prepare_mixin') - - -def module(**changes): - values = {name:getattr(adapter,name) for name in MEMBERS} - values.update(changes) - return SimpleNamespace(**values) +from orchard_cli.apple import device as adapter_module class AdapterContract(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('j71b'), adapter) - self.assertIsInstance(adapter, Adapter) - - def test_missing_operation_is_rejected_not_filled_by_a_fallback(self): - incomplete = module() - del incomplete.connect_saved - with patch.object(devices.importlib, 'import_module', return_value=incomplete): - with self.assertRaisesRegex(ValueError, 'Incomplete adapter contract'): - devices.load('j71b') - - def test_registry_profile_mismatch_is_rejected(self): - with patch.object(devices.importlib, 'import_module', return_value=module(PROFILE='other')): - with self.assertRaisesRegex(ValueError, 'does not match'): - devices.load('j71b') - - def test_invalid_presentation_metadata_is_rejected(self): - for values in ({'NAME':''}, {'NAME':None}, {'REBOOT_MESSAGE':None}): - with patch.object(devices.importlib, 'import_module', return_value=module(**values)): - with self.assertRaisesRegex(ValueError, 'Invalid adapter presentation'): - devices.load('j71b') + def test_every_described_device_satisfies_the_contract_without_discovery(self): + self.assertIn('j71b', devices.profiles()) + with patch.object(adapter_module.usb, 'devices', side_effect=AssertionError('Unexpected USB access')): + for profile in devices.profiles(): + with self.subTest(profile=profile): + adapter = devices.load(profile) + self.assertIsInstance(adapter, Adapter) + self.assertEqual(adapter.PROFILE, profile) + self.assertTrue(adapter.NAME) + self.assertIn(adapter.NAME, adapter.REBOOT_MESSAGE) + + def test_identity_comes_from_the_hardware_descriptor(self): + adapter = devices.load('j71b') + self.assertEqual((adapter.CHIP, adapter.BOARD, adapter.PRODUCT), (0x8010, 0x18, 'J71b Recovery')) + self.assertTrue(adapter.INSTALL) + + def test_unknown_profile_is_refused(self): + with self.assertRaisesRegex(ValueError, 'not installed'): + devices.load('no-such-board') def test_dfu_discovery_returns_explicit_resume_state_without_mutating_observation(self): - observed = {'mode':'dfu', 'location':'usb', 'ecid':1, - 'chip':adapter.CHIP, 'board':adapter.BOARD} - with patch.object(adapter.usb, 'devices', return_value=[observed]): + adapter = devices.load('j71b') + observed = {'mode': 'dfu', 'location': 'usb', 'ecid': 1, + 'chip': adapter.CHIP, 'board': adapter.BOARD} + with patch.object(adapter_module.usb, 'devices', return_value=[observed]): selected = adapter.discover() self.assertIs(selected['resumed'], False) self.assertEqual(selected['ecid'], 1) self.assertNotIn('resumed', observed) - 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): + def test_choose_matches_the_connected_device_against_every_descriptor(self): + adapter = devices.load('j71b') + observed = {'mode': 'dfu', 'location': 'usb', 'ecid': 1, + 'chip': adapter.CHIP, 'board': adapter.BOARD} + with patch.object(adapter_module.usb, 'devices', return_value=[observed]), \ + patch('orchard_cli.apple.usb.devices', return_value=[observed]): loaded, selected = devices.choose() - self.assertIs(loaded, implementation) - self.assertIs(selected, selection) - implementation.discover.assert_called_once_with() + self.assertEqual(loaded.PROFILE, 'j71b') + self.assertEqual(selected['ecid'], 1) + + def test_unknown_connected_device_is_named_in_the_error(self): + observed = {'mode': 'dfu', 'location': 'usb', 'ecid': 1, 'chip': 0x8999, 'board': 0x7f} + with patch('orchard_cli.apple.usb.devices', return_value=[observed]): + with self.assertRaisesRegex(ValueError, 'not one Orchard supports'): + devices.choose() -if __name__ == '__main__':unittest.main() +if __name__ == '__main__': + unittest.main() diff --git a/tests/test_binary_codecs.py b/tests/test_binary_codecs.py index 7c1a55f..4940423 100644 --- a/tests/test_binary_codecs.py +++ b/tests/test_binary_codecs.py @@ -5,7 +5,7 @@ from orchard_cli.apple.adt_tree import parse, u32s from orchard_cli.apple.macho import macho from orchard_cli.apple.osserialize import apple_xml -from orchard_cli.j71b import touch +from orchard_cli.apple import z2fw as touch def property_record(name, value): @@ -64,19 +64,23 @@ def test_touch_integer_bounds_and_word_encoding(self): with self.assertRaises(ValueError): touch.swapped(b'odd') - def test_calibration_packets_have_exact_wire_layout(self): - packet = touch.calibration_packet(0x100000, b'\x01\x02\x03') - self.assertEqual(packet.hex(), '18e1300100000000001000100201000300060000') + def test_blob_packets_decode_and_reject_damage(self): + # HBPP blob packet as apple_z2 builds it for J71b calibration. + packet = bytes.fromhex('18e1300100000000001000100201000300060000') self.assertEqual(touch.decode_blob(packet), (0x100000, b'\x01\x02\x03\0')) for index in (0, 4, 10, 12, len(packet) - 1): damaged = bytearray(packet) damaged[index] ^= 1 with self.subTest(index=index), self.assertRaises(ValueError): touch.decode_blob(bytes(damaged)) - with self.assertRaises(ValueError): - touch.calibration_packet(0xffffffff, b'abcd') - with self.assertRaises(ValueError): - touch.calibration_packet(0, b'') + + def test_calibration_records_have_exact_wire_layout(self): + payload = struct.pack('SysConfigKeyMtClPayloadAAAAAAAAAAA=' + 'Propertymulti-touch-calibrationMaxSize' + '0x600Address0x100400' + 'TypeProperty') if calibration else '' + return ( + 'TEST,6' + 'Address0x400b0000Mask' + '0x1TypeReadModifyWrite' + 'Value0x0SkipCheck' + 'Address0x100000Payload' + f'{b64(bytes(range(1, 11)))}TypeBinary' + + prop + + 'AddressType' + 'Value0x1' + 'TypeConfigConfig' + 'ProtocolZ2Z2 ConfigInter Packet Delay' + '125Min DMA Transfer Size17' + 'SPI ConfigClock Phase1Word Delay' + '0Clock Polarity1' + 'CS Delay10000Clock Period Ms' + '41Power Off On Reset' + 'Device Tree Overrides' + f'otp-address{b64(struct.pack("' + f'hbpp-version{b64(struct.pack("' + f'otp-sn{b64(struct.pack("' + 'Boot TimeoutNormal Boot Ms500' + '').encode() + b'\0' + + +def mtprops_fixture(boot_timeout=True): + """mtprops with an array of preconstructed packets and an OSSerialize firmware config.""" + timeout = ('Boot TimeoutNormal Boot Ms300' + if boot_timeout else '') + config = ( + 'ProtocolZ2Min DMA Transfer Size8' + 'Power Off On ResetZ2 ConfigInter Packet Delay' + '50SPI ConfigClock Phase0' + 'Clock Polarity0Word Delay0' + 'CS Delay45Clock Period Ms62' + + timeout + + 'Calibration Sequence' + 'Providermulti-touch-calibrationAddress' + '0x10009000Max Size1800' + 'Providerorb-gap-calAddress' + '0x1000a000Max Size992' + 'Boot Sequence' + 'Address0x10003060Mask0xffffffff' + 'Value6099' + 'ActionRequestCalibration' + 'Address0x10003000Value3' + '').encode() + b'\0' + return plistlib.dumps({'TEST,2': { + 'PreconstructedBootloadPacketType': 'TESTPKT', + 'Constructed Firmware': [b'\x18\xe1\x30\x01' + bytes(range(16)), b'\x1e\x33' + bytes(14)], + 'Firmware Config': config}}) + + +def convert_packets(data, packet='TESTPKT'): + profile = plistlib.loads(data)['TEST,2'] + return z2fw.convert_packets(profile, apple_xml(profile['Firmware Config']), packet=packet) + + +class PacketEncoding(unittest.TestCase): + def test_hbpp14_blob_matches_the_driver_layout_and_roundtrips(self): + packet = z2fw.encode_blob(0x100000, b'\x01\x02\x03') + self.assertEqual(packet.hex(), '18e1300100000000001000100201000300060000') + self.assertEqual(z2fw.decode_blob(packet), (0x100000, b'\x01\x02\x03\0')) + data = bytes(range(256)) * 3 + self.assertEqual(z2fw.decode_blob(z2fw.encode_blob(0x12345678, data)), (0x12345678, data)) + for address, data in ((0, b''), (0xfffffffe, b'abcd'), (0, bytes(0x40001))): + with self.subTest(address=address, size=len(data)), self.assertRaises(ValueError): + z2fw.encode_blob(address, data) + + def test_register_read_is_two_bit_reversed_raw_transfers(self): + (c1, request), (c2, exchange) = z2fw.register_read(0x400b6020) + self.assertEqual((c1, c2), (z2fw.RAW_XFER, z2fw.RAW_XFER)) + # tx 8, rx 0, flags 1; 1c73, address word-swapped, BE16 sum of the address bytes. + self.assertEqual(request.hex(), '08000000' '00000000' '01000000' '1c73' '6020400b' '00cb') + self.assertEqual(exchange.hex(), '08000000' '08000000' '01000000' '1aa118e118e118e1') + with self.assertRaises(ValueError): + z2fw.raw_xfer(bytes(17), 0) + + def test_otp_preflight_reads_each_selected_word(self): + records = z2fw.otp_preflight_records(0x400b6000, 62, 8) + self.assertEqual(len(records), 3 + 2 * 9) + self.assertEqual(records[3], (z2fw.SEND, z2fw.register_packet(0x400b6010, 0xffffffff, 0xffffffff))) + self.assertEqual(records[4], (z2fw.SEND, z2fw.register_packet(0x400b600c, 0xffffffff, 62))) + self.assertEqual(records[13], (z2fw.SEND, z2fw.register_packet(0x400b600c, 0xffffffff, 63))) + for size in (0, 6, 260): + with self.subTest(size=size), self.assertRaises(ValueError): + z2fw.otp_preflight_records(0x400b6000, 0, size) + + def test_config_valid_mask_follows_present_fields(self): + config = {'Min DMA Transfer Size': 17, 'Power Off On Reset': True, + 'Device Tree Overrides': {'hbpp-version': struct.pack(' Adapter: + return device