Require approval for new Thunderbolt accessories

test/shell.d/fixtures/thunderbolt/bolt-integration.py

@@ -0,0 +1,185 @@
+"""Only the hardware fixture is Python: upstream Bolt uses UMockdev/GObject.
+
+All Omarchy behavior and assertions run in bolt-exercise.sh against real boltd,
+real busctl, a private system bus, and disposable sysfs/firmware/store state.
+"""
+import importlib.machinery
+import importlib.util
+import os
+from pathlib import Path
+import subprocess
+import sys
+import time
+import unittest
+
+root = Path(sys.argv.pop(1))
+loader = importlib.machinery.SourceFileLoader('bolt_upstream', str(Path(os.environ['BOLT_TEST_SOURCE']) / 'tests/test-integration'))
+spec = importlib.util.spec_from_loader(loader.name, loader)
+u = importlib.util.module_from_spec(spec)
+loader.exec_module(u)
+
+
+class ApprovalIntegration(u.BoltTest):
+  def exercise(self, security, numeric=False):
+    self.user_config(AuthMode='disabled', DefaultPolicy='manual')
+    _, host = self.add_domain_host(security=security, iommu='1')
+    path, uid = self.add_device(host, 1, 'Dock', 'Example', authorized=0, key='' if security == 'secure' else None, boot='0')
+    _, other = self.add_device(host, 2, 'Unknown', 'Example', authorized=0, key=None, boot='0')
+    if numeric:
+      # Linux omits these files when the peripheral has no descriptive names.
+      for attribute in ('device_name', 'vendor_name'):
+        Path(self.testbed.get_root_dir(), path.lstrip('/'), attribute).unlink()
+    self.daemon_start()
+    self.polkitd_start()
+    self.polkitd.SetAllowed(['org.freedesktop.bolt.authorize', 'org.freedesktop.bolt.enroll', 'org.freedesktop.bolt.manage'])
+    env = os.environ | {'ROOT': str(root), 'UMOCKDEV_DIR': self.testbed.get_root_dir(),
+                        'TB_TEST_DIR': self.rundir, 'TB_TEST_UID': uid, 'TB_TEST_OTHER': other,
+                        'TB_TEST_KEYS': str(Path(self.dbpath, 'keys')), 'TB_TEST_SECURITY': security}
+    command = ['bash', str(root / 'test/shell.d/fixtures/thunderbolt/bolt-exercise.sh')]
+    subprocess.run(command + ['approve'], env=env, check=True)
+    self.daemon_stop()
+    self.testbed.set_attribute(path, 'authorized', '0')
+    if security == 'secure':
+      self.testbed.set_attribute(path, 'key', '\n')
+    self.daemon_start()
+    subprocess.run(command + ['restart'], env=env, check=True)
+    # The real Bash daemon publishes a fresh readable snapshot and reconnects
+    # to Bolt after a restart. No custom Omarchy D-Bus server is involved.
+    child = subprocess.Popen(command + ['daemon'], env=env)
+    try:
+      for _ in range(100):
+        result = subprocess.run(command + ['status'], env=env, capture_output=True)
+        if result.returncode == 0:
+          break
+        time.sleep(0.1)
+      else:
+        raise RuntimeError('Bash daemon did not publish a valid snapshot')
+      subprocess.run(command + ['once'], env=env, check=True)
+      self.daemon_stop()
+      self.daemon_start()
+      for _ in range(100):
+        if subprocess.run(command + ['status'], env=env, capture_output=True).returncode == 0:
+          break
+        time.sleep(0.1)
+      else:
+        raise RuntimeError('Bash daemon did not publish the new Bolt generation')
+    finally:
+      child.terminate()
+      child.wait(timeout=15)
+    self.daemon_stop()
+
+  def test_user_approval(self):
+    self.exercise('user')
+
+  def test_secure_approval(self):
+    self.exercise('secure')
+
+  def test_numeric_identity(self):
+    self.exercise('user', numeric=True)
+
+  def removal(self, security, boot=False, disconnected=False):
+    self.user_config(AuthMode='disabled')
+    _, host = self.add_domain_host(security=security, iommu='1')
+    path, uid = self.add_device(host, 1, 'Dock', 'Example', authorized=0, key='' if security == 'secure' else None, boot='0')
+    other_path, other = self.add_device(host, 2, 'Existing', 'Example', authorized=0, key=None, boot='0')
+    self.daemon_start()
+    self.polkitd_start()
+    self.polkitd.SetAllowed(['org.freedesktop.bolt.authorize', 'org.freedesktop.bolt.enroll', 'org.freedesktop.bolt.manage'])
+    env = os.environ | {'ROOT': str(root), 'UMOCKDEV_DIR': self.testbed.get_root_dir(),
+                        'TB_TEST_DIR': self.rundir, 'TB_TEST_UID': uid, 'TB_TEST_OTHER': other,
+                        'TB_TEST_STORE': self.dbpath, 'TB_TEST_SECURITY': security}
+    command = ['bash', str(root / 'test/shell.d/fixtures/thunderbolt/bolt-removal.sh')]
+    service = 'omarchy-thunderbolt-authorization.service'
+    child = None
+
+    def operation(action, unit):
+      nonlocal child
+      if unit == 'bolt.service':
+        if action == 'stop':
+          self.daemon_stop()
+        elif self.daemon is None:
+          self.daemon_start()
+      elif unit == service:
+        if action == 'stop' and child is not None:
+          child.terminate()
+          child.wait(timeout=15)
+          child = None
+        elif action == 'start' and child is None:
+          child = subprocess.Popen(['bash', str(root / 'test/shell.d/fixtures/thunderbolt/bolt-exercise.sh'), 'daemon'], env=env)
+      else:
+        raise RuntimeError('Unexpected private service: ' + unit)
+      marker = Path(self.rundir, unit + '.active')
+      if action == 'start':
+        marker.touch()
+      else:
+        marker.unlink(missing_ok=True)
+
+    def run(action):
+      process = subprocess.Popen(command + [action], env=env)
+      control = Path(self.rundir, 'control')
+      deadline = time.monotonic() + 90
+      try:
+        while process.poll() is None:
+          if time.monotonic() > deadline:
+            raise RuntimeError('Private setup timed out: ' + action)
+          if control.exists():
+            operation(*control.read_text().strip().split())
+            control.unlink()
+          time.sleep(0.05)
+        if process.returncode:
+          raise subprocess.CalledProcessError(process.returncode, command + [action])
+      finally:
+        if process.poll() is None:
+          process.terminate()
+          process.wait(timeout=15)
+
+    def wait_for(action):
+      for _ in range(50):
+        result = subprocess.run(command + [action], env=env, capture_output=True, text=True)
+        if result.returncode == 0:
+          print(result.stdout, end='', flush=True)
+          return
+        time.sleep(0.1)
+      raise RuntimeError(action + ': ' + result.stdout + result.stderr)
+
+    def unplug():
+      for device_path in (path, other_path):
+        self.testbed.uevent(device_path, 'remove')
+        self.testbed.remove_device(device_path)
+      wait_for('disconnected')
+
+    try:
+      Path(self.rundir, 'bolt.service.active').touch()
+      run('approve')
+      operation('start', service)
+      Path(self.rundir, service + '.enabled').touch()
+      if boot:
+        run('boot-enable')
+      run('rollback')
+      if disconnected:
+        unplug()
+      run('disable')
+      if not disconnected:
+        unplug()
+      for number, device_uid, name in ((1, uid, 'Dock'), (2, other, 'Existing')):
+        attributes = ['device_name', name, 'device', '0x23', 'vendor_name', 'Example',
+                      'vendor', '0x23', 'authorized', '0', 'unique_id', device_uid, 'boot', '0']
+        if number == 1 and security == 'secure':
+          attributes += ['key', '\n']
+        self.testbed.add_device('thunderbolt', '0-' + str(number), host, attributes, ['DEVTYPE', 'thunderbolt_device'])
+      wait_for('reconnected')
+    finally:
+      operation('stop', service)
+      self.daemon_stop()
+
+  def test_user_removal(self):
+    self.removal('user')
+
+  def test_secure_removal(self):
+    self.removal('secure', boot=True, disconnected=True)
+
+
+names = sys.argv[1:] or ('test_user_approval', 'test_secure_approval', 'test_numeric_identity', 'test_user_removal', 'test_secure_removal')
+suite = unittest.TestSuite(ApprovalIntegration(name) for name in names)
+result = unittest.TextTestRunner(verbosity=2).run(suite)
+sys.exit(not result.wasSuccessful())