186 lines
7.6 KiB
Python
186 lines
7.6 KiB
Python
"""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())
|