Batch AGNOS GPIO Permission Rules

This commit is contained in:
firestarsdog
2026-10-04 22:56:08 -04:00
parent f6a01d2126
commit a71528c485
2 changed files with 74 additions and 2 deletions
+5
View File
@@ -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"
@@ -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)):