From 2e11638b0a1921c6c6ce5bb99d4c96209932f911 Mon Sep 17 00:00:00 2001 From: Shayne Legault <8741557+shaynelegault@users.noreply.github.com> Date: Thu, 29 Oct 2026 10:18:10 -0600 Subject: [PATCH] Add iPhone 6 RAM boot and USB NFS host link --- NOTICE | 4 + docs/n61-nfs-link.md | 114 ++++++++++ docs/ram-development.md | 16 ++ src/orchard_cli/apple/boot.py | 11 +- src/orchard_cli/apple/usb.py | 2 + src/orchard_cli/apple/z2fw.py | 61 +++++- src/orchard_cli/boot.py | 7 +- src/orchard_cli/build.py | 68 +++++- src/orchard_cli/command.py | 22 ++ src/orchard_cli/firmware.py | 17 +- src/orchard_cli/kernel.py | 101 ++++++++- src/orchard_cli/nfs_host.py | 313 ++++++++++++++++++++++++++++ src/orchard_cli/provenance.py | 57 +++++ src/orchard_cli/ram.py | 117 ++++++++++- src/orchard_cli/recipe.py | 12 ++ src/orchard_cli/transports/pongo.py | 10 +- tests/test_adapter_contract.py | 25 +++ tests/test_device_state.py | 42 ++++ tests/test_kernel_receipts.py | 176 ++++++++++++++++ tests/test_nfs_host.py | 126 +++++++++++ tests/test_package_boundaries.py | 2 +- tests/test_pongo_preflight.py | 33 +++ tests/test_provenance.py | 47 +++++ tests/test_ram.py | 105 +++++++++- tests/test_runtime.py | 179 ++++++++++++++-- tests/test_shared_build_cache.py | 268 +++++++++++++++++++++++- tests/test_z2fw.py | 77 +++++++ 27 files changed, 1947 insertions(+), 65 deletions(-) create mode 100644 docs/n61-nfs-link.md create mode 100644 src/orchard_cli/nfs_host.py create mode 100644 src/orchard_cli/provenance.py create mode 100644 tests/test_kernel_receipts.py create mode 100644 tests/test_nfs_host.py create mode 100644 tests/test_pongo_preflight.py create mode 100644 tests/test_provenance.py diff --git a/NOTICE b/NOTICE index 1410926..33f8732 100644 --- a/NOTICE +++ b/NOTICE @@ -6,3 +6,7 @@ framing are informed by Corellium's Project Sandcastle and Paul Praschl's Gen2 mtprops and MTFW HBPP item conversions (mtfw.c, makez2fw.c): https://github.com/Pauli1Go/HoolockLinux-linux-firmware Reference revision: 944541e67b4c603df328c77d30093551de802c17. +convert_gen1() ports the legacy Gen1 bootstrap from Corellium LLC's +GPL-2.0-or-later hx-touchd/mtfw/mtfw.c (Copyright (C) 2020 Corellium LLC): +https://github.com/corellium/projectsandcastle +Reference revision: 03db9c6ae04141eb940f3b9f56d446f50d57fadf. diff --git a/docs/n61-nfs-link.md b/docs/n61-nfs-link.md new file mode 100644 index 0000000..dd66a1a --- /dev/null +++ b/docs/n61-nfs-link.md @@ -0,0 +1,114 @@ +# Experimental iPhone 6 NFS host link + +The N61 RAM experiment uses a USB network link to a private NFS root. Its host +address is `10.61.0.1/30`; the phone uses `10.61.0.2`. This command manages the +macOS host address for that link. The NFS server and any forwarding process +remain separate host resources. + +Stage the RAM runtime with `--handoff manual` and the exact `nfs-root.json` +from the private NFS root. If the root is on a Linux server, copy that identity +file to the Mac without changing its bytes and compare the SHA-256 values. Use +the Mac copy for every host-link command in this session. + +With the iPhone in DFU, run `orchard devices` and verify its ECID, chip `0x7000` +(28672) and board `0x06` (6) against the selected phone. Set `N61_LOCATION` in +the Mac shell to the reported numeric USB location. Boot the matching RAM image +with `orchard run /path/to/staged-ram-directory` and retain the same cable and +location through the transition. Stop to inspect the connection if the location +changes or another boot device appears. + +Set `N61_ROOT_IDENTITY` to the Mac copy of `nfs-root.json` and +`N61_LINK_STATE` to a new private receipt path. Inspect the resulting gadget: + +```sh +orchard nfs-link inspect --root-identity "$N61_ROOT_IDENTITY" --location "$N61_LOCATION" +``` + +The result includes `endpoint.serial_port` and `endpoint.boot_id`. Set +`N61_SERIAL_PORT` to the reported port and open it from the Mac: + +```sh +screen "$N61_SERIAL_PORT" 115200 +``` + +Run the following command in the phone's rescue shell: + +```sh +cat /proc/sys/kernel/random/boot_id +``` + +Compare the result with `endpoint.boot_id`. Keep the serial session open and +use a second Mac terminal for host commands. Set `N61_BOOT_ID` there to the +matching UUID, along with `N61_LOCATION`, `N61_ROOT_IDENTITY` and +`N61_LINK_STATE` from the first terminal. Then configure the host address: + +```sh +orchard nfs-link apply --root-identity "$N61_ROOT_IDENTITY" \ + --location "$N61_LOCATION" --boot-id "$N61_BOOT_ID" --state "$N61_LINK_STATE" +``` + +The command verifies the root, boot UUID, USB location, serial port and network +interface before requesting administrator access. It rejects conflicting routes, +an unowned existing address and ambiguous boot-device discovery. If address +setup fails after the intent is recorded, inspect the host and the receipt before +another attempt. + +Start the NFS server and any required forwarding process for this session. +For the qualified network settings, the phone reaches the configured NFSv4.0 +export through `10.61.0.1:32061`. For a server bound to loopback on a separate +Linux host, one SSH terminal can forward the port after the host address exists: + +```sh +ssh -N -o ExitOnForwardFailure=yes \ + -L 10.61.0.1:32061:127.0.0.1:32061 "$NFS_SERVER_SSH" +``` + +Set `NFS_SERVER_SSH` to the server's SSH destination and add its port and key +options when needed. Verify the listening address or forward reaches +the intended server and that the export exposes the staged root with the same +identity bytes and root ownership. Keep the server and any forward running +throughout the Linux session. + +Recheck the selected gadget and the owned host address using the same receipt: + +```sh +orchard nfs-link inspect --root-identity "$N61_ROOT_IDENTITY" \ + --location "$N61_LOCATION" --boot-id "$N61_BOOT_ID" --state "$N61_LINK_STATE" +``` + +The result should report `already-owned`. On the checked serial shell, compare +`/proc/sys/kernel/random/boot_id` with `N61_BOOT_ID` again. When the link and +export are ready, request handoff once: + +```sh +[ ! -e /run/orchard/nfs-handoff-attempted ] && + [ ! -e /run/orchard/nfs-handoff ] && + touch /run/orchard/nfs-handoff +``` + +The runtime makes one mount attempt and keeps the RAM rescue shell available +if handoff fails. Inspect the failure and device state before another boot. + +After shutting down the Linux session and disconnecting its NFS gadget, retire +the address owned by that receipt: + +```sh +orchard nfs-link cleanup --root-identity "$N61_ROOT_IDENTITY" --state "$N61_LINK_STATE" +``` + +Cleanup refuses to run while any Apple `05ac:4144` experimental gadget remains +connected, including the RAM-only shell. It checks the recorded interface and +hardware address before removing an alias. If the interface has disappeared, +the receipt is retired without changing another interface. The NFS server, +forwarding processes and other host exports remain under their owners' control. + +This gadget advertises the profile, root UUID and boot UUID. It does not yet +expose the device ECID or implement Orchard's serial-pinned SSH bootstrap and +device protocol. The operator must retain the verified DFU location throughout +the transition. To qualify full recovery, the RAM runtime needs to parse the +chosen `unique-chip-id` from the live ADT and bind it to the selected ECID, +root UUID and kernel boot UUID. It must provision an ephemeral client key over +the checked serial endpoint, return its boot-scoped SSH host key over that same +endpoint, and answer the shared recovery status request through the pinned SSH +session. The host must check the same device and boot at each transition. +Treat this link as a bring-up interface until those checks pass on the phone. diff --git a/docs/ram-development.md b/docs/ram-development.md index 72f1045..806597d 100644 --- a/docs/ram-development.md +++ b/docs/ram-development.md @@ -134,6 +134,22 @@ base producers remain reusable, including the charging-worker cache. A changed base producer does not inherit that compatibility. Cache reuse still checks the complete file inventory and hashes. +Loader workspaces are separated by device, the complete m1n1 recipe and the build +environment identity. The source tree recorded after the initial build identifies +that recipe's baseline. Tracked and untracked source edits select their own cached +loader without changing the checkout's Git index. A workspace without a recorded +baseline uses its source tree identity as well. Loader builds sharing a workspace +wait for one another, even when they use different component caches. If a nested +submodule has local changes, build the loader explicitly and pass its path with +`--loader`. + +The RAM build receipt records the CLI and hardware package content hashes, plus +the Git head and dirty state when those packages come from tracked checkouts. It +records the kernel component receipt and a copy of the device firmware recipe +when those are available. Explicit older kernel inputs without a receipt retain +their image and selected DTB hashes. The receipt states whether the firmware +recipe supplied files to the RAM image. + Developer receipts are not publisher signatures or hardware qualification. Explicit component paths are trusted inputs: callers must keep kernel, DTB and modules coherent. Never automatically retry uncertain boot or storage mutations. diff --git a/src/orchard_cli/apple/boot.py b/src/orchard_cli/apple/boot.py index f288588..8e4dcdd 100644 --- a/src/orchard_cli/apple/boot.py +++ b/src/orchard_cli/apple/boot.py @@ -57,11 +57,14 @@ def run_process_group(argv, *, timeout, **kwargs): with subprocess.Popen(argv, start_new_session=True, **kwargs) as process: try: returncode = process.wait(timeout=timeout) + if returncode: + raise subprocess.CalledProcessError(returncode, argv) except BaseException: - os.killpg(process.pid, signal.SIGKILL) + try: + os.killpg(process.pid, signal.SIGKILL) + except ProcessLookupError: + pass raise - if returncode: - raise subprocess.CalledProcessError(returncode, argv) def boot(device, assets, manifest, progress, *, profile, chip, board, product): @@ -73,6 +76,8 @@ def boot(device, assets, manifest, progress, *, profile, chip, board, product): if not executable: raise ValueError('palera1n is required on the host to boot Pongo') usb.selected(device['location'], device['ecid'], chip=chip, board=board, mode='dfu') + pongo.preflight() + if device['mode'] == 'dfu': progress.phase('Booting Pongo') progress.phase('Wait for the Apple logo to appear. If it does not disappear after a few seconds, ' 'disconnect and reconnect the cable at the iPad once.') diff --git a/src/orchard_cli/apple/usb.py b/src/orchard_cli/apple/usb.py index 6ca3bc6..e75a4a0 100644 --- a/src/orchard_cli/apple/usb.py +++ b/src/orchard_cli/apple/usb.py @@ -40,6 +40,8 @@ def devices(): ports = [] while descendants: child = descendants.pop() + if child.get('IOObjectClass') == 'IOUSBHostDevice': + continue descendants.extend(child.get('IORegistryEntryChildren', [])) if child.get('IOObjectClass') == 'IOSerialBSDClient': port = child.get('IOCalloutDevice') diff --git a/src/orchard_cli/apple/z2fw.py b/src/orchard_cli/apple/z2fw.py index 6888c48..3278bef 100644 --- a/src/orchard_cli/apple/z2fw.py +++ b/src/orchard_cli/apple/z2fw.py @@ -5,13 +5,15 @@ The file is the same for every unit of a model: apple_z2 fills each calibration record from the DT properties m1n1 copies from the ADT at boot. -Three input forms: +Input forms: - convert(): an mtprops profile whose "Constructed Firmware" is one preconstructed HBPP14 packet (J71b). - convert_packets(): an mtprops profile whose "Constructed Firmware" is an array of preconstructed packets (iPhone 7 Plus); makez2fw's Gen2 path. - convert_mtfw(): an MTFW image (IM4P payload) listing HBPP load items per personality (iPad 7); makez2fw's HBPP item path. +- convert_gen1(): a legacy Z2 mtprops data packet (iPhone 6), with the + Gen1 bootstrap from Corellium Project Sandcastle. convert_packets() and convert_mtfw() are ports of Paul Praschl's (Pauli1Go) GPL-2.0-or-later makez2fw (mtfw.c, makez2fw.c), which builds on Corellium's @@ -36,7 +38,8 @@ PROVIDERS = {'multi-touch-calibration': 0, 'orb-gap-cal': 1, 'orb-force-cal': 2, 'shape-dynamic-accel-cal': 3, 'multitouch-to-display-offset': 4, 'firefly-calibration': 5, 'pressure-effect-correction': 6, - 'firefly-p100-calibration': 7, 'ext-multi-touch-calibration': 8} + 'firefly-p100-calibration': 7, 'ext-multi-touch-calibration': 8, + 'prox-calibration': 9} def uint(value, name, maximum=0xffffffff): @@ -68,21 +71,21 @@ def from_xe32(data): return lo | hi << 16 -def decode_blob(packet): +def decode_blob(packet, *, hbpp_version=0x104): if len(packet) < 20 or packet[:4] != b'\x18\xe1\x30\x01': - raise ValueError('Not a prefixed HBPP14 blob') - words_minus_one = struct.unpack_from('>H', packet, 4)[0] - size = (words_minus_one + 1) * 4 + raise ValueError('Not a prefixed HBPP blob') + words = struct.unpack_from('>H', packet, 4)[0] + size = (words + (hbpp_version >= 0x104)) * 4 if len(packet) != 16 + size: - raise ValueError('HBPP14 length mismatch or extra trailing packets') + raise ValueError('HBPP length mismatch or extra trailing packets') if sum(packet[4:10]) != struct.unpack_from('>H', packet, 10)[0]: - raise ValueError('HBPP14 header checksum mismatch') + raise ValueError('HBPP header checksum mismatch') payload = swapped(packet[12:-4]) if sum(payload) & 0xffffffff != from_xe32(packet[-4:]): - raise ValueError('HBPP14 payload checksum mismatch') + raise ValueError('HBPP payload checksum mismatch') address = from_xe32(packet[6:10]) if address + size > 0x100000000: - raise ValueError('HBPP14 address range overflow') + raise ValueError('HBPP address range overflow') return address, payload @@ -307,6 +310,44 @@ def container(records): return output +def convert_gen1(profile, *, packet, touch_maximum, prox_maximum, fll_mval, ref_clk_div, clk32_clock_enable): + """Legacy Z2 data-form mtprops with dynamic per-unit calibration. + + Corellium projectsandcastle 03db9c6ae04141eb940f3b9f56d446f50d57fadf, + hx-touchd/mtfw/mtfw.c:169-219 defines this Gen1 command order. INIT + preserves its unacknowledged exchanges; SEND requests a separate ACK. + Calibration records replace its SysCfg loads at the same positions. + The N61 packet uses full payload-word counts, before HBPP 0x104. + Board recipes supply the clock values from Apple's matching kernel + personality; the legacy mtprops packet contains no Firmware Config. + """ + if packet != 'Z2' or profile.get('PreconstructedBootloadPacketType') != packet: + raise ValueError('Unexpected Gen1 packet type') + payload = profile.get('Constructed Firmware') + if not isinstance(payload, bytes): + raise ValueError('Expected data-form Gen1 constructed firmware') + if 'Firmware Config' in profile: + raise ValueError('Gen1 firmware with an unsupported config') + address, firmware = decode_blob(payload, hbpp_version=0x103) + if address != 0: + raise ValueError('Unexpected Gen1 firmware load address') + providers = [] + records = [(INIT, b'\x19\xc1')] + [(INIT, HBPP_INIT + b'\x18\xe1' * 6)] * 3 + records.append(calibration_record('prox-calibration', 0x10009600, prox_maximum, providers)) + records.append(calibration_record('multi-touch-calibration', 0x10009000, touch_maximum, providers)) + records.append((SEND, payload)) + boot = [(0x10003060, uint(fll_mval, 'FLL multiplier')), + (0x1000305c, uint(ref_clk_div, 'reference clock divider')), (0x10003058, 6), + (0x10003000, 3), (0x10003518, uint(clk32_clock_enable, '32K clock enable'))] + records += [(SEND, register_packet(address, 0xffffffff, value)) for address, value in boot] + records += [(SEND, REQUEST_CALIBRATION), (INIT, bytes.fromhex('1d5334001000000100000045'))] + return container(records), {'packet': packet, 'generation': 1, + 'firmware_packet_preserved_verbatim': True, + 'firmware_packet_bytes': len(payload), 'firmware_payload_bytes': len(firmware), + 'firmware_address': address, 'calibration': providers, + 'register_operations': len(boot), 'command_sequence': [c for c, p in records]} + + def convert_packets(profile, config, *, packet): """mtprops whose constructed firmware is an array of preconstructed packets. diff --git a/src/orchard_cli/boot.py b/src/orchard_cli/boot.py index 19a0e55..cecd4b8 100644 --- a/src/orchard_cli/boot.py +++ b/src/orchard_cli/boot.py @@ -26,7 +26,7 @@ def boot_ram(output, *, yes=False): """Boot once via existing USB/Pongo transport. No recovery API required.""" from orchard_hardware import device as hardware_device from .apple import usb - from .apple.boot import wait_mode + from .apple.boot import run_process_group, wait_mode from .state import operation_lock from .transports import pongo @@ -64,11 +64,12 @@ def boot_ram(output, *, yes=False): confirm_boot(mismatch=mismatch) verify_ram(output) usb.selected(device['location'], ecid, chip=chip, board=board, mode=device['mode']) + pongo.preflight() if device['mode'] == 'dfu': print('Starting Pongo once. Inspect any uncertain failure before retrying.') with (output / 'palera1n.log').open('wb') as log: - subprocess.run([executable, '-p', '-k', str(output / 'pongo.bin')], - check=True, timeout=120, stdout=log, stderr=subprocess.STDOUT) + run_process_group([executable, '-p', '-k', str(output / 'pongo.bin')], + timeout=120, stdout=log, stderr=subprocess.STDOUT) wait_mode(device, 'pongo', chip=chip, board=board, product='', seconds=60) usb.selected(device['location'], ecid, chip=chip, board=board) pongo.upload(output / 'linux.bin', device['location'], receipt['files']['linux.bin']) diff --git a/src/orchard_cli/build.py b/src/orchard_cli/build.py index daf3f92..3014e57 100644 --- a/src/orchard_cli/build.py +++ b/src/orchard_cli/build.py @@ -24,6 +24,7 @@ from . import sources, kernel as kernels _BASE_PRODUCER = kernels.build_base +_DEVICE_PRODUCER = kernels.build_device def configured_device(directory): @@ -93,11 +94,13 @@ def kernel_base_key(hardware, device, cache): 'kernel/upstream.lock', 'kernel/configs/' + platform, 'kernel/series/' + platform, 'kernel/series.py', 'kernel/patches', 'kernel/build', 'kernel/Containerfile', 'kernel/install-toolchain', - 'kernel/stage-modules.py', 'buildenv.py', + 'kernel/stage-modules.py', 'kernel/write-receipt.py', 'kernel/sources.toml', + '__init__.py', 'buildenv.py', ]) # Source preparation is already represented by its resulting tree. Hash # only the binary/SDK producer, not unrelated functions in kernel.py. - producer = hashlib.sha256(inspect.getsource(_BASE_PRODUCER).encode()).hexdigest() + producer = hashlib.sha256(''.join(inspect.getsource(item) for item in + (_BASE_PRODUCER, kernels._verify_producer, kernels._tree_digest, kernels._record)).encode()).hexdigest() sdk = digest(Path(__file__).with_name('kernel_sdk.py')) mode = os.environ.get('ORCHARD_BUILD_NATIVE', '0') + ':' + os.environ.get('ORCHARD_BUILD_ENV_ID', '') return hashlib.sha256((recipes + producer + sdk + tree + mode).encode()).hexdigest() @@ -105,9 +108,14 @@ def kernel_base_key(hardware, device, cache): def kernel_key(hardware, device, base_key): """Device layer: DTB selection and external drivers, reusing the compiled base.""" - recipes = recipe_key(hardware, ['devices/' + device + '/device.toml', 'kernel/modules/' + device, + platform = kernels.layout(hardware, device)[0] + recipes = recipe_key(hardware, ['devices/' + device, 'kernel/modules/' + device, + 'kernel/modules/' + platform, 'kernel/write-receipt.py', + 'kernel/sources.toml', '__init__.py', 'kernel/build-modules', 'kernel/stage-modules.py']) - return hashlib.sha256((base_key + recipes).encode()).hexdigest() + producer = hashlib.sha256(''.join(inspect.getsource(item) for item in + (_DEVICE_PRODUCER, kernels._verify_producer, kernels._tree_digest, kernels._record)).encode()).hexdigest() + return hashlib.sha256((base_key + recipes + producer).encode()).hexdigest() _HELD_LOCKS = set() @@ -230,6 +238,39 @@ def has_loader_patches(hardware): return any((Path(hardware) / 'boot/m1n1/patches').rglob('*.patch')) +def _loader_source_tree(source): + """Snapshot local edits without changing the checkout or its index.""" + source = Path(source) + if not (source / '.git').exists(): + raise ValueError(f'Loader source is not a Git checkout: {source}; it was preserved') + dirty = subprocess.check_output(['git', '-C', str(source), 'submodule', 'foreach', '--quiet', + '--recursive', 'git status --porcelain --untracked-files=all'], + stderr=subprocess.PIPE) + if dirty.strip(): + raise ValueError('Loader submodule has local changes; build it explicitly and supply --loader') + return sources.working_tree(source) + + +def _loader_recipe(hardware, device): + recipe = recipe_key(hardware, ['boot/m1n1', 'buildenv.py']) + mode = os.environ.get('ORCHARD_BUILD_NATIVE', '0') + ':' + os.environ.get('ORCHARD_BUILD_ENV_ID', '') + # Older entries were keyed only by recipes, even for edited workspaces. + recipe = hashlib.sha256(('loader-v2:' + recipe + ':' + mode).encode()).hexdigest() + return recipe, hardware / '.work/m1n1' / device / recipe + + +def _loader_key(hardware, device): + recipe, work = _loader_recipe(hardware, device) + source = work / 'source' + if not source.exists(): + return recipe + tree = _loader_source_tree(source) + baseline = work / 'orchard-cli-source-tree' + if baseline.is_file() and baseline.read_text().strip() == tree: + return recipe + return hashlib.sha256((recipe + ':source:' + tree).encode()).hexdigest() + + # What the static tools are built from; the rest of the recovery directory is # recovery runtime and doesn't change them. TOOL_INPUTS = ['boot/initramfs/recovery/' + name for name in @@ -294,12 +335,23 @@ def produce_pongo(stage): busybox = _prepare_tools(hardware, components) / 'busybox' if loader is None: def produce_loader(stage): - argv = [hardware / 'boot/m1n1/build', device, '--output', stage / 'm1n1.bin'] - if (hardware / 'boot/m1n1/.work' / device).exists(): + _, work = _loader_recipe(hardware, device) + source = work / 'source' + before = _loader_source_tree(source) if source.exists() else None + argv = [hardware / 'boot/m1n1/build', device, + '--work-dir', work, '--output', stage / 'm1n1.bin'] + if source.exists(): argv.append('--resume') command(*argv) - directory = cached_component(components, 'm1n1-' + device, - lambda: recipe_key(hardware, ['boot/m1n1', 'buildenv.py']), produce_loader) + after = _loader_source_tree(source) + if before is not None and before != after: + raise ValueError('Loader source changed during build; source was preserved and result was not cached') + if before is None: + (work / 'orchard-cli-source-tree').write_text(after + '\n') + recipe, _ = _loader_recipe(hardware, device) + with component_lock(hardware / '.work/m1n1-locks' / device / recipe): + directory = cached_component(components, 'm1n1-' + device, + lambda: _loader_key(hardware, device), produce_loader) loader = directory / 'm1n1.bin' return {name: Path(path).resolve() for name, path in dict(kernel=kernel, busybox=busybox, loader=loader, pongo=pongo).items()} diff --git a/src/orchard_cli/command.py b/src/orchard_cli/command.py index 5a68b71..cf4c5f2 100644 --- a/src/orchard_cli/command.py +++ b/src/orchard_cli/command.py @@ -55,6 +55,12 @@ def main(): check.add_argument('--boot', action='store_true', help='resolve only boot assets') commands.add_parser('devices', help='list USB devices without modifying them') commands.add_parser('setup-host', help='install missing boot tools and Linux USB permissions; no device access') + nfs_link = commands.add_parser('nfs-link', help='inspect or manage an experimental N61 USB NFS host link') + nfs_link.add_argument('operation', choices=('inspect', 'apply', 'cleanup')) + nfs_link.add_argument('--root-identity', type=Path, required=True, help='private staged NFS root identity') + nfs_link.add_argument('--location', type=lambda value: int(value, 0), help='USB location selected during DFU') + nfs_link.add_argument('--boot-id', help='boot UUID observed during inspection') + nfs_link.add_argument('--state', type=Path, help='private ownership receipt for this session') boot = commands.add_parser('boot', help='boot recovery from a manifest or explicit image') boot.add_argument('source', nargs='?', help='registered distro, local TOML or exact HTTPS URL') boot.add_argument('--manifest', '--release', dest='manifest') @@ -121,6 +127,22 @@ def main(): elif args.action == 'setup-host': from orchard_cli.host_setup import setup result = setup(progress) + elif args.action == 'nfs-link': + from orchard_cli import nfs_host + if args.operation == 'cleanup': + if args.state is None: + raise ValueError('NFS link cleanup requires --state') + result = nfs_host.cleanup(args.root_identity, args.state) + else: + if args.location is None: + raise ValueError('NFS link inspection requires --location') + if args.operation == 'apply': + if args.boot_id is None or args.state is None: + raise ValueError('NFS link setup requires --boot-id and --state') + result = nfs_host.apply(args.root_identity, args.location, args.boot_id, args.state) + else: + result = nfs_host.inspect(args.root_identity, args.location, + expected_boot=args.boot_id, state=args.state) elif args.action == 'boot': adapter, device = choose() if args.image: diff --git a/src/orchard_cli/firmware.py b/src/orchard_cli/firmware.py index 4d63dc3..12edcc0 100644 --- a/src/orchard_cli/firmware.py +++ b/src/orchard_cli/firmware.py @@ -78,15 +78,11 @@ def pack_mixin(prepared, metadata, outdir): return archive, record -def recipe_for(profile, hardware=None): - """A device's firmware recipe, validated and supported. - - From the installed hardware package, or from the hardware checkout at - HARDWARE (a development build against another branch). - """ +def recipe_path_for(profile, hardware=None): + """Find a device's public recipe in the installed package or a checkout.""" if hardware is None: from orchard_hardware import firmware_path - return recipes.check(json.loads(firmware_path(profile).read_text())) + return firmware_path(profile) from .build import tomllib if not re.fullmatch(r'[a-z0-9][a-z0-9_-]*', profile): raise ValueError('Invalid device profile') @@ -99,7 +95,12 @@ def recipe_for(profile, hardware=None): path = descriptor.parent.joinpath(*parts) if any(p.is_symlink() for p in (path, *path.parents) if descriptor.parent in (p, *p.parents)): raise ValueError('Firmware recipe path contains a symlink') - return recipes.check(json.loads(path.read_text())) + return path + + +def recipe_for(profile, hardware=None): + """A device's firmware recipe, validated and supported.""" + return recipes.check(json.loads(recipe_path_for(profile, hardware).read_text())) def ensure(profile, clock, progress, *, allow_download=True, hardware=None): diff --git a/src/orchard_cli/kernel.py b/src/orchard_cli/kernel.py index f64531f..efb8c36 100644 --- a/src/orchard_cli/kernel.py +++ b/src/orchard_cli/kernel.py @@ -7,6 +7,7 @@ """ from contextlib import contextmanager import fcntl +import hashlib import json import os from pathlib import Path @@ -180,6 +181,58 @@ def prepare_source(hardware, cache, device): return source, work / 'output' +def _record(path): + path = Path(path) + if path.is_symlink() or not path.is_file(): + raise ValueError('Expected a regular kernel component file: ' + str(path)) + digest = hashlib.sha256() + with path.open('rb') as stream: + for chunk in iter(lambda: stream.read(1024 * 1024), b''): + digest.update(chunk) + return {'bytes': path.stat().st_size, 'sha256': digest.hexdigest()} + + +def _tree_digest(root): + """Bind a cached SDK or in-tree module set, including executable modes.""" + root = Path(root) + entries = {} + for path in sorted(root.rglob('*')): + if path.is_symlink(): + raise ValueError('Linked cached kernel input: ' + str(path)) + if path.is_file(): + entries[path.relative_to(root).as_posix()] = {**_record(path), 'mode': path.stat().st_mode & 0o777} + elif not path.is_dir(): + raise ValueError('Unsupported cached kernel input: ' + str(path)) + return hashlib.sha256(json.dumps(entries, sort_keys=True, separators=(',', ':')).encode()).hexdigest() + + +def _verify_producer(hardware, device, directory, receipt, tree): + """Check the shared image and every platform DTB against hardware's receipt.""" + platform, _ = layout(hardware, device) + expected_source = receipt.get('source', {}) + work = work_directory(hardware, device) + config = Path(hardware) / 'kernel/configs' / platform / 'config' + if (receipt.get('schema') != 1 or receipt.get('platform') != platform or + expected_source.get('commit') != pin(hardware)['commit'] or + expected_source.get('tree') != tree or + expected_source.get('prepared_tree') != (work / 'prepared-tree').read_text().strip() or + expected_source.get('patch_list_sha256') != _record(work / 'applied-patches')['sha256'] or + expected_source.get('config_sha256') != _record(config)['sha256']): + raise ValueError('Cached kernel producer source or platform differs') + files = receipt.get('files') + if not isinstance(files, dict): + raise ValueError('Cached kernel producer inventory is missing') + dtbs = {p.relative_to(directory).as_posix() for p in (Path(directory) / 'dtbs').rglob('*') if p.is_file()} + expected_dtbs = {name for name in files if name.startswith('dtbs/')} + if not dtbs or dtbs != expected_dtbs: + raise ValueError('Cached kernel producer DTB inventory differs') + for name in {'Image.gz', 'kernel.release'} | dtbs: + if files.get(name) != _record(Path(directory) / name): + raise ValueError('Cached kernel producer component differs: ' + name) + if receipt.get('kernel_release') != (Path(directory) / 'kernel.release').read_text().strip(): + raise ValueError('Cached kernel producer release differs') + + def build_base(hardware, cache, device, output): """One-time fill using hardware's existing full producer, not a new API. @@ -203,7 +256,15 @@ def build_base(hardware, cache, device, output): shutil.copytree(work / 'full/dtbs', output / 'dtbs') shutil.copytree(work / 'in-tree/lib', output / 'lib') export_sdk(compiler_output, output / 'sdk') - (output / 'inputs.json').write_text(json.dumps({'source_tree': tree}) + '\n') + inputs = {'source_tree': tree} + if (hardware / 'kernel/write-receipt.py').is_file(): + producer = work / 'full/receipt.json' + receipt = json.loads(producer.read_text()) + _verify_producer(hardware, device, output, receipt, tree) + shutil.copy2(producer, output / 'producer-receipt.json') + inputs.update(sdk_sha256=_tree_digest(output / 'sdk'), + modules_sha256=_tree_digest(output / 'lib')) + (output / 'inputs.json').write_text(json.dumps(inputs) + '\n') if sources.working_tree(source) != tree: raise ValueError('Kernel source changed during build; refusing result') @@ -215,7 +276,8 @@ def build_device(hardware, cache, device, base, output): platform, dtb = layout(hardware, device) source, _ = prepare_source(hardware, cache, device) tree = sources.working_tree(source) - if json.loads((base / 'inputs.json').read_text())['source_tree'] != tree: + inputs = json.loads((base / 'inputs.json').read_text()) + if inputs['source_tree'] != tree: raise ValueError('Cached kernel source identity differs; rebuild the base') if (base / 'sdk/.config').read_bytes() != (hardware / 'kernel/configs' / platform / 'config').read_bytes(): raise ValueError('Cached kernel configuration differs; rebuild the base') @@ -224,8 +286,21 @@ def build_device(hardware, cache, device, base, output): raise ValueError('Invalid kernel release') if (base / 'sdk/include/config/kernel.release').read_text().strip() != release: raise ValueError('Cached module SDK release differs') + writer = hardware / 'kernel/write-receipt.py' + receipt = None + if writer.is_file(): + receipt = json.loads((base / 'producer-receipt.json').read_text()) + _verify_producer(hardware, device, base, receipt, tree) + if (inputs.get('sdk_sha256') != _tree_digest(base / 'sdk') or + inputs.get('modules_sha256') != _tree_digest(base / 'lib')): + raise ValueError('Cached kernel SDK or in-tree modules differ') with tempfile.TemporaryDirectory(prefix='modules-', dir=work_directory(hardware, device)) as temporary: work = Path(temporary) + patch_list = work_directory(hardware, device) / 'applied-patches' + recipe_file = work / 'recipe.json' + if receipt is not None: + command(sys.executable, writer, '--capture', '--device', device, '--platform', platform, + '--patch-list', patch_list, '--recipe-root', hardware, '--recipe-file', recipe_file) sdk = work / 'output' shutil.copytree(base / 'sdk', sdk) # The unchanged hardware staging recipe scans the compiler-output tree. @@ -241,17 +316,29 @@ def build_device(hardware, cache, device, base, output): '--file', hardware / 'kernel/Containerfile', '--context', hardware / 'kernel', '--image', 'localhost/orchard-hardware-kernel-toolchain:bookworm') external = [] - if (hardware / 'kernel/modules' / device).is_dir(): + for name in dict.fromkeys((platform, device)): + module_sources = hardware / 'kernel/modules' / name + if not module_sources.is_dir(): + continue + destination = work / ('external-' + name) command(hardware / 'kernel/build-modules', '--source', source, '--build', sdk, - '--modules', hardware / 'kernel/modules' / device, - '--localversion', '-orchard-' + platform, '--output', work / 'external') - external = ['--external', work / 'external'] + '--modules', module_sources, + '--localversion', '-orchard-' + platform, '--output', destination) + external += ['--external', destination] command(sys.executable, hardware / 'kernel/stage-modules.py', '--build', sdk, *external, '--output', work / 'staged') output.mkdir(parents=True) for name in ('Image.gz', 'kernel.release'): shutil.copy2(base / name, output / name) - shutil.copy2(base / 'dtbs' / (dtb + '.dtb'), output / (device + '.dtb')) + shutil.copytree(base / 'dtbs', output / 'dtbs') + shutil.copy2(output / 'dtbs' / (dtb + '.dtb'), output / (device + '.dtb')) shutil.copytree(work / 'staged/lib', output / 'lib') + if receipt is not None: + origin = receipt['source'] + command(sys.executable, writer, '--directory', output, '--device', device, + '--platform', platform, '--dtb', dtb, '--source-commit', origin['commit'], + '--source-tree', tree, '--prepared-tree', origin['prepared_tree'], + '--patch-list', patch_list, '--config', hardware / 'kernel/configs' / platform / 'config', + '--recipe-root', hardware, '--recipe-file', recipe_file) if sources.working_tree(source) != tree: raise ValueError('Kernel source changed during module build; refusing result') diff --git a/src/orchard_cli/nfs_host.py b/src/orchard_cli/nfs_host.py new file mode 100644 index 0000000..6667d74 --- /dev/null +++ b/src/orchard_cli/nfs_host.py @@ -0,0 +1,313 @@ +"""Manage the selected experimental NFS gadget's host-side IPv4 alias.""" +import hashlib +import ipaddress +import json +import os +from pathlib import Path +import plistlib +import re +import stat +import subprocess +import sys +import uuid + +from orchard_hardware.boot.initramfs.nfs.stage import identity + + +def _uuid(value): + try: + if str(uuid.UUID(value)) != value: + raise ValueError + except (ValueError, TypeError, AttributeError): + raise ValueError('Expected a canonical UUID') from None + return value + + +def root_identity(path): + path = Path(path) + if path.is_symlink() or not path.is_file(): + raise ValueError('Expected a regular NFS root identity') + raw = path.read_bytes() + record, _ = identity(raw) + if record['profile'] != 'n61' or record['architecture'] != 'aarch64': + raise ValueError('Expected the N61 AArch64 NFS root identity') + return record, hashlib.sha256(raw).hexdigest() + + +def _usb_devices(raw): + if not isinstance(raw, bytes) or len(raw) > 16 * 1024 * 1024: + raise ValueError('Invalid USB registry data') + pending = plistlib.loads(raw) + if not isinstance(pending, list): + raise ValueError('Invalid USB registry data') + pending, found, visited = list(pending), [], 0 + while pending: + node = pending.pop() + visited += 1 + if visited > 100000 or not isinstance(node, dict): + raise ValueError('Invalid USB registry tree') + children = node.get('IORegistryEntryChildren', []) + if not isinstance(children, list): + raise ValueError('Invalid USB registry children') + pending.extend(children) + if node.get('IOObjectClass') == 'IOUSBHostDevice' and node.get('idVendor') == 0x05ac: + found.append(node) + return found + + +def select_macos(raw, location, root_id, *, expected_boot=None): + """Bind one experimental gadget to its selected USB location and root.""" + if type(location) is not int or not 0 < location < 2**32: + raise ValueError('Invalid selected USB location') + _uuid(root_id) + if expected_boot is not None: + _uuid(expected_boot) + devices = _usb_devices(raw) + active = [node for node in devices if node.get('idProduct') in (0x1227, 0x4141, 0x4142, 0x4144)] + if len(active) != 1 or active[0].get('locationID') != location: + raise ValueError('Expected exactly one boot device at the selected USB location') + node = active[0] + if node.get('idProduct') != 0x4144 or node.get('USB Product Name') != 'Orchard n61 NFS': + raise ValueError('The selected USB device is not the N61 NFS gadget') + serial = node.get('USB Serial Number') or '' + match = re.fullmatch(r'PROFILE:n61 ROOT:([0-9a-f-]{36}) BOOT:([0-9a-f-]{36})', serial) + if not match or _uuid(match[1]) != root_id: + raise ValueError('The selected NFS root identity differs') + boot = _uuid(match[2]) + if expected_boot is not None and boot != expected_boot: + raise ValueError('The selected NFS boot identity differs') + pending = list(node.get('IORegistryEntryChildren', [])) + ports, interfaces, macs = [], [], [] + while pending: + child = pending.pop() + if child.get('IOObjectClass') == 'IOUSBHostDevice': + continue + pending.extend(child.get('IORegistryEntryChildren', [])) + if child.get('IOObjectClass') == 'IOSerialBSDClient': + ports.append(child.get('IOCalloutDevice')) + if child.get('IOObjectClass') in ('IOEthernetInterface', 'IOSkywalkLegacyEthernetInterface'): + interfaces.append(child.get('BSD Name')) + if 'IOMACAddress' in child: + macs.append(child['IOMACAddress']) + if (len(ports) != 1 or not isinstance(ports[0], str) or + not re.fullmatch(r'/dev/cu\.[^/\x00\r\n]+', ports[0])): + raise ValueError('Expected exactly one NFS USB serial endpoint') + if (len(interfaces) != 1 or not isinstance(interfaces[0], str) or + not re.fullmatch(r'en[0-9]+', interfaces[0])): + raise ValueError('Expected exactly one NFS USB network interface') + expected_mac = bytes.fromhex('02000a3d0001') + if not macs or any(value != expected_mac for value in macs): + raise ValueError('The selected NFS network hardware address differs') + return {'location': location, 'profile': 'n61', 'root_id': root_id, 'boot_id': boot, + 'serial_port': ports[0], 'interface': interfaces[0], 'mac': '02:00:0a:3d:00:01'} + + +def interfaces(text): + result, current = {}, None + for line in text.splitlines(): + header = re.match(r'^([A-Za-z][A-Za-z0-9._-]{0,31}):', line) + if header: + current = result.setdefault(header[1], {'mac': None, 'addresses': []}) + continue + if current is None: + continue + mac = re.match(r'\s+ether ([0-9a-fA-F:]{17})(?:\s|$)', line) + address = re.match(r'\s+inet ([0-9.]+) netmask (0x[0-9a-fA-F]+|[0-9.]+)(?:\s|$)', line) + if mac: + if current['mac'] is not None: + raise ValueError('Duplicate interface hardware address') + current['mac'] = mac[1].lower() + if address: + mask = str(ipaddress.IPv4Address(int(address[2], 16))) if address[2].startswith('0x') else address[2] + current['addresses'].append(ipaddress.IPv4Interface(address[1] + '/' + mask)) + return result + + +def routes(text): + result, netif = [], None + for line in text.splitlines(): + fields = line.split() + if fields and fields[0] == 'Destination' and 'Netif' in fields: + netif = fields.index('Netif') + continue + if not fields or fields[0] == 'default' or not re.match(r'^[0-9]', fields[0]): + continue + if netif is None or len(fields) <= netif: + raise ValueError('Unrecognized IPv4 route table') + address, slash, prefix = fields[0].partition('/') + octets = address.split('.') + if not 1 <= len(octets) <= 4: + raise ValueError('Unrecognized IPv4 route destination') + width = int(prefix) if slash else len(octets) * 8 + address = '.'.join(octets + ['0'] * (4 - len(octets))) + result.append((ipaddress.IPv4Network(address + '/' + str(width), strict=False), fields[netif])) + if netif is None: + raise ValueError('The IPv4 route header is absent') + return result + + +def plan(endpoint, configuration, route_table, owner=None): + device = endpoint['interface'] + target = ipaddress.IPv4Interface('10.61.0.1/30') + available = interfaces(configuration) + if device not in available or available[device]['mac'] not in (endpoint['mac'], '02:00:00:00:00:00'): + raise ValueError('The selected USB interface differs') + matches = [key for key, value in available.items() if value['mac'] == endpoint['mac']] + if matches not in ([], [device]): + raise ValueError('The USB network hardware address is ambiguous') + present = False + for key, value in available.items(): + for address in value['addresses']: + if address.network.overlaps(target.network): + if key != device or address != target: + raise ValueError('An existing IPv4 address conflicts with the N61 network') + present = True + if owner is not None: + for key in ('location', 'root_id', 'boot_id', 'interface', 'mac'): + if owner.get(key) != endpoint[key]: + raise ValueError('The saved host link belongs to another session') + if (owner.get('kind'), owner.get('state'), owner.get('address')) != ( + 'orchard.n61-nfs-host-link', 'owned', str(target)): + raise ValueError('The saved host link is not a confirmed ownership receipt') + if present and owner is None: + raise ValueError('The N61 address exists without an ownership receipt') + for network, key in routes(route_table): + if network.prefixlen == 0 or not network.overlaps(target.network): + continue + if present and owner is not None and ( + (key == device and network.subnet_of(target.network)) or + (key == 'lo0' and network == ipaddress.IPv4Network('10.61.0.1/32'))): + continue + raise ValueError('An existing IPv4 route conflicts with the N61 network') + return {'action': 'already-owned' if present else 'add-alias', 'endpoint': endpoint, + 'address': str(target), 'command': ['/usr/bin/sudo', '-n', '/sbin/ifconfig', device, + 'inet', '10.61.0.1', 'netmask', '255.255.255.252', 'alias']} + + +def inspect(root_path, location, *, expected_boot=None, state=None): + if sys.platform != 'darwin': + raise ValueError('N61 NFS host link currently supports macOS') + root, root_sha = root_identity(root_path) + raw = subprocess.run(['/usr/sbin/ioreg', '-a', '-l', '-r', '-c', 'IOUSBHostDevice'], + check=True, capture_output=True, timeout=10).stdout + endpoint = select_macos(raw, location, root['root_id'], expected_boot=expected_boot) + if not stat.S_ISCHR(Path(endpoint['serial_port']).stat().st_mode): + raise ValueError('The selected NFS serial endpoint is not a character device') + config = subprocess.run(['/sbin/ifconfig', '-a'], check=True, capture_output=True, text=True, timeout=10).stdout + table = subprocess.run(['/usr/sbin/netstat', '-rn', '-f', 'inet'], + check=True, capture_output=True, text=True, timeout=10).stdout + owner = None + if state is not None and (Path(state).exists() or Path(state).is_symlink()): + path = Path(state) + if path.is_symlink() or not path.is_file() or path.stat().st_mode & 0o077: + raise ValueError('Expected a private regular host link receipt') + owner = json.loads(path.read_text()) + if owner.get('root_identity_sha256') != root_sha: + raise ValueError('The saved root identity differs') + return {**plan(endpoint, config, table, owner), 'root_identity_sha256': root_sha} + + +def _write_new(path, record): + path = Path(path) + path.parent.mkdir(parents=True, exist_ok=True, mode=0o700) + fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL | getattr(os, 'O_NOFOLLOW', 0), 0o600) + with os.fdopen(fd, 'w') as stream: + json.dump(record, stream, sort_keys=True, indent=2) + stream.write('\n') + + +def apply(root_path, location, boot_id, state): + """Add one alias after two checks; preserve an uncertain intent on failure.""" + _uuid(boot_id) + path = Path(state) + first = inspect(root_path, location, expected_boot=boot_id, state=path) + if first['action'] == 'already-owned': + return first + if path.exists() or path.is_symlink(): + raise ValueError('Choose a new private host link receipt') + subprocess.run(['/usr/bin/sudo', '-v'], check=True) + if inspect(root_path, location, expected_boot=boot_id, state=path) != first: + raise ValueError('The USB link changed during authorization') + record = {'schema': 1, 'kind': 'orchard.n61-nfs-host-link', 'state': 'adding', + 'address': first['address'], 'root_identity_sha256': first['root_identity_sha256'], + **first['endpoint']} + _write_new(path, record) + try: + if inspect(root_path, location, expected_boot=boot_id) != first: + raise ValueError('The USB link changed before address setup') + subprocess.run(first['command'], check=True, timeout=15) + record['state'] = 'owned' + # Verify with the candidate ownership record before committing it. + config = subprocess.run(['/sbin/ifconfig', '-a'], check=True, capture_output=True, text=True, timeout=10).stdout + table = subprocess.run(['/usr/sbin/netstat', '-rn', '-f', 'inet'], check=True, + capture_output=True, text=True, timeout=10).stdout + raw = subprocess.run(['/usr/sbin/ioreg', '-a', '-l', '-r', '-c', 'IOUSBHostDevice'], + check=True, capture_output=True, timeout=10).stdout + endpoint = select_macos(raw, location, record['root_id'], expected_boot=boot_id) + root, root_sha = root_identity(root_path) + if (root['root_id'] != record['root_id'] or root_sha != record['root_identity_sha256'] or + endpoint != first['endpoint'] or plan(endpoint, config, table, record)['action'] != 'already-owned'): + raise ValueError('The host address could not be verified') + temporary = path.with_name('.' + path.name + '.' + str(uuid.uuid4())) + _write_new(temporary, record) + os.replace(temporary, path) + except BaseException: + # The original adding receipt requires inspection before another attempt. + raise + return {**first, 'action': 'added', 'ownership_receipt': str(path)} + + +def _has_nfs_gadget(raw): + return any(node.get('idProduct') == 0x4144 for node in _usb_devices(raw)) + + +def _replace(path, record): + temporary = path.with_name('.' + path.name + '.' + str(uuid.uuid4())) + _write_new(temporary, record) + os.replace(temporary, path) + + +def cleanup(root_path, state): + """Retire only an owned alias after the experimental gadget disconnects.""" + if sys.platform != 'darwin': + raise ValueError('N61 NFS host link currently supports macOS') + path = Path(state) + if path.is_symlink() or not path.is_file() or path.stat().st_mode & 0o077: + raise ValueError('Expected a private regular host link receipt') + record = json.loads(path.read_text()) + root, root_sha = root_identity(root_path) + if (record.get('schema'), record.get('kind'), record.get('state'), record.get('profile'), + record.get('root_id'), record.get('root_identity_sha256'), record.get('address'), record.get('mac')) != ( + 1, 'orchard.n61-nfs-host-link', 'owned', 'n61', root['root_id'], root_sha, + '10.61.0.1/30', '02:00:0a:3d:00:01'): + raise ValueError('The host link receipt is not an owned N61 session') + _uuid(record.get('boot_id')) + interface = record.get('interface') + if not isinstance(interface, str) or not re.fullmatch(r'en[0-9]+', interface): + raise ValueError('Invalid saved network interface') + + def inspect_retired(): + raw = subprocess.run(['/usr/sbin/ioreg', '-a', '-l', '-r', '-c', 'IOUSBHostDevice'], + check=True, capture_output=True, timeout=10).stdout + if _has_nfs_gadget(raw): + raise ValueError('Disconnect the NFS gadget before retiring its host address') + config = subprocess.run(['/sbin/ifconfig', '-a'], check=True, capture_output=True, text=True, timeout=10).stdout + row = interfaces(config).get(interface) + if row is None or ipaddress.IPv4Interface('10.61.0.1/30') not in row['addresses']: + return False + if row['mac'] != record['mac']: + raise ValueError('The saved interface no longer has its verified hardware address') + return True + + if inspect_retired(): + subprocess.run(['/usr/bin/sudo', '-v'], check=True) + if not inspect_retired(): + raise ValueError('The host link changed during authorization') + removing = {**record, 'state': 'removing'} + _replace(path, removing) + subprocess.run(['/usr/bin/sudo', '-n', '/sbin/ifconfig', interface, 'inet', '10.61.0.1', '-alias'], + check=True, timeout=15) + if inspect_retired(): + raise ValueError('The owned host address is still present') + _replace(path, {**record, 'state': 'retired'}) + return {'action': 'retired', 'ownership_receipt': str(path)} diff --git a/src/orchard_cli/provenance.py b/src/orchard_cli/provenance.py new file mode 100644 index 0000000..5a2d9ca --- /dev/null +++ b/src/orchard_cli/provenance.py @@ -0,0 +1,57 @@ +"""Record the source contents that produced a local build.""" +import hashlib +import json +import os +from pathlib import Path +import subprocess + + +EXCLUDED = {'.git', '.work', '.venv', '__pycache__', '.mypy_cache', '.pytest_cache', 'build', 'dist'} + + +def snapshot(directory): + """Fingerprint package contents and report a Git head only when this is tracked source.""" + root = Path(directory).resolve() + if not root.is_dir(): + raise ValueError('Source package directory is missing: ' + str(root)) + files = {} + for current, directories, names in os.walk(root): + directories[:] = sorted(name for name in directories if name not in EXCLUDED and not name.endswith('.egg-info')) + for name in directories: + if (Path(current) / name).is_symlink(): + raise ValueError('Linked source package directory: ' + str(Path(current) / name)) + for name in sorted(names): + if name in EXCLUDED or name.endswith('.egg-info'): + continue + path = Path(current) / name + if path.is_symlink() or not path.is_file(): + raise ValueError('Linked or unsupported source package file: ' + str(path)) + digest = hashlib.sha256() + with path.open('rb') as stream: + for chunk in iter(lambda: stream.read(1024 * 1024), b''): + digest.update(chunk) + files[path.relative_to(root).as_posix()] = { + 'bytes': path.stat().st_size, 'mode': path.stat().st_mode & 0o777, 'sha256': digest.hexdigest()} + result = {'sha256': hashlib.sha256(json.dumps(files, sort_keys=True, separators=(',', ':')).encode()).hexdigest(), + 'kind': 'installed'} + try: + top = Path(subprocess.check_output(['git', '-C', str(root), 'rev-parse', '--show-toplevel'], + stderr=subprocess.DEVNULL, text=True).strip()).resolve() + tracked = (root / '__init__.py').relative_to(top) + subprocess.run(['git', '-C', str(top), 'ls-files', '--error-unmatch', '--', str(tracked)], + stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, check=True) + head = subprocess.check_output(['git', '-C', str(top), 'rev-parse', 'HEAD'], + stderr=subprocess.DEVNULL, text=True).strip() + status = subprocess.check_output(['git', '-C', str(top), 'status', '--porcelain', '--untracked-files=all'], + stderr=subprocess.DEVNULL) + except (OSError, ValueError, subprocess.CalledProcessError): + return result + result.update(kind='checkout', git={'head': head, 'clean': not bool(status)}) + return result + + +def cli(): + import orchard_cli + import orchard_cli_schemas + return {'code': snapshot(Path(orchard_cli.__file__).parent), + 'schemas': snapshot(Path(orchard_cli_schemas.__file__).parent)} diff --git a/src/orchard_cli/ram.py b/src/orchard_cli/ram.py index 7eb5766..18f8afe 100644 --- a/src/orchard_cli/ram.py +++ b/src/orchard_cli/ram.py @@ -5,6 +5,7 @@ Device confirmation and upload live in boot.py. """ import json +import hashlib import os from pathlib import Path, PurePosixPath import re @@ -13,7 +14,7 @@ import sys import tempfile -from . import build +from . import build, provenance from .build import command, digest, recipe_key @@ -29,9 +30,76 @@ def check_busybox(path): raise ValueError('BusyBox must be static (no DYNAMIC or INTERP segment)') +def _kernel_evidence(kernel, device, release): + """Verify an available hardware receipt before recording it in the RAM build.""" + image, dtb = kernel / 'Image.gz', kernel / (device + '.dtb') + evidence = {'image_sha256': digest(image), 'dtb_sha256': digest(dtb), 'release': release} + receipt_path = kernel / 'receipt.json' + if not receipt_path.exists() and not receipt_path.is_symlink(): + return evidence, None + if receipt_path.is_symlink() or not receipt_path.is_file(): + raise ValueError('Linked kernel receipt refused') + data = receipt_path.read_bytes() + receipt = json.loads(data) + if (receipt.get('schema'), receipt.get('device'), receipt.get('kernel_release')) != (1, device, release): + raise ValueError('Kernel receipt identity differs') + source, hardware_recipe = receipt.get('source'), receipt.get('hardware_recipe') + if (not isinstance(source, dict) or any( + not isinstance(source.get(name), str) or not re.fullmatch('[0-9a-f]{40}', source[name]) + for name in ('commit', 'tree', 'prepared_tree')) or any( + not isinstance(source.get(name), str) or not re.fullmatch('[0-9a-f]{64}', source[name]) + for name in ('patch_list_sha256', 'config_sha256'))): + raise ValueError('Kernel source provenance is incomplete') + if not isinstance(hardware_recipe, dict) or not isinstance(hardware_recipe.get('files'), dict): + raise ValueError('Kernel hardware recipe is incomplete') + encoded = json.dumps(hardware_recipe['files'], sort_keys=True, separators=(',', ':')).encode() + if hardware_recipe.get('sha256') != hashlib.sha256(encoded).hexdigest(): + raise ValueError('Kernel hardware recipe digest differs') + files = receipt.get('files') + if not isinstance(files, dict) or not {'Image.gz', device + '.dtb', 'kernel.release'} <= files.keys(): + raise ValueError('Kernel receipt inventory is incomplete') + actual = set() + for path in kernel.rglob('*'): + if path.is_symlink(): + raise ValueError('Linked kernel receipt input refused') + if path.is_file() and path != receipt_path: + actual.add(path.relative_to(kernel).as_posix()) + elif not path.is_file() and not path.is_dir(): + raise ValueError('Unsupported kernel receipt input') + if actual != files.keys(): + raise ValueError('Kernel receipt inventory differs') + for name, record in files.items(): + path = kernel / name + if (not isinstance(record, dict) or set(record) != {'bytes', 'sha256'} or + path.stat().st_size != record['bytes'] or digest(path) != record['sha256']): + raise ValueError('Kernel receipt component differs: ' + name) + selected_dtb = receipt.get('dtb') + if (not isinstance(selected_dtb, str) or not re.fullmatch('[a-z0-9][a-z0-9_.-]*', selected_dtb) or + files.get('dtbs/' + selected_dtb + '.dtb') != files[device + '.dtb']): + raise ValueError('Kernel receipt device tree differs') + evidence.update(receipt_sha256=hashlib.sha256(data).hexdigest(), source=receipt.get('source'), + hardware_recipe_sha256=hardware_recipe['sha256']) + return evidence, data + + +def _recipe_evidence(hardware, device): + descriptor = hardware / 'devices' / device / 'device.toml' + if not descriptor.is_file(): + return None + description = build.tomllib.loads(descriptor.read_text()) + if 'firmware' not in description: + return None + from .firmware import recipe_path_for + path = recipe_path_for(device, hardware) + if path.is_symlink() or not path.is_file(): + raise ValueError('Firmware recipe is missing or linked') + return path.read_bytes() + + def assemble_ram(hardware, output, *, kernel, busybox, loader, pongo, init, files=None, hostname='orchard-ram', device, modules=False, - bootargs='console=tty0 loglevel=7 rdinit=/init', sources=()): + bootargs='console=tty0 loglevel=7 rdinit=/init', sources=(), + firmware_included=False, firmware_recipe_sha256=None): """Pack a caller-owned /init and arbitrary files. No firmware or modules are activated here. Files map relative RAM-root destinations to local files. """ @@ -45,6 +113,14 @@ def assemble_ram(hardware, output, *, kernel, busybox, loader, pongo, init, release = (kernel / 'kernel.release').read_text().strip() if not release or '/' in release or len(release.split()) != 1: raise ValueError('Invalid kernel.release') + kernel_record, kernel_receipt = _kernel_evidence(kernel, device, release) + firmware_recipe = _recipe_evidence(hardware, device) + if firmware_included and firmware_recipe is None: + raise ValueError('Included firmware requires a device recipe') + firmware_sha = hashlib.sha256(firmware_recipe).hexdigest() if firmware_recipe is not None else None + if firmware_recipe_sha256 is not None and firmware_sha != firmware_recipe_sha256: + raise ValueError('Firmware recipe changed during the build') + source_record = {'cli': provenance.cli(), 'hardware': provenance.snapshot(hardware)} inputs = [init, *map(Path, (files or {}).values()), *map(Path, sources)] with tempfile.TemporaryDirectory(prefix='build-', dir=output) as temporary: work = Path(temporary) @@ -86,9 +162,29 @@ def assemble_ram(hardware, output, *, kernel, busybox, loader, pongo, init, receipt = {'kind': 'orchard.ram-build', 'device': device, 'kernel_release': release, 'hardware_tested': False, 'files': {name: digest(work / name) for name in ('linux.bin', 'pongo.bin')}, - 'sources': {str(p.resolve()): digest(p) for p in inputs}} + 'sources': {str(p.resolve()): digest(p) for p in inputs}, + 'provenance': {**source_record, 'kernel': kernel_record, + 'firmware': {'included_via_recipe': firmware_included, 'recipe_sha256': firmware_sha}}} + if source_record != {'cli': provenance.cli(), 'hardware': provenance.snapshot(hardware)}: + raise ValueError('CLI or hardware source changed during RAM assembly') + if _kernel_evidence(kernel, device, release) != (kernel_record, kernel_receipt): + raise ValueError('Kernel component changed during RAM assembly') + if _recipe_evidence(hardware, device) != firmware_recipe: + raise ValueError('Firmware recipe changed during RAM assembly') + snapshots = {} + if kernel_receipt is not None: + snapshots['kernel-receipt.json'] = kernel_receipt + if firmware_recipe is not None: + snapshots['firmware-recipe.json'] = firmware_recipe + for name, data in snapshots.items(): + (work / name).write_bytes(data) for name in ('linux.bin', 'linux.bin.json', 'pongo.bin', 'initramfs.cpio.gz'): os.replace(work / name, output / name) + for name in ('kernel-receipt.json', 'firmware-recipe.json'): + if name in snapshots: + os.replace(work / name, output / name) + else: + (output / name).unlink(missing_ok=True) (output / 'build.json').write_text(json.dumps(receipt, indent=2) + '\n') print(f'Built {output / "linux.bin"}. No device was accessed.') return output @@ -131,11 +227,16 @@ def build_directory(directory, *, hardware=None, cache=None, kernel=None, for tool in sorted(tools.iterdir()): if tool.name != 'busybox' and tool.is_file(): files.setdefault(TOOL_DESTINATIONS.get(tool.name, 'usr/bin') + '/' + tool.name, tool) + firmware_sha = None if firmware: from . import firmware as firmwares from .progress import Clock, Progress recipe = firmwares.recipe_for(device, hardware) + recipe_path = firmwares.recipe_path_for(device, hardware) + firmware_sha = digest(recipe_path) outputs = firmwares.ensure(device, Clock(), Progress(), hardware=hardware) + if digest(recipe_path) != firmware_sha: + raise ValueError('Firmware recipe changed during acquisition') for item in recipe['outputs']: if item['destination'] in files: raise ValueError(f'rootfs/ and the {device} firmware both provide {item["destination"]}') @@ -144,7 +245,8 @@ def build_directory(directory, *, hardware=None, cache=None, kernel=None, busybox=busybox, loader=loader, pongo=pongo) hostname = re.sub(r'[^A-Za-z0-9_-]', '-', directory.name).strip('-_')[:63] or 'orchard-ram' assemble_ram(hardware, output, **components, init=init, files=files, - modules=modules, hostname=hostname, device=device) + modules=modules, hostname=hostname, device=device, firmware_included=firmware, + firmware_recipe_sha256=firmware_sha) receipt = json.loads((output / 'build.json').read_text()) # Catch newly added/removed overlay files too, not only changes to old inputs. receipt['overlay'] = {'path': str(overlay), 'sha256': recipe_key(directory, ['rootfs'], ignore=())} @@ -162,6 +264,13 @@ def verify_ram(output): path = output / name if path.is_symlink() or digest(path) != expected: raise ValueError(f'{path} changed; rebuild before booting') + evidence = receipt.get('provenance', {}) + for name, expected in (('kernel-receipt.json', evidence.get('kernel', {}).get('receipt_sha256')), + ('firmware-recipe.json', evidence.get('firmware', {}).get('recipe_sha256'))): + if expected is not None: + path = output / name + if path.is_symlink() or digest(path) != expected: + raise ValueError(f'{path} changed; rebuild before booting') # Old charger receipts remain bootable across this host-side refactor. Their # source list included host launchers, which are not part of the RAM payload. if 'overlay' in receipt: diff --git a/src/orchard_cli/recipe.py b/src/orchard_cli/recipe.py index b49add9..8abc056 100644 --- a/src/orchard_cli/recipe.py +++ b/src/orchard_cli/recipe.py @@ -63,6 +63,14 @@ def _z2fw_packets(step, data, expected): return output +def _z2fw_gen1(step, data, expected): + output, _ = z2fw.convert_gen1(_touch_profile(data, step['profile']), packet=step['packet'], + touch_maximum=step['touch_maximum'], prox_maximum=step['prox_maximum'], + fll_mval=step['fll_mval'], ref_clk_div=step['ref_clk_div'], + clk32_clock_enable=step['clk32_clock_enable']) + return output + + def _z2fw_mtfw(step, data, expected): output, _ = z2fw.convert_mtfw(apple_xml(data), step['profile'], otp_preflight=step['otp_preflight']) return output @@ -86,6 +94,10 @@ def _z2fw_mtfw(step, data, expected): 'calibration_size': int}), # mtprops with an array of preconstructed packets (iPhone 7 Plus), makez2fw's layout. 'z2fw-from-mtprops@2': (_z2fw_packets, {'profile': str, 'packet': str}), + # Legacy Gen1 data-form mtprops (iPhone 6), Corellium's bootstrap and board calibration bounds. + 'z2fw-from-mtprops@3': (_z2fw_gen1, {'profile': str, 'packet': str, + 'touch_maximum': int, 'prox_maximum': int, + 'fll_mval': int, 'ref_clk_div': int, 'clk32_clock_enable': int}), # MTFW (IM4P payload) with HBPP load items per personality (iPad 7), makez2fw's layout. 'z2fw-from-mtfw@1': (_z2fw_mtfw, {'profile': str, 'otp_preflight': bool}), } diff --git a/src/orchard_cli/transports/pongo.py b/src/orchard_cli/transports/pongo.py index 3c4c8ca..8fbf0ca 100644 --- a/src/orchard_cli/transports/pongo.py +++ b/src/orchard_cli/transports/pongo.py @@ -122,7 +122,10 @@ def __init__(self): self.claimed = False def _bind(self, name, result, *args): - fn = getattr(self.lib, name) + try: + fn = getattr(self.lib, name) + except AttributeError as error: + raise PongoError(f'libusb-1.0 is missing required symbol {name}') from error fn.argtypes = args fn.restype = result @@ -243,6 +246,11 @@ def matches_target(target, bus, ports): raise PongoError('Unsupported USB host') +def preflight(): + """Check the library and required symbols without opening a device.""" + LibUSB() + + def wait_prompt(usb, seconds=15): deadline = time.monotonic() + seconds while time.monotonic() < deadline: diff --git a/tests/test_adapter_contract.py b/tests/test_adapter_contract.py index a88b5ae..84008ac 100644 --- a/tests/test_adapter_contract.py +++ b/tests/test_adapter_contract.py @@ -2,6 +2,8 @@ import unittest from unittest.mock import patch +import orchard_hardware + from orchard_cli.contracts import Adapter from orchard_cli import devices from orchard_cli.apple import device as adapter_module @@ -54,6 +56,29 @@ def test_unknown_connected_device_is_named_in_the_error(self): with self.assertRaisesRegex(ValueError, 'not one Orchard supports'): devices.choose() + def test_n61_uses_generic_adapter_and_keeps_installation_and_recovery_disabled(self): + descriptor = {'device': {'profile': 'n61', 'name': 'iPhone 6', 'apple_products': ['iPhone7,2']}, + 'match': {'dfu': {'chip': 0x7000, 'board': 0x06}}, + 'firmware': {'recipe': 'firmware.json'}, + 'support': {'install': False, 'status': 'experimental'}} + original = orchard_hardware.device + observed = {'mode': 'dfu', 'location': 'usb', 'ecid': 1, 'chip': 0x7000, 'board': 0x06} + with patch.object(orchard_hardware, 'profiles', return_value=['j71b', 'n61']), \ + patch.object(orchard_hardware, 'device', side_effect=lambda name: + descriptor if name == 'n61' else original(name)), \ + patch.object(adapter_module.usb, 'devices', return_value=[observed]): + adapter, selected = devices.choose() + self.assertEqual((adapter.PROFILE, adapter.APPLE_PRODUCT), ('n61', 'iPhone7,2')) + self.assertFalse(adapter.INSTALL) + self.assertIsNone(adapter.PRODUCT) + self.assertFalse(selected['resumed']) + self.assertEqual(selected['ecid'], observed['ecid']) + with self.assertRaisesRegex(ValueError, 'no Orchard recovery system'): + adapter.recovery_endpoint() + observed['board'] = 0x07 + with self.assertRaisesRegex(ValueError, 'not one Orchard supports'): + devices.choose() + if __name__ == '__main__': unittest.main() diff --git a/tests/test_device_state.py b/tests/test_device_state.py index 5212902..362bd96 100644 --- a/tests/test_device_state.py +++ b/tests/test_device_state.py @@ -131,6 +131,48 @@ def test_macos_usb_descriptor_identifies_device_and_boot(self): self.assertEqual((rows[0]['ecid'],rows[0]['boot_id']),(2,BOOT_B)) self.assertEqual(rows[0]['serial_port'],'/dev/cu.device') + def test_macos_hubs_do_not_inherit_child_device_serial_ports(self): + phone = {'IOObjectClass':'IOUSBHostDevice','idVendor':0x05ac,'idProduct':0x4144, + 'locationID':123,'USB Product Name':'Orchard n61_ram', + 'IORegistryEntryChildren':[ + {'IOObjectClass':'IOUSBHostInterface','IORegistryEntryChildren':[ + {'IOObjectClass':'IOSerialBSDClient','IOCalloutDevice':'/dev/cu.phone'}]}]} + other = {'IOObjectClass':'IOUSBHostDevice','idVendor':0x1234,'idProduct':1, + 'locationID':124,'IORegistryEntryChildren':[ + {'IOObjectClass':'IOSerialBSDClient','IOCalloutDevice':'/dev/cu.other'}]} + inner = {'IOObjectClass':'IOUSBHostDevice','idVendor':0x05ac,'idProduct':0x8004, + 'locationID':120,'USB Product Name':'AppleUSB2Hub', + 'IORegistryEntryChildren':[phone,other]} + outer = {'IOObjectClass':'IOUSBHostDevice','idVendor':0x05ac,'idProduct':0x8004, + 'locationID':100,'USB Product Name':'AppleUSB2Hub', + 'IORegistryEntryChildren':[inner]} + raw = plistlib.dumps([outer]) + with patch.object(usb.sys,'platform','darwin'), patch.object(usb.subprocess,'run',return_value= + subprocess.CompletedProcess([],0,raw,b'')): + rows = {row['location']:row for row in usb.devices()} + self.assertEqual(set(rows),{100,120,123}) + self.assertIsNone(rows[100]['serial_port']) + self.assertIsNone(rows[120]['serial_port']) + self.assertEqual(rows[123]['serial_port'],'/dev/cu.phone') + self.assertEqual(rows[123]['product'],'Orchard n61_ram') + + def test_a8_usb_identity_is_parsed_numerically_on_both_hosts(self): + serial = 'CPID:7000 BDID:06 ECID:0000123456789ABC' + expected = (0x7000, 0x06, 0x123456789abc) + node = {'IOObjectClass':'IOUSBHostDevice', 'idVendor':0x05ac, 'idProduct':0x1227, + 'locationID':123, 'USB Serial Number':serial, 'USB Product Name':'Apple DFU'} + with patch.object(usb.sys,'platform','darwin'), patch.object(usb.subprocess,'run',return_value= + subprocess.CompletedProcess([],0,plistlib.dumps([node]),b'')): + mac = usb.devices()[0] + self.assertEqual(tuple(mac[k] for k in ('chip','board','ecid')),expected) + + node = self.root/'1-1' + node.mkdir() + for name,value in {'idVendor':'05ac','idProduct':'1227','serial':serial}.items(): + (node/name).write_text(value) + linux = linux_usb.devices(self.root)[0] + self.assertEqual(tuple(linux[k] for k in ('chip','board','ecid')),expected) + def test_linux_usb_descriptor_identifies_device_and_boot_without_serial_commands(self): node = self.root/'1-2' node.mkdir() diff --git a/tests/test_kernel_receipts.py b/tests/test_kernel_receipts.py new file mode 100644 index 0000000..04aeba9 --- /dev/null +++ b/tests/test_kernel_receipts.py @@ -0,0 +1,176 @@ +"""The cached kernel and its device output retain hardware's receipt contract.""" +import hashlib +import importlib.util +import json +from pathlib import Path +import shutil +import subprocess +import sys +import tempfile +import unittest +from unittest.mock import patch + +import orchard_hardware +from orchard_cli import build, kernel, kernel_sdk + +import hardware_layout + + +class KernelReceipts(unittest.TestCase): + def setUp(self): + temporary = tempfile.TemporaryDirectory() + self.addCleanup(temporary.cleanup) + self.root = Path(temporary.name).resolve() + self.hardware = self.root / 'hardware' + hardware_layout.write(self.hardware, device='n61', platform='apple-4k', dtb='t7000-n61') + writer = Path(orchard_hardware.__file__).resolve().parent / 'kernel/write-receipt.py' + if not writer.is_file(): + self.skipTest('Installed hardware has no kernel receipt writer') + shutil.copyfile(writer, self.hardware / 'kernel/write-receipt.py') + self.writer = self.hardware / 'kernel/write-receipt.py' + spec = importlib.util.spec_from_file_location('hardware_kernel_receipt', writer) + self.receipt = importlib.util.module_from_spec(spec) + spec.loader.exec_module(self.receipt) + for name in ('__init__.py', 'buildenv.py', 'kernel/Containerfile', 'kernel/install-toolchain', + 'kernel/build', 'kernel/build-modules', 'kernel/stage-modules.py', 'kernel/sources.toml'): + path = self.hardware / name + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(name) + (self.hardware / 'kernel/upstream.lock').write_text('url=file:///fixture\ncommit=' + 'a' * 40 + '\n') + (self.hardware / 'devices/n61/device.toml').write_text( + '[kernel]\nplatform = "apple-4k"\ndtb = "t7000-n61"\n[firmware]\nrecipe = "firmware.json"\n') + (self.hardware / 'devices/n61/firmware.json').write_text('firmware recipe') + other = self.hardware / 'devices/j81/device.toml' + other.parent.mkdir(parents=True) + other.write_text('[kernel]\nplatform = "apple-4k"\ndtb = "t7001-j81"\n') + for name in ('apple-4k', 'n61'): + path = self.hardware / 'kernel/modules' / name / 'module/Makefile' + path.parent.mkdir(parents=True) + path.write_text('module') + self.work = kernel.work_directory(self.hardware, 'n61') + self.work.mkdir(parents=True) + (self.work / 'applied-patches').write_text('') + (self.work / 'prepared-tree').write_text('c' * 40 + '\n') + self.source = self.root / 'source' + self.source.mkdir() + self.compiler = self.root / 'compiler' + for name in kernel_sdk.REQUIRED: + path = self.compiler / name + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text('sdk fixture') + for name in kernel_sdk.DIRECTORIES: + (self.compiler / name).mkdir(parents=True, exist_ok=True) + config = self.hardware / 'kernel/configs/apple-4k/config' + config.write_text('CONFIG_ARM64_4K_PAGES=y\n') + (self.compiler / '.config').write_bytes(config.read_bytes()) + self.release = '7.3.0-rc1-orchard-apple-4k' + (self.compiler / 'include/config/kernel.release').write_text(self.release + '\n') + self.calls = [] + + def write_modules(self, target, external=()): + modules = target / 'lib/modules' / self.release + (modules / 'kernel/drivers').mkdir(parents=True) + (modules / 'kernel/drivers/core.ko').write_bytes(b'core module') + for root in external: + for path in root.rglob('*.ko'): + dest = modules / 'extra' / path.name + dest.parent.mkdir(parents=True, exist_ok=True) + shutil.copyfile(path, dest) + (modules / 'modules.dep').write_text('kernel/drivers/core.ko:\n') + + def command(self, *argv, **kwargs): + self.calls.append(argv) + if Path(argv[0]) == self.hardware / 'kernel/build': + full = Path(argv[argv.index('--output') + 1]) + (full / 'dtbs').mkdir(parents=True) + (full / 'Image.gz').write_bytes(b'kernel image') + (full / 'kernel.release').write_text(self.release + '\n') + (full / 'n61.dtb').write_bytes(b'n61 dtb') + (full / 'dtbs/t7000-n61.dtb').write_bytes(b'n61 dtb') + (full / 'dtbs/t7001-j81.dtb').write_bytes(b'other dtb') + self.write_modules(full) + snapshot = self.root / 'producer-recipe.json' + snapshot.write_text(json.dumps(self.receipt.recipe_snapshot( + self.hardware, 'n61', 'apple-4k', self.work / 'applied-patches'))) + self.receipt.write_receipt(full, 'n61', 'apple-4k', 't7000-n61', 'a' * 40, + 'b' * 40, 'c' * 40, self.work / 'applied-patches', + self.hardware / 'kernel/configs/apple-4k/config', self.hardware, snapshot) + elif Path(argv[0]) == self.hardware / 'kernel/build-modules': + destination = Path(argv[argv.index('--output') + 1]) + destination.mkdir() + name = Path(argv[argv.index('--modules') + 1]).name + (destination / (name + '.ko')).write_bytes((name + ' module').encode()) + elif len(argv) > 1 and Path(argv[1]) == self.hardware / 'kernel/stage-modules.py': + destination = Path(argv[argv.index('--output') + 1]) + external = [Path(argv[index + 1]) for index, arg in enumerate(argv) if arg == '--external'] + self.write_modules(destination, external) + elif len(argv) > 1 and Path(argv[1]) == self.writer: + subprocess.run(list(map(str, argv)), check=True, **kwargs) + else: + self.assertIn('orchard_hardware.buildenv', argv) + + def produce(self): + base = self.root / 'base' + with patch.object(kernel, 'prepare_source', return_value=(self.source, self.compiler)), \ + patch.object(kernel.sources, 'working_tree', return_value='b' * 40), \ + patch.object(build, 'command', side_effect=self.command): + kernel.build_base(self.hardware, self.root / 'cache', 'n61', base) + return base + + def device(self, base, output, device='n61'): + with patch.object(kernel, 'prepare_source', return_value=(self.source, self.compiler)), \ + patch.object(kernel.sources, 'working_tree', return_value='b' * 40), \ + patch.object(build, 'command', side_effect=self.command): + kernel.build_device(self.hardware, self.root / 'cache', device, base, output) + + def test_device_receipt_covers_cached_image_dtbs_and_rebuilt_modules(self): + base = self.produce() + output = self.root / 'result' + self.device(base, output) + receipt = json.loads((output / 'receipt.json').read_text()) + self.assertEqual((receipt['device'], receipt['platform'], receipt['dtb']), + ('n61', 'apple-4k', 't7000-n61')) + self.assertEqual((receipt['source']['tree'], receipt['source']['prepared_tree']), + ('b' * 40, 'c' * 40)) + self.assertIn('devices/n61/firmware.json', receipt['hardware_recipe']['files']) + self.assertEqual(receipt['files']['n61.dtb'], receipt['files']['dtbs/t7000-n61.dtb']) + for name in ('Image.gz', 'n61.dtb', 'dtbs/t7000-n61.dtb', 'dtbs/t7001-j81.dtb', + 'lib/modules/' + self.release + '/extra/apple-4k.ko', + 'lib/modules/' + self.release + '/extra/n61.ko'): + path = output / name + self.assertEqual(receipt['files'][name], {'bytes': path.stat().st_size, + 'sha256': hashlib.sha256(path.read_bytes()).hexdigest()}) + self.assertEqual(len([call for call in self.calls if Path(call[0]) == self.hardware / 'kernel/build']), 1) + + def test_second_device_reuses_base_with_its_own_dtb_and_module_receipt(self): + base = self.produce() + phone, other = self.root / 'phone', self.root / 'other' + self.device(base, phone) + self.device(base, other, device='j81') + receipt = json.loads((other / 'receipt.json').read_text()) + self.assertEqual((receipt['device'], receipt['platform'], receipt['dtb']), + ('j81', 'apple-4k', 't7001-j81')) + self.assertEqual((other / 'j81.dtb').read_bytes(), b'other dtb') + self.assertEqual((phone / 'Image.gz').read_bytes(), (other / 'Image.gz').read_bytes()) + self.assertIn('lib/modules/' + self.release + '/extra/apple-4k.ko', receipt['files']) + self.assertNotIn('lib/modules/' + self.release + '/extra/n61.ko', receipt['files']) + self.assertEqual(len([call for call in self.calls if Path(call[0]) == self.hardware / 'kernel/build']), 1) + + def test_corrupted_sdk_dtb_and_source_marker_are_rejected(self): + base = self.produce() + for relative in ('sdk/Module.symvers', 'lib/modules/' + self.release + '/kernel/drivers/core.ko', + 'dtbs/t7000-n61.dtb'): + path = base / relative + original = path.read_bytes() + path.write_bytes(original + b'changed') + with self.subTest(path=relative), self.assertRaises(ValueError): + self.device(base, self.root / 'rejected') + self.assertFalse((self.root / 'rejected').exists()) + path.write_bytes(original) + (self.work / 'prepared-tree').write_text('d' * 40 + '\n') + with self.assertRaisesRegex(ValueError, 'producer source or platform differs'): + self.device(base, self.root / 'rejected') + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/test_nfs_host.py b/tests/test_nfs_host.py new file mode 100644 index 0000000..a86137e --- /dev/null +++ b/tests/test_nfs_host.py @@ -0,0 +1,126 @@ +"""N61 host-link selection and ownership checks use synthetic host fixtures.""" +import json +from pathlib import Path +import plistlib +import subprocess +import tempfile +from types import SimpleNamespace +import unittest +from unittest.mock import patch + +from orchard_cli import nfs_host + +ROOT = '11111111-1111-4111-8111-111111111111' +BOOT = '22222222-2222-4222-8222-222222222222' +LOCATION = 34734080 +MAC = '02:00:0a:3d:00:01' +ROUTES = 'Destination Gateway Flags Netif Expire\ndefault 192.0.2.1 UG en0\n' +CONFIG = 'en0: flags=8863\n\tether aa:bb:cc:dd:ee:ff\n\tinet 192.0.2.10 netmask 0xffffff00\nen9: flags=8863\n\tether ' + MAC + '\n' +ASSIGNED = CONFIG + '\tinet 10.61.0.1 netmask 0xfffffffc\n' + + +def gadget(): + return {'IOObjectClass': 'IOUSBHostDevice', 'idVendor': 0x05ac, 'idProduct': 0x4144, + 'locationID': LOCATION, 'USB Product Name': 'Orchard n61 NFS', + 'USB Serial Number': 'PROFILE:n61 ROOT:' + ROOT + ' BOOT:' + BOOT, + 'IORegistryEntryChildren': [ + {'IOObjectClass': 'IOSerialBSDClient', 'IOCalloutDevice': '/dev/cu.usbmodem-test'}, + {'IOObjectClass': 'IOEthernetInterface', 'BSD Name': 'en9', + 'IOMACAddress': bytes.fromhex('02000a3d0001')}]} + + +def data(*devices): + return plistlib.dumps(list(devices)) + + +class GadgetSelection(unittest.TestCase): + def test_requires_selected_root_boot_and_unique_device(self): + endpoint = nfs_host.select_macos(data(gadget()), LOCATION, ROOT, expected_boot=BOOT) + self.assertEqual((endpoint['interface'], endpoint['mac'], endpoint['boot_id']), ('en9', MAC, BOOT)) + with self.assertRaisesRegex(ValueError, 'boot identity'): + nfs_host.select_macos(data(gadget()), LOCATION, ROOT, expected_boot=ROOT) + with self.assertRaisesRegex(ValueError, 'root identity'): + nfs_host.select_macos(data(gadget()), LOCATION, BOOT) + other = {'IOObjectClass': 'IOUSBHostDevice', 'idVendor': 0x05ac, + 'idProduct': 0x1227, 'locationID': LOCATION + 1} + with self.assertRaisesRegex(ValueError, 'exactly one boot device'): + nfs_host.select_macos(data(gadget(), other), LOCATION, ROOT) + + def test_nested_device_cannot_supply_the_parents_serial_port(self): + parent = gadget() + parent['IORegistryEntryChildren'] = [parent['IORegistryEntryChildren'][1], { + 'IOObjectClass': 'IOUSBHostDevice', 'idVendor': 0x05ac, 'idProduct': 0x1000, + 'IORegistryEntryChildren': [{'IOObjectClass': 'IOSerialBSDClient', + 'IOCalloutDevice': '/dev/cu.unrelated'}]}] + with self.assertRaisesRegex(ValueError, 'serial endpoint'): + nfs_host.select_macos(data(parent), LOCATION, ROOT) + + def test_invalid_usb_tree_cannot_select_a_device(self): + with self.assertRaisesRegex(ValueError, 'registry data'): + nfs_host.select_macos(plistlib.dumps({'unexpected': True}), LOCATION, ROOT) + with self.assertRaisesRegex(ValueError, 'registry tree'): + nfs_host.select_macos(plistlib.dumps(['unexpected']), LOCATION, ROOT) + + +class HostLink(unittest.TestCase): + def setUp(self): + self.endpoint = nfs_host.select_macos(data(gadget()), LOCATION, ROOT, expected_boot=BOOT) + self.owner = {'schema': 1, 'kind': 'orchard.n61-nfs-host-link', 'state': 'owned', + 'address': '10.61.0.1/30', 'root_identity_sha256': 'a' * 64, **self.endpoint} + self.tmp = tempfile.TemporaryDirectory() + self.addCleanup(self.tmp.cleanup) + self.state = Path(self.tmp.name) / 'link.json' + + def test_plans_owned_alias_and_rejects_foreign_addresses_and_routes(self): + self.assertEqual(nfs_host.plan(self.endpoint, CONFIG, ROUTES)['action'], 'add-alias') + with self.assertRaisesRegex(ValueError, 'without an ownership receipt'): + nfs_host.plan(self.endpoint, ASSIGNED, ROUTES) + own_route = ROUTES + '10.61.0/30 link#9 UCS en9\n' + self.assertEqual(nfs_host.plan(self.endpoint, ASSIGNED, own_route, self.owner)['action'], 'already-owned') + with self.assertRaisesRegex(ValueError, 'another session'): + nfs_host.plan(self.endpoint, ASSIGNED, own_route, {**self.owner, 'boot_id': ROOT}) + with self.assertRaisesRegex(ValueError, 'route conflicts'): + nfs_host.plan(self.endpoint, CONFIG, ROUTES + '10.61.0/30 link#8 UCS en8\n') + with self.assertRaisesRegex(ValueError, 'address conflicts'): + nfs_host.plan(self.endpoint, CONFIG + '\tinet 10.61.0.2 netmask 0xfffffffc\n', ROUTES) + + def test_failed_alias_operation_leaves_uncertain_intent(self): + first = {'action': 'add-alias', 'endpoint': self.endpoint, 'address': '10.61.0.1/30', + 'root_identity_sha256': 'a' * 64, 'command': ['sudo', 'ifconfig']} + def run(argv, **kwargs): + if argv == first['command']: + raise subprocess.CalledProcessError(1, argv) + return SimpleNamespace(returncode=0) + with patch.object(nfs_host, 'inspect', return_value=first), patch.object(nfs_host.subprocess, 'run', side_effect=run): + with self.assertRaises(subprocess.CalledProcessError): + nfs_host.apply(Path('/root-id'), LOCATION, BOOT, self.state) + self.assertEqual(json.loads(self.state.read_text())['state'], 'adding') + self.assertEqual(self.state.stat().st_mode & 0o777, 0o600) + + def test_cleanup_refuses_active_gadget_and_retires_only_owned_alias(self): + self.state.write_text(json.dumps(self.owner)) + self.state.chmod(0o600) + current = {'connected': True, 'assigned': True, 'removed': False} + def run(argv, **kwargs): + if argv[0] == '/usr/sbin/ioreg': + return SimpleNamespace(stdout=data(gadget()) if current['connected'] else data()) + if argv[0] == '/sbin/ifconfig': + return SimpleNamespace(stdout=ASSIGNED if current['assigned'] else CONFIG) + if argv[:3] == ['/usr/bin/sudo', '-n', '/sbin/ifconfig']: + current['removed'] = True + current['assigned'] = False + return SimpleNamespace(returncode=0) + with patch.object(nfs_host.sys, 'platform', 'darwin'), \ + patch.object(nfs_host, 'root_identity', return_value=({'root_id': ROOT}, 'a' * 64)), \ + patch.object(nfs_host.subprocess, 'run', side_effect=run): + with self.assertRaisesRegex(ValueError, 'Disconnect the NFS gadget'): + nfs_host.cleanup(Path('/root-id'), self.state) + self.assertFalse(current['removed']) + current['connected'] = False + self.assertEqual(nfs_host.cleanup(Path('/root-id'), self.state)['action'], 'retired') + self.assertTrue(current['removed']) + self.assertEqual(json.loads(self.state.read_text())['state'], 'retired') + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/test_package_boundaries.py b/tests/test_package_boundaries.py index c7caee5..323d66d 100644 --- a/tests/test_package_boundaries.py +++ b/tests/test_package_boundaries.py @@ -40,7 +40,7 @@ def find_spec(self, name, path=None, target=None): from orchard_hardware.apple import adt from orchard_cli.command import main sys.argv=['orchard','distros'];main() -sys.argv=['orchard','resolve',MANIFEST];main() +sys.argv=['orchard','resolve',MANIFEST,'--profile','j71b'];main() '''.replace('MANIFEST', repr(str(FIXTURE.resolve()))) with tempfile.TemporaryDirectory() as store: result = subprocess.run([sys.executable, '-I', '-c', script], cwd=store, diff --git a/tests/test_pongo_preflight.py b/tests/test_pongo_preflight.py new file mode 100644 index 0000000..ac7c7df --- /dev/null +++ b/tests/test_pongo_preflight.py @@ -0,0 +1,33 @@ +"""Preflight checks the library without interacting with USB devices.""" +from types import SimpleNamespace +import unittest +from unittest.mock import MagicMock, patch + +from orchard_cli.transports import pongo + + +class PongoPreflightTests(unittest.TestCase): + def test_preflight_binds_symbols_without_initializing_usb(self): + library = MagicMock() + with patch.object(pongo, 'library_path', return_value=library): + pongo.preflight() + library.libusb_init.assert_not_called() + library.libusb_get_device_list.assert_not_called() + library.libusb_open.assert_not_called() + + def test_preflight_rejects_a_missing_library(self): + with patch.object(pongo, 'find_library', return_value=None), \ + patch.object(pongo.shutil, 'which', return_value=None), \ + patch.object(pongo.C, 'CDLL', side_effect=OSError('not found')) as load, \ + self.assertRaisesRegex(pongo.PongoError, 'libusb-1.0 is unavailable'): + pongo.preflight() + load.assert_called() + + def test_preflight_rejects_an_incompatible_library(self): + with patch.object(pongo, 'library_path', return_value=SimpleNamespace()), \ + self.assertRaisesRegex(pongo.PongoError, 'missing required symbol libusb_init'): + pongo.preflight() + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/test_provenance.py b/tests/test_provenance.py new file mode 100644 index 0000000..a1f5bac --- /dev/null +++ b/tests/test_provenance.py @@ -0,0 +1,47 @@ +"""Build provenance distinguishes clean checkouts, local edits and installed files.""" +from pathlib import Path +import subprocess +import tempfile +import unittest + +from orchard_cli import provenance + + +class SourceProvenance(unittest.TestCase): + def setUp(self): + temporary = tempfile.TemporaryDirectory() + self.addCleanup(temporary.cleanup) + self.root = Path(temporary.name).resolve() + self.package = self.root / 'package' + self.package.mkdir() + (self.package / '__init__.py').write_text('first\n') + + def test_installed_files_do_not_claim_a_git_revision(self): + result = provenance.snapshot(self.package) + self.assertEqual(result['kind'], 'installed') + self.assertNotIn('git', result) + + def test_checkout_records_head_and_flags_edits(self): + subprocess.run(['git', 'init', '-q', str(self.root)], check=True) + subprocess.run(['git', '-C', str(self.root), 'add', 'package'], check=True) + subprocess.run(['git', '-C', str(self.root), '-c', 'user.name=Test', + '-c', 'user.email=test@example.invalid', '-c', 'commit.gpgsign=false', + 'commit', '-qm', 'fixture'], check=True) + clean = provenance.snapshot(self.package) + self.assertEqual(clean['kind'], 'checkout') + self.assertTrue(clean['git']['clean']) + self.assertEqual(len(clean['git']['head']), 40) + (self.package / '__init__.py').write_text('changed\n') + dirty = provenance.snapshot(self.package) + self.assertFalse(dirty['git']['clean']) + self.assertEqual(dirty['git']['head'], clean['git']['head']) + self.assertNotEqual(dirty['sha256'], clean['sha256']) + + def test_linked_source_files_are_refused(self): + (self.package / 'linked').symlink_to(self.package / '__init__.py') + with self.assertRaisesRegex(ValueError, 'Linked'): + provenance.snapshot(self.package) + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/test_ram.py b/tests/test_ram.py index b059047..3370a21 100644 --- a/tests/test_ram.py +++ b/tests/test_ram.py @@ -1,6 +1,7 @@ """RAM building/boot tests. Synthetic components; real packer, mocked USB.""" from contextlib import nullcontext import gzip +import hashlib import io import json from pathlib import Path @@ -15,6 +16,7 @@ import orchard_hardware from orchard_cli import boot, build, ram +from orchard_cli.transports import pongo class RamTests(unittest.TestCase): @@ -60,6 +62,10 @@ def setUp(self): tools = patch.object(build, 'prepare_tools', return_value=self.tools) tools.start() self.addCleanup(tools.stop) + self.real_pongo_preflight = pongo.preflight + preflight = patch.object(pongo, 'preflight') + self.pongo_preflight = preflight.start() + self.addCleanup(preflight.stop) def build(self, **kwargs): options = dict(init=self.init, files={'usr/bin/test': self.script}, modules=True, device='j71b') @@ -79,6 +85,41 @@ def test_real_packaging_and_repeat(self): self.build() ram.verify_ram(self.output) + def test_receipt_records_sources_and_detects_changed_provenance_artifacts(self): + (self.kernel / 'dtbs').mkdir() + (self.kernel / 'dtbs/t8010-j71b.dtb').write_bytes((self.kernel / 'j71b.dtb').read_bytes()) + files = {} + for path in self.kernel.rglob('*'): + if path.is_file(): + files[path.relative_to(self.kernel).as_posix()] = { + 'bytes': path.stat().st_size, 'sha256': build.digest(path)} + recipe_files = {'devices/j71b/device.toml': {'bytes': 1, 'sha256': 'e' * 64}} + recipe_digest = hashlib.sha256(json.dumps(recipe_files, sort_keys=True, separators=(',', ':')).encode()).hexdigest() + kernel_receipt = {'schema': 1, 'device': 'j71b', 'dtb': 't8010-j71b', 'kernel_release': 'test-release', + 'source': {'commit': 'a' * 40, 'tree': 'f' * 40, 'prepared_tree': 'b' * 40, + 'patch_list_sha256': 'c' * 64, 'config_sha256': 'd' * 64}, + 'hardware_recipe': {'sha256': recipe_digest, 'files': recipe_files}, 'files': files} + (self.kernel / 'receipt.json').write_text(json.dumps(kernel_receipt)) + self.build() + receipt = ram.verify_ram(self.output) + evidence = receipt['provenance'] + self.assertEqual(evidence['kernel']['source']['tree'], 'f' * 40) + self.assertEqual(evidence['kernel']['receipt_sha256'], build.digest(self.output / 'kernel-receipt.json')) + self.assertEqual(evidence['firmware']['recipe_sha256'], build.digest(self.output / 'firmware-recipe.json')) + self.assertFalse(evidence['firmware']['included_via_recipe']) + self.assertIn('sha256', evidence['hardware']) + self.assertIn('sha256', evidence['cli']['code']) + for name in ('kernel-receipt.json', 'firmware-recipe.json'): + path = self.output / name + original = path.read_bytes() + path.write_bytes(original + b'changed') + with self.subTest(name=name), self.assertRaisesRegex(ValueError, 'changed'): + ram.verify_ram(self.output) + path.write_bytes(original) + (self.kernel / 'Image.gz').write_bytes(b'changed after receipt') + with self.assertRaisesRegex(ValueError, 'component differs'): + self.build() + def test_custom_init_and_no_modules(self): custom = self.base / 'init' custom.write_text('#!/bin/busybox sh\necho custom\n') @@ -98,6 +139,12 @@ def test_changed_payload_and_source_are_refused(self): with self.assertRaises(ValueError): ram.verify_ram(self.output) + def test_changed_cli_source_during_assembly_rejects_the_build(self): + with patch.object(ram.provenance, 'cli', side_effect=[{'source': 'first'}, {'source': 'changed'}]), \ + self.assertRaisesRegex(ValueError, 'source changed during RAM assembly'): + self.build() + self.assertFalse((self.output / 'build.json').exists()) + def test_host_launcher_edit_does_not_invalidate_build(self): self.build() (self.base / 'run.py').write_text('# host-only change') @@ -180,6 +227,7 @@ def test_pongo_boot_uses_no_rpc_or_readiness_claim(self): patch('builtins.input', return_value=''), patch('builtins.print') as output: boot.boot_ram(self.output) self.assertEqual(selected.call_count, 2) + self.pongo_preflight.assert_called_once() upload.assert_called_once() self.assertIn('NOT been verified', str(output.call_args_list)) @@ -331,14 +379,66 @@ def test_mismatched_dfu_boot_tracks_actual_device_through_pongo(self): patch('orchard_cli.state.operation_lock', return_value=nullcontext()), \ patch('orchard_cli.apple.boot.wait_mode') as wait, \ patch.object(boot.shutil, 'which', return_value='/fake/palera1n'), \ - patch.object(boot.subprocess, 'run') as execute: + patch('orchard_cli.apple.boot.run_process_group') as execute: boot.boot_ram(self.output, yes=True) selected.assert_any_call('fake', 123, chip=0x8011, board=0x04, mode='dfu') selected.assert_any_call('fake', 123, chip=0x8011, board=0x04) wait.assert_called_once_with(device, 'pongo', chip=0x8011, board=0x04, product='', seconds=60) execute.assert_called_once() + self.assertEqual(execute.call_args.args[0], ['/fake/palera1n', '-p', '-k', str(self.output / 'pongo.bin')]) + self.assertEqual(execute.call_args.kwargs['timeout'], 120) + self.assertEqual(Path(execute.call_args.kwargs['stdout'].name), self.output / 'palera1n.log') + self.assertEqual(execute.call_args.kwargs['stderr'], subprocess.STDOUT) + self.pongo_preflight.assert_called_once() upload.assert_called_once() + def test_missing_libusb_stops_before_bootstrap_or_pongo_upload(self): + self.build() + for mode in ('dfu', 'pongo'): + device = dict(mode=mode, chip=0x8010, board=0x18, ecid=123, location='fake') + with self.subTest(mode=mode), \ + patch.object(pongo, 'preflight', side_effect=self.real_pongo_preflight), \ + patch.object(pongo, 'library_path', side_effect=pongo.PongoError('libusb-1.0 is unavailable')), \ + patch.object(pongo.LibUSB, 'open') as open_usb, \ + patch('orchard_cli.apple.usb.sole_target', return_value=device), \ + patch('orchard_cli.apple.usb.selected') as selected, \ + patch('orchard_cli.transports.pongo.upload') as upload, \ + patch('orchard_cli.state.operation_lock', return_value=nullcontext()), \ + patch('orchard_cli.apple.boot.wait_mode') as wait, \ + patch.object(boot, 'verify_ram', wraps=ram.verify_ram) as verify, \ + patch.object(boot.shutil, 'which', return_value='/fake/palera1n'), \ + patch('orchard_cli.apple.boot.run_process_group') as execute, \ + self.assertRaisesRegex(pongo.PongoError, 'libusb-1.0 is unavailable'): + boot.boot_ram(self.output, yes=True) + self.assertEqual(verify.call_count, 2) + selected.assert_called_once_with('fake', 123, chip=0x8010, board=0x18, mode=mode) + execute.assert_not_called() + wait.assert_not_called() + upload.assert_not_called() + open_usb.assert_not_called() + self.assertFalse((self.output / 'palera1n.log').exists()) + + def test_dfu_bootstrap_failure_never_waits_or_uploads(self): + self.build() + device = dict(mode='dfu', chip=0x8010, board=0x18, ecid=123, location='fake') + failures = (subprocess.TimeoutExpired('palera1n', 120), + subprocess.CalledProcessError(2, 'palera1n'), KeyboardInterrupt()) + for failure in failures: + with self.subTest(error=type(failure).__name__), \ + patch('orchard_cli.apple.usb.sole_target', return_value=device), \ + patch('orchard_cli.apple.usb.selected') as selected, \ + patch('orchard_cli.transports.pongo.upload') as upload, \ + patch('orchard_cli.state.operation_lock', return_value=nullcontext()), \ + patch('orchard_cli.apple.boot.wait_mode') as wait, \ + patch.object(boot.shutil, 'which', return_value='/fake/palera1n'), \ + patch('orchard_cli.apple.boot.run_process_group', side_effect=failure) as execute: + with self.assertRaises(type(failure)): + boot.boot_ram(self.output, yes=True) + execute.assert_called_once() + selected.assert_called_once_with('fake', 123, chip=0x8010, board=0x18, mode='dfu') + wait.assert_not_called() + upload.assert_not_called() + def test_mismatch_refusal_stops_before_usb_mutation(self): self.build() device = dict(mode='pongo', chip=0x8011, board=0x04, ecid=123, location='fake') @@ -475,6 +575,9 @@ def test_directory_build_firmware_follows_the_recipe(self): archive = gzip.decompress((directory / 'out/initramfs.cpio.gz').read_bytes()) self.assertIn(b'lib/firmware/brcm/test.bin', archive) self.assertIn(b'firmware blob', archive) + evidence = ram.verify_ram(directory / 'out')['provenance']['firmware'] + self.assertTrue(evidence['included_via_recipe']) + self.assertEqual(evidence['recipe_sha256'], build.digest(directory / 'out/firmware-recipe.json')) (directory / 'rootfs/lib/firmware/brcm').mkdir(parents=True) (directory / 'rootfs/lib/firmware/brcm/test.bin').write_bytes(b'mine') with patch.object(build, 'prepare_components', return_value=self.components), \ diff --git a/tests/test_runtime.py b/tests/test_runtime.py index 91aae7d..055346f 100644 --- a/tests/test_runtime.py +++ b/tests/test_runtime.py @@ -4,6 +4,7 @@ import json import os from pathlib import Path +import signal import shutil import subprocess import sys @@ -35,6 +36,67 @@ def tearDown(self): self.cache.stop() self.temporary.cleanup() + def assert_process_stopped(self, pid): + def exists(): + try: + os.kill(pid, 0) + except ProcessLookupError: + return False + except PermissionError as error: + self.fail(f'Cannot inspect test process {pid}: {error}') + return True + + for _ in range(50): + if not exists(): + return + try: + observed = subprocess.run(['ps', '-o', 'stat=', '-p', str(pid)], + capture_output=True, text=True) + except OSError as error: + self.fail(f'Cannot inspect test process {pid}: {error}') + state = observed.stdout.strip() + if observed.stderr.strip(): + self.fail(f'Cannot inspect test process {pid}: {observed.stderr.strip()}') + if observed.returncode: + # ps returns 1 with empty output for an absent PID on macOS and + # Linux. Check the PID again to distinguish an exit race. + if observed.returncode == 1 and not state and not exists(): + return + self.fail(f'Cannot inspect test process {pid}: ps exited {observed.returncode}') + if not state or len(state.split()) != 1: + self.fail(f'Cannot inspect test process {pid}: invalid ps output {state!r}') + if state.startswith(('Z', 'X')): + return + time.sleep(0.05) + self.fail('bootstrap child remained running') + + def test_process_check_accepts_only_absence_or_zombie(self): + with patch('os.kill', side_effect=ProcessLookupError), patch('subprocess.run') as ps: + self.assert_process_stopped(123) + ps.assert_not_called() + with patch('os.kill'), patch('subprocess.run', return_value= + subprocess.CompletedProcess([], 0, 'Z\n', '')): + self.assert_process_stopped(123) + with patch('os.kill', side_effect=[None, ProcessLookupError]), \ + patch('subprocess.run', return_value=subprocess.CompletedProcess([], 1, '', '')): + self.assert_process_stopped(123) + + def test_process_check_rejects_inspection_errors(self): + for result in (subprocess.CompletedProcess([], 1, '', 'operation not permitted'), + subprocess.CompletedProcess([], 1, '', ''), + subprocess.CompletedProcess([], 0, '', '')): + with self.subTest(result=result), patch('os.kill'), patch('subprocess.run', return_value=result), \ + self.assertRaisesRegex(AssertionError, 'Cannot inspect test process'): + self.assert_process_stopped(123) + with patch('os.kill', side_effect=PermissionError('operation not permitted')), \ + self.assertRaisesRegex(AssertionError, 'Cannot inspect test process'): + self.assert_process_stopped(123) + with patch('os.kill') as probe, patch('time.sleep'), patch('subprocess.run', return_value= + subprocess.CompletedProcess([], 0, 'S\n', '')), \ + self.assertRaisesRegex(AssertionError, 'remained running'): + self.assert_process_stopped(123) + self.assertTrue(all(call.args == (123, 0) for call in probe.call_args_list)) + def test_adapter_boot_sequence_timeouts_and_no_retry_are_preserved(self): from orchard_cli import devices from orchard_cli.apple import device as adapter_module @@ -48,8 +110,10 @@ def test_adapter_boot_sequence_timeouts_and_no_retry_are_preserved(self): patch.object(adapter_module.usb, 'selected') as selected, \ patch.object(boot, 'wait_mode') as wait, \ patch.object(boot, 'run_process_group') as bootstrap, \ + patch.object(boot.pongo, 'preflight') as preflight, \ patch.object(boot.pongo, 'upload') as upload: adapter.boot(device, assets, record, Mock()) + preflight.assert_called_once() self.assertEqual(bootstrap.call_count, 1) self.assertEqual(bootstrap.call_args.args[0], ['/tools/palera1n', '-p', '-k', str(assets['pongo'])]) self.assertEqual(bootstrap.call_args.kwargs['timeout'], 120) @@ -63,26 +127,109 @@ def test_adapter_boot_sequence_timeouts_and_no_retry_are_preserved(self): self.assertEqual(bootstrap.call_count, 2) # one success, one failed call; no retry. upload.assert_not_called() + def test_adapter_missing_libusb_stops_before_bootstrap_or_upload(self): + from orchard_cli.apple import boot + from orchard_cli.transports import pongo + assets = {'pongo': self.root / 'pongo', 'recovery': self.root / 'boot'} + record = {'assets': {'recovery': {'sha256': 'expected-digest'}}} + for mode in ('dfu', 'pongo'): + device = {'mode': mode, 'location': 'selected-port', 'ecid': 123} + with self.subTest(mode=mode), \ + patch.object(pongo, 'library_path', side_effect=pongo.PongoError('libusb-1.0 is unavailable')), \ + patch.object(pongo.LibUSB, 'open') as open_usb, \ + patch.object(boot.shutil, 'which', return_value='/tools/palera1n'), \ + patch.object(boot.usb, 'selected') as selected, \ + patch.object(boot, 'wait_mode') as wait, \ + patch.object(boot, 'run_process_group') as bootstrap, \ + patch.object(pongo, 'upload') as upload, \ + self.assertRaisesRegex(pongo.PongoError, 'libusb-1.0 is unavailable'): + boot.boot(device, assets, record, Mock(), profile='j71b', chip=1, board=2, product='fixture') + if mode == 'dfu': + selected.assert_called_once_with('selected-port', 123, chip=1, board=2, mode='dfu') + else: + selected.assert_not_called() + bootstrap.assert_not_called() + wait.assert_not_called() + upload.assert_not_called() + open_usb.assert_not_called() + + def test_adapter_running_recovery_does_not_need_libusb(self): + from orchard_cli.apple import boot + with patch.object(boot.pongo, 'preflight') as preflight, \ + patch.object(boot.pongo, 'upload') as upload: + boot.boot({'mode': 'linux'}, {}, {}, Mock(), profile='j71b', chip=1, board=2, product='fixture') + preflight.assert_not_called() + upload.assert_not_called() + def test_bootstrap_timeout_stops_spawned_children(self): # palera1n posix_spawns checkra1n; a surviving checkra1n would exploit # the next DFU device behind a later command's back. from orchard_cli.apple import boot pidfile = self.root / 'child.pid' - script = f'sleep 300 & echo $! > {pidfile}; wait' - with self.assertRaises(subprocess.TimeoutExpired): - boot.run_process_group(['sh', '-c', script], timeout=1) - child = int(pidfile.read_text()) - for _ in range(50): - try: - state = Path(f'/proc/{child}/stat').read_text().rsplit(')', 1)[1].split()[0] - except (FileNotFoundError, ProcessLookupError): # gone mid-read - break - if state == 'Z': - break - time.sleep(0.05) - else: - os.kill(child, 9) - self.fail('bootstrap child outlived the timeout') + groupfile = self.root / 'group.pid' + script = 'echo $$ > "$2"; sleep 300 & echo $! > "$1"; wait' + try: + with self.assertRaises(subprocess.TimeoutExpired): + boot.run_process_group(['sh', '-c', script, 'fixture', str(pidfile), str(groupfile)], timeout=1) + self.assert_process_stopped(int(pidfile.read_text())) + finally: + if groupfile.exists(): + try: + os.killpg(int(groupfile.read_text()), signal.SIGKILL) + except ProcessLookupError: + pass + + def test_bootstrap_interruption_stops_spawned_children(self): + child_file = self.root / 'interrupt-child.pid' + group_file = self.root / 'interrupt-group.pid' + runner = ''' +import sys +from orchard_cli.apple.boot import run_process_group +script = 'echo $$ > "$2"; sleep 300 & echo $! > "$1"; wait' +try: + run_process_group(['sh', '-c', script, 'fixture', sys.argv[1], sys.argv[2]], timeout=10) +except KeyboardInterrupt: + sys.exit(0) +sys.exit(2) +''' + worker = subprocess.Popen([sys.executable, '-c', runner, str(child_file), str(group_file)], + stdout=subprocess.PIPE, stderr=subprocess.PIPE) + try: + deadline = time.monotonic() + 5 + while not child_file.exists() and worker.poll() is None and time.monotonic() < deadline: + time.sleep(0.02) + self.assertTrue(child_file.exists(), 'bootstrap child did not start') + child = int(child_file.read_text()) + os.kill(worker.pid, signal.SIGINT) + stdout, stderr = worker.communicate(timeout=5) + self.assertEqual(worker.returncode, 0, (stdout + stderr).decode()) + self.assert_process_stopped(child) + finally: + if worker.poll() is None: + worker.kill() + worker.wait() + if group_file.exists(): + try: + os.killpg(int(group_file.read_text()), signal.SIGKILL) + except ProcessLookupError: + pass + + def test_bootstrap_nonzero_exit_stops_spawned_children(self): + from orchard_cli.apple import boot + pidfile = self.root / 'failed-child.pid' + groupfile = self.root / 'failed-group.pid' + script = 'echo $$ > "$2"; sleep 300 & echo $! > "$1"; exit 7' + try: + with self.assertRaises(subprocess.CalledProcessError) as failure: + boot.run_process_group(['sh', '-c', script, 'fixture', str(pidfile), str(groupfile)], timeout=5) + self.assertEqual(failure.exception.returncode, 7) + self.assert_process_stopped(int(pidfile.read_text())) + finally: + if groupfile.exists(): + try: + os.killpg(int(groupfile.read_text()), signal.SIGKILL) + except ProcessLookupError: + pass def test_bootstrap_failure_is_reported(self): from orchard_cli.apple import boot @@ -278,7 +425,7 @@ def guard(event, args): sys.addaudithook(guard) from orchard_cli.command import main -sys.argv = ['orchard', 'resolve', INPUT] +sys.argv = ['orchard', 'resolve', INPUT, '--profile', 'j71b'] main() '''.replace('BLOCKED', repr([str(path) for path in blocked])).replace('INPUT', repr(str(inputs / 'index.toml'))) env = {**os.environ, 'ORCHARD_HOME': str(self.root / 'isolated-store'), 'PYTHONDONTWRITEBYTECODE': '1'} diff --git a/tests/test_shared_build_cache.py b/tests/test_shared_build_cache.py index 92ecfa0..ad06a8c 100644 --- a/tests/test_shared_build_cache.py +++ b/tests/test_shared_build_cache.py @@ -2,10 +2,14 @@ from contextlib import redirect_stdout import io import hashlib +import json import os from pathlib import Path import shutil +import subprocess +import sys import tempfile +import time import unittest from unittest.mock import patch @@ -85,6 +89,32 @@ def prepare(self, hardware): self.produce(hw, cache, device, out, base)): return build.prepare_components(hardware, self.cache, device='j71b', **self.overrides) + def test_recovery_tools_cache_tracks_tool_inputs_not_runtime(self): + for relative in build.TOOL_INPUTS: + path = self.one / relative + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text('original tool input') + runtime = self.one / 'boot/initramfs/recovery/init' + runtime.write_text('original runtime') + calls = [] + + def produce_tools(*args, env): + calls.append(args) + tools = Path(env['OMA_RECOVERY_TOOLS_WORK']) / 'bin' + tools.mkdir(parents=True) + (tools / 'busybox').write_text('built tool') + + with patch.object(build, 'command', side_effect=produce_tools): + first = build.prepare_tools(self.one, self.cache, device='j71b') + runtime.write_text('changed runtime') + self.assertEqual(build.prepare_tools(self.one, self.cache, device='j71b'), first) + self.assertEqual(len(calls), 1) + (self.one / build.TOOL_INPUTS[0]).write_text('changed tool input') + second = build.prepare_tools(self.one, self.cache, device='j71b') + self.assertNotEqual(first, second) + self.assertEqual(len(calls), 2) + self.assertEqual((second / 'busybox').read_text(), 'built tool') + def test_fresh_second_checkout_neither_downloads_checks_out_nor_compiles_linux(self): first = self.prepare(self.one) self.assertEqual(self.calls, ['kernel-base', 'external-modules-and-dtb']) @@ -166,16 +196,20 @@ def test_external_driver_change_reuses_kernel_base_and_sdk(self): self.assertEqual((second['kernel'] / 'lib/modules/test-release/external.ko').read_text(), 'changed external driver') - def test_readme_other_device_and_firmware_changes_do_not_rebuild_kernel(self): + def test_device_firmware_changes_refresh_receipt_without_rebuilding_kernel_base(self): first = self.prepare(self.one) - for relative in ('README.md', 'devices/j71b/firmware.json', 'kernel/configs/other/config', - 'kernel/modules/other/driver.c'): + for relative in ('README.md', 'kernel/configs/other/config', 'kernel/modules/other/driver.c'): path = self.two / relative path.parent.mkdir(parents=True, exist_ok=True) path.write_text('unrelated change') second = self.prepare(self.two) self.assertEqual(first['kernel'], second['kernel']) self.assertEqual(len(self.calls), 2) + (self.two / 'devices/j71b/firmware.json').write_text('changed firmware recipe') + third = self.prepare(self.two) + self.assertNotEqual(first['kernel'], third['kernel']) + self.assertEqual(self.calls.count('kernel-base'), 1) + self.assertEqual(self.calls.count('external-modules-and-dtb'), 2) def test_config_and_local_in_tree_edits_rebuild_the_base_without_reset(self): self.prepare(self.one) @@ -206,6 +240,234 @@ def test_dtb_selection_only_rebuilds_device_layer(self): self.assertEqual(self.calls.count('kernel-base'), 1) self.assertEqual(self.calls.count('external-modules-and-dtb'), 2) + def test_page_size_platforms_keep_separate_kernels_and_module_sdks(self): + hardware_layout.write(self.one, device='j71b', platform='apple-16k', dtb='t8010-j71b') + hardware_layout.write(self.one, device='n61', platform='apple-4k', dtb='t7000-n61') + (self.one / 'kernel/configs/apple-16k/config').write_text('CONFIG_ARM64_16K_PAGES=y\n') + (self.one / 'kernel/configs/apple-4k/config').write_text('CONFIG_ARM64_4K_PAGES=y\n') + calls = [] + + def produce_base(hardware, cache, device, output): + platform = kernel.layout(hardware, device)[0] + calls.append(('base', platform)) + output.mkdir(parents=True) + (output / 'Image.gz').write_text(platform) + (output / 'kernel.release').write_text('test-' + platform) + (output / 'sdk').mkdir() + (output / 'sdk/Module.symvers').write_text(platform) + (output / 'dtbs').mkdir() + dtb = kernel.layout(hardware, device)[1] + (output / 'dtbs' / (dtb + '.dtb')).write_text(dtb + ':' + platform) + + def produce_device(hardware, cache, device, base, output): + platform = kernel.layout(hardware, device)[0] + self.assertEqual((base / 'sdk/Module.symvers').read_text(), platform) + calls.append(('device', device)) + shutil.copytree(base, output) + dtb = kernel.layout(hardware, device)[1] + shutil.copyfile(base / 'dtbs' / (dtb + '.dtb'), output / (device + '.dtb')) + + with patch.object(kernel, 'build_base', side_effect=produce_base), \ + patch.object(kernel, 'build_device', side_effect=produce_device): + large_pages = build.prepare_kernel(self.one, self.cache, device='j71b') + small_pages = build.prepare_kernel(self.one, self.cache, device='n61') + reused = build.prepare_kernel(self.one, self.cache, device='n61') + + self.assertNotEqual(large_pages, small_pages) + self.assertEqual(reused, small_pages) + self.assertEqual((large_pages / 'kernel.release').read_text(), 'test-apple-16k') + self.assertEqual((small_pages / 'kernel.release').read_text(), 'test-apple-4k') + self.assertEqual((large_pages / 'j71b.dtb').read_text(), 't8010-j71b:apple-16k') + self.assertEqual((small_pages / 'n61.dtb').read_text(), 't7000-n61:apple-4k') + self.assertEqual(calls, [('base', 'apple-16k'), ('device', 'j71b'), + ('base', 'apple-4k'), ('device', 'n61')]) + + def test_loader_recipe_and_device_select_isolated_resumable_workspaces(self): + builder = self.one / 'boot/m1n1/build' + builder.parent.mkdir(parents=True) + builder.write_text('#!' + sys.executable + '\n' + ''' +import json +from pathlib import Path +import subprocess +import sys + +args = sys.argv[1:] +device = args[0] +work = Path(args[args.index('--work-dir') + 1]) +output = Path(args[args.index('--output') + 1]) +resume = '--resume' in args +assert (work / 'source').exists() if resume else not work.exists() +(work / 'source').mkdir(parents=True, exist_ok=resume) +marker = work / 'source/preserve.txt' +if not marker.exists(): + marker.write_text(device + ':' + work.name) + subprocess.run(['git', 'init', '-q', str(marker.parent)], check=True) + subprocess.run(['git', '-C', str(marker.parent), 'add', 'preserve.txt'], check=True) + subprocess.run(['git', '-C', str(marker.parent), '-c', 'user.name=Test', + '-c', 'user.email=test@example.invalid', '-c', 'commit.gpgsign=false', + 'commit', '-qm', 'loader fixture'], check=True) +if (work / 'change-during-build').exists(): + marker.write_text('changed during build') +log = Path(__file__).resolve().parents[2] / '.work/loader-calls.jsonl' +with log.open('a') as stream: + stream.write(json.dumps({'device': device, 'work': str(work), 'resume': resume}) + '\\n') +output.write_text(marker.read_text()) +''') + builder.chmod(0o755) + patch_file = builder.parent / 'patches/0001.patch' + patch_file.parent.mkdir() + patch_file.write_text('original loader recipe\n') + legacy = builder.parent / '.work/j71b/source/legacy.txt' + legacy.parent.mkdir(parents=True) + legacy.write_text('preserve earlier workspace\n') + overrides = {key: value for key, value in self.overrides.items() if key != 'loader'} + overrides['kernel'] = self.root / 'explicit-kernel' + + def prepare(device='j71b'): + return build.prepare_components(self.one, self.cache, device=device, **overrides) + + def calls(): + return [json.loads(line) for line in (self.one / '.work/loader-calls.jsonl').read_text().splitlines()] + + original_key, original_work = build._loader_recipe(self.one, 'j71b') + first = prepare() + self.assertEqual(calls(), [{'device': 'j71b', 'work': str(original_work), 'resume': False}]) + self.assertEqual(prepare()['loader'], first['loader']) + self.assertEqual(len(calls()), 1) + + source = original_work / 'source/preserve.txt' + original = source.read_text() + (original_work / 'compiler-output.o').write_text('preserved object') + source.write_text('editable loader source') + edited = prepare()['loader'] + self.assertNotEqual(edited, first['loader']) + self.assertEqual(edited.read_text(), 'editable loader source') + self.assertEqual(calls()[-1], {'device': 'j71b', 'work': str(original_work), 'resume': True}) + self.assertEqual(prepare()['loader'], edited) + self.assertEqual((original_work / 'compiler-output.o').read_text(), 'preserved object') + source.write_text(original) + self.assertEqual(prepare()['loader'], first['loader']) + + baseline = original_work / 'orchard-cli-source-tree' + recorded = baseline.read_text() + baseline.unlink() + source.write_text('unrecorded workspace edit') + unknown = prepare()['loader'] + self.assertNotEqual(unknown, first['loader']) + self.assertEqual(unknown.read_text(), 'unrecorded workspace edit') + self.assertFalse(baseline.exists()) + source.write_text(original) + baseline.write_text(recorded) + + source.write_text('edit before a build') + (original_work / 'change-during-build').touch() + with self.assertRaisesRegex(ValueError, 'source changed during build'): + prepare() + self.assertEqual(source.read_text(), 'changed during build') + (original_work / 'change-during-build').unlink() + source.write_text(original) + + shutil.rmtree(first['loader'].parent) + prepare() + self.assertEqual(calls()[-1], {'device': 'j71b', 'work': str(original_work), 'resume': True}) + preserved = (original_work / 'source/preserve.txt').read_bytes() + patch_file.write_text('changed loader recipe\n') + changed_key, _ = build._loader_recipe(self.one, 'j71b') + changed = prepare() + self.assertNotEqual(changed['loader'], first['loader']) + self.assertEqual(calls()[-1], {'device': 'j71b', + 'work': str(self.one / '.work/m1n1/j71b' / changed_key), 'resume': False}) + prepare('n61') + self.assertEqual(calls()[-1], {'device': 'n61', + 'work': str(self.one / '.work/m1n1/n61' / changed_key), 'resume': False}) + with patch.dict(os.environ, {'ORCHARD_BUILD_NATIVE':'1', 'ORCHARD_BUILD_ENV_ID':'toolchain-b'}): + alternate_key, alternate_work = build._loader_recipe(self.one, 'j71b') + alternate = prepare()['loader'] + self.assertNotEqual(alternate_key, changed_key) + self.assertNotEqual(alternate, changed['loader']) + self.assertEqual(calls()[-1], {'device':'j71b', 'work':str(alternate_work), 'resume':False}) + self.assertEqual((original_work / 'source/preserve.txt').read_bytes(), preserved) + self.assertEqual(legacy.read_text(), 'preserve earlier workspace\n') + + def test_loader_rejects_dirty_submodules_without_changing_them(self): + sub = self.root / 'submodule' + source = self.root / 'loader-repo' + for directory in (sub, source): + subprocess.run(['git', 'init', '-q', str(directory)], check=True) + (sub / 'driver.c').write_text('original\n') + subprocess.run(['git', '-C', str(sub), 'add', 'driver.c'], check=True) + subprocess.run(['git', '-C', str(sub), '-c', 'user.name=Test', + '-c', 'user.email=test@example.invalid', '-c', 'commit.gpgsign=false', + 'commit', '-qm', 'fixture'], check=True) + subprocess.run(['git', '-C', str(source), '-c', 'protocol.file.allow=always', + 'submodule', 'add', '-q', str(sub), 'dependency'], check=True) + subprocess.run(['git', '-C', str(source), '-c', 'user.name=Test', + '-c', 'user.email=test@example.invalid', '-c', 'commit.gpgsign=false', + 'commit', '-qm', 'loader'], check=True) + build._loader_source_tree(source) + local_edit = source / 'dependency/driver.c' + local_edit.write_text('local experiment\n') + with self.assertRaisesRegex(ValueError, 'submodule has local changes'): + build._loader_source_tree(source) + self.assertEqual(local_edit.read_text(), 'local experiment\n') + + def test_loader_workspace_is_serialized_across_distinct_component_caches(self): + builder = self.one / 'boot/m1n1/build' + builder.parent.mkdir(parents=True) + builder.write_text('#!' + sys.executable + '\n' + ''' +import fcntl +from pathlib import Path +import subprocess +import sys +import time + +args = sys.argv[1:] +work = Path(args[args.index('--work-dir') + 1]) +output = Path(args[args.index('--output') + 1]) +guard = Path(__file__).resolve().parents[2] / '.work/race-guard' +guard.parent.mkdir(parents=True, exist_ok=True) +with guard.open('w') as lock: + fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB) + (guard.parent / 'build-started').touch() + time.sleep(0.8) + source = work / 'source' + if not source.exists(): + source.mkdir(parents=True) + subprocess.run(['git', 'init', '-q', str(source)], check=True) + (source / 'code').write_text('loader') + subprocess.run(['git', '-C', str(source), 'add', 'code'], check=True) + subprocess.run(['git', '-C', str(source), '-c', 'user.name=Test', + '-c', 'user.email=test@example.invalid', '-c', 'commit.gpgsign=false', + 'commit', '-qm', 'loader'], check=True) + output.write_text((source / 'code').read_text()) +''') + builder.chmod(0o755) + patch_file = builder.parent / 'patches/0001.patch' + patch_file.parent.mkdir() + patch_file.write_text('loader recipe\n') + program = ''' +from pathlib import Path +import sys +from orchard_cli import build +root, hardware, cache = map(Path, sys.argv[1:]) +build.prepare_components(hardware, cache, device='j71b', kernel=root / 'kernel', + busybox=root / 'busybox', pongo=root / 'pongo') +''' + first = subprocess.Popen([sys.executable, '-c', program, str(self.root), str(self.one), + str(self.root / 'cache-one')], stdout=subprocess.PIPE, stderr=subprocess.PIPE) + self.addCleanup(lambda: first.poll() is None and first.kill()) + marker = self.one / '.work/build-started' + deadline = time.monotonic() + 10 + while not marker.exists() and first.poll() is None and time.monotonic() < deadline: + time.sleep(0.01) + self.assertTrue(marker.exists(), 'first loader build did not start') + second = subprocess.Popen([sys.executable, '-c', program, str(self.root), str(self.one), + str(self.root / 'cache-two')], stdout=subprocess.PIPE, stderr=subprocess.PIPE) + self.addCleanup(lambda: second.poll() is None and second.kill()) + for process in (first, second): + stdout, stderr = process.communicate(timeout=20) + self.assertEqual(process.returncode, 0, (stdout + stderr).decode()) + if __name__ == '__main__': unittest.main() diff --git a/tests/test_z2fw.py b/tests/test_z2fw.py index eb6be8f..5acc02f 100644 --- a/tests/test_z2fw.py +++ b/tests/test_z2fw.py @@ -82,6 +82,12 @@ def mtprops_fixture(boot_timeout=True): 'Firmware Config': config}}) +def gen1_fixture(): + """Legacy data-form mtprops; eight synthetic payload bytes, full word count.""" + return {'PreconstructedBootloadPacketType': 'Z2', 'Constructed Firmware': + bytes.fromhex('18e13001' '0002' '00000000' '0002' '0100030205040706' '001c0000')} + + def convert_packets(data, packet='TESTPKT'): profile = plistlib.loads(data)['TEST,2'] return z2fw.convert_packets(profile, apple_xml(profile['Firmware Config']), packet=packet) @@ -207,6 +213,63 @@ def test_refuses_other_packet_types_and_data_form(self): z2fw.convert_packets(profile, apple_xml(profile['Firmware Config']), packet='TESTPKT') +class Gen1Conversion(unittest.TestCase): + def convert(self, profile=None, **kwargs): + return z2fw.convert_gen1(gen1_fixture() if profile is None else profile, + **({'packet': 'Z2', 'touch_maximum': 1024, 'prox_maximum': 128, + 'fll_mval': 6099, 'ref_clk_div': 2, 'clk32_clock_enable': 1} | kwargs)) + + def test_corellium_gen1_sequence_preserves_wire_bytes_and_dynamic_calibration(self): + output, info = self.convert() + exchange = bytes.fromhex('1aa118e118e118e118e118e118e118e1') + records = z2fw.decode_container(output) + self.assertEqual(records, [ + (z2fw.INIT, bytes.fromhex('19c1')), + (z2fw.INIT, exchange), (z2fw.INIT, exchange), (z2fw.INIT, exchange), + (z2fw.CALIBRATION, struct.pack('