Reset the USB port when the device does not return after checkm8

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)
 
 

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

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'
 
 

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 = ' - <Info>: Checkmate!\n - <Verbose>: Device should now reconnect in download mode\n' \
+            ' - <Verbose>: DFU mode device disconnected\n'
+RETURNED = ' - <Info>: 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(' - <Info>: 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()