From a71528c4856ecef4e685a7fa3ed889ee9c43629a Mon Sep 17 00:00:00 2001 From: firestarsdog <229254897+firestarsdog@users.noreply.github.com> Date: Sun, 4 Oct 2026 22:56:08 -0400 Subject: [PATCH] Batch AGNOS GPIO Permission Rules --- launch_chffrplus.sh | 5 ++ .../updated/tests/test_retained_agnos.py | 71 ++++++++++++++++++- 2 files changed, 74 insertions(+), 2 deletions(-) diff --git a/launch_chffrplus.sh b/launch_chffrplus.sh index 0407572418..be72c6491d 100755 --- a/launch_chffrplus.sh +++ b/launch_chffrplus.sh @@ -42,6 +42,11 @@ function agnos_init { SUBSYSTEM=="usb", ATTRS{idVendor}=="3801", ATTRS{idProduct}=="ddcc", MODE="0666" SUBSYSTEM=="usb", ATTRS{idVendor}=="3801", ATTRS{idProduct}=="ddee", MODE="0666" EOF + # Batch AGNOS GPIO permissions so coldplug does not block Panda USB events. + if [ -f /etc/udev/rules.d/99-gpio.rules ] && ! mountpoint -q /etc/udev/rules.d/99-gpio.rules; then + sed "s/{} \\\\;/{} +/g" /etc/udev/rules.d/99-gpio.rules > /run/udev/rules.d/99-gpio.rules + mount --bind /run/udev/rules.d/99-gpio.rules /etc/udev/rules.d/99-gpio.rules + fi udevadm control --reload-rules for product in ddcc ddee; do udevadm trigger --subsystem-match=usb --attr-match=idVendor=3801 --attr-match=idProduct="$product" diff --git a/openpilot/system/updated/tests/test_retained_agnos.py b/openpilot/system/updated/tests/test_retained_agnos.py index 16a61384fa..e4dfa2c8ce 100644 --- a/openpilot/system/updated/tests/test_retained_agnos.py +++ b/openpilot/system/updated/tests/test_retained_agnos.py @@ -2,6 +2,8 @@ import os import re from pathlib import Path +import shlex +import shutil import subprocess import sys import tempfile @@ -92,12 +94,28 @@ if [ "$UDEV_TEST_FAILURE" = hang ]; then sleep 60; fi if [ "$UDEV_TEST_FAILURE" = "$1" ]; then exit 1; fi ''') udevadm.chmod(0o755) + gpio_rules = self.root / 'stock-gpio.rules' + if not gpio_rules.exists(): + gpio_rules.write_text('SUBSYSTEM=="gpio*", PROGRAM="/bin/sh -c \'find -L /sys/class/gpio/ -maxdepth 2 ' + + '-exec chown root:gpio {} \\; -exec chmod 770 {} \\; || true\'"\n' + + 'SUBSYSTEM=="gpio", KERNEL=="gpiochip[0]", GROUP="gpio", MODE="660"\n') + for name, body in { + 'mountpoint': '[ -f "$UDEV_TEST_MOUNTED" ]', + 'mount': '''[ "$UDEV_TEST_FAILURE" != mount ] || exit 1 +echo "$*" >> "$UDEV_TEST_MOUNT_LOG" +cp "$2" "$3" +touch "$UDEV_TEST_MOUNTED"''', + }.items(): + command = self.root / name + command.write_text('#!/bin/sh\n' + body + '\n') + command.chmod(0o755) script = self.root / 'test.sh' # Replace only device observations in a copy of the actual function. Every # effectful command is a test double; no launch manager or device is run. function = function.replace('/VERSION', str(version)).replace('/AGNOS', str(marker)) function = function.replace('/sys/firmware/devicetree/base/model', str(device_model)) function = function.replace('/run/udev/rules.d', str(self.root / 'udev-rules')) + function = function.replace('/etc/udev/rules.d/99-gpio.rules', str(gpio_rules)) function = function.replace('timeout -k 1s 5s', 'timeout -k 0.1s 0.2s') script.write_text('''source "$1/launch_env.sh" DIR="$1" @@ -110,7 +128,8 @@ read() { echo prompt >> "$log"; return 1; } op_run_command() { echo effect >> "$log"; } ''' + function + '\n' + tail + '\n') environment = dict(os.environ, PATH=f'{self.root}:{os.environ["PATH"]}', - UDEV_TEST_LOG=str(self.root / 'udev.log'), UDEV_TEST_FAILURE=udev_failure) + UDEV_TEST_LOG=str(self.root / 'udev.log'), UDEV_TEST_FAILURE=udev_failure, + UDEV_TEST_MOUNTED=str(self.root / 'mounted'), UDEV_TEST_MOUNT_LOG=str(self.root / 'mount.log')) return subprocess.run(['bash', str(script), str(self.root)], env=environment, capture_output=True, text=True, timeout=5) @@ -129,10 +148,16 @@ op_run_command() { echo effect >> "$log"; } if model != 'comma tici': self.assertFalse((self.root / 'udev-rules').exists()) self.assertFalse((self.root / 'udev.log').exists()) + self.assertFalse((self.root / 'mount.log').exists()) rules = self.root / 'udev-rules/99-starpilot-panda.rules' expected = ('SUBSYSTEM=="usb", ATTRS{idVendor}=="3801", ATTRS{idProduct}=="ddcc", MODE="0666"\n' + 'SUBSYSTEM=="usb", ATTRS{idVendor}=="3801", ATTRS{idProduct}=="ddee", MODE="0666"\n') self.assertEqual(rules.read_text(), expected) + gpio_rules = self.root / 'stock-gpio.rules' + self.assertEqual(gpio_rules.read_text().count('{} +'), 2) + self.assertNotIn('\\;', gpio_rules.read_text()) + self.assertIn('KERNEL=="gpiochip[0]", GROUP="gpio", MODE="660"', gpio_rules.read_text()) + self.assertEqual(len((self.root / 'mount.log').read_text().splitlines()), 1) commands = ['control --reload-rules', *[ f'trigger --subsystem-match=usb --attr-match=idVendor=3801 --attr-match=idProduct={product}' for product in ('ddcc', 'ddee')]] @@ -141,10 +166,11 @@ op_run_command() { echo effect >> "$log"; } self.assertEqual(result.returncode, 0, result.stderr) self.assertEqual(rules.read_text(), expected) self.assertEqual((self.root / 'udev.log').read_text().splitlines(), commands * 2) + self.assertEqual(len((self.root / 'mount.log').read_text().splitlines()), 1) def test_panda_usb_permission_failure_never_blocks_manager_start(self): function = shell_function(ROOT / 'launch_chffrplus.sh', 'agnos_init') - for failure in ('control', 'trigger', 'hang'): + for failure in ('mount', 'control', 'trigger', 'hang'): with self.subTest(failure=failure): result = self.run_shell(function, 'agnos_init && echo manager-start', '19.8.2', model='comma tici', udev_failure=failure) @@ -161,6 +187,47 @@ op_run_command() { echo effect >> "$log"; } self.assertIn('Panda USB permission setup failed; continuing startup.', result.stdout) self.assertIn('manager-start', result.stdout) + def test_gpio_permissions_batch_commands_without_changing_targets(self): + function = shell_function(ROOT / 'launch_chffrplus.sh', 'agnos_init') + self.run_shell(function, 'agnos_init', '19.8.2', model='comma tici') + rule = (self.root / 'stock-gpio.rules').read_text().splitlines()[0] + command = shlex.split(re.search(r'PROGRAM="(.*)"', rule)[1])[2] + self.assertEqual(command.count('{} +'), 2) + + gpio = self.root / 'gpio' + gpio.mkdir() + pin = self.root / 'pin' + pin.mkdir() + for name in ('direction', 'value', 'edge', 'uevent'): + (pin / name).touch() + (pin / 'subsystem').symlink_to(gpio, target_is_directory=True) + (gpio / 'gpio134').symlink_to(pin, target_is_directory=True) + (gpio / 'export').touch() + command = command.replace('/sys/class/gpio/', str(gpio) + '/').replace('root:gpio', f'{os.getuid()}:{os.getgid()}') + commands = self.root / 'permission-commands' + commands.mkdir() + log = self.root / 'permission.log' + for name in ('chown', 'chmod'): + wrapper = commands / name + wrapper.write_text(f'#!/bin/sh\necho {name} >> "$GPIO_TEST_LOG"\nexec {shlex.quote(shutil.which(name))} "$@"\n') + wrapper.chmod(0o755) + environment = dict(os.environ, PATH=f'{commands}:{os.environ["PATH"]}', GPIO_TEST_LOG=str(log)) + paths = [gpio, gpio / 'export', pin, *(pin / name for name in ('direction', 'value', 'edge', 'uevent'))] + results = [] + for program in (command.replace('{} +', r'{} \;'), command): + for path in paths: + path.chmod(0o755 if path.is_dir() else 0o644) + log.write_text('') + result = subprocess.run(['sh', '-c', program], env=environment, capture_output=True, text=True, timeout=5) + self.assertEqual(result.returncode, 0, result.stderr) + results.append([(path.stat().st_mode, path.stat().st_uid, path.stat().st_gid) for path in paths]) + if program == command: + self.assertEqual(log.read_text().splitlines(), ['chown', 'chmod']) + else: + self.assertGreater(len(log.read_text().splitlines()), 2) + self.assertEqual(results[0], results[1]) + self.assertTrue(all(path.stat().st_mode & 0o777 == 0o770 for path in paths)) + def test_developer_helper_retained_match_or_mismatch_never_prompts(self): function = shell_function(ROOT / 'tools/op.sh', 'op_check_agnos_update') for current, code in (('19.8.2', 0), ('19.8', 1)):