diff --git a/.github/workflows/build.yml b/.github/workflows/build.yml index e9e7dec..0e0df31 100644 --- a/.github/workflows/build.yml +++ b/.github/workflows/build.yml @@ -10,7 +10,7 @@ on: hardware_tag: description: Hardware version to pair with this CLI required: true - default: v0.1.10 + default: v0.1.11 hardware_run_id: description: Optional hardware workflow run providing the tested wheel before its first release required: false @@ -24,7 +24,7 @@ jobs: package: runs-on: ubuntu-24.04 env: - HARDWARE_TAG: ${{ inputs.hardware_tag || 'v0.1.10' }} + HARDWARE_TAG: ${{ inputs.hardware_tag || 'v0.1.11' }} steps: - uses: actions/checkout@v4 with: diff --git a/README.md b/README.md index b982f5f..5830513 100644 --- a/README.md +++ b/README.md @@ -52,6 +52,7 @@ see the [launcher build and publication instructions](https://github.com/Orchard ## Documentation +- [iPad mini 1 private tethered boot and NFS handoff](docs/p105.md) - [Release publishing and updates](docs/releases.md) - [Device operation protocol](docs/device-protocol.md) - [Host adapter contract](docs/adapter-contract.md) diff --git a/docs/adapter-contract.md b/docs/adapter-contract.md index 8eeb9a3..5fb74cb 100644 --- a/docs/adapter-contract.md +++ b/docs/adapter-contract.md @@ -13,25 +13,30 @@ | `connect(device)` | Context-managed serial bootstrap and pinned SSH access bound to the selected ECID/boot. No host-side device storage/service recipes. | | `connect_saved(device)` | Context-managed use of the selected device/boot's retained pinned credentials. Never reprovision credentials or drive a login prompt. | | `forget_access(scope)` | Remove only the confirmed device/boot's cached host credentials. Device-side revocation and its confirmed receipt happen first. | -| `capture_adt(device=None)` | Fresh, bounded, private Apple ADT capture for this adapter's model. When supplied, bind the confirmed device selection to the captured identity. Do not log or implicitly retain the raw tree. | +| `capture_inputs(device=None)` | Return a `PrivateCapture` with kind, profile, ECID and private bytes. J71b captures a fresh ADT; P105 checks the live selection against the owner's hashed MtCl capture. Raw inputs stay private. | | `firmware_inputs(clock, progress)` | Return verified cached/acquired host firmware inputs; account download time separately. No device hardware or storage mutation. | -| `prepare_mixin(adt, firmware_dir, outdir)` | Return `(Path, MixinReceipt)` for a private data-only archive bound to the selected device/recipe. Preserve firmware hash, containment, calibration and output guards. | +| `prepare_mixin(inputs, firmware_dir, outdir)` | Return `(Path, MixinReceipt)` for a private data-only archive bound to the selected device/recipe. Preserve firmware hash, containment, calibration and output guards. | Arguments and return types are defined in `contracts.py`, not repeated in a second runtime schema. Image-facing protocol validation, manifest validation and hardware identity verification remain in their existing owners. ## Records -- `SelectedDevice`: required `mode`, `location`, non-null validated `ecid` and explicit boolean `resumed`; optional observed chip/board/product/serial details. `resumed` is true for running recovery and false for DFU/Pongo. Common orchestration requires this explicit state. A USB location may change across enumeration and is not persistent device identity. +- `SelectedDevice`: required `mode`, `location`, non-null validated `ecid` and explicit boolean `resumed`; optional observed chip/board/product/serial details and iOS UDID. Modes include DFU, Pongo, iOS, iBEC and Linux. A running Linux recovery has `resumed=true`. After a P105 upload, the adapter updates the selection with the observed Linux boot identity before connecting. USB locations can change across enumeration. - `Connection`: SSH arguments, validated `BootState` and a current-endpoint check performed before every RPC. Private state is keyed by profile/ECID/boot, never by USB port. -- `BootAssets`: local paths for the selected operation. Running recovery needs no upload; other states require recovery, and DFU additionally requires Pongo. Installation root is not interpreted by the adapter's boot method. +- `BootAssets`: verified local paths for the selected boot method. J71b uses recovery and Pongo. P105 uses recovery, staging loader, private iBEC and kloader. The installation root belongs to the installation workflow. - `BootRequest`: minimum boot input, including the recovery SHA receipt. Both manifest-resolved and explicit-file boot use it; no source-specific fallback adapter API exists. - `MixinReceipt`: schema 1, `private-mixin`, exact compressed byte count and SHA-256. +- `PrivateCapture`: kind `apple-adt` or `p105-mtcl`, adapter profile, validated ECID and private bytes. P105's JSON payload records its panel capture and digest. Keep its device binding with the private input. TypedDict records describe the boundary. They do not replace validation of untrusted USB data or manifests, and do not imply that every dictionary constructed by an arbitrary caller is safe. ## Enforcement -The registry checks required member presence, profile identity and presentation metadata without invoking discovery. Python runtime Protocol checks do not verify function signatures; static checking does that. The real J71b module is assigned directly to `Adapter` in `tests/typecheck_adapter.py`, so signature drift is rejected without creating a wrapper implementation. +The registry checks required member presence, profile identity and presentation +metadata without invoking discovery. Static checks assign both registered +modules directly to `Adapter` in `tests/typecheck_adapter.py`. Ambiguous USB +discovery stops automatic selection across adapters. `--profile` preserves an +explicit selection through the common workflows. From `cli/`: diff --git a/docs/device-protocol.md b/docs/device-protocol.md index 4f62117..c004e6e 100644 --- a/docs/device-protocol.md +++ b/docs/device-protocol.md @@ -62,3 +62,32 @@ Host code must not infer device paths or encode module loads, filesystem offsets - Pre-exec PID 1 failures attempt to restore moved pseudo-filesystem aliases and restart key-only recovery diagnostics, never retrying handoff/storage. Reported failures stop host readiness polling immediately. Kernel failure or failure inside the final root-switch executable can still require DFU recovery. - The desktop shell reports readiness only after constructing its root and successfully capturing a compositor frame to `/dev/null`. The device checks the boot ID, live shell process and owned Wayland socket. - No screenshot is retained. Readiness does not attest touch gestures, brightness interaction or Wi-Fi association. +## P105 tethered recovery + +P105 uses protocol 1 with a limited operation set: + +| Operation | Behavior | +| --- | --- | +| `status` | Report profile, provisioned ECID, live kernel boot ID, root mode, readiness and the flash fence | +| `diagnostics` | Add the kernel version, MTD inventory and a bounded kernel log | +| `bootstrap` | Provision one diagnostic key through the physical RAM serial console; return the pinned host key and USB SSH endpoint | +| `handoff` | Queue one NFS pivot bound to profile, ECID, boot ID and operation UUID | + +`handoff` takes `nfs_root` in the form `10.55.0.1:/absolute/export/path`. +Recovery accepts an authenticated request before PID 1 mounts the export. It +checks the P105 systemd marker and ARM executable, then prepares hardware and +RAM diagnostics before moving mounts. A rejected mount or root validation leaves the RAM console and +its diagnostic state available. Requests receive one dispatch per attempt. + +The USB serial number carries the owner's ECID and the current kernel boot +ID. The ECID is provisioned from the verified iOS selection; P105's static +device tree supplies its board identity. The ordinary kernel disables its +NAND controller. Each ready status checks the live board and controller fence. + +During a mount move, an in-flight status can report `mode=handoff` from the +protected queued identity. It stays unready until the live checks succeed in +the NFS root. The serial-pinned SSH key travels in RAM across this transition. +The host stores access by profile, ECID and boot ID. + +The internal storage operations described for J71b below remain outside the +P105 operation set. The host checks the reported operations before dispatch. diff --git a/docs/p105.md b/docs/p105.md new file mode 100644 index 0000000..2cdd60b --- /dev/null +++ b/docs/p105.md @@ -0,0 +1,140 @@ +# iPad mini 1 Wi-Fi: tethered recovery and NFS + +This contribution pairs CLI 0.1.11 with hardware 0.1.11. Install both local +wheels while these branches await publication. The hardware package supplies +the P105 resources required by the adapter. + +The `p105` adapter supports `iPad2,5` / P105AP running the qualified jailbreak +on iOS 9.3.5 build 13G36. It stages the verified kloader and plaintext iBEC over +a UDID-bound SSH proxy, waits for the same ECID in Recovery, then sends the +private bundle and staging loader once. Reconnect the USB cable when prompted. + +Recovery reports its provisioned ECID and the live kernel boot ID over USB. +That ECID comes from the owner's verified iOS selection. P105 supplies panel +calibration through MtCl; J71b supplies a live ADT. SSH bootstrap pins the +recovery host key through the physical serial console. Saved credentials are +scoped to the selected device and boot. + +The adapter supports status, bounded diagnostics and a handoff to the owner's +existing USB NFS export. Internal storage preparation awaits a qualified A5 +storage backend. Keep the host serving NFS throughout the desktop session. + +Physical qualification reached the Omarchy desktop and preserved the RAM USB +rescue console through repeated boots. Touch remains slow. One repeat required +a touchscreen service restart after calibration failed; that driver issue +remains open. See the companion hardware validation record for the measured +scope. Diagnostic readiness reports the reached systemd targets; inspect +diagnostics and individual services when input fails. + +## Prepare the private recovery + +Use the P105 resources from the companion hardware contribution. In +`minimal-linux`, `p105/build-base.py` builds the signed ARMv7 Alpine package +base. Its ARM64 package manager uses the official ARMv7 signing keys and +extracts target packages without executing them. Retain its base receipt. + +`p105/prepare-recovery.py` combines that base with an explicit owner initramfs, +the matching panel calibration and the verified target record. The hardware +producer recognizes its `etc/orchard/p105.json`, keeps the desktop kernel +prerequisites, and selects a USB gadget whose serial number carries the boot +identity. The RAM environment waits for an authenticated NFS request. + +```sh +python3 p105/build-base.py --output /private/p105/alpine-base.tar +python3 p105/prepare-recovery.py \ + --base /private/p105/alpine-base.tar \ + --base-receipt /private/p105/alpine-base.json \ + --initramfs /private/p105/owner-initramfs \ + --calibration /private/p105/owner-MtCl.bin \ + --target /private/p105/target.json \ + --output /private/p105/recovery-root +``` + +Build that root through the installed hardware package's `kernel/p105/build`, +passing it as `--private-initramfs` with the separate `--calibration` input. +The resulting components and their receipt stay private. Compose their boot +directory with the installed CLI: + +```sh +orchard p105-manifest --components /private/p105/components \ + --ibec /private/p105/iBEC.autogo --kloader /private/p105/kloader \ + --output /private/p105/boot +``` + +This creates an owner-local `boot.toml` and copies the verified boot assets. +The resolver requires local files and the manifest's private flag for this +boot method. The qualified iBEC and kloader hashes are checked again before +staging. A changed component receipt or another device's ECID stops the boot. + +## Owner configuration + +Set `ORCHARD_P105_CONFIG` to an owner-only JSON file, or place it at +`ORCHARD_HOME/private/p105.json`. Its `schema` is 1. The required fields are: + +| Field | Owner input | +| --- | --- | +| `target` | Object containing the selected `udid` and integer `ecid` | +| `tools` | Directory containing `ideviceinfo`, `iproxy`, `irecovery`, and `sshpass` when password authentication is used | +| `ios.known_hosts` | Private file containing the trusted iOS key under alias `p105-ios-usb` | +| `ios.identity` or `ios.password_file` | Explicit owner-only SSH identity or password file | +| `components` | Directory containing the verified private recovery components | +| `firmware_root` | Owner initramfs with the pinned firmware destinations | +| `calibration` | Object containing `file`, integer `ecid` and the capture's `sha256` | + +All paths are absolute. Keep the calibration capture and configuration bound +to the same iPad. `orchard --profile p105 capture-inputs --output FILE` writes +a protected MtCl record after checking the selected device. Private mixins +accept this record with `--inputs`; J71b retains its `--adt` input. + +The host also needs SSH, `ssh-keygen`, `showmount` and Podman for preparation. +USB utilities can come from the owner's vetted Legacy iOS Kit installation. +Provisioning the jailbreak and trusting the initial iOS SSH key remain owner +setup steps. + +## Boot and hand off + +Prepare the Arch ARMv7 root with the hardware runtime's `p105-systemd-v1` +marker and Python 3.9 or later. Export it over NFSv3/TCP to `10.55.0.2`. +The host address is `10.55.0.1/24` on the selected USB link. + +```sh +orchard --profile p105 boot /private/p105/boot/boot.toml \ + --nfs-root 10.55.0.1:/absolute/exported/root +orchard --profile p105 status +orchard --profile p105 diagnostics +``` + +Omit `--nfs-root` to stop in RAM recovery. Provision its first diagnostic +session with `orchard --profile p105 status --recovery`. A later +`orchard --profile p105 handoff --nfs-root HOST:/PATH` uses the saved session. + +The host verifies the requested export before dispatch. Recovery accepts one +handoff per boot and validates the systemd root before moving mounts. The +installed root receives the hardware payload and RAM diagnostic keys. A +separate diagnostic listener keeps the serial-pinned key through the pivot; +the root's existing SSH configuration retains its own files beneath the +temporary authorization bind mount. The physical rescue console runs from RAM. + +Host reports preserve an uncertain outcome before each upload or handoff. +After an uncertain upload, inspect the device and return to iOS before starting +a fresh attempt. A timeout while waiting for the cable can continue from iBEC: +repeat the same command within 15 minutes. Orchard requires the newest saved +attempt to be awaiting iBEC, with matching ECID and boot files. An upload phase +with an unknown outcome blocks continuation. Uploads and NFS requests receive +one dispatch per attempt. + +## Host link ownership + +Orchard obtains the network interface and its MAC from the selected USB +device's registry or sysfs children while checking its device and boot identity. +This handles macOS reporting a masked MAC through `ifconfig`. An existing host +address is retained. A missing address can be added with the host's normal +administrator prompt; its private +`network.json` records ownership. After shutting down Linux, run: + +```sh +orchard p105-network-cleanup --record /path/to/this/boot/network.json +``` + +Cleanup removes an address only when that record says Orchard added it. The +owner continues to manage the existing NFS export and its service. diff --git a/docs/releases.md b/docs/releases.md index e516590..36f4b42 100644 --- a/docs/releases.md +++ b/docs/releases.md @@ -21,7 +21,7 @@ python -m pip install --upgrade pip build python -m build --wheel ``` -Build hardware first, then CLI. Wheel versions come from each repository's `pyproject.toml`. The release pair prepared here is CLI `0.1.10` and hardware `0.1.10`. It includes the expanded J71b Wi-Fi firmware recipe; consuming images must require Orchard `0.1.10` or newer. +Build hardware first, then CLI. Wheel versions come from each repository's `pyproject.toml`. The release pair prepared here is CLI `0.1.11` and hardware `0.1.11`. This pair adds private P105 tethered recovery and NFS handoff. P105 requires the companion hardware resources; public image assembly remains a separate release step. ## Prepare assets locally @@ -29,9 +29,9 @@ From the CLI repository, pass explicit paths to both wheels: ```sh python tools/prepare_release.py \ - --cli-wheel dist/orchard_cli-0.1.10-py3-none-any.whl \ - --hardware-wheel /path/to/orchard_hardware-0.1.10-py3-none-any.whl \ - --output out/publication/host-0.1.10 + --cli-wheel dist/orchard_cli-0.1.11-py3-none-any.whl \ + --hardware-wheel /path/to/orchard_hardware-0.1.11-py3-none-any.whl \ + --output out/publication/host-0.1.11 ``` Use the actual wheel filenames for later versions. The tool reads their metadata and computes hashes. Output must be a new directory: @@ -44,8 +44,8 @@ Before publishing, install the pair into a fresh virtual environment, run `pytho ```sh python3 -m venv out/release-test out/release-test/bin/python -m pip install --only-binary=:all: \ - dist/orchard_cli-0.1.10-py3-none-any.whl \ - /path/to/orchard_hardware-0.1.10-py3-none-any.whl + dist/orchard_cli-0.1.11-py3-none-any.whl \ + /path/to/orchard_hardware-0.1.11-py3-none-any.whl out/release-test/bin/python -m pip check PYTHONDONTWRITEBYTECODE=1 out/release-test/bin/python -m unittest discover -s tests -v out/release-test/bin/orchard resolve tests/fixtures/first/index.toml diff --git a/pyproject.toml b/pyproject.toml index 87f13ae..2254e1d 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,13 +4,13 @@ build-backend = "setuptools.build_meta" [project] name = "orchard-cli" -version = "0.1.10" +version = "0.1.11" description = "Apple device Linux boot and installation tooling" readme = "README.md" license = "GPL-2.0-or-later AND MIT" license-files = ["LICENSE", "NOTICE", "LICENSES/*"] requires-python = ">=3.9" -dependencies = ["orchard-hardware>=0.1.10,<0.2", "jsonschema>=4.25,<5", "tomli>=2; python_version < '3.11'"] +dependencies = ["orchard-hardware>=0.1.11,<0.2", "jsonschema>=4.25,<5", "tomli>=2; python_version < '3.11'"] [project.urls] Repository = "https://github.com/OrchardLinux/cli" @@ -23,7 +23,7 @@ orch = "orchard_cli.command:main" [tool.setuptools] include-package-data = false -packages = ["orchard_cli", "orchard_cli.j71b", "orchard_cli.apple", "orchard_cli.transports", "orchard_cli_schemas"] +packages = ["orchard_cli", "orchard_cli.j71b", "orchard_cli.p105", "orchard_cli.apple", "orchard_cli.transports", "orchard_cli_schemas"] package-dir = {"" = "src", orchard_cli_schemas = "schemas"} [tool.setuptools.package-data] @@ -36,7 +36,7 @@ dev = ["mypy==1.18.2"] [tool.mypy] python_version = "3.9" mypy_path = "src" -files = ["src/orchard_cli/contracts.py", "src/orchard_cli/devices.py", "src/orchard_cli/j71b/adapter.py", "src/orchard_cli/j71b/inputs.py", "src/orchard_cli/j71b/mixin.py", "src/orchard_cli/service.py", "src/orchard_cli/command.py", "tests/typecheck_adapter.py"] +files = ["src/orchard_cli/contracts.py", "src/orchard_cli/devices.py", "src/orchard_cli/j71b/adapter.py", "src/orchard_cli/j71b/inputs.py", "src/orchard_cli/j71b/mixin.py", "src/orchard_cli/p105", "src/orchard_cli/service.py", "src/orchard_cli/command.py", "tests/typecheck_adapter.py"] check_untyped_defs = true follow_imports = "silent" ignore_missing_imports = true diff --git a/schemas/v1/boot.schema.json b/schemas/v1/boot.schema.json index 23dc3ec..f8a02f1 100644 --- a/schemas/v1/boot.schema.json +++ b/schemas/v1/boot.schema.json @@ -38,6 +38,23 @@ "version": { "type": "string", "minLength": 1 + }, + "loader": { + "$ref": "#/$defs/receipt" + }, + "ibec": { + "$ref": "#/$defs/receipt" + }, + "kloader": { + "$ref": "#/$defs/receipt" + }, + "private": { + "type": "boolean" + }, + "target_ecid": { + "type": "integer", + "minimum": 1, + "maximum": 18446744073709551615 } }, "required": [ @@ -67,6 +84,35 @@ "pongo" ] } + }, + { + "if": { + "properties": { + "boot_method": { + "const": "jailbreak-kdfu" + } + }, + "required": [ + "boot_method" + ] + }, + "then": { + "required": [ + "loader", + "ibec", + "kloader", + "private", + "target_ecid" + ], + "properties": { + "device": { + "const": "p105" + }, + "private": { + "const": true + } + } + } } ], "$defs": { diff --git a/src/orchard_cli/apple/linux_usb.py b/src/orchard_cli/apple/linux_usb.py index c363f08..801caad 100644 --- a/src/orchard_cli/apple/linux_usb.py +++ b/src/orchard_cli/apple/linux_usb.py @@ -21,15 +21,28 @@ def devices(root=Path('/sys/bus/usb/devices')): ecid = re.search(r'\bECID:([0-9a-fA-F]+)\b', serial) chip = re.search(r'\bCPID:([0-9a-fA-F]+)\b', serial) board = re.search(r'\bBDID:([0-9a-fA-F]+)\b', serial) - ports = {str(Path('/dev') / p.name) for p in node.glob('**/tty/ttyACM*')} + interfaces = [p for p in node.glob(node.name + ':*') if p.is_dir()] + ports = {str(Path('/dev') / p.name) for interface in interfaces for p in interface.glob('**/tty/ttyACM*')} + networks = [p for interface in interfaces for p in interface.glob('net/*') if p.is_dir()] if len(ports) > 1: raise ValueError('Ambiguous USB serial ports on a device') + if len(networks) > 1: + raise ValueError('Ambiguous USB network interfaces on a device') product = (node / 'product').read_text().strip() if (node / 'product').exists() else '' - found.append(dict(location=node.name, mode=MODES.get(pid, 'other'), + observation = dict(location=node.name, mode=MODES.get(pid, 'other'), ecid=int(ecid.group(1), 16) if ecid else None, boot_id=str(uuid.UUID(boot.group(1))) if boot else None, chip=int(chip.group(1), 16) if chip else None, board=int(board.group(1), 16) if board else None, pwned='YOLO:checkra1n' in serial, product=product, - serial_port=next(iter(ports), None))) + serial_port=next(iter(ports), None)) + if re.fullmatch(r'[0-9a-fA-F]{40}', serial): + observation['udid'] = serial.lower() + if networks: + mac = (networks[0] / 'address').read_text().strip().lower() + if not re.fullmatch(r'[0-9a-f]{2}(?::[0-9a-f]{2}){5}', mac): + raise ValueError('Invalid USB network interface address') + observation['network_interface'] = networks[0].name + observation['network_mac'] = mac + found.append(observation) return found diff --git a/src/orchard_cli/apple/usb.py b/src/orchard_cli/apple/usb.py index 6ca3bc6..72b9cc2 100644 --- a/src/orchard_cli/apple/usb.py +++ b/src/orchard_cli/apple/usb.py @@ -6,7 +6,7 @@ import uuid from .linux_usb import devices as linux_devices -from orchard_cli.state import EndpointUnavailable +from orchard_cli.state import EndpointUnavailable, AmbiguousDevices MODES = {0x1227: 'dfu', 0x4141: 'pongo', 0x4142: 'linux'} @@ -38,22 +38,40 @@ def devices(): for label in ('ECID', 'CPID', 'BDID')] descendants = list(node.get('IORegistryEntryChildren', [])) ports = [] + network_names, network_macs = set(), set() while descendants: child = descendants.pop() + if child.get('IOObjectClass') == 'IOUSBHostDevice': + continue descendants.extend(child.get('IORegistryEntryChildren', [])) + if child.get('IOObjectClass') in ('IOEthernetInterface', 'IOSkywalkLegacyEthernetInterface'): + name = child.get('BSD Name') + if isinstance(name, str) and re.fullmatch(r'[A-Za-z0-9][A-Za-z0-9._-]{0,31}', name): + network_names.add(name) + mac = child.get('IOMACAddress') + if isinstance(mac, bytes) and len(mac) == 6: + network_macs.add(':'.join(f'{value:02x}' for value in mac)) if child.get('IOObjectClass') == 'IOSerialBSDClient': port = child.get('IOCalloutDevice') if isinstance(port, str) and port.startswith('/dev/cu.'): ports.append(port) if len(ports) > 1: raise DeviceError('Ambiguous USB serial ports on the selected device') - found.append({'location': location, 'mode': MODES.get(node.get('idProduct'), 'other'), + if len(network_names) > 1 or len(network_macs) > 1: + raise DeviceError('Ambiguous USB network interfaces on a device') + observation = {'location': location, 'mode': MODES.get(node.get('idProduct'), 'other'), 'ecid': int(fields[0].group(1), 16) if fields[0] else None, 'boot_id': str(uuid.UUID(boot.group(1))) if boot else None, 'chip': int(fields[1].group(1), 16) if fields[1] else None, 'board': int(fields[2].group(1), 16) if fields[2] else None, 'product': node.get('USB Product Name'), - 'serial_port': ports[0] if ports else None}) + 'serial_port': ports[0] if ports else None} + if re.fullmatch(r'[0-9a-fA-F]{40}', serial): + observation['udid'] = serial.lower() + if network_names and network_macs: + observation['network_interface'] = next(iter(network_names)) + observation['network_mac'] = next(iter(network_macs)) + found.append(observation) return found @@ -62,7 +80,7 @@ def sole_target(): if not targets: raise EndpointUnavailable('Selected USB target is not connected') if len(targets) != 1: - raise ValueError('Connect exactly one target in DFU, Pongo or Linux; device selection is not supported') + raise AmbiguousDevices('Connect exactly one target in DFU, Pongo or Linux; device selection is not supported') return targets[0] diff --git a/src/orchard_cli/command.py b/src/orchard_cli/command.py index eb83fbe..1bb8e4b 100644 --- a/src/orchard_cli/command.py +++ b/src/orchard_cli/command.py @@ -22,6 +22,7 @@ def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument('--version', action='version', version=version('orchard-cli')) parser.add_argument('--verbose', action='store_true', help='show internal device events') + parser.add_argument('--profile', choices=('j71b', 'p105'), help='select a device profile explicitly') commands = parser.add_subparsers(dest='action', required=True) build = commands.add_parser('build', help='build DIRECTORY/init and optional DIRECTORY/rootfs into a RAM payload') build.add_argument('directory', type=Path, help='any directory containing init and optional rootfs/') @@ -49,7 +50,7 @@ def main(): distros.add_argument('source', nargs='?') check = commands.add_parser('resolve', help='verify inputs without accessing a device') check.add_argument('source') - check.add_argument('--profile', default='j71b') + check.add_argument('--profile', dest='resolve_profile') 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') @@ -59,14 +60,27 @@ def main(): boot.add_argument('--image', type=Path) boot.add_argument('--receipt', type=Path) boot.add_argument('--pongo-image', type=Path) + boot.add_argument('--nfs-root', help='hand recovery off to an existing host USB NFS export') adt = commands.add_parser('capture-adt', help='save a fresh recovery ADT privately') adt.add_argument('--output', type=Path, required=True) + capture = commands.add_parser('capture-inputs', help='save private hardware inputs for the selected device') + capture.add_argument('--output', type=Path, required=True) + network_cleanup = commands.add_parser('p105-network-cleanup', help='remove an address added by an earlier P105 boot') + network_cleanup.add_argument('--record', type=Path, required=True) + private_boot = commands.add_parser('p105-manifest', help='compose a private P105 boot directory from verified components') + for name in ('components','ibec','kloader','output'): + private_boot.add_argument('--'+name, type=Path, required=True) mixin = commands.add_parser('mixin', help='prepare a data-only private mixin') - for name in ('adt', 'firmware-dir', 'output'): + mixin_inputs = mixin.add_mutually_exclusive_group(required=True) + mixin_inputs.add_argument('--adt', type=Path, help='J71b ADT capture') + mixin_inputs.add_argument('--inputs', type=Path, help='device-specific private input capture') + for name in ('firmware-dir', 'output'): mixin.add_argument('--' + name, type=Path, required=True) for action in ('status', 'diagnostics', 'targets', 'destination', 'prepare', 'install-root', 'handoff', 'abort'): command = commands.add_parser(action, help='invoke the device-owned ' + action + ' operation') command.add_argument('--recovery', action='store_true', help='provision access through the recovery serial shell') + if action == 'handoff': + command.add_argument('--nfs-root', help='existing host USB NFS export') if action in ('destination', 'install-root'): command.add_argument('--target', required=True, help='observed partition identifier') command.add_argument('--token', required=True, help='device-issued selection token') @@ -94,31 +108,52 @@ def main(): run_directory(args.directory, yes=args.yes) return with ExitStack() as stack: - if args.action not in ('install', 'launch', 'distros', 'resolve', 'devices', 'mixin', 'setup-host'): + if args.action not in ('install', 'launch', 'distros', 'resolve', 'devices', 'mixin', 'setup-host', 'p105-manifest'): stack.enter_context(operation_lock()) if args.action == 'install': source = args.manifest or args.source if not source: raise ValueError('Specify a distro, index or manifest: orchard install SOURCE') - result = install(source, progress=progress) + result = install(source, progress=progress, **({'profile':args.profile} if args.profile else {})) elif args.action == 'launch': from orchard_cli.launch import launch result = launch(args.manifest, directory=args.directory, target=args.target, - allow_firmware_download=args.allow_firmware_download, progress=progress) + allow_firmware_download=args.allow_firmware_download, progress=progress, + **({'profile':args.profile} if args.profile else {})) elif args.action == 'distros': if bool(args.name) != bool(args.source): raise ValueError('Use orchard distros NAME INDEX_URL_OR_FILE, or orchard distros to list') result = register(args.name, args.source) if args.name else catalogue() elif args.action == 'resolve': - record, assets = resolve(args.source, args.profile, progress, purpose='boot' if args.boot else 'install') + record, assets = resolve(args.source, args.resolve_profile or args.profile or 'j71b', progress, + purpose='boot' if args.boot else 'install') result = {'manifest': record, 'files': {key: str(path) for key, path in assets.items()}} elif args.action == 'devices': - result = load().list_devices() + if args.profile: + result = load(args.profile).list_devices() + else: + from orchard_cli.devices import SUPPORTED + rows = [row for profile in SUPPORTED for row in load(profile).list_devices()] + result = list({json.dumps(row, sort_keys=True): row for row in rows}.values()) + elif args.action == 'p105-network-cleanup': + from orchard_cli.p105.network import cleanup + result = cleanup(args.record) + elif args.action == 'p105-manifest': + from orchard_cli.p105.prepare import manifest + result = manifest(args.components,args.ibec,args.kloader,args.output) elif args.action == 'setup-host': - from orchard_cli.host_setup import setup - result = setup(progress) + if args.profile == 'p105': + from orchard_cli.p105.config import tool + result = {'profile':'p105','tools':{name:tool(name) for name in + ('ideviceinfo','iproxy','irecovery')}} + else: + from orchard_cli.host_setup import setup + result = setup(progress) elif args.action == 'boot': - adapter, device = choose() + adapter, device = choose(args.profile) if args.profile else choose() + if args.nfs_root: + from orchard_cli.p105.network import nfs_root + nfs_root(args.nfs_root) if args.image: if not args.receipt: raise ValueError('--image requires --receipt') @@ -137,24 +172,42 @@ def main(): record, assets = resolve(source, adapter.PROFILE, progress, purpose='boot') adapter.boot(device, assets, record, progress) result = {'recovery_observed': True} - elif args.action == 'capture-adt': - adapter, device = choose() + if args.nfs_root: + from orchard_cli.service import handoff_nfs + with adapter.connect(device) as nfs_connection: + result = handoff_nfs(nfs_connection, args.nfs_root, progress) + elif args.action in ('capture-adt', 'capture-inputs'): + adapter, device = choose(args.profile) if args.profile else choose() if args.output.exists() or args.output.is_symlink(): raise ValueError('ADT destination already exists') - data = adapter.capture_adt(device) + captured = adapter.capture_inputs(device) + if args.action == 'capture-adt' and captured['kind'] != 'apple-adt': + raise ValueError('This device uses panel calibration; use orchard capture-inputs') + if captured['ecid'] != device['ecid'] or captured['profile'] != adapter.PROFILE: + raise ValueError('Private capture belongs to another device') + data = captured['data'] args.output.parent.mkdir(parents=True, exist_ok=True, mode=0o700) with os.fdopen(os.open(args.output, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600), 'wb') as stream: stream.write(data) result = {'saved': str(args.output), 'bytes': len(data)} elif args.action == 'mixin': - adapter = load() - archive, result = adapter.prepare_mixin(args.adt, args.firmware_dir, args.output) + adapter = load(args.profile) + archive, result = adapter.prepare_mixin(args.inputs or args.adt, args.firmware_dir, args.output) result = {**result, 'path': str(archive)} else: - adapter, device = choose() + adapter, device = choose(args.profile) if args.profile else choose() connection = adapter.connect if args.recovery else adapter.connect_saved with connection(device) as ssh: client = Client(ssh, progress) + if args.action == 'handoff' and args.nfs_root: + from orchard_cli.service import handoff_nfs + result = handoff_nfs(ssh, args.nfs_root, progress) + print(json.dumps(result, indent=2)) + return + if args.action not in ('status', 'diagnostics'): + available = client.call('status').get('operations') + if available is not None and args.action not in available: + raise ValueError('Operation unavailable in this recovery; use its reported operations') request = {} if args.action in ('destination', 'install-root'): request = {'target': args.target, 'selection_token': args.token, 'directory': args.directory} diff --git a/src/orchard_cli/contracts.py b/src/orchard_cli/contracts.py index 0a32e53..6f27b2b 100644 --- a/src/orchard_cli/contracts.py +++ b/src/orchard_cli/contracts.py @@ -1,7 +1,7 @@ """Typed host adapter boundary, not a device/plugin or boot-method framework. Adapters remain ordinary modules. These records describe the current supported -USB/checkm8 workflow; common orchestration does not interpret chip/board values. +USB boot workflows; common orchestration does not interpret chip/board values. Runtime artifact/identity verification remains in the existing implementations. """ from contextlib import AbstractContextManager @@ -22,12 +22,15 @@ class _SelectionDetails(TypedDict, total=False): serial_port: Optional[str] pwned: bool boot_id: Optional[str] + udid: str + network_interface: str + network_mac: str class SelectedDevice(_SelectionDetails): """Validated discovery result; resumed is explicit, never inferred by the host.""" location: USBLocation - mode: Literal['dfu', 'pongo', 'linux'] + mode: Literal['dfu', 'pongo', 'linux', 'ios', 'ibec'] ecid: int resumed: bool @@ -41,13 +44,21 @@ class BootAssets(TypedDict, total=False): pongo: Path recovery: Path root: Path + loader: Path + ibec: Path + kloader: Path class BootReceipt(TypedDict): sha256: str -class BootRequest(TypedDict): +class _BootDetails(TypedDict, total=False): + target_ecid: int + boot_method: str + + +class BootRequest(_BootDetails): """Minimum adapter boot input, shared by resolved and explicit-file requests.""" assets: Mapping[str, BootReceipt] @@ -59,6 +70,13 @@ class MixinReceipt(TypedDict): sha256: str +class PrivateCapture(TypedDict): + kind: Literal['apple-adt', 'p105-mtcl'] + profile: str + ecid: int + data: bytes + + @dataclass(frozen=True) class Connection: argv: List[str] @@ -93,7 +111,7 @@ def connect_saved(self, device: SelectedDevice) -> AbstractContextManager[Connec def forget_access(self, scope: BootState) -> None: ... - def capture_adt(self, device: Optional[SelectedDevice] = None) -> bytes: ... + def capture_inputs(self, device: Optional[SelectedDevice] = None) -> PrivateCapture: ... def firmware_inputs(self, clock: Clock, progress: Progress, *, allow_download: bool = True) -> Path: ... diff --git a/src/orchard_cli/devices.py b/src/orchard_cli/devices.py index aa8a55f..c764ba9 100644 --- a/src/orchard_cli/devices.py +++ b/src/orchard_cli/devices.py @@ -3,8 +3,9 @@ from typing import List, Optional, Tuple from .contracts import Adapter, SelectedDevice +from .state import AmbiguousDevices -SUPPORTED = ('j71b',) +SUPPORTED = ('j71b', 'p105') def load(profile: Optional[str] = None) -> Adapter: @@ -21,13 +22,15 @@ def load(profile: Optional[str] = None) -> Adapter: return adapter -def choose() -> Tuple[Adapter, SelectedDevice]: +def choose(profile: Optional[str] = None) -> Tuple[Adapter, SelectedDevice]: matches: List[Tuple[Adapter, SelectedDevice]] = [] errors: List[str] = [] - for profile in SUPPORTED: + for profile in ((profile,) if profile else SUPPORTED): adapter = load(profile) try: matches.append((adapter, adapter.discover())) + except AmbiguousDevices: + raise except ValueError as error: errors.append(str(error)) if len(matches) != 1: diff --git a/src/orchard_cli/j71b/adapter.py b/src/orchard_cli/j71b/adapter.py index 51c7da9..983065e 100644 --- a/src/orchard_cli/j71b/adapter.py +++ b/src/orchard_cli/j71b/adapter.py @@ -2,7 +2,7 @@ from contextlib import AbstractContextManager from typing import Any, Dict, List, Optional, cast -from orchard_cli.contracts import BootAssets, BootRequest, Connection, SelectedDevice +from orchard_cli.contracts import BootAssets, BootRequest, Connection, SelectedDevice, PrivateCapture from orchard_cli.state import BootState, DeviceState, EndpointUnavailable from orchard_cli.progress import Progress from orchard_cli.apple import usb, boot as apple_boot, adt_capture @@ -53,6 +53,11 @@ def capture_adt(device: Optional[SelectedDevice] = None) -> bytes: return data +def capture_inputs(device: Optional[SelectedDevice] = None) -> PrivateCapture: + data = capture_adt(device) + return {'kind': 'apple-adt', 'profile': PROFILE, 'ecid': validate_adt(data), 'data': data} + + def list_devices() -> List[Dict[str, Any]]: return usb.devices() diff --git a/src/orchard_cli/launch.py b/src/orchard_cli/launch.py index e21d672..2e84645 100644 --- a/src/orchard_cli/launch.py +++ b/src/orchard_cli/launch.py @@ -12,9 +12,9 @@ @exclusive_operation def launch(manifest=None, *, directory=None, target=None, allow_firmware_download=False, - input_fn=input, progress=None): + input_fn=input, progress=None, profile=None): clock, progress = Clock(), progress or Progress() - adapter, device = choose() + adapter, device = choose(profile) if profile else choose() if ask(clock, f'Launch an installed system on {adapter.NAME}? [y/N] ', input_fn).lower() not in ('y','yes'): return {'cancelled':True} manifest = manifest if manifest is not None else launch_manifest(adapter.PROFILE) @@ -33,7 +33,10 @@ def launch(manifest=None, *, directory=None, target=None, allow_firmware_downloa with tempfile.TemporaryDirectory(prefix='launch-',dir=private) as temporary: work = Path(temporary) adt = work / 'device.adt' - adt.write_bytes(adapter.capture_adt(device)) + capture = adapter.capture_inputs(device) + if capture['profile'] != adapter.PROFILE or capture['ecid'] != device['ecid']: + raise ValueError('Private input capture differs from the selected device') + adt.write_bytes(capture['data']) adt.chmod(0o600) mixin, receipt = adapter.prepare_mixin(adt,firmware,work/'mixin') with mixin.open('rb') as source: diff --git a/src/orchard_cli/p105/__init__.py b/src/orchard_cli/p105/__init__.py new file mode 100644 index 0000000..a4a0c19 --- /dev/null +++ b/src/orchard_cli/p105/__init__.py @@ -0,0 +1 @@ +"""Original iPad mini Wi-Fi host support.""" diff --git a/src/orchard_cli/p105/adapter.py b/src/orchard_cli/p105/adapter.py new file mode 100644 index 0000000..3400759 --- /dev/null +++ b/src/orchard_cli/p105/adapter.py @@ -0,0 +1,134 @@ +"""P105 policy for the native Orchard boot and pinned recovery transport.""" +import base64 +from contextlib import AbstractContextManager, contextmanager +import hashlib +import io +import json +from pathlib import Path +import tarfile +from typing import Any, Dict, Iterator, List, Optional, Tuple, cast + +from orchard_hardware import firmware_path +from orchard_cli.contracts import (BootAssets, BootRequest, Connection, MixinReceipt, + PrivateCapture, SelectedDevice) +from orchard_cli.progress import Clock, Progress +from orchard_cli.state import BootState, DeviceState, AmbiguousDevices +from orchard_cli.transports import ssh +from .config import PROFILE, NAME, PRODUCT, regular, settings +from . import usb, network +from .boot import boot + +REBOOT_MESSAGE = 'Return the iPad to iOS, then start a fresh Orchard P105 boot.' + + +def list_devices() -> List[Dict[str, Any]]: + return usb.observations() + + +def discover() -> SelectedDevice: + rows = list_devices() + if len(rows) > 1: + raise AmbiguousDevices('Connect exactly one supported P105 device') + if len(rows) != 1: + raise ValueError('Connect exactly one supported P105 device') + row = rows[0] + DeviceState(PROFILE, row['ecid']) + if row['mode'] == 'linux': + row = usb.recovery() + BootState(DeviceState(PROFILE, row['ecid']), row['boot_id']) + return cast(SelectedDevice, {**row, 'resumed': row['mode'] == 'linux'}) + + +@contextmanager +def connect(device: SelectedDevice) -> Iterator[Connection]: + endpoint = usb.recovery() + if endpoint['ecid'] != device['ecid']: + raise ValueError('Selected P105 changed before SSH bootstrap') + scope = BootState(DeviceState(PROFILE, device['ecid']), endpoint['boot_id']) + network.ensure(scope) + with ssh.session(scope.device, usb.recovery, expected_boot=device.get('boot_id')) as connection: + yield connection + + +def connect_saved(device: SelectedDevice) -> AbstractContextManager[Connection]: + if device['mode'] != 'linux' or not device.get('boot_id'): + raise ValueError('P105 diagnostics require the selected running Linux boot') + return ssh.saved_session(DeviceState(PROFILE, device['ecid']), usb.recovery, + expected_boot=device['boot_id']) + + +def forget_access(scope: BootState) -> None: + if scope.device.profile != PROFILE: + raise ValueError('P105 access scope differs') + ssh.forget_access(scope) + + +def capture_inputs(device: Optional[SelectedDevice] = None) -> PrivateCapture: + device = device or discover() + if device['mode'] == 'ios': + usb.verify_ios(dict(device)) + elif device['mode'] == 'linux': + observed = usb.recovery() + if observed['ecid'] != device['ecid'] or observed['boot_id'] != device.get('boot_id'): + raise ValueError('Selected P105 changed before private input capture') + else: + raise ValueError('Capture P105 inputs from its verified iOS or Linux environment') + config = settings() + calibration = config['calibration'] + if calibration['ecid'] != device['ecid'] or config['target']['ecid'] != device['ecid']: + raise ValueError('Panel calibration record belongs to another device') + data = regular(Path(calibration['file']), private=True).read_bytes() + if not 0 < len(data) <= 65536 or hashlib.sha256(data).hexdigest() != calibration['sha256']: + raise ValueError('Owner panel calibration differs from its capture receipt') + record = {'schema': 1, 'kind': 'p105-mtcl', 'profile': PROFILE, 'ecid': device['ecid'], + 'calibration': base64.b64encode(data).decode(), 'sha256': calibration['sha256']} + return {'kind': 'p105-mtcl', 'profile': PROFILE, 'ecid': device['ecid'], + 'data': (json.dumps(record)+'\n').encode()} + + +def firmware_inputs(clock: Clock, progress: Progress, *, allow_download: bool = True) -> Path: + path = Path(settings()['firmware_root']) + if path.is_symlink() or not path.is_dir(): + raise ValueError('Select the verified private P105 firmware root') + recipe = json.loads(firmware_path(PROFILE).read_text()) + for entry in recipe['files'] + recipe['generated']: + source = path / entry['destination'] + if not source.resolve().is_relative_to(path.resolve()): + raise ValueError('Firmware escapes the owner root') + if hashlib.sha256(regular(source).read_bytes()).hexdigest() != entry['sha256']: + raise ValueError('P105 firmware differs from the pinned recipe: ' + entry['name']) + return path + + +def prepare_mixin(inputs: Path, firmware_dir: Path, outdir: Path) -> Tuple[Path, MixinReceipt]: + record = json.loads(regular(inputs, private=True).read_text()) + if record.get('kind') != 'p105-mtcl' or record.get('profile') != PROFILE or record.get('schema') != 1: + raise ValueError('Expected a P105 MtCl capture record') + DeviceState(PROFILE, record['ecid']) + data = base64.b64decode(record['calibration'], validate=True) + if not 0 < len(data) <= 65536 or hashlib.sha256(data).hexdigest() != record['sha256']: + raise ValueError('P105 panel capture integrity check failed') + recipe = json.loads(firmware_path(PROFILE).read_text()) + payload = {'lib/firmware/mtcal.bin': data} + for entry in recipe['files']+recipe['generated']: + path = firmware_dir / entry['destination'] + if not path.resolve().is_relative_to(firmware_dir.resolve()): + raise ValueError('Firmware escapes the owner root') + value = regular(path).read_bytes() + if hashlib.sha256(value).hexdigest() != entry['sha256']: + raise ValueError('P105 firmware differs from the pinned recipe') + payload[entry['destination']] = value + payload['manifest.json'] = json.dumps({'schema': 1, 'profile': PROFILE, 'ecid': record['ecid'], + 'recipe_sha256': hashlib.sha256(firmware_path(PROFILE).read_bytes()).hexdigest()}).encode() + if outdir.exists() or outdir.is_symlink(): + raise ValueError('Choose a fresh private mixin directory') + outdir.mkdir(parents=True, mode=0o700) + output = outdir/'p105-mixin.tar.gz' + with tarfile.open(output, 'w:gz') as archive: + for name, value in sorted(payload.items()): + member = tarfile.TarInfo(name) + member.size, member.mode = len(value), 0o644 + archive.addfile(member, io.BytesIO(value)) + output.chmod(0o600) + return output, {'schema': 1, 'kind': 'private-mixin', 'bytes': output.stat().st_size, + 'sha256': hashlib.sha256(output.read_bytes()).hexdigest()} diff --git a/src/orchard_cli/p105/boot.py b/src/orchard_cli/p105/boot.py new file mode 100644 index 0000000..163b36c --- /dev/null +++ b/src/orchard_cli/p105/boot.py @@ -0,0 +1,233 @@ +"""Qualified direct-iBEC transport with a durable record before each mutation.""" +from contextlib import contextmanager +import hashlib +import json +import os +from pathlib import Path +import re +import secrets +import socket +import subprocess +import time +from typing import Any, Dict, Iterator, Mapping, Tuple, cast + +from orchard_cli.contracts import BootAssets, BootRequest, SelectedDevice +from orchard_cli.progress import Progress +from orchard_cli.state import DeviceState, private_directory +from . import usb +from .config import PROFILE, regular, settings, tool + +KLOADER_SHA = 'bec212b4d99a1e3b9fa4add663fa884a2435045d1e47051415f4e26e197f8132' +IBEC_SHA = 'e987b101800bedf03c086aa55854ff594fcce1b0f55ea1424f3ccf180a40abce' + + +def save(path: Path, value: Dict[str, Any]) -> None: + temporary = path.with_suffix('.pending') + fd = os.open(temporary, os.O_WRONLY | os.O_CREAT | os.O_TRUNC | os.O_NOFOLLOW, 0o600) + with os.fdopen(fd, 'w') as stream: + json.dump(value, stream) + stream.flush() + os.fsync(stream.fileno()) + os.replace(temporary, path) + + +class IOSConnection: + def __init__(self, device: SelectedDevice, port: int, config: Dict[str, Any]): + self.device, self.port, self.config = device, port, config + + def run(self, command: str, content: bytes = b'', *, transition: bool = False) -> bytes: + usb.verify_ios(dict(self.device)) + ios = self.config['ios'] + hosts = regular(Path(ios['known_hosts']), private=True) + argv = ['ssh', '-F', '/dev/null', '-p', str(self.port), '-o', 'HostKeyAlias=p105-ios-usb', + '-o', 'UserKnownHostsFile=' + str(hosts), '-o', 'StrictHostKeyChecking=yes', + '-o', 'HostKeyAlgorithms=+ssh-rsa', '-o', 'ConnectTimeout=8', + '-o', 'IdentitiesOnly=yes', '-o', 'ControlMaster=no'] + read_fd = write_fd = -1 + try: + if 'identity' in ios: + argv += ['-i', str(regular(Path(ios['identity']), private=True)), '-o', 'BatchMode=yes'] + else: + password = regular(Path(ios['password_file']), private=True).read_bytes().strip() + if not password or len(password) > 1024 or b'\n' in password or b'\r' in password: + raise ValueError('Invalid private iOS SSH password input') + read_fd, write_fd = os.pipe() + os.write(write_fd, password+b'\n') + os.close(write_fd) + write_fd = -1 + argv = [tool('sshpass'), '-d', str(read_fd), *argv, '-o', 'PreferredAuthentications=password', + '-o', 'NumberOfPasswordPrompts=1'] + result = subprocess.run([*argv, 'root@127.0.0.1', command], input=content, + capture_output=True, timeout=45, pass_fds=(read_fd,) if read_fd >= 0 else ()) + if transition: + if result.returncode not in (0, 255) or b'ORCHARD_DIRECT_IBEC_EXEC' not in result.stdout.splitlines(): + raise RuntimeError('iBEC transition outcome is unknown; inspect the device before another boot') + elif result.returncode: + raise RuntimeError('Pinned iOS SSH operation failed') + return result.stdout + finally: + for fd in (read_fd, write_fd): + if fd >= 0: + os.close(fd) + + +@contextmanager +def ios_connection(device: SelectedDevice) -> Iterator[IOSConnection]: + config = settings() + if config['target'] != {'udid': device.get('udid'), 'ecid': device['ecid']}: + raise ValueError('Owner configuration belongs to another iPad') + usb.verify_ios(dict(device)) + with socket.socket() as sock: + sock.bind(('127.0.0.1', 0)) + port = sock.getsockname()[1] + proxy = subprocess.Popen([tool('iproxy'), '-u', device['udid'], str(port), '22', '-s', '127.0.0.1'], + stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) + try: + deadline = time.monotonic()+10 + while True: + if proxy.poll() is not None: + raise RuntimeError('Selected iOS USB proxy exited') + try: + with socket.create_connection(('127.0.0.1', port), timeout=0.2): + break + except OSError: + if time.monotonic() >= deadline: + raise RuntimeError('Selected iOS USB proxy timed out') + time.sleep(0.1) + yield IOSConnection(device, port, config) + finally: + proxy.terminate() + try: + proxy.wait(timeout=5) + except subprocess.TimeoutExpired: + proxy.kill() + proxy.wait() + + +def verify_assets(device: SelectedDevice, assets: BootAssets, manifest: BootRequest) -> None: + paths = cast(Mapping[str, Path], assets) + if manifest.get('target_ecid') != device['ecid']: + raise ValueError('Private boot manifest belongs to another device') + components = Path(settings()['components']) + receipt = json.loads(regular(components/'components.json', private=True).read_text()) + if (receipt.get('device_ecid') != device['ecid'] or receipt.get('mode') != 'orchard-recovery' or + receipt.get('private') is not True or receipt.get('publishable') is not False or + receipt.get('flash_controller') != 'disabled'): + raise ValueError('P105 components lack the selected recovery identity or flash fence') + for role, name in (('recovery','staging-bundle.bin'),('loader','staging-loader.bin')): + if receipt['components'][name]['sha256'] != manifest['assets'][role]['sha256']: + raise ValueError('Boot manifest differs from the verified P105 component receipt') + for role in ('recovery', 'loader', 'ibec', 'kloader'): + path = regular(paths[role]) + if hashlib.sha256(path.read_bytes()).hexdigest() != manifest['assets'][role]['sha256']: + raise ValueError('P105 boot asset changed: ' + role) + for role, digest in (('kloader', KLOADER_SHA), ('ibec', IBEC_SHA)): + if manifest['assets'][role]['sha256'] != digest: + raise ValueError('P105 requires the qualified ' + role + ' recipe') + loader = assets['loader'].read_bytes() + bundle = assets['recovery'].read_bytes() + if not 0 < len(loader) <= 32768 or not bundle.startswith(loader.ljust(32768, b'\0')): + raise ValueError('P105 bundle and staging loader differ') + + +def upload(device: SelectedDevice, path: Path) -> None: + usb.ibec(device['ecid']) + result = subprocess.run([tool('irecovery'), '-i', str(device['ecid']), '-f', str(path)], + capture_output=True, timeout=180) + if result.returncode or not re.search(rb'(? Tuple[Path, Dict[str, Any]]: + candidates = list(folder.glob('*.json')) + if not candidates: + raise ValueError('iBEC requires a saved Orchard transition; start from iOS') + receipt = max(candidates, key=lambda path: path.stat().st_mtime_ns) + regular(receipt, private=True) + record = json.loads(receipt.read_text()) + age = time.time() - receipt.stat().st_mtime + if (not isinstance(record, dict) or record.get('phase') != 'awaiting-ibec' or + record.get('ecid') != device['ecid'] or record.get('assets') != dict(manifest['assets']) or + not 0 <= age <= 900): + raise ValueError('Only the latest verified iBEC wait can continue; inspect unknown outcomes before a fresh iOS boot') + return receipt, record + + +def enter_ios(device: SelectedDevice, assets: BootAssets, manifest: BootRequest, + receipt: Path, record: Dict[str, Any], progress: Progress) -> None: + paths = cast(Mapping[str, Path], assets) + with ios_connection(device) as connection: + directory = '/tmp/orchard-p105-ibec-'+secrets.token_hex(16) + connection.run('umask 077; mkdir '+directory) + for role, name in (('kloader', 'kloader'), ('ibec', 'ibec.raw')): + content = paths[role].read_bytes() + target = directory+'/'+name + connection.run('set -C; umask 077; cat > '+target+' && chmod '+('700' if role == 'kloader' else '600')+' '+target, content) + if connection.run('cat '+target) != content: + raise ValueError('Staged iOS payload differs from its source') + record['phase'] = 'transition-outcome-unknown' + save(receipt, record) + checks = ['test -d '+directory, 'test ! -L '+directory] + for role, name in (('kloader', 'kloader'), ('ibec', 'ibec.raw')): + target = directory+'/'+name + checks.append('test -f '+target+'; test ! -L '+target+'; d=$(/usr/bin/openssl dgst -sha256 < '+target+'); ' + '[ "${d##* }" = "'+manifest['assets'][role]['sha256']+'" ]') + checks += ['echo ORCHARD_DIRECT_IBEC_EXEC', 'exec '+directory+'/kloader '+directory+'/ibec.raw'] + progress.phase('Entering iBEC on the selected iPad') + connection.run('set -eu; '+'; '.join(checks), transition=True) + record['phase'] = 'awaiting-ibec' + save(receipt, record) + + +def boot(device: SelectedDevice, assets: BootAssets, manifest: BootRequest, progress: Progress) -> None: + if manifest.get('target_ecid') != device['ecid']: + raise ValueError('Private boot manifest belongs to another device') + if device['mode'] == 'linux': + current = usb.recovery() + if current['ecid'] != device['ecid'] or current['boot_id'] != device.get('boot_id'): + raise ValueError('Selected running P105 boot changed') + return + verify_assets(device, assets, manifest) + paths = cast(Mapping[str, Path], assets) + folder = private_directory(DeviceState(PROFILE, device['ecid']).root / 'boot-attempts') + if device['mode'] == 'ibec': + receipt, record = pending_attempt(folder, device, manifest) + progress.phase('Continuing the saved iBEC wait for the same verified boot files') + elif device['mode'] == 'ios': + record = {'ecid': device['ecid'], 'assets': dict(manifest['assets']), 'phase': 'validated'} + receipt = folder / (secrets.token_hex(16)+'.json') + save(receipt, record) + enter_ios(device, assets, manifest, receipt, record, progress) + else: + raise ValueError('Start a P105 boot from iOS or its saved iBEC transition') + progress.phase('Reconnect the iPad USB cable once; waiting for the selected iBEC') + deadline = time.monotonic()+600 + while True: + try: + usb.ibec(device['ecid']) + break + except ValueError: + if time.monotonic() >= deadline: + raise RuntimeError('iBEC wait expired; reconnect USB and repeat this command to continue its saved wait') + time.sleep(1) + for role in ('recovery', 'loader'): + record['phase'] = role+'-upload-outcome-unknown' + save(receipt, record) + upload(device, paths[role]) + record['phase'] = 'awaiting-linux' + save(receipt, record) + progress.phase('Waiting for identified P105 RAM recovery') + deadline = time.monotonic()+120 + while True: + try: + current = usb.recovery() + if current['ecid'] != device['ecid']: + raise RuntimeError('Another device appeared after the P105 upload') + record.update(phase='complete', boot_id=current['boot_id']) + save(receipt, record) + device.update(cast(SelectedDevice, {**current, 'resumed': True})) + return + except ValueError: + if time.monotonic() >= deadline: + raise RuntimeError('P105 recovery did not enumerate; retain the saved boot attempt for inspection') + time.sleep(0.5) diff --git a/src/orchard_cli/p105/config.py b/src/orchard_cli/p105/config.py new file mode 100644 index 0000000..609c881 --- /dev/null +++ b/src/orchard_cli/p105/config.py @@ -0,0 +1,41 @@ +"""Explicit owner inputs. Importing the adapter never reads this configuration.""" +import json +import os +from pathlib import Path +import stat +from typing import Any, Dict + +from orchard_cli.cache import cache_root +from orchard_cli.state import DeviceState + +PROFILE = 'p105' +NAME = 'iPad mini 1 (Wi-Fi)' +PRODUCT = 'Orchard-P105' +CHIP, BOARD = 0x8942, 0x0a + + +def regular(path: Path, *, private: bool = False) -> Path: + info = path.lstat() + if not stat.S_ISREG(info.st_mode) or (private and (info.st_mode & 0o077 or info.st_uid != os.getuid())): + raise ValueError('Expected an owner-only regular input' if private else 'Expected a regular input') + return path + + +def settings() -> Dict[str, Any]: + path = Path(os.environ.get('ORCHARD_P105_CONFIG', cache_root() / 'private/p105.json')).expanduser().absolute() + regular(path, private=True) + value = json.loads(path.read_text()) + if value.get('schema') != 1 or not isinstance(value.get('target'), dict): + raise ValueError('Expected P105 owner configuration schema 1') + DeviceState(PROFILE, value['target']['ecid']) + import re + if not re.fullmatch('[0-9a-fA-F]{40}', value['target'].get('udid', '')): + raise ValueError('Invalid configured P105 UDID') + return value + + +def tool(name: str) -> str: + path = Path(settings()['tools']) / name + if not regular(path).stat().st_mode & 0o111: + raise ValueError('P105 host tool is not executable: ' + name) + return str(path) diff --git a/src/orchard_cli/p105/network.py b/src/orchard_cli/p105/network.py new file mode 100644 index 0000000..3b706c9 --- /dev/null +++ b/src/orchard_cli/p105/network.py @@ -0,0 +1,129 @@ +"""Own only the selected USB link's address; use an existing owner NFS export.""" +import json +import os +from pathlib import Path +import re +import shlex +import subprocess +import sys +import time +from typing import Any, Dict + +from orchard_cli.state import BootState, private_directory +from . import usb +from .boot import save + +HOST_MAC = '02:10:5a:05:00:01' +HOST = '10.55.0.1' + + +def interfaces(selected: str = '') -> Dict[str, str]: + if sys.platform == 'darwin': + text = subprocess.check_output(['/sbin/ifconfig', '-a'], text=True, timeout=10) + result = {} + for block in re.split(r'(?=^[A-Za-z0-9][A-Za-z0-9._-]*:)', text, flags=re.M): + name = block.split(':', 1)[0] + if (selected and name == selected) or (not selected and re.search(r'\bether '+HOST_MAC+r'\b', block, re.I)): + result[name] = block + return result + if sys.platform == 'linux': + rows = json.loads(subprocess.check_output(['ip', '-j', 'address'], text=True, timeout=10)) + return {r['ifname']: json.dumps(r) for r in rows if (selected and r['ifname'] == selected) or + (not selected and r.get('address', '').lower() == HOST_MAC)} + raise ValueError('P105 USB networking supports macOS and Linux hosts') + + +def privileged(argv: list) -> None: + if os.geteuid() == 0: + subprocess.run(argv, check=True, timeout=20) + elif sys.platform == 'darwin': + script = 'do shell script '+json.dumps(shlex.join(argv))+' with administrator privileges' + subprocess.run(['/usr/bin/osascript', '-e', script], check=True, timeout=120) + else: + subprocess.run(['sudo', '--', *argv], check=True, timeout=120) + + +def address_present(text: str) -> bool: + return bool(re.search(r'(? Path: + deadline = time.monotonic()+15 + while True: + endpoint = usb.recovery() + if not scope.matches(endpoint): + raise ValueError('USB identity changed before network setup') + selected = endpoint.get('network_interface', '') + if selected and endpoint.get('network_mac') != HOST_MAC: + raise ValueError('Selected P105 USB network address differs from the gadget recipe') + if selected and not re.fullmatch('[A-Za-z0-9][A-Za-z0-9._-]{0,31}', selected): + raise ValueError('Unexpected selected USB interface name') + rows = interfaces(selected) if selected else {} + if len(rows) == 1: + break + if len(rows) > 1 or time.monotonic() >= deadline: + raise ValueError('Expected exactly one P105 USB network interface') + time.sleep(0.25) + name, text = next(iter(rows.items())) + if not re.fullmatch('[A-Za-z0-9][A-Za-z0-9._-]{0,31}', name): + raise ValueError('Unexpected USB interface name') + path = private_directory(scope.root)/'network.json' + if path.exists(): + previous = json.loads(path.read_text()) + if previous.get('phase') == 'address-outcome-unknown': + raise ValueError('USB address outcome is unknown; inspect its saved record before another setup') + if previous.get('changed'): + if previous['interface'] != name or not address_present(text): + raise ValueError('Owned USB network state changed; inspect its saved record') + return path + record: Dict[str, Any] = {'profile': 'p105', 'ecid': scope.device.ecid, 'boot_id': scope.boot_id, + 'interface': name, 'mac': HOST_MAC, 'changed': False} + if not address_present(text): + record['phase'] = 'address-outcome-unknown' + save(path, record) + command = (['/sbin/ifconfig', name, 'inet', HOST, 'netmask', '255.255.255.0', 'alias'] + if sys.platform == 'darwin' else ['ip', 'address', 'add', HOST+'/24', 'dev', name]) + privileged(command) + current = usb.recovery() + if (not scope.matches(current) or current.get('network_interface') != name or + not address_present(interfaces(name).get(name, ''))): + raise ValueError('Selected USB address setup could not be verified') + record['changed'] = True + record['phase'] = 'ready' + save(path, record) + return path + + +def nfs_root(value: str) -> str: + if not re.fullmatch(r'10\.55\.0\.1:/[A-Za-z0-9._/-]+', value) or '/..' in value or value.endswith('/'): + raise ValueError('Expected the host USB export as 10.55.0.1:/absolute/export/path') + return value + + +def check_export(value: str) -> None: + path = nfs_root(value).split(':', 1)[1] + result = subprocess.run(['showmount', '-e', HOST], capture_output=True, text=True, timeout=15) + if result.returncode or not any(line.split() and line.split()[0] == path for line in result.stdout.splitlines()): + raise ValueError('The requested NFS root is absent from the existing host export') + + +def cleanup(path: Path) -> Dict[str, Any]: + from .config import regular + record = json.loads(regular(path, private=True).read_text()) + if record.get('profile') != 'p105' or record.get('mac') != HOST_MAC or record.get('phase') != 'ready': + raise ValueError('Expected a completed P105 network ownership record') + name = record['interface'] + if record['changed']: + if any(usb.product_matches(row.get('product')) for row in usb.apple_usb.devices()): + raise ValueError('Shut down the P105 Linux session before removing its USB address') + rows = interfaces() + if name not in rows and not interfaces(name): + save(path, {**record, 'phase': 'cleaned', 'changed': False}) + return {'cleaned': True, 'usb_link': 'retired', 'nfs_service': 'existing owner service retained'} + if name not in rows or not address_present(rows[name]): + raise ValueError('Owned network address changed; cleanup stopped') + command = (['/sbin/ifconfig', name, 'inet', HOST, '-alias'] if sys.platform == 'darwin' else + ['ip', 'address', 'delete', HOST+'/24', 'dev', name]) + privileged(command) + save(path, {**record, 'phase': 'cleaned', 'changed': False}) + return {'cleaned': True, 'nfs_service': 'existing owner service retained'} diff --git a/src/orchard_cli/p105/prepare.py b/src/orchard_cli/p105/prepare.py new file mode 100644 index 0000000..4fef2c1 --- /dev/null +++ b/src/orchard_cli/p105/prepare.py @@ -0,0 +1,44 @@ +"""Make an owner-local boot manifest from qualified private components.""" +import hashlib +import json +import os +from pathlib import Path +import shutil +from typing import Any, Dict + +from .boot import IBEC_SHA, KLOADER_SHA +from .config import regular + + +def manifest(components: Path, ibec: Path, kloader: Path, output: Path) -> Dict[str, Any]: + record = json.loads(regular(components/'components.json', private=True).read_text()) + if (record.get('profile') != 'p105' or record.get('private') is not True or + record.get('publishable') is not False or record.get('mode') != 'orchard-recovery' or + type(record.get('device_ecid')) is not int or not 0 < record['device_ecid'] < 2**64 or + record.get('flash_controller') != 'disabled'): + raise ValueError('Expected owner-bound P105 Orchard recovery components') + inputs = {'boot':components/'staging-bundle.bin','loader':components/'staging-loader.bin', + 'ibec':ibec,'kloader':kloader} + hashes = {role:hashlib.sha256(regular(path).read_bytes()).hexdigest() for role,path in inputs.items()} + if hashes['ibec'] != IBEC_SHA or hashes['kloader'] != KLOADER_SHA: + raise ValueError('P105 needs the qualified iBEC and kloader inputs') + for role in ('boot','loader'): + if hashes[role] != record['components'][inputs[role].name]['sha256']: + raise ValueError('Private boot components changed') + if output.exists() or output.is_symlink(): + raise ValueError('Choose a fresh owner boot directory') + os.umask(0o077) + output.mkdir(parents=True,mode=0o700) + rows = ['schema = 1','kind = "orchard.boot"','device = "p105"', + 'boot_method = "jailbreak-kdfu"','min_orchard = "0.1.11"','protocol = 1', + 'private = true','target_ecid = '+str(record['device_ecid'])] + for role,path in inputs.items(): + name = {'boot':'staging-bundle.bin','loader':'staging-loader.bin','ibec':'ibec.raw','kloader':'kloader'}[role] + shutil.copyfile(path,output/name) + (output/name).chmod(0o600) + rows += ['','['+role+']','file = '+json.dumps(name),'bytes = '+str(path.stat().st_size), + 'sha256 = '+json.dumps(hashes[role])] + destination = output/'boot.toml' + destination.write_text('\n'.join(rows)+'\n') + destination.chmod(0o600) + return {'manifest':str(destination),'private':True,'publishable':False} diff --git a/src/orchard_cli/p105/usb.py b/src/orchard_cli/p105/usb.py new file mode 100644 index 0000000..3066f3a --- /dev/null +++ b/src/orchard_cli/p105/usb.py @@ -0,0 +1,86 @@ +"""Read-only iOS and iBEC discovery alongside Orchard's identified USB gadget.""" +import plistlib +import re +import subprocess +from typing import Any, Dict, List + +from orchard_cli.apple import usb as apple_usb +from .config import BOARD, CHIP, PRODUCT, settings, tool + + +def product_matches(value: Any) -> bool: + # macOS's USB Product Name replaces the descriptor's hyphen with an underscore. + return value in (PRODUCT, PRODUCT.replace('-', '_')) + + +def query(*args: str) -> str: + result = subprocess.run(list(args), capture_output=True, timeout=15, check=False) + if result.returncode: + raise ValueError('Selected P105 USB query failed') + return result.stdout.decode().strip() + + +def ios(udid: str) -> Dict[str, Any]: + if not re.fullmatch('[0-9a-fA-F]{40}', udid): + raise ValueError('Invalid iOS device identifier') + data = plistlib.loads(query(tool('ideviceinfo'), '-u', udid, '-x').encode()) + if (data.get('ProductType') != 'iPad2,5' or str(data.get('HardwareModel', '')).lower() != 'p105ap' or + str(data.get('UniqueDeviceID', '')).lower() != udid.lower()): + raise ValueError('Selected device is not iPad2,5 / P105AP') + ecid = data.get('UniqueChipID') + if type(ecid) is not int or not 0 < ecid < 2**64: + raise ValueError('Invalid iOS chip identity') + return {'mode': 'ios', 'location': udid.lower(), 'udid': udid.lower(), 'ecid': ecid, + 'chip': CHIP, 'board': BOARD, 'product': 'iPad2,5', + 'ios_version': data.get('ProductVersion'), 'ios_build': data.get('BuildVersion')} + + +def verify_ios(device: Dict[str, Any]) -> None: + current = ios(device['udid']) + if current['ecid'] != device['ecid']: + raise ValueError('iOS device identity changed') + if (current['ios_version'], current['ios_build']) != ('9.3.5', '13G36'): + raise ValueError('This P105 boot path requires iOS 9.3.5 build 13G36') + + +def ibec(ecid: int) -> Dict[str, Any]: + text = query(tool('irecovery'), '-i', str(ecid), '-q') + fields: Dict[str, str] = {} + for line in text.splitlines(): + key, sep, value = line.partition(':') + if sep and key in ('ECID', 'PRODUCT', 'MODE', 'CPID', 'BDID'): + if key in fields: + raise ValueError('Ambiguous recovery identity') + fields[key] = value.strip() + if (fields.get('PRODUCT'), fields.get('MODE', '').lower()) != ('iPad2,5', 'recovery'): + raise ValueError('Expected the selected P105 in iBEC Recovery') + if int(fields.get('ECID', '0'), 0) != ecid: + raise ValueError('iBEC device identity changed') + return {'mode': 'ibec', 'location': str(ecid), 'ecid': ecid, 'chip': CHIP, 'board': BOARD} + + +def recovery() -> Dict[str, Any]: + matches = [row for row in apple_usb.devices() if row.get('mode') == 'linux' and product_matches(row.get('product'))] + if len(matches) != 1: + raise ValueError('Connect exactly one Orchard P105 recovery') + row = matches[0] + if not row.get('ecid') or not row.get('boot_id') or not row.get('serial_port'): + raise ValueError('P105 recovery lacks its device and boot identity') + return row + + +def observations() -> List[Dict[str, Any]]: + observed = apple_usb.devices() + result = [row for row in observed if product_matches(row.get('product'))] + identifiers = [row['udid'] for row in observed if row.get('udid')] + for identifier in identifiers: + try: + result.append(ios(identifier)) + except ValueError: + continue + if not result: + try: + result.append(ibec(settings()['target']['ecid'])) + except (ValueError, OSError): + pass + return result diff --git a/src/orchard_cli/releases.py b/src/orchard_cli/releases.py index b118c9e..de6658c 100644 --- a/src/orchard_cli/releases.py +++ b/src/orchard_cli/releases.py @@ -172,8 +172,16 @@ def select_manifest(source, profile, progress=None, clock=None, purpose='install allowed = ('orchard.install',) if purpose == 'install' else ('orchard.install', 'orchard.boot') if document['kind'] not in allowed or document['device'] != profile: raise ValueError('Manifest is not compatible with this device/operation') - if document['boot_method'] != 'checkm8-pongo': - raise ValueError('This Orchard build implements only checkm8-pongo') + method = document['boot_method'] + if method not in ('checkm8-pongo', 'jailbreak-kdfu') or method not in hardware_device(profile)['boot']['methods']: + raise ValueError('Unsupported boot method for this device') + if method == 'jailbreak-kdfu': + if purpose == 'install': + raise ValueError('P105 supports RAM recovery and NFS handoff; internal installation awaits storage support') + if bundle is None or not document.get('private'): + raise ValueError('P105 boot requires an owner-local private manifest') + if any('file' not in document[role] for role in ('boot', 'loader', 'ibec', 'kloader')): + raise ValueError('P105 private boot assets must be local files') if document['protocol'] != 1: raise ValueError('Unsupported device protocol') actual = tuple(int(part) for part in version('orchard-cli').split('.')) @@ -187,17 +195,22 @@ def select_manifest(source, profile, progress=None, clock=None, purpose='install def resolve(manifest, profile, progress=None, clock=None, purpose='install'): document, bundle = select_manifest(manifest, profile, progress, clock, purpose) - roles = {'pongo': 'pongo', 'recovery': 'boot'} + roles = ({'pongo': 'pongo', 'recovery': 'boot'} if document['boot_method'] == 'checkm8-pongo' else + {'recovery': 'boot', 'loader': 'loader', 'ibec': 'ibec', 'kloader': 'kloader'}) if purpose == 'install': roles['root'] = 'root' paths, receipts = {}, {role: document[field] for role, field in roles.items()} - labels = {'pongo': 'Pongo', 'recovery': 'Recovery', 'root': document.get('name', 'Linux')} + labels = {'pongo': 'Pongo', 'recovery': 'Recovery', 'root': document.get('name', 'Linux'), + 'loader': 'Staging loader', 'ibec': 'Private iBEC', 'kloader': 'iOS loader'} with ThreadPoolExecutor(max_workers=len(roles)) as pool: futures = {role: pool.submit(materialize, receipt, bundle, progress, clock, label=labels[role]) for role, receipt in receipts.items()} for role, future in futures.items(): paths[role] = future.result() record = {'protocol': document['protocol'], 'profile': profile, 'assets': receipts, + 'boot_method': document['boot_method'], 'release': document.get('version'), 'name': document.get('name', 'Linux'), 'default_root': document.get('install', {}).get('default_root', '/linux')} + if 'target_ecid' in document: + record['target_ecid'] = document['target_ecid'] return record, paths diff --git a/src/orchard_cli/service.py b/src/orchard_cli/service.py index 0e832ce..d31b95f 100644 --- a/src/orchard_cli/service.py +++ b/src/orchard_cli/service.py @@ -163,6 +163,40 @@ def wait_ready(client, progress, timeout=90, *, context='Root installation compl time.sleep(0.5) +def handoff_nfs(connection, source, progress): + from .p105.network import check_export + scope = connection.scope + client = Client(connection, progress) + status = client.call('status') + if not scope.matches(status) or status.get('profile') != scope.device.profile: + raise ValueError('NFS handoff identity differs from the selected boot') + if 'handoff-nfs' not in status.get('capabilities', []): + raise ValueError('Selected recovery does not offer an NFS handoff') + check_export(source) + path = private_directory(scope.root)/'last-nfs-handoff.json' + if path.exists(): + raise ValueError('NFS handoff was already attempted in this boot; inspect its saved report') + request = {'profile': scope.device.profile, 'ecid': scope.device.ecid, + 'boot_id': scope.boot_id, 'nfs_root': source, 'operation_id': str(uuid.uuid4())} + report = {**request, 'status': 'unknown', 'recovery_status': status} + save_report(path, report) + try: + result = client.call('handoff', request, timeout=20) + if (result.get('status') != 'accepted' or + any(result.get(key) != value for key, value in request.items())): + raise ValueError('NFS handoff receipt differs from this request') + progress.phase('NFS handoff accepted; waiting for the installed system') + ready = wait_ready(client, progress, timeout=120, context='NFS handoff was accepted') + if not scope.matches(ready) or ready.get('root_filesystem') not in ('nfs', 'nfs4'): + raise ValueError('Installed NFS system differs from this recovery boot') + report.update(status='complete', installed_status=ready) + save_report(path, report) + return ready + except BaseException as error: + save_report(path, {**report, 'error': str(error)}) + raise + + def finish_ssh_access(client, adapter: Adapter, status, report, path, clock, input_fn, progress, *, scope): if not scope.matches(status) or status.get('profile') != scope.device.profile: raise ValueError('Readiness identity differs from the credential scope') @@ -206,9 +240,9 @@ def finish_ssh_access(client, adapter: Adapter, status, report, path, clock, inp @exclusive_operation -def install(manifest=None, *, input_fn=input, progress=None): +def install(manifest=None, *, input_fn=input, progress=None, profile=None): clock, progress = Clock(), progress or Progress() - adapter, device = choose() # Definition of installation timing begins here. + adapter, device = choose(profile) if profile else choose() answer = ask(clock, f'Install Linux on {adapter.NAME}? [Y/n] ', input_fn) if answer.lower() not in ('', 'y', 'yes'): progress.phase('Cancelled; no download or device mutation performed') @@ -251,8 +285,11 @@ def install(manifest=None, *, input_fn=input, progress=None): if firmware is None: firmware = adapter.firmware_inputs(clock, progress) progress.phase('Preparing device hardware…') - adt = work / 'device.adt' - adt.write_bytes(adapter.capture_adt(device)) + capture = adapter.capture_inputs(device) + if capture['profile'] != adapter.PROFILE or capture['ecid'] != device['ecid']: + raise ValueError('Private input capture differs from the selected device') + adt = work / ('device.adt' if capture['kind'] == 'apple-adt' else 'device-inputs.json') + adt.write_bytes(capture['data']) adt.chmod(0o600) mixin, mixin_record = adapter.prepare_mixin(adt, firmware, work / 'mixin') with mixin.open('rb') as source: diff --git a/src/orchard_cli/state.py b/src/orchard_cli/state.py index c3ab58a..369dd67 100644 --- a/src/orchard_cli/state.py +++ b/src/orchard_cli/state.py @@ -15,6 +15,10 @@ class EndpointUnavailable(RuntimeError): """The selected USB endpoint is temporarily absent, not a different identity.""" +class AmbiguousDevices(ValueError): + """Discovery found multiple candidates and cannot select a device.""" + + def private_directory(path: Path) -> Path: """No linked private-state components; credentials never inherit public modes.""" root = cache_root() / 'private' diff --git a/tests/test_adapter_contract.py b/tests/test_adapter_contract.py index 85507b3..9999c19 100644 --- a/tests/test_adapter_contract.py +++ b/tests/test_adapter_contract.py @@ -8,7 +8,7 @@ from orchard_cli.j71b import adapter MEMBERS = ('PROFILE', 'NAME', 'REBOOT_MESSAGE', 'list_devices', 'discover', 'boot', - 'connect', 'connect_saved', 'forget_access', 'capture_adt', 'firmware_inputs', 'prepare_mixin') + 'connect', 'connect_saved', 'forget_access', 'capture_inputs', 'firmware_inputs', 'prepare_mixin') def module(**changes): @@ -53,7 +53,7 @@ def test_dfu_discovery_returns_explicit_resume_state_without_mutating_observatio def test_choose_passes_the_validated_selection_through_unchanged(self): selection = {'mode':'dfu','location':'usb','ecid':1,'resumed':False} implementation = module(discover=Mock(return_value=selection)) - with patch.object(devices, 'load', return_value=implementation): + with patch.object(devices, 'load', return_value=implementation), patch.object(devices, 'SUPPORTED', ('j71b',)): loaded, selected = devices.choose() self.assertIs(loaded, implementation) self.assertIs(selected, selection) diff --git a/tests/test_device_state.py b/tests/test_device_state.py index 5212902..f00827c 100644 --- a/tests/test_device_state.py +++ b/tests/test_device_state.py @@ -137,7 +137,7 @@ def test_linux_usb_descriptor_identifies_device_and_boot_without_serial_commands for name,value in {'idVendor':'05ac','idProduct':'4142','product':'J71b Recovery', 'serial':f'ECID:0000000000000002 BOOT:{BOOT_B}'}.items(): (node/name).write_text(value) - (node/'iface/tty/ttyACM0').mkdir(parents=True) + (node/(node.name+':1.0')/'tty/ttyACM0').mkdir(parents=True) rows = linux_usb.devices(self.root) self.assertEqual(len(rows),1) self.assertEqual((rows[0]['ecid'],rows[0]['boot_id']),(2,BOOT_B)) diff --git a/tests/test_download_ui.py b/tests/test_download_ui.py index e123b0e..54ad1c2 100644 --- a/tests/test_download_ui.py +++ b/tests/test_download_ui.py @@ -64,7 +64,7 @@ def test_internal_device_events_are_private_diagnostics(self): def test_release_assets_start_concurrently(self): barrier = threading.Barrier(3) - document = {'protocol': 1, 'version': '1', 'name': 'Test Linux', + document = {'protocol': 1, 'version': '1', 'name': 'Test Linux', 'boot_method': 'checkm8-pongo', 'pongo': {}, 'boot': {}, 'root': {}} def materialize(*args, **kwargs): barrier.wait(timeout=2) diff --git a/tests/test_host_workflow.py b/tests/test_host_workflow.py index aa811a6..79afc47 100644 --- a/tests/test_host_workflow.py +++ b/tests/test_host_workflow.py @@ -55,7 +55,7 @@ def guided_run(self, filesystem): image.write_bytes(b'root-stream') mixin.write_bytes(b'private-data') adapter = Mock(NAME='Supported tablet', PROFILE='device') - adapter.capture_adt.return_value = b'fresh-adt' + adapter.capture_inputs.return_value = {'kind':'apple-adt','profile':'device','ecid':1,'data':b'fresh-adt'} adapter.prepare_mixin.return_value = (mixin, {'bytes': 12, 'sha256': 'mixin-digest'}) import threading downloads_started = threading.Barrier(2) diff --git a/tests/test_launch.py b/tests/test_launch.py index 850eb6c..0ea97ea 100644 --- a/tests/test_launch.py +++ b/tests/test_launch.py @@ -25,7 +25,7 @@ def run_flow(self, *, lost=False, discovered=True, answer='launch minimal', mani receipt={'release':'alpine','root_sha256':'a'*64,'directory':'minimal'} adapter=Mock(PROFILE='j71b',NAME='iPad') adapter.connect.return_value=nullcontext(connection) - adapter.capture_adt.return_value=b'fresh ADT' + adapter.capture_inputs.return_value={'kind':'apple-adt','profile':'j71b','ecid':1,'data':b'fresh ADT'} def mixin(adt,firmware,dest): dest.mkdir();path=dest/'mixin.tar.gz';path.write_bytes(b'x') return path,{'bytes':1,'sha256':'a'*64} @@ -43,7 +43,7 @@ def call(operation,request=None,**kwargs): raise AssertionError('Launch must not use an installation/storage mutation RPC: '+operation) client.call.side_effect=call ready={'ready':True,'release':'alpine','ecid':1,'boot_id':BOOT,'profile':'j71b'} - with patch.object(launch,'choose',return_value=(adapter,{'resumed':False})), \ + with patch.object(launch,'choose',return_value=(adapter,{'resumed':False,'ecid':1})), \ patch.object(launch,'resolve',return_value=({'release':'launcher'}, {'recovery':Path('boot')})) as resolve, \ patch.object(launch,'launch_manifest',return_value='https://example.invalid/j71b.boot.toml') as selected_manifest, \ patch.object(launch,'Client',return_value=client), \ diff --git a/tests/test_p105.py b/tests/test_p105.py new file mode 100644 index 0000000..57a87b4 --- /dev/null +++ b/tests/test_p105.py @@ -0,0 +1,275 @@ +"""P105 selection, private boot assets and handoff boundaries without hardware.""" +from contextlib import nullcontext +import hashlib +import io +import json +import os +from pathlib import Path +import subprocess +import tempfile +import unittest +from unittest.mock import Mock, patch +import uuid + +from orchard_cli import devices, releases, service +from orchard_cli.contracts import Adapter, Connection +from orchard_cli.p105 import adapter, boot, network, usb +from orchard_cli.progress import Progress +from orchard_cli.protocol import DeviceError +from orchard_cli.state import BootState, DeviceState + +BOOT = '11111111-1111-4111-8111-111111111111' + + +class P105(unittest.TestCase): + def setUp(self): + self.temp = tempfile.TemporaryDirectory() + self.addCleanup(self.temp.cleanup) + self.root = Path(self.temp.name).resolve() + self.env = patch.dict(os.environ, ORCHARD_HOME=str(self.root/'store')) + self.env.start() + self.addCleanup(self.env.stop) + + def test_registered_adapter_import_has_no_device_or_config_access(self): + with patch.object(adapter, 'discover', side_effect=AssertionError('USB access')), \ + patch('orchard_cli.p105.config.settings', side_effect=AssertionError('private read')): + self.assertIs(devices.load('p105'), adapter) + self.assertIsInstance(adapter, Adapter) + + def test_explicit_install_profile_is_preserved_before_confirmation(self): + implementation=Mock(PROFILE='p105',NAME='iPad mini') + with patch.object(service,'choose',return_value=(implementation,{'mode':'ios'})) as selected: + result=service.install('unused',profile='p105',input_fn=lambda _:'n',progress=Progress(io.StringIO())) + self.assertTrue(result['cancelled']) + selected.assert_called_once_with('p105') + implementation.boot.assert_not_called() + + def test_firmware_inputs_are_verified_before_returning_the_cache(self): + payload=self.root/'firmware';payload.mkdir();blob=payload/'blob';blob.write_bytes(b'firmware') + recipe=self.root/'recipe.json';recipe.write_text(json.dumps({'files':[{'name':'test', + 'destination':'blob','sha256':hashlib.sha256(b'firmware').hexdigest()}],'generated':[]})) + with patch.object(adapter,'settings',return_value={'firmware_root':str(payload)}), \ + patch.object(adapter,'firmware_path',return_value=recipe): + self.assertEqual(adapter.firmware_inputs(Mock(),Mock(),allow_download=False),payload) + blob.write_bytes(b'changed') + with self.assertRaisesRegex(ValueError,'pinned recipe'):adapter.firmware_inputs(Mock(),Mock()) + + def test_discovery_refuses_multiple_p105_devices(self): + one = {'mode':'ios','location':'one','ecid':1,'udid':'1'*40} + with patch.object(adapter, 'list_devices', return_value=[one,{**one,'ecid':2}]): + with self.assertRaisesRegex(ValueError, 'exactly one'): + adapter.discover() + + def test_unrelated_ios_product_is_rejected(self): + import plistlib + data = {'ProductType':'iPhone10,6','HardwareModel':'d221ap','UniqueDeviceID':'1'*40,'UniqueChipID':1} + with patch.object(usb, 'tool', return_value='ideviceinfo'), \ + patch.object(usb, 'query', return_value=plistlib.dumps(data).decode()): + with self.assertRaisesRegex(ValueError, 'iPad2,5'): + usb.ios('1'*40) + + def test_ios_discovery_uses_connected_usb_identifiers(self): + observed=[{'mode':'other','udid':'1'*40,'product':'iPad'}, + {'mode':'other','udid':'2'*40,'product':'iPhone'}] + ipad={'mode':'ios','ecid':1,'udid':'1'*40} + def query(identifier): + if identifier=='1'*40:return ipad + raise ValueError('unrelated iPhone') + with patch.object(usb.apple_usb,'devices',return_value=observed),patch.object(usb,'ios',side_effect=query): + self.assertEqual(usb.observations(),[ipad]) + + def test_recovery_requires_one_identified_serial_endpoint(self): + row = {'mode':'linux','product':'Orchard-P105','ecid':1,'boot_id':BOOT,'serial_port':'/dev/ttyACM0'} + with patch.object(usb.apple_usb, 'devices', return_value=[row,row]): + with self.assertRaises(ValueError):usb.recovery() + with patch.object(usb.apple_usb, 'devices', return_value=[{**row,'boot_id':None}]): + with self.assertRaisesRegex(ValueError, 'identity'):usb.recovery() + + def test_ibec_query_is_targeted_and_rejects_changed_ecid(self): + text='PRODUCT: iPad2,5\nMODE: Recovery\nECID: 0x2\n' + with patch.object(usb, 'tool', return_value='irecovery'), patch.object(usb, 'query', return_value=text) as query: + with self.assertRaisesRegex(ValueError, 'identity changed'):usb.ibec(1) + self.assertEqual(query.call_args.args,('irecovery','-i','1','-q')) + + def test_wrong_private_target_is_refused_before_state_or_transport(self): + with patch.object(boot, 'ios_connection') as ios, patch.object(boot, 'upload') as upload: + with self.assertRaisesRegex(ValueError, 'another device'): + boot.boot({'mode':'ios','ecid':1}, {}, {'target_ecid':2,'assets':{}}, Mock()) + ios.assert_not_called();upload.assert_not_called() + self.assertFalse((self.root/'store').exists()) + + def test_uncertain_upload_is_never_retried(self): + result = subprocess.CompletedProcess([],0,stdout=b'99%',stderr=b'') + with patch.object(boot.usb,'ibec') as selected, patch.object(boot,'tool',return_value='irecovery'), \ + patch.object(boot.subprocess,'run',return_value=result) as send: + with self.assertRaisesRegex(RuntimeError,'unknown'): + boot.upload({'ecid':1},Path('bundle')) + selected.assert_called_once_with(1) + send.assert_called_once() + + def test_upload_timeout_propagates_without_retry(self): + with patch.object(boot.usb,'ibec'), patch.object(boot,'tool',return_value='irecovery'), \ + patch.object(boot.subprocess,'run',side_effect=subprocess.TimeoutExpired('irecovery',180)) as send: + with self.assertRaises(subprocess.TimeoutExpired):boot.upload({'ecid':1},Path('bundle')) + send.assert_called_once() + + def test_saved_ibec_wait_continues_before_either_upload(self): + folder=self.root/'attempts';folder.mkdir() + receipt=folder/'wait.json';manifest={'target_ecid':1,'assets':{'recovery':{'sha256':'a'*64}}} + record={'ecid':1,'assets':manifest['assets'],'phase':'awaiting-ibec'} + boot.save(receipt,record) + with patch.object(boot.time,'time',return_value=receipt.stat().st_mtime+1): + path,result=boot.pending_attempt(folder,{'ecid':1},manifest) + self.assertEqual(path,receipt);self.assertEqual(result,record) + + def test_saved_wait_rejects_uncertain_upload_changed_assets_or_stale_state(self): + folder=self.root/'attempts';folder.mkdir();receipt=folder/'wait.json' + manifest={'target_ecid':1,'assets':{'recovery':{'sha256':'a'*64}}} + base={'ecid':1,'assets':manifest['assets'],'phase':'awaiting-ibec'} + for change in ({'phase':'recovery-upload-outcome-unknown'},{'ecid':2},{'assets':{}},{}): + boot.save(receipt,{**base,**change}) + age=901 if not change else 1 + with self.subTest(change=change),patch.object(boot.time,'time',return_value=receipt.stat().st_mtime+age): + with self.assertRaisesRegex(ValueError,'latest verified'):boot.pending_attempt(folder,{'ecid':1},manifest) + + def test_ibec_continuation_uses_one_saved_attempt_and_each_payload_once(self): + from orchard_cli.state import private_directory + selected={'mode':'ibec','ecid':1,'location':'usb','resumed':False} + assets={'recovery':self.root/'bundle','loader':self.root/'loader'} + manifest={'target_ecid':1,'assets':{'recovery':{'sha256':'a'*64}}} + folder=private_directory(DeviceState('p105',1).root/'boot-attempts');receipt=folder/'wait.json' + boot.save(receipt,{'ecid':1,'assets':manifest['assets'],'phase':'awaiting-ibec'}) + endpoint={'mode':'linux','ecid':1,'boot_id':BOOT,'location':'usb','serial_port':'/dev/ttyACM0'} + phases=[] + def send(device,path):phases.append(json.loads(receipt.read_text())['phase']) + with patch.object(boot,'verify_assets'),patch.object(boot,'ios_connection') as ios, \ + patch.object(usb,'ibec'),patch.object(usb,'recovery',return_value=endpoint), \ + patch.object(boot,'upload',side_effect=send) as upload: + boot.boot(selected,assets,manifest,Mock()) + ios.assert_not_called();self.assertEqual(upload.call_count,2) + self.assertEqual(phases,['recovery-upload-outcome-unknown','loader-upload-outcome-unknown']) + self.assertEqual(json.loads(receipt.read_text())['phase'],'complete') + self.assertEqual(selected['boot_id'],BOOT) + + def test_running_recovery_is_observed_without_upload(self): + selected={'mode':'linux','ecid':1,'boot_id':BOOT,'resumed':True} + with patch.object(usb,'recovery',return_value=selected), patch.object(boot,'upload') as upload: + boot.boot(selected,{}, {'target_ecid':1,'assets':{}},Mock()) + upload.assert_not_called() + + def test_running_recovery_rejects_another_devices_manifest(self): + selected={'mode':'linux','ecid':1,'boot_id':BOOT,'resumed':True} + with patch.object(usb,'recovery') as query: + with self.assertRaisesRegex(ValueError,'another device'): + boot.boot(selected,{}, {'target_ecid':2,'assets':{}},Mock()) + query.assert_not_called() + + def test_macos_product_spelling_keeps_the_same_identity_checks(self): + row={'mode':'linux','product':'Orchard_P105','ecid':1,'boot_id':BOOT,'serial_port':'/dev/ttyACM0'} + with patch.object(usb.apple_usb,'devices',return_value=[row]): + self.assertEqual(usb.recovery(),row) + with patch.object(usb.apple_usb,'devices',return_value=[{**row,'boot_id':None}]): + with self.assertRaisesRegex(ValueError,'identity'):usb.recovery() + + def test_local_p105_manifest_resolves_without_pongo(self): + rows=['schema=1','kind="orchard.boot"','device="p105"','boot_method="jailbreak-kdfu"', + 'min_orchard="0.1.0"','protocol=1','private=true','target_ecid=1'] + for role in ('boot','loader','ibec','kloader'): + data=role.encode();(self.root/role).write_bytes(data) + rows += ['['+role+']','file="'+role+'"','bytes='+str(len(data)), + 'sha256="'+hashlib.sha256(data).hexdigest()+'"'] + manifest=self.root/'boot.toml';manifest.write_text('\n'.join(rows)+'\n') + record,assets=releases.resolve(manifest,'p105',purpose='boot') + self.assertEqual(set(assets),{'recovery','loader','ibec','kloader'}) + self.assertEqual(record['target_ecid'],1) + with self.assertRaises(ValueError):releases.resolve(manifest,'p105',purpose='install') + manifest.write_text(manifest.read_text().replace('private=true','private=false')) + with self.assertRaises(ValueError):releases.resolve(manifest,'p105',purpose='boot') + + def test_nfs_root_rejects_remote_server_and_shell_or_path_tricks(self): + for source in ('other:/root','10.55.0.1:/','10.55.0.1:/root\nother', + '10.55.0.1:/root/../else','10.55.0.1:/root;true'): + with self.subTest(source=source),self.assertRaises(ValueError):network.nfs_root(source) + + def test_existing_host_address_is_retained_during_cleanup(self): + scope=BootState(DeviceState('p105',1),BOOT) + with patch.object(usb,'recovery',return_value={'ecid':1,'boot_id':BOOT,'network_interface':'en9','network_mac':network.HOST_MAC}), \ + patch.object(network,'interfaces',return_value={'en9':'ether '+network.HOST_MAC+' inet 10.55.0.1'}), \ + patch.object(network,'privileged') as admin: + path=network.ensure(scope) + self.assertFalse(json.loads(path.read_text())['changed']) + self.assertTrue(network.cleanup(path)['cleaned']) + admin.assert_not_called() + + def test_selected_usb_registry_link_survives_masked_ifconfig_mac(self): + scope=BootState(DeviceState('p105',1),BOOT) + endpoint={'ecid':1,'boot_id':BOOT,'network_interface':'en9','network_mac':network.HOST_MAC} + with patch.object(usb,'recovery',return_value=endpoint), \ + patch.object(network,'interfaces',return_value={'en9':'ether 02:00:00:00:00:00 inet 10.55.0.1'}) as query, \ + patch.object(network,'privileged') as admin: + record=json.loads(network.ensure(scope).read_text()) + query.assert_called_with('en9') + admin.assert_not_called() + self.assertFalse(record['changed']) + + def test_macos_network_metadata_belongs_to_the_usb_parent(self): + import plistlib + def gadget(ecid,name,interface,port): + return {'IOObjectClass':'IOUSBHostDevice','idVendor':0x05ac,'idProduct':0x4142, + 'locationID':ecid,'USB Serial Number':f'ECID:{ecid:x} BOOT:{BOOT}', + 'USB Product Name':name,'IORegistryEntryChildren':[ + {'IOObjectClass':'IOSerialBSDClient','IOCalloutDevice':port}, + {'IOObjectClass':'IOUserNetworkEthernet','IOMACAddress':bytes.fromhex('02105a050001'), + 'IORegistryEntryChildren':[{'IOObjectClass':'IOSkywalkLegacyEthernetInterface','BSD Name':interface}]}]} + raw=plistlib.dumps([{'IOObjectClass':'IOUSBHostDevice','idVendor':0x05ac,'idProduct':0x800c, + 'locationID':99,'USB Product Name':'Hub','IORegistryEntryChildren':[ + gadget(1,'Orchard_P105','en9','/dev/cu.p105'),gadget(2,'Other','en10','/dev/cu.other')]}]) + with patch.object(usb.apple_usb.sys,'platform','darwin'),patch.object(usb.apple_usb.subprocess,'run', + return_value=subprocess.CompletedProcess([],0,raw,b'')): + rows=usb.apple_usb.devices() + selected=next(r for r in rows if r['product']=='Orchard_P105') + self.assertEqual(selected['network_interface'],'en9') + self.assertEqual(selected['serial_port'],'/dev/cu.p105') + self.assertNotIn('network_interface',next(r for r in rows if r['product']=='Hub')) + + def test_linux_network_metadata_comes_from_that_devices_interface(self): + from orchard_cli.apple import linux_usb + node=self.root/'1-2';node.mkdir() + for name,value in {'idVendor':'05ac','idProduct':'4142','product':'Orchard-P105', + 'serial':f'ECID:1 BOOT:{BOOT}'}.items():(node/name).write_text(value) + net=node/'1-2:1.0/net/usb0';net.mkdir(parents=True);(net/'address').write_text(network.HOST_MAC) + nested=node/'1-2.1/1-2.1:1.0/net/wrong';nested.mkdir(parents=True) + (nested/'address').write_text('aa:bb:cc:dd:ee:ff') + rows=linux_usb.devices(self.root) + self.assertEqual(rows[0]['network_interface'],'usb0') + self.assertEqual(rows[0]['network_mac'],network.HOST_MAC) + + def test_cleanup_refuses_an_attached_p105_even_if_selection_is_ambiguous(self): + scope=BootState(DeviceState('p105',1),BOOT) + path=scope.root/'network.json';path.parent.mkdir(parents=True) + path.write_text(json.dumps({'profile':'p105','ecid':1,'boot_id':BOOT,'mac':network.HOST_MAC, + 'phase':'ready','interface':'en9','changed':True}));path.chmod(0o600) + with patch.object(usb.apple_usb,'devices',return_value=[{'product':'Orchard_P105'},{'product':'Orchard-P105'}]), \ + patch.object(network,'privileged') as admin: + with self.assertRaisesRegex(ValueError,'Shut down'):network.cleanup(path) + admin.assert_not_called() + + def test_handoff_uncertainty_is_saved_and_never_replayed(self): + scope=BootState(DeviceState('p105',1),BOOT) + connection=Connection([],scope,Mock()) + client=Mock() + client.call.side_effect=[{'profile':'p105','ecid':1,'boot_id':BOOT,'capabilities':['handoff-nfs']},DeviceError('lost response')] + with patch.object(service,'Client',return_value=client),patch.object(network,'check_export'): + with self.assertRaises(DeviceError):service.handoff_nfs(connection,'10.55.0.1:/root',Mock()) + report=json.loads((scope.root/'last-nfs-handoff.json').read_text()) + self.assertEqual(report['status'],'unknown') + self.assertEqual(client.call.call_count,2) + client.call.side_effect=None + client.call.return_value={'profile':'p105','ecid':1,'boot_id':BOOT,'capabilities':['handoff-nfs']} + with patch.object(service,'Client',return_value=client),patch.object(network,'check_export'): + with self.assertRaisesRegex(ValueError,'already attempted'): + service.handoff_nfs(connection,'10.55.0.1:/root',Mock()) + self.assertEqual(client.call.call_count,3) + + +if __name__ == '__main__':unittest.main() diff --git a/tests/typecheck_adapter.py b/tests/typecheck_adapter.py index 8e0e328..e95b858 100644 --- a/tests/typecheck_adapter.py +++ b/tests/typecheck_adapter.py @@ -1,5 +1,7 @@ """Static conformance check: the registered module must satisfy Adapter.""" from orchard_cli.contracts import Adapter from orchard_cli.j71b import adapter +from orchard_cli.p105 import adapter as p105 J71B: Adapter = adapter +P105: Adapter = p105