From d3cad49ee37d12f318a86d2ed21f8d1261a13941 Mon Sep 17 00:00:00 2001 From: Dalton Alexandre Date: Thu, 29 Oct 2026 09:20:03 -0700 Subject: [PATCH] Reset the USB port when the device does not return after checkm8 After "Checkmate!" the device drops off USB and must come back in download mode. On some Linux hosts it never does until the cable is replugged at the device within palera1n's 20-second window. Disabling and re-enabling the hub port forces the re-detection: on a J71b the device came back 4 s after a reset where three bare attempts never did. setup-host's udev rule makes the port's disable control writable for the USB access group when a device enters DFU. During the Pongo bootstrap, Orchard watches palera1n's log and resets that port up to three times if the device has not returned 2.5 s after checkm8. Hosts without the control (macOS, or no rule) behave as before. Co-Authored-By: Claude Opus 5.5 --- src/orchard_cli/apple/boot.py | 22 ++++++-- src/orchard_cli/apple/replug.py | 77 ++++++++++++++++++++++++++++ src/orchard_cli/host_setup.py | 4 ++ tests/test_replug.py | 90 +++++++++++++++++++++++++++++++++ 4 files changed, 190 insertions(+), 3 deletions(-) create mode 100644 src/orchard_cli/apple/replug.py create mode 100644 tests/test_replug.py diff --git a/src/orchard_cli/apple/boot.py b/src/orchard_cli/apple/boot.py index 14ae628..649bcd9 100644 --- a/src/orchard_cli/apple/boot.py +++ b/src/orchard_cli/apple/boot.py @@ -9,7 +9,7 @@ from orchard_cli.state import DeviceState, private_directory from orchard_cli.transports import pongo -from . import usb +from . import replug, usb def wait_mode(device, mode, *, chip, board, product, seconds=30): @@ -88,15 +88,31 @@ def start_pongo(device, image, log, say, *, chip, board, product, seconds=30): if not executable: raise ValueError('palera1n is required on the host to boot Pongo') usb.selected(device['location'], device['ecid'], chip=chip, board=board, mode='dfu') + # Resolved while the device is present: its sysfs node goes away after checkm8. + control = replug.port_control(device['location']) say('Booting Pongo') - say('Wait for the Apple logo to appear. If it does not disappear after a few seconds, ' - 'disconnect and reconnect the cable at the device once.') + if control: + say('Wait for the Apple logo to appear. Orchard resets the USB port if the device does not ' + 'come back; if the logo stays for about 15 seconds, disconnect and reconnect the cable ' + 'at the device once.') + else: + say('Wait for the Apple logo to appear. If it does not disappear after a few seconds, ' + 'disconnect and reconnect the cable at the device once.') + watcher = replug.PortReset(log.name, control) if control else None + if watcher: + watcher.start() try: run_process_group([executable, '-p', '-k', str(image)], stdout=log, stderr=subprocess.STDOUT, timeout=120) except (OSError, subprocess.SubprocessError) as error: raise RuntimeError('Pongo bootstrap failed or outcome unknown; inspect USB before any retry. ' f'Log: {log.name}') from error + finally: + if watcher: + watcher.stop() + watcher.join() + if watcher and watcher.resets: + say(f'USB port reset {watcher.resets}x to bring the device back after checkm8') return wait_mode(device, 'pongo', chip=chip, board=board, product=product, seconds=seconds) diff --git a/src/orchard_cli/apple/replug.py b/src/orchard_cli/apple/replug.py new file mode 100644 index 0000000..5216bc7 --- /dev/null +++ b/src/orchard_cli/apple/replug.py @@ -0,0 +1,77 @@ +"""Linux: reset the device's USB port when it does not come back after checkm8. + +After "Checkmate!" the device drops off USB and must re-enumerate in download +mode. Some Linux hosts never see it return until the cable is replugged at the +device. Disabling and re-enabling the hub port forces that re-detection. The +port's disable control is root-only; setup-host's udev rule makes it writable +for the USB access group when a device enters DFU. +""" +import os +from pathlib import Path +import threading + +SYSFS = Path('/sys/bus/usb/devices') +DISCONNECTED = 'DFU mode device disconnected' +RETURNED = ('Entered download mode', 'Download mode device found') + + +def port_control(location, root=SYSFS): + """The writable disable control of the port a device is on, or None.""" + try: + control = (root / location / 'port').resolve(strict=True) / 'disable' + except (OSError, RuntimeError): + return None + return control if control.is_file() and os.access(control, os.W_OK) else None + + +class PortReset(threading.Thread): + """Watch palera1n's log; reset the port if the device does not return.""" + + def __init__(self, log, control, *, settle=2.5, wait=5.0, attempts=3): + super().__init__(daemon=True) + self.log, self.control = Path(log), Path(control) + self.settle, self.wait, self.attempts = settle, wait, attempts + self.resets = 0 + self._stop = threading.Event() + + def stop(self): + self._stop.set() + + def _after_checkmate(self): + try: + text = self.log.read_text(errors='replace') + except OSError: + return None + index = text.find('Checkmate!') + return None if index < 0 else text[index:] + + def _returned(self): + text = self._after_checkmate() + return bool(text) and any(line in text for line in RETURNED) + + def _await(self, seconds, condition): + for _ in range(max(1, int(seconds * 10))): + if condition(): + return True + if self._stop.wait(0.1): + return False + return condition() + + def run(self): + dropped = lambda: DISCONNECTED in (self._after_checkmate() or '') + while not self._stop.is_set() and not dropped(): + self._stop.wait(0.1) + if self._stop.is_set() or self._await(self.settle, self._returned): + return + for _ in range(self.attempts): + if self._stop.is_set(): + return + try: + self.control.write_text('1') + self._stop.wait(0.5) + self.control.write_text('0') + except OSError: + return + self.resets += 1 + if self._await(self.wait, self._returned): + return diff --git a/src/orchard_cli/host_setup.py b/src/orchard_cli/host_setup.py index 40a0b6d..a82416f 100644 --- a/src/orchard_cli/host_setup.py +++ b/src/orchard_cli/host_setup.py @@ -65,6 +65,10 @@ def usb_rules(group): f'MODE="0660", GROUP="{group}", TAG+="uaccess"') lines.append('SUBSYSTEM=="tty", ATTRS{idVendor}=="05ac", ATTRS{idProduct}=="4142", ' f'MODE="0660", GROUP="{group}", TAG+="uaccess"') + # Let Orchard reset the port of a device in DFU when it does not re-enumerate after checkm8. + lines.append('ACTION=="add", SUBSYSTEM=="usb", ENV{DEVTYPE}=="usb_device", ATTR{idVendor}=="05ac", ' + 'ATTR{idProduct}=="1227", RUN+="/bin/sh -c \'chgrp ' + f'{group} /sys%p/port/disable && chmod g+w /sys%p/port/disable\'"') return '\n'.join(lines) + '\n' diff --git a/tests/test_replug.py b/tests/test_replug.py new file mode 100644 index 0000000..1436c02 --- /dev/null +++ b/tests/test_replug.py @@ -0,0 +1,90 @@ +"""USB port reset after checkm8: fake sysfs and palera1n logs; never touches a device.""" +import os +from pathlib import Path +import tempfile +import threading +import time +import unittest + +from orchard_cli.apple import replug + +CHECKMATE = ' - : Checkmate!\n - : Device should now reconnect in download mode\n' \ + ' - : DFU mode device disconnected\n' +RETURNED = ' - : Entered download mode\n' + + +class PortReset(unittest.TestCase): + def setUp(self): + self.temporary = tempfile.TemporaryDirectory() + self.root = Path(self.temporary.name) + port = self.root / 'usb1-port2' + port.mkdir() + self.control = port / 'disable' + self.control.write_text('0') + (self.root / '1-2').mkdir() + (self.root / '1-2' / 'port').symlink_to('../usb1-port2') + self.log = self.root / 'palera1n.log' + self.log.write_text(' - : Waiting for devices\n') + + def tearDown(self): + self.temporary.cleanup() + + def watch(self, **timing): + watcher = replug.PortReset(self.log, self.control, **{'settle': 0.2, 'wait': 0.3, **timing}) + watcher.start() + return watcher + + def test_port_control_requires_a_writable_disable_attribute(self): + self.assertEqual(replug.port_control('1-2', root=self.root), self.control.resolve()) + self.assertIsNone(replug.port_control('1-9', root=self.root)) + self.control.chmod(0o444) + if os.access(self.control, os.W_OK): + self.skipTest('running as root: permissions are not enforced') + self.assertIsNone(replug.port_control('1-2', root=self.root)) + + def test_no_reset_when_the_device_returns_by_itself(self): + self.control.write_text('untouched') + self.log.write_text(CHECKMATE + RETURNED) + watcher = self.watch() + watcher.join(5) + self.assertFalse(watcher.is_alive()) + self.assertEqual(watcher.resets, 0) + self.assertEqual(self.control.read_text(), 'untouched') + + def test_resets_until_the_device_returns_and_leaves_the_port_enabled(self): + self.log.write_text(CHECKMATE) + watcher = self.watch() + + def returns_after_first_reset(): + while watcher.resets < 1: + time.sleep(0.02) + with self.log.open('a') as log: + log.write(RETURNED) + helper = threading.Thread(target=returns_after_first_reset) + helper.start() + watcher.join(5) + helper.join(5) + self.assertEqual(watcher.resets, 1) + self.assertEqual(self.control.read_text(), '0') + + def test_gives_up_after_the_attempts_and_leaves_the_port_enabled(self): + self.log.write_text(CHECKMATE) + watcher = self.watch(attempts=2) + watcher.join(5) + self.assertFalse(watcher.is_alive()) + self.assertEqual(watcher.resets, 2) + self.assertEqual(self.control.read_text(), '0') + + def test_stop_before_checkmate_does_nothing(self): + self.control.write_text('untouched') + watcher = self.watch() + time.sleep(0.2) + watcher.stop() + watcher.join(5) + self.assertFalse(watcher.is_alive()) + self.assertEqual(watcher.resets, 0) + self.assertEqual(self.control.read_text(), 'untouched') + + +if __name__ == '__main__': + unittest.main()