diff --git a/boot/initramfs/root/init b/boot/initramfs/root/init index c804ca1d..9556f126 100755 --- a/boot/initramfs/root/init +++ b/boot/initramfs/root/init @@ -1,6 +1,6 @@ #!/bin/busybox sh # orchard's initrd for installed systems: load storage, mount the root that -# root=PARTUUID= names, start its init. The same for every distribution and +# root=PARTUUID= names (on a GPT or LwVM disk), start its init. The same for every distribution and # init system; only BusyBox and the platform's storage modules are inside. export PATH=/bin /bin/busybox --install -s /bin @@ -53,6 +53,16 @@ gpt_partuuid() { echo "$4$3$2$1-$6$5-$8$7-$9${10}-${11}${12}${13}${14}${15}${16}" } +# lwvm_partuuid DISK NUMBER: the GUID of LwVM entry NUMBER (32-bit boards' iOS +# disk), from the header the kernel's LwVM reader reads too; its partition +# numbers are the entries' numbers. +lwvm_partuuid() { + [ "$(od -An -tx1 -v -N 16 "$1" | tr -d ' \n')" = 6a9088cf8afd630ae351e24887e0b98b ] || return 1 + set -- $(od -An -tx1 -v -j $((512 + ($2 - 1) * 128 + 16)) -N 16 "$1") + [ "$*" != '00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00' ] || return 1 + echo "$1$2$3$4-$5$6-$7$8-$9${10}-${11}${12}${13}${14}${15}${16}" +} + device= seconds=0 while [ -z "$device" ]; do @@ -61,7 +71,10 @@ while [ -z "$device" ]; do part=${number%/partition} disk=$(basename "$(dirname "$(readlink -f "$part")")") sector=$(cat "/sys/class/block/$disk/queue/logical_block_size") - if [ "$(gpt_partuuid "/dev/$disk" "$sector" "$(cat "$number")")" = "$root" ]; then + # A GPT first, as always; only a disk without one is asked for LwVM. + id=$(gpt_partuuid "/dev/$disk" "$sector" "$(cat "$number")") || + id=$(lwvm_partuuid "/dev/$disk" "$(cat "$number")") + if [ "$id" = "$root" ]; then device=/dev/${part##*/} break fi diff --git a/devices/p105/README.md b/devices/p105/README.md index 167dba85..0c172860 100644 --- a/devices/p105/README.md +++ b/devices/p105/README.md @@ -34,9 +34,12 @@ public recipe or release. `kernel/patches/armv7/mtd`; the disk is `applesftlN`, on the 64 GB model only. Reclamation (#93) is unfinished: until it lands, writes stop once the blank flash blocks are used up. -- A Linux root on LwVM: the kernel reads LwVM partitions - (`kernel/patches/armv7/block`), but the shared initrd finds its root by - reading a GPT itself and needs to learn LwVM partitions. +- Installing: `orchard install` writes a root into an existing Linux LwVM + partition (the shared initrd finds it by its LwVM GUID); creating that + partition belongs to the iOS resize step, which does not exist yet. There + is no boot partition: starting the installed system later needs the + Cascadia bundle the install leaves on the host, placed for iOS's boot + chain. Starting any payload needs `orchard-launcher` (recovery-autogo). - USB: the cold-handoff topic (#90) awaits a device boot of its exact revision, and the iPad 3 (J1) a regression boot of the shared PHY code; see [USB handoff](../../docs/bring-up/p105-usb-handoff.md). diff --git a/devices/p105/device.toml b/devices/p105/device.toml index 6e8daab3..fa6c9bf8 100644 --- a/devices/p105/device.toml +++ b/devices/p105/device.toml @@ -24,6 +24,13 @@ dtb = "s5l8942x-p105" # arch/arm/boot/dts/apple/.dtb [firmware] recipe = "firmware.json" +# The iOS disk's translation layer writes append-only: until reclamation lands +# (OrchardLinux/orchard#93), freed space is never reused and writes fail once +# the blank flash runs out. Drop this table when it lands; the installer +# warns from it and allows the install either way. +[storage] +reclaims = false + [support] status = "experimental" diff --git a/docs/boot-contract.md b/docs/boot-contract.md index aa7c72fc..361a36a1 100644 --- a/docs/boot-contract.md +++ b/docs/boot-contract.md @@ -50,6 +50,19 @@ which one and opens a shell on the console. - **Partition identity only.** Roots are named by PARTUUID in their entries, never by partition number or position. - **Inside its root partition, a system uses its native layout,** for example upstream Omarchy's btrfs subvolumes. +## Partitions (LwVM, 32-bit boards, interim) + +The iOS disk of 32-bit boards is partitioned with LwVM, which the kernel reads +(`kernel/patches/armv7/block`): partitions keep their LwVM entry number, and the +entry's GUID is their PARTUUID. Until the iOS resize step creates the Linux +entry, the installer only replaces an existing, unencrypted Linux entry +(`0fc63daf-8483-4772-8e79-3d69d8477de4`) in place and never writes the LwVM map. +There is no boot partition yet: the entry and its files are written to the +root's own `/boot`, in the layout below, for a later boot partition or loader to +take over. Meanwhile the host keeps the Cascadia bundle that starts the +installed system (its tree's `bootargs` carry `root=PARTUUID=`), which has to be +placed where iOS's boot chain finds it. + ## Entries The layout follows BLS: @@ -68,6 +81,8 @@ devicetree /6a1f…/6.18.2-orchard/t8010-j71b.dtb options root=PARTUUID=4d6c…-… rw quiet ``` +`architecture` is `aa64`, or `arm` for a 32-bit zImage. + | Key | Orchard rule | |---|---| | `linux` | Required. arm64 `Image`, raw or gzip-compressed | diff --git a/schemas/v1/device.schema.json b/schemas/v1/device.schema.json index 35bc4cf0..77db1dee 100644 --- a/schemas/v1/device.schema.json +++ b/schemas/v1/device.schema.json @@ -135,6 +135,15 @@ ], "additionalProperties": false }, + "storage": { + "type": "object", + "properties": { + "reclaims": { + "type": "boolean" + } + }, + "additionalProperties": false + }, "support": { "type": "object", "properties": { diff --git a/src/orchard/apple/device.py b/src/orchard/apple/device.py index b2a98c13..18ffdd36 100644 --- a/src/orchard/apple/device.py +++ b/src/orchard/apple/device.py @@ -14,7 +14,7 @@ from orchard.apple import entries, usb, boot as apple_boot, adt_capture from orchard.apple.adt_tree import parse from orchard.transports import ssh -from orchard import firmware, hardware +from orchard import firmware, hardware, launcher from .adt import validate as validate_adt @@ -40,6 +40,12 @@ def __init__(self, descriptor: Dict[str, Any]): self.INSTALL = not self.INSTALL_BLOCKERS self.STATUS = support.get('status', 'experimental') self.FEATURES = dict(descriptor.get('features', {})) + # Facts about the internal storage that the installer tells the person (service.caveats). + self.STORAGE = dict(descriptor.get('storage', {})) + # The boot method orchard-launcher starts this board with, when no method of orchard's own fits. + listed = descriptor.get('boot', {}).get('methods', []) + self.LAUNCHER_CHAIN = None if any(name in hardware.IMPLEMENTED_METHODS for name in listed) else \ + next((name for name in listed if launcher.delivers(name)), None) 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) @@ -83,6 +89,8 @@ def discover(self) -> SelectedDevice: return cast(SelectedDevice, {**device, 'resumed': False}) def boot_inputs(self) -> List[Any]: + if self.LAUNCHER_CHAIN: + return [] # the launcher brings what its chain needs return entries.inputs(entries.select(self._descriptor)) def boot(self, device: SelectedDevice, assets: BootAssets, diff --git a/src/orchard/apple/linux_usb.py b/src/orchard/apple/linux_usb.py index 36c8ea0c..cdc4735c 100644 --- a/src/orchard/apple/linux_usb.py +++ b/src/orchard/apple/linux_usb.py @@ -7,6 +7,19 @@ MODES = {0x1227: 'dfu', 0x4141: 'pongo', 0x4142: 'linux', 0x4144: 'ram', 0x4145: 'system'} +def serial_tags(serial): + """What an Apple USB serial string says beyond ECID, CPID and BDID. + + serial_number: the device's serial number (DFU's SRNM:[...]); pwned: the + exploit that left its PWND:[...] tag (checkm8, ...), or checkra1n for its + YOLO:checkra1n tag; None when absent. + """ + number = re.search(r'\bSRNM:\[([A-Za-z0-9]{1,32})\]', serial) + pwned = re.search(r'\bPWND:\[([^\]]{1,64})\]', serial) + return {'serial_number': number.group(1) if number else None, + 'pwned': pwned.group(1) if pwned else ('checkra1n' if 'YOLO:checkra1n' in serial else None)} + + def devices(root=Path('/sys/bus/usb/devices')): found = [] for node in root.iterdir(): @@ -32,8 +45,8 @@ def devices(root=Path('/sys/bus/usb/devices')): boot_id=str(uuid.UUID(boot.group(1))) if boot else None, chip=int(chip.group(1), 16) if chip else None, board=int(board.group(1), 16) if board else None, - pwned='YOLO:checkra1n' in serial, profile=profile.group(1) if profile else None, product=product, - serial_port=next(iter(ports), None))) + profile=profile.group(1) if profile else None, product=product, + serial_port=next(iter(ports), None), **serial_tags(serial))) return found diff --git a/src/orchard/apple/usb.py b/src/orchard/apple/usb.py index 76791e13..fa4090fd 100644 --- a/src/orchard/apple/usb.py +++ b/src/orchard/apple/usb.py @@ -5,7 +5,7 @@ import sys import uuid -from .linux_usb import devices as linux_devices, m1n1_ports as linux_m1n1_ports +from .linux_usb import devices as linux_devices, m1n1_ports as linux_m1n1_ports, serial_tags from orchard.state import EndpointUnavailable # 0x4144: a RAM payload's USB session (orchard/usb-session). @@ -58,7 +58,7 @@ def devices(): 'board': int(fields[2].group(1), 16) if fields[2] else None, 'profile': profile.group(1) if profile else None, 'product': node.get('USB Product Name'), - 'serial_port': ports[0] if ports else None}) + 'serial_port': ports[0] if ports else None, **serial_tags(serial)}) return found diff --git a/src/orchard/boot.py b/src/orchard/boot.py index b8d7bc33..f18139ab 100644 --- a/src/orchard/boot.py +++ b/src/orchard/boot.py @@ -71,6 +71,14 @@ def boot_ram(output, *, yes=False, logs=None): print('WARNING: The intended device for this build is unknown.') print('Compatibility with your connected device has not been established.') print('It may fail to boot or stop responding until you restart it.') + if receipt.get('loader_format') == 'cascadia': + # 32-bit boards: orchard-launcher takes the device from DFU to the payload (launcher.py). + if not yes: + confirm_boot(mismatch=mismatch) + verify_ram(output) + start_cascadia(output, receipt, device) + print('The launcher reports the payload started; Linux readiness has NOT been verified.') + return # Before the question, so that it can finish while it is being read. entry = entries.select(description if not mismatch else None) fetching = Background(fetch_chain, entries.inputs(entry) if device['mode'] == 'dfu' else [], Progress()) @@ -90,6 +98,21 @@ def boot_ram(output, *, yes=False, logs=None): print('or screen /dev/ttyACM0 115200 (Linux; verify the actual port).') +def start_cascadia(output, receipt, device): + """The Cascadia payload in OUTPUT through orchard-launcher, with the device's serial number in its tree.""" + import tempfile + from . import launcher + from .hardware import data_root + from .image import cascadia + from .progress import Progress + image = Path(output) / 'linux.bin' + with tempfile.TemporaryDirectory(prefix='payload-') as directory: + if device.get('serial_number'): + image = Path(directory) / 'linux.bin' + image.write_bytes(cascadia.with_serial((Path(output) / 'linux.bin').read_bytes(), device['serial_number'])) + launcher.load(image, chain=receipt.get('boot_method', 'recovery-autogo'), catalog=data_root(), progress=Progress()) + + def run_directory(directory, *, yes=False, build=True, device=None, platform=None, **build_options): """Build what is missing or outdated for the attached board, then boot it once. diff --git a/src/orchard/contracts.py b/src/orchard/contracts.py index 76dc456f..85252899 100644 --- a/src/orchard/contracts.py +++ b/src/orchard/contracts.py @@ -20,7 +20,8 @@ class _SelectionDetails(TypedDict, total=False): board: Optional[int] product: Optional[str] serial_port: Optional[str] - pwned: bool + pwned: Optional[str] + serial_number: Optional[str] boot_id: Optional[str] diff --git a/src/orchard/hardware.py b/src/orchard/hardware.py index 9eea2e88..3f62a820 100644 --- a/src/orchard/hardware.py +++ b/src/orchard/hardware.py @@ -97,11 +97,16 @@ def install_blockers(profile, described=None): as soon as it could work, even if other features don't work yet. """ described = device(profile) if described is None else described + from . import launcher blockers = [] if 'kernel' not in described: blockers.append('it has no kernel or device tree yet') - if not any(_method_usable(profile, name) for name in described.get('boot', {}).get('methods', [])): - blockers.append('this Orchard has no boot method for it yet') + listed = described.get('boot', {}).get('methods', []) + if not any(_method_usable(profile, name) for name in listed): + if any(launcher.delivers(name) and _method_listed(profile, name) for name in listed) and launcher.find() is None: + blockers.append(f'it starts through {launcher.NAME}, which is not installed') + else: + blockers.append('this Orchard has no boot method for it yet') if 'firmware' not in described: blockers.append('it has no firmware recipe yet') if described.get('features', {}).get('storage_write', 'missing') == 'missing': @@ -125,9 +130,18 @@ def has_recovery(described, root=None): return False +def method_implemented(name): + """Whether this Orchard can run boot method NAME: its own code, or orchard-launcher when installed.""" + from . import launcher + return name in IMPLEMENTED_METHODS or (launcher.delivers(name) and launcher.find() is not None) + + def _method_usable(profile, name): - if name not in IMPLEMENTED_METHODS: - return False + return method_implemented(name) and _method_listed(profile, name) + + +def _method_listed(profile, name): + """Whether the board has what method NAME needs: its boot.json recipe, where the method takes one.""" if not methods()[name]['boot_json']: return True recipe = device_path(profile).parent / 'boot.json' diff --git a/src/orchard/image/cascadia.py b/src/orchard/image/cascadia.py index afee424a..b6fe2b57 100644 --- a/src/orchard/image/cascadia.py +++ b/src/orchard/image/cascadia.py @@ -6,13 +6,17 @@ where the initramfs is; it copies it to the profile's initrd address before it turns off the MMU and enters the unchanged zImage, as Project Cascadia's linux_boot does. The DTB's /chosen linux,initrd-start/-end, placeholders in -the board's tree, are set in place to that copy. +the board's tree, are set in place to that copy. The host may also give the +device's serial number, which goes to the tree's root as serial-number (what +m1n1 copies from Apple's tree on 64-bit boards), and boot arguments for an +installed system, which replace /chosen bootargs. This module uses only the standard library and also runs by file path. """ import argparse import hashlib import json from pathlib import Path +import re import struct import tomllib import zlib @@ -284,6 +288,37 @@ def set_initrd(data, start, size): return bytes(result) +SERIAL = re.compile(r'[A-Za-z0-9]{1,32}') + + +def set_properties(data, *, serial=None, bootargs=None): + """DATA with the root's serial-number and /chosen bootargs set; unchanged when neither is given.""" + if serial is None and bootargs is None: + return data + root, reservations, cpu = read_fdt(data) + if serial is not None: + if not SERIAL.fullmatch(serial): + raise ValueError('Expected a serial number of letters and digits') + root['properties']['serial-number'] = serial.encode() + b'\0' + if bootargs is not None: + if any(ord(c) < 0x20 or ord(c) > 0x7e for c in bootargs): + raise ValueError('Boot arguments must be printable ASCII on one line') + chosen = [child for child in root['children'] if child['name'] == 'chosen'] + if not chosen: + raise ValueError('The DTB has no /chosen for boot arguments') + chosen[0]['properties']['bootargs'] = bootargs.encode() + b'\0' + return write_fdt(root, reservations, cpu) + + +def board_bootargs(data): + """The board tree's own /chosen bootargs, or ''.""" + root, _, _ = read_fdt(data) + for child in root['children']: + if child['name'] == 'chosen': + return child['properties'].get('bootargs', b'').rstrip(b'\0').decode('ascii', 'replace') + return '' + + def decompression_bound(kernel, dtb_size, described): """Conservatively bound the v6.12 ARM decompressor's relocated workspace. @@ -318,12 +353,14 @@ def decompression_bound(kernel, dtb_size, described): return bound -def assemble(loader, kernel, dtb, initramfs, described, *, elf=None): +def assemble(loader, kernel, dtb, initramfs, described, *, elf=None, serial=None, bootargs=None): """(image, receipt): LOADER, the unchanged zImage KERNEL with DTB appended, then INITRAMFS. The loader copies INITRAMFS to the profile's initrd address before it - turns off the MMU, and the DTB's /chosen says where it is.""" + turns off the MMU, and the DTB's /chosen says where it is. SERIAL and + BOOTARGS: see set_properties.""" validate_profile(described) + source_dtb, dtb = dtb, set_properties(dtb, serial=serial, bootargs=bootargs) verify_loader(loader, described, elf) if (len(kernel) < 0x30 or struct.unpack_from(' 1: + raise D.InstallError('disks', 'More than one internal disk: ' + ', '.join(found)) + if found: + return '/dev/' + found[0] + if time.monotonic() > deadline: + raise ValueError(f'Storage did not come up within {seconds} s') + time.sleep(0.1) + + def read_disk(connection, disk, offset, size, block): return shell(connection, f'dd if={disk} bs={block} skip={offset // block} count={size // block} 2>/dev/null') @@ -242,6 +298,8 @@ def probe(start): def inspect(connection, disk): """The layout install_layout plans from, as install_device.py inspect reports it.""" + if read_disk(connection, disk, 0, lwvm.HEADER_BYTES, lwvm.HEADER_BYTES)[:16] == lwvm.MAGIC: + return inspect_lwvm(connection, disk) block, header, entries = read_table(connection, disk) info, partitions, free = D.parse_gpt(disk, block, header, entries) for part in partitions: @@ -249,6 +307,69 @@ def inspect(connection, disk): return {'disk': info, 'partitions': partitions, 'free': free} +def kernel_partitions(connection, disk): + """{number: (start, sectors)} of the partitions the kernel found on DISK, in 512-byte sectors.""" + name = Path(disk).name + text = shell(connection, f'for p in /sys/class/block/{name}/{name}*; do [ -e "$p/partition" ] && ' + 'echo $(cat "$p/partition") $(cat "$p/start") $(cat "$p/size"); done; true').decode() + return {int(number): (int(start), int(size)) for number, start, size in (line.split() for line in text.splitlines())} + + +def inspect_lwvm(connection, disk): + """An LwVM disk's layout: the partitions the kernel added, with the type and identity of their LwVM entries. + + The kernel's LwVM reader exposes each partition's place (sysfs) but not its type + or GUID, so the host reads the header itself; both copies must agree. Scattered + partitions, which the kernel skips, are not offered, and a partition whose place + differs between the kernel and the header stops the install. + """ + name = Path(disk).name + sector = int(shell(connection, f'cat /sys/class/block/{name}/queue/logical_block_size')) + media = int(shell(connection, f'cat /sys/class/block/{name}/size')) * lwvm.SECTOR + chunk = lwvm.chunk_size(media) + try: + snapshot = lwvm.parse(read_disk(connection, disk, 0, lwvm.HEADER_BYTES, lwvm.HEADER_BYTES), + read_disk(connection, disk, chunk - lwvm.HEADER_BYTES, lwvm.HEADER_BYTES, + lwvm.HEADER_BYTES), media) + except lwvm.Error as error: + raise D.InstallError('lwvm', f'{disk}: {error}') from error + found = kernel_partitions(connection, disk) + partitions = [] + for part in snapshot['partitions']: + if part['number'] not in found: + continue + start, size = found[part['number']] + extents = part['extents'] + if (len(extents) != 1 or extents[0]['physical'] != start * lwvm.SECTOR or + part['bytes'] != size * lwvm.SECTOR): + raise D.InstallError('lwvm', f'{disk}: partition {part["number"]} lies elsewhere than its LwVM entry says') + row = {'number': part['number'], 'start': start * lwvm.SECTOR // sector, + 'blocks': size * lwvm.SECTOR // sector, 'type': part['type'], 'partuuid': part['partuuid'], + 'name': part['name'], 'attributes': part['attributes']} + row.update(probe(shell(connection, f'dd if={D.node(disk, part["number"])} bs=4096 count=1 2>/dev/null || true'))) + partitions.append(row) + info = {'path': disk, 'sector': sector, 'table': 'lwvm', 'table_sha256': snapshot['header_sha256'], + 'media_bytes': media, 'chunk_bytes': chunk} + return {'disk': info, 'partitions': sorted(partitions, key=lambda p: p['start']), 'free': []} + + +def lwvm_target(connection, plan, disk): + """The node of the planned LwVM partition, after checking the live disk against the plan. + + Nothing is written to the table: the root replaces the partition's contents only. + """ + layout = inspect_lwvm(connection, disk) + if layout['disk']['table_sha256'] != plan['disk']['table_sha256'] or layout['disk']['sector'] != plan['disk']['sector']: + raise D.InstallError('layout-changed', 'The LwVM table changed since it was inspected; inspect again') + root = plan['root'] + match = [p for p in layout['partitions'] if p['number'] == root['number']] + if (not match or (match[0]['partuuid'], match[0]['start'], match[0]['blocks']) != + (root['partuuid'], root['start'], root['blocks']) or + match[0]['type'] != lwvm.LINUX_TYPE or match[0]['attributes']): + raise D.InstallError('plan-mismatch', 'The partition to replace is not the inspected Linux partition') + return D.node(disk, root['number']), layout['partitions'] + + def entry_bytes(kind, partuuid, start, blocks, name): return (uuid.UUID(kind).bytes_le + uuid.UUID(partuuid).bytes_le + struct.pack(' /run/orchard/done.part && mv /run/orchard/done.part /run/orchard/done") diff --git a/src/orchard/install_layout.py b/src/orchard/install_layout.py index 097b3851..ae59cc3d 100644 --- a/src/orchard/install_layout.py +++ b/src/orchard/install_layout.py @@ -8,6 +8,10 @@ - A target is either a free range or one existing Linux partition to replace. - The disk has at most one XBOOTLDR boot partition; it is created inside the target when missing, and an existing one is reused. +- On an LwVM disk (32-bit boards) the target is an existing Linux partition, + which is replaced in place, and there is no boot partition: the boot entry + stays in the root's /boot. The installer never changes the LwVM map; making + room for Linux is iOS's resize step. """ from dataclasses import dataclass from typing import Dict, List, Optional @@ -58,9 +62,16 @@ def boot_partition(layout) -> Optional[dict]: return boots[0] if boots else None +def lwvm(layout) -> bool: + """Whether the disk is partitioned with LwVM (32-bit boards): no boot partition, no new partitions.""" + return layout['disk'].get('table') == 'lwvm' + + def options(layout) -> List[Option]: """Every allowed target, in disk order.""" block = layout['disk']['sector'] + if lwvm(layout): + return lwvm_options(layout) boot = boot_partition(layout) needed = MIN_ROOT_BYTES + (0 if boot else BOOT_BYTES) align = MIB // block @@ -86,6 +97,21 @@ def options(layout) -> List[Option]: return sorted(result, key=lambda o: o.start) +def lwvm_options(layout) -> List[Option]: + """The Linux partitions of an LwVM disk that a root fits in; LwVM attributes (encryption) rule one out.""" + block = layout['disk']['sector'] + result = [] + for part in layout['partitions']: + if part['type'] != LINUX_DATA or part.get('attributes') or part['blocks'] * block < MIN_ROOT_BYTES: + continue + label = part.get('label') or part.get('name') or '' + fs = part.get('fs') or 'no filesystem' + result.append(Option('replace', part['start'], part['blocks'], + f'replace Linux partition {part["number"]} ({fs}{", " + label if label else ""}, ' + f'{part["blocks"] * block / GIB:.1f} GiB) - erases it', partition=part)) + return sorted(result, key=lambda o: o.start) + + def plan(layout, option: Option, image: Dict, entry: Dict, layer: Dict, firmware: Optional[Dict] = None, account: Optional[Dict] = None) -> Dict: """The plan `install_device.py run` executes for one chosen option. @@ -93,6 +119,8 @@ def plan(layout, option: Option, image: Dict, entry: Dict, layer: Dict, firmware IMAGE is the distro's root image, LAYER orchard's platform layer (kernel, DTBs, modules) that goes into the root after it. """ + if lwvm(layout): + return lwvm_plan(layout, option, image, entry, layer, firmware, account) block = layout['disk']['sector'] align = MIB // block boot = boot_partition(layout) @@ -132,3 +160,32 @@ def plan(layout, option: Option, image: Dict, entry: Dict, layer: Dict, firmware 'firmware': firmware, 'account': account, # the release's install.account; secrets travel separately } + + +def lwvm_plan(layout, option, image, entry, layer, firmware, account): + """The plan for an LwVM disk: the root fills the chosen Linux partition, which keeps its place and identity. + + No partition is created or removed and there is no boot partition: `boot` + is None, and the boot entry goes to the root's own /boot. + """ + block = layout['disk']['sector'] + part = option.partition + if option.kind != 'replace' or part is None or part['type'] != LINUX_DATA or part.get('attributes'): + raise LayoutError('On an LwVM disk only an existing, unencrypted Linux partition can be replaced') + if part['blocks'] * block < (image['expanded_bytes'] + layer['expanded_bytes']) * 1.25: + raise LayoutError('Target is too small for this system image') + return { + 'version': PLAN_VERSION, + 'disk': {'path': layout['disk']['path'], 'sector': block, 'table': 'lwvm', + 'table_sha256': layout['disk']['table_sha256']}, + 'remove': None, + 'boot': None, + 'root': {'number': part['number'], 'start': part['start'], 'blocks': part['blocks'], + 'partuuid': part['partuuid'], 'fs_uuid': str(uuid.uuid4()), 'fs': 'ext4', + 'label': entry['id'][:16]}, + 'image': image, + 'layer': layer, + 'entry': entry, + 'firmware': firmware, + 'account': account, + } diff --git a/src/orchard/launcher.py b/src/orchard/launcher.py new file mode 100644 index 00000000..7fcc5598 --- /dev/null +++ b/src/orchard/launcher.py @@ -0,0 +1,84 @@ +"""Hand a boot bundle to orchard-launcher, which takes the device from DFU to the bundle running. + +Boot methods orchard's Python does not implement itself (a patched iBSS and iBEC on 32-bit +boards: recovery-autogo) are the launcher's. This module is the whole seam, so its shape can +change in one place: + + orchard-launcher load BUNDLE --chain CHAIN --catalog DIR + +prints one JSON object per line: {"event": "step", ...} while it works, +{"event": "needs_user", "message": ...} when the person has to act (DFU, replug), and last +{"result": "started"} or {"result": "error", "reason": ..., "message": ...}. +""" +import json +import os +from pathlib import Path +import shutil +import subprocess + +NAME = 'orchard-launcher' +# Boot methods (methods.toml) that the launcher delivers. +METHODS = frozenset({'recovery-autogo'}) + + +class LauncherError(RuntimeError): + def __init__(self, reason, message): + super().__init__(message) + self.reason = reason + + +def find(): + """The launcher binary: ORCHARD_LAUNCHER, else orchard-launcher on PATH; None if neither.""" + configured = os.environ.get('ORCHARD_LAUNCHER') + if configured: + path = Path(configured).expanduser() + return path if path.is_file() and os.access(path, os.X_OK) else None + found = shutil.which(NAME) + return Path(found) if found else None + + +def delivers(method): + """Whether METHOD is started by the launcher rather than by orchard's own code.""" + return method in METHODS + + +def load(bundle, *, chain, catalog, progress): + """Start BUNDLE on the attached device through CHAIN; returns once the launcher says it started. + + Never retried: a failed start raises with the launcher's reason, and the + device is left as the launcher left it. + """ + executable = find() + if executable is None: + raise LauncherError('missing', f'Starting this device needs {NAME}, which is not installed. ' + f'Install it, or point ORCHARD_LAUNCHER at it, then run the command again.') + argv = [str(executable), 'load', str(bundle), '--chain', chain, '--catalog', str(catalog)] + result, other = None, [] + with subprocess.Popen(argv, stdin=subprocess.DEVNULL, stdout=subprocess.PIPE, stderr=subprocess.PIPE, + text=True, bufsize=1) as process: + try: + for line in process.stdout: + try: + message = json.loads(line) + except ValueError: + other.append(line.rstrip()) + continue + if not isinstance(message, dict): + continue + if message.get('event') == 'step': + progress.step(str(message.get('message') or message.get('step') or 'Starting')) + elif message.get('event') == 'needs_user': + progress.say(str(message.get('message', 'The device needs you'))) + elif 'result' in message: + result = message + except BaseException: + process.kill() + raise + errors = process.stderr.read() + if result is None: + detail = (errors.strip() or '\n'.join(other)).splitlines()[-5:] + raise LauncherError('no-result', f'{NAME} stopped without a result (exit {process.returncode})' + + (': ' + ' '.join(detail) if detail else '')) + if result['result'] != 'started': + raise LauncherError(str(result.get('reason', 'error')), str(result.get('message') or f'{NAME} could not start the device')) + return result diff --git a/src/orchard/lwvm.py b/src/orchard/lwvm.py new file mode 100644 index 00000000..524082c1 --- /dev/null +++ b/src/orchard/lwvm.py @@ -0,0 +1,108 @@ +"""Checked LwVM snapshots, from Shayne Legault's LwVM installer (shaynelegault/orchard#94). + +All offsets are bytes. UUIDs in LwVM descriptors use network byte order. +""" +import hashlib +import struct +import uuid +import zlib + +MAGIC = bytes.fromhex('6a9088cf8afd630ae351e24887e0b98b') +HEADER_BYTES = 4096 +LINUX_TYPE = '0fc63daf-8483-4772-8e79-3d69d8477de4' +HFS_TYPE = '48465300-0000-11aa-aa11-00306543ecac' +SECTOR = 512 + + +class Error(ValueError): + pass + + +def digest(data): + return hashlib.sha256(data).hexdigest() + + +def chunk_size(media_bytes): + if type(media_bytes) is not int or media_bytes < 8192 or media_bytes % HEADER_BYTES: + raise Error('LwVM medium size is unsupported') + shifted = (media_bytes - 1) >> 10 + chunk = 1 << shifted.bit_length() + if not shifted or shifted > 0xffffffff or chunk < 2 * HEADER_BYTES or media_bytes // chunk < 2: + raise Error('LwVM chunk geometry is unsupported') + return chunk + + +def checksum(header): + checked = bytearray(header) + checked[0x2c:0x30] = bytes(4) + return zlib.crc32(checked) + + +def parse(primary, backup, media_bytes=None): + if len(primary) != HEADER_BYTES or primary != backup or primary[:16] != MAGIC: + raise Error('LwVM header pair is incomplete or disagrees') + if checksum(primary) != struct.unpack_from('= end or end > size or begin % HEADER_BYTES or end % HEADER_BYTES: + raise Error('LwVM partition range is invalid') + if any(begin < p['end'] and end > p['begin'] for p in parts): + raise Error('LwVM virtual partition ranges overlap') + identities.add(str(identity)) + parts.append({'number': index + 1, 'type': str(kind), 'partuuid': str(identity), + 'begin': begin, 'end': end, 'bytes': end - begin, 'attributes': flags, + 'name': raw[56:128].decode('utf-16-le', 'strict').rstrip('\0'), + 'chunks': {}, 'extents': []}) + words = struct.unpack_from('<1024H', primary, 0x800) + for physical, word in enumerate(words): + owner, logical = word >> 12, word & 0xfff + if owner == 15: + if not ((physical == 0 and word == 0xf000) or + (0 < physical < size // chunk and word == 0xf111) or + (physical >= size // chunk and word == 0xf3ff)): + raise Error('LwVM has an unknown allocation sentinel') + continue + if owner >= count or physical == 0 or physical >= size // chunk: + raise Error('LwVM chunk owner or physical address is invalid') + owned = parts[owner]['chunks'] + if logical in owned: + raise Error('LwVM logical chunk is aliased') + owned[logical] = physical + for part in parts: + needed = (part['bytes'] + chunk - 1) // chunk + if set(part['chunks']) != set(range(needed)): + raise Error('LwVM partition chunk coverage is incomplete') + for logical, physical in sorted(part['chunks'].items()): + length = min(chunk, part['bytes'] - logical * chunk) + offset = physical * chunk + if offset + length > size: + raise Error('LwVM partition extent exceeds the medium') + extents = part['extents'] + if extents and extents[-1]['physical'] + extents[-1]['bytes'] == offset: + extents[-1]['bytes'] += length + else: + extents.append({'logical': logical * chunk, 'physical': offset, 'bytes': length}) + return {'media_bytes': size, 'chunk_bytes': chunk, 'backup_offset': chunk - HEADER_BYTES, + 'header_sha256': digest(primary), 'partitions': parts} + + +def partition(snapshot, identity): + identity = str(uuid.UUID(identity)) + found = [p for p in snapshot['partitions'] if p['partuuid'] == identity] + if len(found) != 1: + raise Error('LwVM partition identity was not found') + return found[0] diff --git a/src/orchard/releases.py b/src/orchard/releases.py index 9bbd5456..f604e89c 100644 --- a/src/orchard/releases.py +++ b/src/orchard/releases.py @@ -9,7 +9,8 @@ import tomllib from . import version -from .hardware import IMPLEMENTED_METHODS, device as hardware_device +from .hardware import IMPLEMENTED_METHODS, device as hardware_device, method_implemented +from .launcher import delivers as launcher_delivers from .cache import cache_root from .downloads import download from .http import open_url @@ -164,8 +165,9 @@ def select_manifest(source, profile, progress=None, clock=None, purpose='install raise ValueError('Manifest is not compatible with this device') if document['boot_method'] not in described.get('boot', {}).get('methods', []): raise ValueError(f'{profile} does not boot with {document["boot_method"]}') - if document['boot_method'] not in IMPLEMENTED_METHODS: - raise ValueError('This Orchard build implements only ' + ', '.join(sorted(IMPLEMENTED_METHODS))) + if not method_implemented(document['boot_method']): + raise ValueError('This Orchard build implements only ' + ', '.join(sorted(IMPLEMENTED_METHODS)) + + (f'; {document["boot_method"]} needs orchard-launcher' if launcher_delivers(document['boot_method']) else '')) if document['protocol'] != 1: raise ValueError('Unsupported device protocol; this manifest is for another Orchard version') actual = tuple(int(part) for part in version().split('.')) diff --git a/src/orchard/service.py b/src/orchard/service.py index dcedcd94..c9f6d46e 100644 --- a/src/orchard/service.py +++ b/src/orchard/service.py @@ -83,6 +83,9 @@ def caveats(adapter): names = [label for name, label in labels.items() if adapter.FEATURES.get(name) == state] if names: rows.append(f'- {title}: ' + ', '.join(names)) + if getattr(adapter, 'STORAGE', {}).get('reclaims') is False: + rows.append('- Storage: freed space is not reused yet. Everything written, this install included, ' + 'uses up blank flash for good; once it runs out, nothing more can be written.') lines = ['Caveats:', *rows] if rows else [] if adapter.STATUS != 'supported': lines.append('This device has not been tested yet, proceed at your own risk!') @@ -109,6 +112,9 @@ def ask_account(account, clock, secrets_fn, input_fn, progress): def choose_target(layout, clock, *, name='Linux', input_fn=input, output=print): """Pick one install target from the device's layout; replacing a partition is typed out.""" options = install_layout.options(layout) + if not options and install_layout.lwvm(layout): + raise ValueError(f'No unencrypted Linux partition of at least {install_layout.MIN_ROOT_BYTES / 1024**3:.0f} GiB. ' + 'The installer never resizes iOS; a Linux partition has to exist already.') if not options: needed = (install_layout.MIN_ROOT_BYTES + install_layout.BOOT_BYTES) / 1024**3 raise ValueError(f'No free space or Linux partition of at least {needed:.0f} GiB. ' @@ -123,7 +129,7 @@ def choose_target(layout, clock, *, name='Linux', input_fn=input, output=print): option = options[int(answer) - 1] if option.kind == 'replace': phrase = f'erase {option.partition["number"]}' - if option.partition['type'] == install_layout.LINUX_DATA: + if option.partition['type'] == install_layout.LINUX_DATA and not install_layout.lwvm(layout): # Installs from the previous installer were one Linux-data partition; this # installer writes a boot partition and an arm64 root instead. output('This looks like an install from the previous Orchard installer. It cannot be ' @@ -188,7 +194,7 @@ def install_direct(adapter, device, name, files, firmware_files, boot_files, cha from .state import DeviceState if device['resumed']: raise ValueError('This release installs without a recovery. Put the device in DFU mode, then run: orchard install') - disk = os.environ.get('ORCHARD_INSTALL_DISK', direct.DISK) + disk = os.environ.get('ORCHARD_INSTALL_DISK') interfaces = direct.link_local_interfaces() state = private_directory(DeviceState(adapter.PROFILE, device['ecid']).root) progress.section('Getting ready', keep_rows=True) @@ -199,7 +205,7 @@ def install_direct(adapter, device, name, files, firmware_files, boot_files, cha client, host_key, authorized, dropbear_host_key = direct.make_keys(directory) extra = direct.add_on(boot_assets['install_tools'], device['ecid'], authorized, dropbear_host_key) image_file = Path(directory) / 'payload.bin' - image_file.write_bytes(direct.boot_image(adapter, boot_assets, extra)) + image_file.write_bytes(direct.boot_image(adapter, boot_assets, extra, device)) with progress.alongside('Device'): found = direct.start(adapter, device, image_file, boot_assets, progress) progress.mark('Install mode on USB') @@ -207,7 +213,8 @@ def install_direct(adapter, device, name, files, firmware_files, boot_files, cha progress.step('Connecting over SSH') with direct.session(adapter.PROFILE, device, found, directory, client, host_key, interfaces) as connection: progress.mark('SSH connected') - layout = direct.inspect(connection, disk) + layout = direct.inspect(connection, disk or direct.find_disk(connection)) + disk = layout['disk']['path'] progress.mark('Disk read') progress.done() option = choose_target(layout, clock, name=name, input_fn=input_fn, output=progress.say) @@ -234,16 +241,21 @@ def install_direct(adapter, device, name, files, firmware_files, boot_files, cha **{key: layer[key] for key in ('bytes', 'sha256', 'expanded_bytes')}}, firmware={ 'path': remote.DIRECTORY + '/firmware.tar.gz', 'bytes': mixin_record['bytes'], 'sha256': mixin_record['sha256']}) - nodes, partitions = direct.write_partitions(connection, plan, disk) - progress.step('Partitions created') - root_node = nodes[plan['root']['partuuid']] - if plan['boot']['create']: - sector = layout['disk']['sector'] - for offset, data in direct.D.fat32_regions(plan['boot']['blocks'], sector, int(plan['boot']['fs_uuid'], 16)): - direct.write_at(connection, disk, sector, plan['boot']['start'] + offset // sector, bytes(data)) - boot_node = nodes[plan['boot']['partuuid']] + if plan['boot'] is None: + # LwVM: the Linux partition is reused as it is; the table is not written. + root_node, partitions = direct.lwvm_target(connection, plan, disk) + boot_node = None else: - boot_node = direct.D.node(disk, plan['boot']['number']) + nodes, partitions = direct.write_partitions(connection, plan, disk) + progress.step('Partitions created') + root_node = nodes[plan['root']['partuuid']] + if plan['boot']['create']: + sector = layout['disk']['sector'] + for offset, data in direct.D.fat32_regions(plan['boot']['blocks'], sector, int(plan['boot']['fs_uuid'], 16)): + direct.write_at(connection, disk, sector, plan['boot']['start'] + offset // sector, bytes(data)) + boot_node = nodes[plan['boot']['partuuid']] + else: + boot_node = direct.D.node(disk, plan['boot']['number']) ranges = json.loads(assets['map'].read_text())['ranges'] with assets['root'].open('rb') as source: timings = direct.stream_image(connection, root_node, image, source, @@ -264,14 +276,20 @@ def install_direct(adapter, device, name, files, firmware_files, boot_files, cha events = remote.run(inside, ['finish', remote.DIRECTORY + '/plan.json'], progress=progress) timeline = direct.shell(connection, 'cat /run/orchard/timeline').decode() progress.mark('Finished inside the root') - direct.unmount_new_root(connection, plan['root']['partuuid']) + direct.unmount_new_root(connection, plan['root']['partuuid'], boot=boot_node is not None) progress.mark('Unmounted, boot continues') report = {'device_timeline': timeline, 'profile': connection.scope.device.profile, 'ecid': connection.scope.device.ecid, 'boot_id': connection.scope.boot_id, 'installed': True, 'release': record['release'], - 'root': plan['root']['partuuid'], 'boot': plan['boot']['partuuid'], + 'root': plan['root']['partuuid'], 'boot': (plan['boot'] or {}).get('partuuid'), 'written': timings, 'events': events, 'timing': clock.result()} path = private_directory(connection.scope.root) / 'last-install.json' save_report(path, report) + if plan['boot'] is None and 'cascadia' in boot_assets: + # No boot partition yet: later starts need this bundle placed where iOS's boot chain finds it. + bundle = path.with_name(f'boot-{plan["root"]["partuuid"].split("-")[0]}.bin') + bundle.write_bytes(direct.installed_bundle(adapter, boot_assets, device, plan)) + report['boot_bundle'] = str(bundle) + save_report(path, report) progress.done('Installed') progress.section('Starting ' + name) if account and account['ssh_keys']: @@ -286,8 +304,10 @@ def install_direct(adapter, device, name, files, firmware_files, boot_files, cha progress.done() login = (f'\nSign in over SSH: ssh {account["user"]}@172.16.42.1' if report.get('ready') and account and account['ssh_keys'] else '') - progress.say(f'\n{name} is {"ready" if report.get("ready") else "starting"}.{login}\n' - 'To start it again later, put the device in DFU mode and run: orchard launch') + later = ('To start it again later, put the device in DFU mode and run: orchard launch' + if 'boot_bundle' not in report else + f'Starting it again later needs its boot bundle in place, which orchard cannot do yet: {report["boot_bundle"]}') + progress.say(f'\n{name} is {"ready" if report.get("ready") else "starting"}.{login}\n{later}') return report diff --git a/tests/test_armv7_direct.py b/tests/test_armv7_direct.py index bb5d5a72..aa83aa80 100644 --- a/tests/test_armv7_direct.py +++ b/tests/test_armv7_direct.py @@ -92,11 +92,29 @@ def test_64_bit_boards_get_exactly_m1n1s_payload(self): direct.start(adapter, {}, self.work / 'image', assets, None) adapter.boot_payload.assert_called_once_with({}, self.work / 'image', self.work / 'pongo', None) - def test_delivering_a_cascadia_bundle_is_not_orchards_yet(self): - with self.assertRaisesRegex(ValueError, 'patched iBEC'): + def test_a_cascadia_bundle_goes_to_orchard_launcher(self): + self.adapter.LAUNCHER_CHAIN = 'recovery-autogo' + with mock.patch.dict('os.environ', {'ORCHARD_LAUNCHER': str(self.work / 'absent')}), \ + self.assertRaisesRegex(RuntimeError, 'needs orchard-launcher, which is not installed'): direct.start(self.adapter, {}, self.work / 'image', self.assets, None) + with mock.patch('orchard.launcher.load') as load, \ + mock.patch('orchard.apple.boot.wait_mode', return_value={'boot_id': 'B'}) as wait: + self.assertEqual(direct.start(self.adapter, {'ecid': 1}, self.work / 'image', self.assets, None), {'boot_id': 'B'}) + self.assertEqual((load.call_args.args, load.call_args.kwargs['chain']), ((self.work / 'image',), 'recovery-autogo')) + self.assertEqual(wait.call_args.args[:2], ({'ecid': 1}, 'install')) self.adapter.boot_payload.assert_not_called() + def test_the_devices_serial_number_goes_into_the_tree(self): + image = direct.boot_image(self.adapter, self.assets, b'ADD-ON', {'serial_number': 'F4KESER1AL00'}) + config = cascadia.configuration(image) + root = cascadia.read_fdt(image[config['dtb_offset']:config['dtb_offset'] + config['dtb_size']])[0] + self.assertEqual(root['properties']['serial-number'], b'F4KESER1AL00\0') + self.assertEqual(image[config['initrd_offset']:], b'INITRD\0\0ADD-ON') + # No serial known: the tree is the board's, byte for byte. + plain = direct.boot_image(self.adapter, self.assets, b'ADD-ON', {}) + self.assertEqual(plain, direct.boot_image(self.adapter, self.assets, b'ADD-ON')) + self.assertNotIn(b'serial-number', plain) + class Release(unittest.TestCase): def test_a_32_bit_release_names_its_boards_loaders_instead_of_m1n1(self): diff --git a/tests/test_armv7_install.py b/tests/test_armv7_install.py new file mode 100644 index 00000000..4aafeabe --- /dev/null +++ b/tests/test_armv7_install.py @@ -0,0 +1,504 @@ +"""Installing on a 32-bit board: the iOS disk's LwVM, a root in an existing Linux partition, no boot partition. + +Synthetic disks and invented identities only; no device is touched. The 64-bit +path is checked next to it, to show that it did not change. +""" +import io +import json +from pathlib import Path +import re +import struct +import subprocess +import sys +import tempfile +import unittest +from unittest import mock +import uuid +import zlib + +from orchard import install_device as D +from orchard import install_direct as direct +from orchard import install_layout as L +from orchard import launcher, lwvm, service +from orchard.apple import linux_usb +from orchard.hardware import ROOT + +sys.path.insert(0, str(Path(__file__).resolve().parent)) + +MEDIA = 64_000_000_000 # the 64 GB model's disk +CHUNK = lwvm.chunk_size(MEDIA) # 64 MiB +HFS = uuid.UUID(lwvm.HFS_TYPE).bytes +LINUX = uuid.UUID(lwvm.LINUX_TYPE).bytes +SYSTEM_ID, DATA_ID, LINUX_ID = ('11111111-2222-4333-8444-555555555555', '22222222-3333-4444-8555-666666666666', + '33333333-4444-4555-8666-777777777777') +# (name, type, GUID, attributes, physical chunks in logical order) +PARTS = (('System', HFS, SYSTEM_ID, 0, list(range(1, 21))), + ('Data', HFS, DATA_ID, 0x3000000000000, list(range(21, 801))), + ('Linux', LINUX, LINUX_ID, 0, list(range(801, 953)))) + + +def header(parts=PARTS): + """An LwVM header for PARTS, chunks given per partition in logical order.""" + h = bytearray(lwvm.HEADER_BYTES) + h[:16] = lwvm.MAGIC + h[16:32] = uuid.UUID('aaaaaaaa-bbbb-4ccc-8ddd-eeeeeeeeeeee').bytes + struct.pack_into('/dev/null \|\| true', command) + if match: + data = bytearray(4096) + if int(match.group(1)) in self.ext4: + data[1080:1082] = b'\x53\xef' + data[1144:1150] = b'linux' + return bytes(data) + raise AssertionError('unexpected command: ' + command) + + def read_disk(self, connection, disk, offset, size, block): + self.commands.append(f'read {disk} {offset}') + if offset == 0: + return self.primary + if offset == CHUNK - lwvm.HEADER_BYTES: + return self.backup + raise AssertionError(f'unexpected read at {offset}') + + def __enter__(self): + self.patches = [mock.patch.object(direct, 'shell', self.shell), mock.patch.object(direct, 'read_disk', self.read_disk)] + for patch in self.patches: + patch.start() + return self + + def __exit__(self, *_): + for patch in self.patches: + patch.stop() + + +class DiskDiscovery(unittest.TestCase): + def test_the_one_internal_disk_is_found_nvme_or_the_ios_flash(self): + for names, expected in ((['loop0', 'nvme0n1', 'zram0'], '/dev/nvme0n1'), (['applesftl0', 'mtdblock0'], '/dev/applesftl0')): + with self.subTest(names=names), FakeDevice(disks=names): + self.assertEqual(direct.find_disk(None), expected) + + def test_two_internal_disks_stop_the_install(self): + with FakeDevice(disks=['nvme0n1', 'applesftl0']), self.assertRaisesRegex(D.InstallError, 'More than one'): + direct.find_disk(None) + + def test_it_waits_for_storage_then_gives_up(self): + with FakeDevice(disks=[]), mock.patch.object(direct.time, 'sleep'), \ + mock.patch.object(direct.time, 'monotonic', side_effect=[0, 1, 31]), \ + self.assertRaisesRegex(ValueError, 'did not come up within 30 s'): + direct.find_disk(None) + + +class LwvmLayout(unittest.TestCase): + def test_the_layout_is_the_kernels_partitions_with_their_lwvm_identity(self): + with FakeDevice(ext4={3}) as device: + layout = direct.inspect(None, '/dev/applesftl0') + self.assertEqual(layout['disk']['table'], 'lwvm') + self.assertEqual(layout['disk']['sector'], 512) + self.assertEqual(layout['free'], []) + rows = {p['number']: p for p in layout['partitions']} + self.assertEqual(sorted(rows), [1, 2, 3]) + self.assertEqual((rows[3]['partuuid'], rows[3]['type'], rows[3]['name'], rows[3]['fs']), + (LINUX_ID, lwvm.LINUX_TYPE, 'Linux', 'ext4')) + self.assertEqual((rows[3]['start'], rows[3]['blocks']), (801 * CHUNK // 512, 152 * CHUNK // 512)) + self.assertTrue(rows[2]['attributes']) + # Only reads: dd from the disk and partitions, cat of sysfs. + self.assertFalse([c for c in device.commands if 'of=' in c or 'blockdev' in c or '>' in c.replace('2>/dev/null', '')]) + + def test_only_an_unencrypted_linux_partition_is_offered_and_kept_in_place(self): + with FakeDevice() as _: + layout = direct.inspect(None, '/dev/applesftl0') + (option,) = L.options(layout) + self.assertEqual((option.kind, option.partition['number'], option.creates_boot), ('replace', 3, False)) + self.assertIn('replace Linux partition 3', option.description) + plan = L.plan(layout, option, {'bytes': 1, 'sha256': 'x', 'expanded_bytes': 1 << 30}, + {'id': 'omarchy', 'title': 'Omarchy', 'options': 'rootwait', 'devicetree': 's5l8942x-p105.dtb'}, + {'path': 'p', 'bytes': 1, 'sha256': 'y', 'expanded_bytes': 1 << 20}) + self.assertIsNone(plan['boot']) + self.assertIsNone(plan['remove']) + self.assertEqual(plan['disk']['table'], 'lwvm') + self.assertEqual((plan['root']['number'], plan['root']['partuuid'], plan['root']['start'], plan['root']['blocks']), + (3, LINUX_ID, 801 * CHUNK // 512, 152 * CHUNK // 512)) + encrypted = json.loads(json.dumps(layout)) + encrypted['partitions'][2]['attributes'] = 1 + self.assertEqual(L.options(encrypted), []) + with self.assertRaisesRegex(ValueError, 'never resizes iOS'): + service.choose_target(encrypted, service.Clock(), input_fn=lambda _: '1', output=lambda *_: None) + + def test_replacing_it_must_be_typed_out(self): + with FakeDevice(): + layout = direct.inspect(None, '/dev/applesftl0') + said = [] + answers = iter(['1', 'erase 3']) + option = service.choose_target(layout, service.Clock(), input_fn=lambda _: next(answers), output=said.append) + self.assertEqual(option.partition['number'], 3) + self.assertFalse(any('previous Orchard installer' in line for line in said)) + answers = iter(['1', 'yes']) + with self.assertRaisesRegex(ValueError, 'Not confirmed'): + service.choose_target(layout, service.Clock(), input_fn=lambda _: next(answers), output=said.append) + + def test_a_scattered_partition_the_kernel_skipped_is_not_offered(self): + parts = (*PARTS[:2], ('Linux', LINUX, LINUX_ID, 0, [*range(850, 953), *range(801, 850)])) + with FakeDevice(primary=header(parts), kernel=kernel_view(parts)): + layout = direct.inspect(None, '/dev/applesftl0') + self.assertEqual([p['number'] for p in layout['partitions']], [1, 2]) + + def test_headers_that_disagree_or_a_kernel_that_disagrees_stop_the_install(self): + other = bytearray(header()) + other[0x200 + 2 * 0x80 + 56:0x200 + 2 * 0x80 + 66] = 'Other'.encode('utf-16-le') + with FakeDevice(backup=bytes(other)), self.assertRaisesRegex(D.InstallError, 'disagrees'): + direct.inspect(None, '/dev/applesftl0') + moved = {**kernel_view(), 3: (800 * CHUNK // 512, 152 * CHUNK // 512)} + with FakeDevice(kernel=moved), self.assertRaisesRegex(D.InstallError, 'lies elsewhere'): + direct.inspect(None, '/dev/applesftl0') + + def test_the_live_disk_is_checked_against_the_plan_before_writing(self): + with FakeDevice(): + layout = direct.inspect(None, '/dev/applesftl0') + plan = L.plan(layout, L.options(layout)[0], {'bytes': 1, 'sha256': 'x', 'expanded_bytes': 1}, + {'id': 'omarchy', 'title': 'O', 'devicetree': 'd'}, {'bytes': 1, 'sha256': 'y', 'expanded_bytes': 1}) + node, partitions = direct.lwvm_target(None, plan, '/dev/applesftl0') + self.assertEqual(node, '/dev/applesftl0p3') + self.assertEqual(len(partitions), 3) + changed = dict(plan, disk=dict(plan['disk'], table_sha256='0' * 64)) + with FakeDevice(), self.assertRaisesRegex(D.InstallError, 'changed since it was inspected'): + direct.lwvm_target(None, changed, '/dev/applesftl0') + elsewhere = dict(plan, root=dict(plan['root'], partuuid=DATA_ID)) + with FakeDevice(), self.assertRaisesRegex(D.InstallError, 'not the inspected Linux partition'): + direct.lwvm_target(None, elsewhere, '/dev/applesftl0') + + def test_a_gpt_disk_is_inspected_as_before(self): + with mock.patch.object(direct, 'read_disk', return_value=bytes(4096)), \ + mock.patch.object(direct, 'read_table', side_effect=RuntimeError('GPT path')) as table: + with self.assertRaisesRegex(RuntimeError, 'GPT path'): + direct.inspect(None, '/dev/nvme0n1') + table.assert_called_once_with(None, '/dev/nvme0n1') + + +class RootLookup(unittest.TestCase): + """orchard's initrd finds a root on the partitions the kernel's LwVM reader added.""" + INIT = ROOT / 'boot/initramfs/root/init' + + def function(self, name): + return re.search(rf'^{name}\(\) \{{.*?^\}}', self.INIT.read_text(), re.M | re.S).group(0) + + def lookup(self, image, number, sector=512): + """The init's own lookup line, GPT first, then LwVM.""" + script = (self.function('gpt_partuuid') + '\n' + self.function('lwvm_partuuid') + '\n' + + 'id=$(gpt_partuuid "$0" "$1" "$2") || id=$(lwvm_partuuid "$0" "$2"); echo "$id"') + result = subprocess.run(['sh', '-c', script, str(image), str(sector), str(number)], capture_output=True, text=True) + return result.stdout.strip() or None + + def test_every_lwvm_entry_reads_as_the_partuuid_the_kernel_gives_it(self): + with tempfile.TemporaryDirectory() as directory: + image = Path(directory) / 'lwvm' + image.write_bytes(header() + bytes(8192)) + self.assertEqual([self.lookup(image, n) for n in (1, 2, 3, 4)], [SYSTEM_ID, DATA_ID, LINUX_ID, None]) + + def test_gpt_disks_read_exactly_as_before(self): + from test_root_initrd import gpt + guids = [str(uuid.uuid4()) for _ in range(2)] + with tempfile.TemporaryDirectory() as directory: + for sector in (512, 4096): + image = Path(directory) / f'gpt-{sector}' + gpt(image, sector, [guids[0], None, guids[1]]) + with self.subTest(sector=sector): + self.assertEqual([self.lookup(image, n, sector) for n in (1, 2, 3)], [guids[0], None, guids[1]]) + blank = Path(directory) / 'blank' + blank.write_bytes(bytes(16384)) + self.assertIsNone(self.lookup(blank, 1, 4096)) + + def test_the_init_tries_gpt_first(self): + text = self.INIT.read_text() + self.assertIn('id=$(gpt_partuuid "/dev/$disk" "$sector" "$(cat "$number")") ||\n' + ' id=$(lwvm_partuuid "/dev/$disk" "$(cat "$number")")', text) + + +class Finish(unittest.TestCase): + """Inside the new root without a boot partition: the entry stays in the root's /boot.""" + def run_finish(self, kernel, boot): + work = tempfile.TemporaryDirectory() + self.addCleanup(work.cleanup) + root = Path(work.name) / 'root' + source = root / 'usr/lib/modules/6.12' + (source / 'dtb').mkdir(parents=True) + (source / 'vmlinuz').write_bytes(kernel) + (source / 'initrd').write_bytes(b'initrd') + (source / 'dtb/s5l8942x-p105.dtb').write_bytes(b'dtb') + (root / 'boot').mkdir() + (root / 'etc').mkdir() + layer = Path(work.name) / 'layer.tar.gz' + layer.write_bytes(b'') + plan = {'layer': {'path': str(layer), 'bytes': 0, 'sha256': '', 'expanded_bytes': 0}, 'firmware': None, + 'account': None, 'boot': boot, + 'root': {'partuuid': LINUX_ID, 'fs_uuid': '5f0c2a1e-0000-4000-8000-000000000001'}, + 'entry': {'id': 'omarchy', 'title': 'Omarchy', 'options': 'rootwait', 'devicetree': 's5l8942x-p105.dtb'}} + with mock.patch.object(D, 'install_layer', return_value=0), mock.patch.object(D, 'emit'): + D.finish(plan, root, {SYSTEM_ID}, None) + return root + + def test_the_root_names_itself_by_filesystem_and_keeps_the_entry_in_boot(self): + from test_cascadia_loader import kernel + root = self.run_finish(kernel(), None) + self.assertEqual((root / 'etc/fstab').read_text(), 'UUID=5f0c2a1e-0000-4000-8000-000000000001 / ext4 rw,relatime 0 1\n') + conf = (root / 'boot/loader/entries/omarchy-33333333-6.12.conf').read_text() + self.assertIn(f'options root=PARTUUID={LINUX_ID} rw rootwait\n', conf) + self.assertIn('architecture arm\n', conf) + self.assertEqual((root / 'boot/omarchy-33333333/6.12/s5l8942x-p105.dtb').read_bytes(), b'dtb') + + def test_64_bit_entries_and_fstab_are_as_before(self): + root = self.run_finish(b'KERNEL', {'partuuid': '44444444-0000-4000-8000-000000000001'}) + self.assertEqual((root / 'etc/fstab').read_text(), + f'PARTUUID={LINUX_ID} / ext4 rw,relatime 0 1\n' + 'PARTUUID=44444444-0000-4000-8000-000000000001 /boot vfat rw,relatime,fmask=0077,dmask=0077 0 2\n') + self.assertIn('architecture aa64\n', (root / 'boot/loader/entries/omarchy-33333333-6.12.conf').read_text()) + + +class InstalledBundle(unittest.TestCase): + def test_the_installed_system_starts_from_its_root_with_the_boards_own_arguments(self): + board = ('earlycon=s3c6400,0x32500000 lpj=5000000 root=/dev/ram0 rw rootwait panic=10 ' + 'g_cdc.host_addr=02:10:5a:05:00:01 g_cdc.dev_addr=02:10:5a:05:00:02') + self.assertEqual(direct.installed_bootargs(board, LINUX_ID, 'rootwait fbcon=map:9'), + f'earlycon=s3c6400,0x32500000 lpj=5000000 rootwait panic=10 root=PARTUUID={LINUX_ID} rw rootwait fbcon=map:9') + + def test_the_bundle_carries_bootargs_and_serial_and_no_add_on(self): + from test_armv7_direct import P105, armv7_layer + from test_cascadia_loader import loader, tree + from orchard.image import cascadia + with tempfile.TemporaryDirectory() as directory: + work = Path(directory) + armv7_layer(work / 'layer', cascadia.write_fdt(tree())) + (work / 'loader').write_bytes(loader(P105)) + adapter = mock.Mock(PROFILE='p105', DEVICETREE='s5l8942x-p105.dtb') + plan = {'root': {'partuuid': LINUX_ID}, 'entry': {'options': 'rootwait'}} + image = direct.installed_bundle(adapter, {'cascadia': work / 'loader', 'layer': work / 'layer'}, + {'serial_number': 'F4KESER1AL00'}, plan) + config = cascadia.configuration(image) + root = cascadia.read_fdt(image[config['dtb_offset']:config['dtb_offset'] + config['dtb_size']])[0] + self.assertEqual(root['children'][0]['properties']['bootargs'], f'old root=PARTUUID={LINUX_ID} rw rootwait\0'.encode()) + self.assertEqual(root['properties']['serial-number'], b'F4KESER1AL00\0') + self.assertEqual(image[config['initrd_offset']:], b'INITRD\0\0') # the initrd alone, padded + + def test_a_ram_payload_gets_the_serial_number_just_before_the_boot(self): + from test_armv7_direct import P105 + from test_cascadia_loader import kernel, loader, tree + from orchard.image import cascadia + image, _ = cascadia.assemble(loader(P105), kernel(), cascadia.write_fdt(tree()), b'RAMDISK', P105) + patched = cascadia.with_serial(image, 'F4KESER1AL00') + config = cascadia.configuration(patched) + root = cascadia.read_fdt(patched[config['dtb_offset']:config['dtb_offset'] + config['dtb_size']])[0] + self.assertEqual(root['properties']['serial-number'], b'F4KESER1AL00\0') + self.assertEqual(patched[config['initrd_offset']:], b'RAMDISK') + self.assertEqual(patched[:0x8000][:4], image[:4]) + with self.assertRaises(ValueError): + cascadia.with_serial(image, 'not a serial;') + + +class Launcher(unittest.TestCase): + def setUp(self): + work = tempfile.TemporaryDirectory() + self.addCleanup(work.cleanup) + self.work = Path(work.name) + + def fake(self, lines, code=0): + path = self.work / 'orchard-launcher' + path.write_text('#!/bin/sh\necho "$@" > "$(dirname "$0")/argv"\n' + + ''.join(f"echo '{line}'\n" for line in lines) + f'exit {code}\n') + path.chmod(0o755) + return path + + def load(self, path): + progress = mock.Mock() + with mock.patch.dict('os.environ', {'ORCHARD_LAUNCHER': str(path)}): + result = launcher.load(self.work / 'bundle.bin', chain='recovery-autogo', catalog=self.work / 'catalog', + progress=progress) + return result, progress + + def test_steps_and_requests_are_shown_and_started_returns(self): + path = self.fake(['{"event":"step","step":"ibss","message":"Sending iBSS"}', + '{"event":"needs_user","message":"Put the iPad in DFU mode"}', 'plain text', '{"result":"started"}']) + result, progress = self.load(path) + self.assertEqual(result, {'result': 'started'}) + progress.step.assert_called_once_with('Sending iBSS') + progress.say.assert_called_once_with('Put the iPad in DFU mode') + self.assertEqual((self.work / 'argv').read_text().split(), + ['load', str(self.work / 'bundle.bin'), '--chain', 'recovery-autogo', '--catalog', str(self.work / 'catalog')]) + + def test_an_error_result_or_no_result_raises_with_the_reason(self): + with self.assertRaisesRegex(launcher.LauncherError, 'iBEC was refused') as caught: + self.load(self.fake(['{"result":"error","reason":"ibec","message":"iBEC was refused"}'], code=1)) + self.assertEqual(caught.exception.reason, 'ibec') + with self.assertRaisesRegex(launcher.LauncherError, 'without a result'): + self.load(self.fake(['{"event":"step","step":"dfu"}'], code=3)) + + def test_a_missing_launcher_says_so(self): + with self.assertRaisesRegex(launcher.LauncherError, 'not installed'): + self.load(self.work / 'absent') + + +class BoardData(unittest.TestCase): + def test_p105_starts_through_the_launcher_and_warns_about_its_storage(self): + from orchard import devices, hardware + adapter = devices.load('p105') + self.assertEqual(adapter.LAUNCHER_CHAIN, 'recovery-autogo') + self.assertEqual(adapter.boot_inputs(), []) + caveats = '\n'.join(service.caveats(adapter)) + self.assertIn('freed space is not reused yet', caveats) + with mock.patch.dict('os.environ', {'ORCHARD_LAUNCHER': str(Path(tempfile.gettempdir()) / 'absent-launcher')}): + self.assertIn('it starts through orchard-launcher, which is not installed', hardware.install_blockers('p105')) + with tempfile.NamedTemporaryFile() as binary: + Path(binary.name).chmod(0o755) + with mock.patch.dict('os.environ', {'ORCHARD_LAUNCHER': binary.name}): + self.assertEqual(hardware.install_blockers('p105'), []) + + def test_64_bit_boards_are_unchanged(self): + from orchard import devices + adapter = devices.load('j71b') + self.assertIsNone(adapter.LAUNCHER_CHAIN) + self.assertEqual([item.name for item in adapter.boot_inputs()], ['palera1n']) + self.assertFalse(any('freed space' in line for line in service.caveats(adapter))) + + +class RamBoot(unittest.TestCase): + def test_orchard_run_hands_a_cascadia_payload_with_the_serial_to_the_launcher(self): + from orchard import boot + with tempfile.TemporaryDirectory() as directory: + output = Path(directory) + (output / 'linux.bin').write_bytes(b'PAYLOAD') + device = {'mode': 'dfu', 'chip': 0x8942, 'board': 0x0a, 'ecid': 1, 'location': 1, 'serial_number': 'F4KESER1AL00'} + sent = [] + with mock.patch.object(boot, 'verify_ram', return_value={'boards': ['p105'], 'loader_format': 'cascadia', + 'boot_method': 'recovery-autogo'}), \ + mock.patch('orchard.apple.usb.sole_target', return_value=device), \ + mock.patch('orchard.image.cascadia.with_serial', side_effect=lambda image, serial: image + serial.encode()), \ + mock.patch('orchard.launcher.load', side_effect=lambda image, **kw: sent.append((image.read_bytes(), kw['chain']))), \ + mock.patch('orchard.apple.boot.start_pongo') as pongo, mock.patch('builtins.print'): + boot.boot_ram(output, yes=True) + self.assertEqual(sent, [(b'PAYLOADF4KESER1AL00', 'recovery-autogo')]) + pongo.assert_not_called() + + +class UsbListing(unittest.TestCase): + def test_dfu_tags_give_the_serial_number_and_the_exploit(self): + serial = 'CPID:8942 CPRV:15 CPFM:03 SCEP:10 BDID:0A ECID:000012345678ABCD IBFL:00 SRNM:[F4KESER1AL00] PWND:[checkm8]' + self.assertEqual(linux_usb.serial_tags(serial), {'serial_number': 'F4KESER1AL00', 'pwned': 'checkm8'}) + self.assertEqual(linux_usb.serial_tags('CPID:8010 ECID:1'), {'serial_number': None, 'pwned': None}) + self.assertEqual(linux_usb.serial_tags('CPID:8010 YOLO:checkra1n'), {'serial_number': None, 'pwned': 'checkra1n'}) + + +class Flow(unittest.TestCase): + """service.install_direct end to end with the device mocked: LwVM writes the root only, GPT as before.""" + def run_flow(self, layout, plan_boot): + from orchard.progress import Clock, Progress + work = tempfile.TemporaryDirectory() + self.addCleanup(work.cleanup) + work = Path(work.name) + for name in ('root', 'layer', 'map'): + (work / name).write_bytes(b'{"ranges": [[0, 1]]}' if name == 'map' else b'x') + record = {'release': '1', 'name': 'Omarchy', 'assets': {'root': {'bytes': 1, 'sha256': 'r', 'expanded_bytes': 1}, + 'layer': {'bytes': 1, 'sha256': 'l', 'expanded_bytes': 1}}} + assets = {'root': work / 'root', 'layer': work / 'layer', 'map': work / 'map'} + boot_assets = {'install_tools': work / 'tools', 'layer': work / 'layer', + **({'cascadia': work / 'loader'} if plan_boot is None else {'m1n1': work / 'm1n1', 'pongo': work / 'pongo'})} + done = lambda value: mock.Mock(done=lambda: True, result=lambda: value) + adapter = mock.Mock(PROFILE='p105' if plan_boot is None else 'j71b', DEVICETREE='tree.dtb', NAME='Tablet') + adapter.prepare_mixin.return_value = (work / 'layer', {'bytes': 1, 'sha256': 'f'}) + connection = mock.Mock(argv=['ssh'], scope=mock.Mock(boot_id='B', root=work / 'state', + device=mock.Mock(profile='p', ecid=1))) + session = mock.MagicMock() + session.return_value.__enter__.return_value = connection + calls = dict(make_keys=mock.Mock(return_value=('c', 'h', 'a', b'k')), + add_on=mock.Mock(return_value=b''), boot_image=mock.Mock(return_value=b'IMG'), + start=mock.Mock(return_value={'boot_id': 'B'}), session=session, + find_disk=mock.Mock(return_value=layout['disk']['path']), + inspect=mock.Mock(return_value=layout), + lwvm_target=mock.Mock(return_value=('/dev/applesftl0p3', layout['partitions'])), + write_partitions=mock.Mock(side_effect=lambda c, plan, d: ({plan['root']['partuuid']: '/dev/nvme0n1p5', + plan['boot']['partuuid']: '/dev/nvme0n1p4'}, [])), + write_at=mock.Mock(), stream_image=mock.Mock(return_value={}), grow=mock.Mock(), + mount_new_root=mock.Mock(), unmount_new_root=mock.Mock(), + shell=mock.Mock(return_value=b''), installed_bundle=mock.Mock(return_value=b'BUNDLE'), + link_local_interfaces=mock.Mock(return_value=set())) + with mock.patch.multiple(direct, **calls), \ + mock.patch.object(service.remote, 'put'), mock.patch.object(service.remote, 'run', return_value=[]), \ + mock.patch.object(service, 'private_directory', side_effect=lambda p: Path(p).mkdir(parents=True, exist_ok=True) or Path(p)), \ + mock.patch.dict('os.environ', {}, clear=False): + answers = iter(['1', 'erase 3' if plan_boot is None else '']) + report = service.install_direct(adapter, {'resumed': False, 'ecid': 1, 'serial_number': 'F4KESER1AL00'}, 'Omarchy', + done((record, assets)), done({}), done(({'account': None}, boot_assets)), [], + None, {'ssh_keys': []}, clock=Clock(), input_fn=lambda _: next(answers), + secrets_fn=None, progress=Progress(io.StringIO())) + return report, calls + + def test_lwvm_writes_the_root_only_and_keeps_a_boot_bundle_on_the_host(self): + with FakeDevice(): + layout = direct.inspect(None, '/dev/applesftl0') + report, calls = self.run_flow(layout, None) + calls['write_partitions'].assert_not_called() + calls['write_at'].assert_not_called() + calls['lwvm_target'].assert_called_once() + self.assertEqual(calls['stream_image'].call_args.args[1], '/dev/applesftl0p3') + self.assertEqual(calls['mount_new_root'].call_args.args[1:], ('/dev/applesftl0p3', None)) + self.assertEqual(calls['unmount_new_root'].call_args.kwargs, {'boot': False}) + self.assertEqual(calls['unmount_new_root'].call_args.args[1], LINUX_ID) + self.assertEqual((report['root'], report['boot']), (LINUX_ID, None)) + self.assertEqual(Path(report['boot_bundle']).read_bytes(), b'BUNDLE') + # The device's serial number reaches the install payload's tree. + self.assertEqual(calls['boot_image'].call_args.args[3]['serial_number'], 'F4KESER1AL00') + + def test_gpt_installs_as_before(self): + layout = {'disk': {'path': '/dev/nvme0n1', 'sector': 4096, 'first': 6, 'last': 1 << 22, 'table_sha256': 'h'}, + 'partitions': [], 'free': [{'start': 256, 'blocks': 1 << 22}]} + report, calls = self.run_flow(layout, True) + calls['lwvm_target'].assert_not_called() + calls['write_partitions'].assert_called_once() + self.assertTrue(calls['write_at'].called) # the new FAT32 boot partition + self.assertEqual(calls['mount_new_root'].call_args.args[1:], ('/dev/nvme0n1p5', '/dev/nvme0n1p4')) + self.assertEqual(calls['unmount_new_root'].call_args.kwargs, {'boot': True}) + self.assertNotIn('boot_bundle', report) + self.assertIsNotNone(report['boot']) + + +if __name__ == '__main__': + unittest.main()