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()