From 029b55b66bf2089c769fdfaaa9e861c9d5a6b289 Mon Sep 17 00:00:00 2001 From: Shayne Legault <8741557+shaynelegault@users.noreply.github.com> Date: Sat, 7 Nov 2026 22:25:12 +0800 Subject: [PATCH 1/3] install: read checked LwVM headers on the host The header parser of Shayne Legault's LwVM installer: both header copies must agree, carry a valid CRC and describe a sane chunk map. Only the reading part is taken (chunk_size, checksum, parse, partition); the HFS shrink proof and the allocation stay out until the resize step is placed. From shaynelegault/orchard#94, src/orchard/lwvm.py at b5c804a. Assisted-by: LLM --- src/orchard/lwvm.py | 108 ++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 108 insertions(+) create mode 100644 src/orchard/lwvm.py 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] From a0c4a82e5e30e00574e1378cb4a432e4c527bc4a Mon Sep 17 00:00:00 2001 From: Henrik Klee Date: Sat, 7 Nov 2026 22:38:16 +0800 Subject: [PATCH 2/3] install: install on 32-bit boards into an existing Linux LwVM partition The recovery-less install now also runs on the iPad mini 1 (P105) side, ready for when a patched iBEC can be delivered: - The add-on finds the one internal disk (NVMe on 64-bit boards, applesftlN on 32-bit ones) instead of assuming /dev/nvme0n1. - On an LwVM disk the layout is the partitions the kernel's LwVM reader added, with type and GUID from the header (both copies must agree). Only an existing, unencrypted Linux partition is offered, replaced in place after a typed confirmation; the LwVM map is never written, and there is no boot partition: the BLS entry stays in the root's /boot (architecture arm), fstab names the root by filesystem UUID, and the host keeps the Cascadia bundle that starts the installed system. - orchard's initrd finds a root on LwVM partitions by their LwVM GUID when a disk has no GPT; GPT disks are looked up as before. - The Cascadia bundle carries the device's serial number (DFU SRNM) as the tree's serial-number, as m1n1 does on 64-bit boards, so the Wi-Fi address is stable per unit; `orchard devices` shows it and the PWND:[...] tag. - Starting a Cascadia bundle (install or `orchard run`) goes through orchard-launcher (src/orchard/launcher.py); a board that needs it cannot install until it is installed, and says so. - P105's device data records that its storage does not reclaim space yet; the installer warns that the disk fills up for good. Assisted-by: LLM --- boot/initramfs/root/init | 17 +- devices/p105/README.md | 9 +- devices/p105/device.toml | 7 + docs/boot-contract.md | 15 + schemas/v1/device.schema.json | 9 + src/orchard/apple/device.py | 10 +- src/orchard/apple/linux_usb.py | 17 +- src/orchard/apple/usb.py | 4 +- src/orchard/boot.py | 23 ++ src/orchard/contracts.py | 3 +- src/orchard/hardware.py | 22 +- src/orchard/image/cascadia.py | 69 ++++- src/orchard/install_device.py | 27 +- src/orchard/install_direct.py | 150 ++++++++-- src/orchard/install_layout.py | 57 ++++ src/orchard/launcher.py | 84 ++++++ src/orchard/releases.py | 8 +- src/orchard/service.py | 54 ++-- tests/test_armv7_direct.py | 22 +- tests/test_armv7_install.py | 503 +++++++++++++++++++++++++++++++++ 20 files changed, 1046 insertions(+), 64 deletions(-) create mode 100644 src/orchard/launcher.py create mode 100644 tests/test_armv7_install.py 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 cde705f6..e073614f 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 +294,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 +303,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/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..9b6a2fd1 --- /dev/null +++ b/tests/test_armv7_install.py @@ -0,0 +1,503 @@ +"""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' + 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() From 782104d146bce58b670670a29f6d25c9d9cc64c5 Mon Sep 17 00:00:00 2001 From: Henrik Klee Date: Sat, 7 Nov 2026 23:00:23 +0800 Subject: [PATCH 3/3] install: no CDC gadget arguments in the installed 32-bit bundle The apple-armv7 kernel has the built-in CDC gadget off now (a066863); orchard's initrd binds its own configfs gadget. Assisted-by: LLM --- src/orchard/install_direct.py | 8 ++++++-- tests/test_armv7_install.py | 3 ++- 2 files changed, 8 insertions(+), 3 deletions(-) diff --git a/src/orchard/install_direct.py b/src/orchard/install_direct.py index 1ad5fda3..36f2c2b8 100644 --- a/src/orchard/install_direct.py +++ b/src/orchard/install_direct.py @@ -185,8 +185,12 @@ def boot_image(adapter, assets, extra, device=None): def installed_bootargs(board, partuuid, options): - """A 32-bit installed system's boot arguments: the board tree's own, with its root replaced by the entry's.""" - kept = [word for word in board.split() if not word.startswith(('root=', 'rdinit=')) and word not in ('rw', 'ro')] + """A 32-bit installed system's boot arguments: the board tree's own, with its root replaced by the entry's. + + The old built-in CDC gadget's arguments go too: the kernel has it off, and orchard's initrd + binds its own configfs gadget (USB serial and network). + """ + kept = [word for word in board.split() if not word.startswith(('root=', 'rdinit=', 'g_cdc.')) and word not in ('rw', 'ro')] return ' '.join([*kept, f'root=PARTUUID={partuuid}', 'rw', *options.split()]) diff --git a/tests/test_armv7_install.py b/tests/test_armv7_install.py index 9b6a2fd1..4aafeabe 100644 --- a/tests/test_armv7_install.py +++ b/tests/test_armv7_install.py @@ -297,7 +297,8 @@ def test_64_bit_entries_and_fstab_are_as_before(self): 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' + 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')