Allow dynamic lookup cgroup rules based on Host udev (#2652)

supervisor/hardware/const.py

@@ -23,8 +23,11 @@ class UdevSubsystem(str, Enum):
     GPIOMEM = "gpiomem"
     VCHIQ = "vchiq"
     GRAPHICS = "graphics"
-    CEC = "CEC"
+    CEC = "cec"
     DRM = "drm"
+    HIDRAW = "hidraw"
+    RPI_HEVCMEM = "rpivid-hevcmem"
+    RPI_H264MEM = "rpivid-h264mem"
 
 
 class PolicyGroup(str, Enum):

supervisor/hardware/policy.py

@@ -16,19 +16,13 @@ _CGROUPS: Dict[PolicyGroup, List[int]] = {
     PolicyGroup.UART: [
         204,  # ttyAMA / ttySAC (tty)
         188,  # ttyUSB (tty)
-        166,  # ttyACM (tty)
-        244   # ttyAML (tty)
+        166   # ttyACM (tty)
     ],
     PolicyGroup.GPIO: [
-        254,  # gpiochip (gpio)
-        245   # gpiomem (gpiomem)
     ],
     PolicyGroup.VIDEO: [
-        239,
-        29,
+        29,   # /dev/fb (graphics)
         81,
-        251,
-        242,  # vchiq (vchiq)
         226
     ],
     PolicyGroup.AUDIO: [
@@ -37,13 +31,35 @@ _CGROUPS: Dict[PolicyGroup, List[int]] = {
     PolicyGroup.USB: [
         189,  # /dev/bus/usb (usb)
         180,  # hiddev (usbmisc)
-        243   # hidraw (hidraw)
     ],
     PolicyGroup.BLUETOOTH: [
         13    # /dev/input (input)
     ]
 }
 
+_CGROUPS_DYNAMIC_MAJOR: Dict[PolicyGroup, List[UdevSubsystem]] = {
+    PolicyGroup.USB: [
+        UdevSubsystem.HIDRAW
+    ],
+    PolicyGroup.GPIO: [
+        UdevSubsystem.GPIO,
+        UdevSubsystem.GPIOMEM
+    ],
+    PolicyGroup.VIDEO: [
+        UdevSubsystem.VCHIQ,
+        UdevSubsystem.MEDIA,
+        UdevSubsystem.CEC,
+        UdevSubsystem.RPI_H264MEM,
+        UdevSubsystem.RPI_HEVCMEM
+    ]
+}
+
+_CGROUPS_DYNAMIC_MINOR: Dict[PolicyGroup, List[UdevSubsystem]] = {
+    PolicyGroup.UART: [
+        UdevSubsystem.SERIAL
+    ]
+}
+
 # fmt: on
 
 
@@ -60,12 +76,33 @@ class HwPolicy(CoreSysAttributes):
 
     def get_cgroups_rules(self, group: PolicyGroup) -> List[str]:
         """Generate cgroups rules for a policy group."""
-        return [f"c {dev}:* rwm" for dev in _CGROUPS.get(group, [])]
+        cgroups: List[str] = [f"c {dev}:* rwm" for dev in _CGROUPS[group]]
+
+        # Lookup dynamic device groups from host
+        if group in _CGROUPS_DYNAMIC_MAJOR:
+            majors = {
+                device.cgroups_major
+                for device in self.sys_hardware.devices
+                if device.subsystem in _CGROUPS_DYNAMIC_MAJOR[group]
+                and device.cgroups_major not in _CGROUPS[group]
+            }
+            cgroups.extend([f"c {dev}:* rwm" for dev in majors])
+
+        # Lookup dynamic devices from host
+        if group in _CGROUPS_DYNAMIC_MINOR:
+            for device in self.sys_hardware.devices:
+                if (
+                    device.subsystem not in _CGROUPS_DYNAMIC_MINOR[group]
+                    or device.cgroups_major in _CGROUPS[group]
+                ):
+                    continue
+                cgroups.append(self.get_cgroups_rule(device))
+
+        return cgroups
 
     def get_cgroups_rule(self, device: Device) -> str:
         """Generate a cgroups rule for given device."""
         cgroup_type = "c" if device.subsystem != UdevSubsystem.DISK else "b"
-
         return f"{cgroup_type} {device.cgroups_major}:{device.cgroups_minor} rwm"
 
     def get_full_access(self) -> str:

tests/hardware/test_policy.py

@@ -35,11 +35,8 @@ def test_device_policy(coresys):
 def test_policy_group(coresys):
     """Test policy group generator."""
     assert coresys.hardware.policy.get_cgroups_rules(PolicyGroup.VIDEO) == [
-        "c 239:* rwm",
         "c 29:* rwm",
         "c 81:* rwm",
-        "c 251:* rwm",
-        "c 242:* rwm",
         "c 226:* rwm",
     ]
 
@@ -83,3 +80,93 @@ def test_allowed_access(coresys):
     )
 
     assert coresys.hardware.policy.allowed_for_access(device)
+
+
+def test_dynamic_group_alloc_minor(coresys):
+    """Test dynamic cgroup generation based on minor."""
+    for device in (
+        Device(
+            "ttyACM0",
+            Path("/dev/ttyACM0"),
+            Path("/sys/bus/usb/001"),
+            "tty",
+            [],
+            {"MAJOR": "204", "MINOR": "10"},
+        ),
+        Device(
+            "ttyUSB0",
+            Path("/dev/ttyUSB0"),
+            Path("/sys/bus/usb/000"),
+            "tty",
+            [Path("/dev/ttyS1"), Path("/dev/serial/by-id/xyx")],
+            {"MAJOR": "188", "MINOR": "10"},
+        ),
+        Device(
+            "ttyS0",
+            Path("/dev/ttyS0"),
+            Path("/sys/bus/usb/002"),
+            "tty",
+            [],
+            {"MAJOR": "4", "MINOR": "65"},
+        ),
+        Device(
+            "video1",
+            Path("/dev/video1"),
+            Path("/sys/bus/usb/003"),
+            "misc",
+            [],
+            {"MAJOR": "38", "MINOR": "10"},
+        ),
+    ):
+        coresys.hardware.update_device(device)
+
+    assert coresys.hardware.policy.get_cgroups_rules(PolicyGroup.UART) == [
+        "c 204:* rwm",
+        "c 188:* rwm",
+        "c 166:* rwm",
+        "c 4:65 rwm",
+    ]
+
+
+def test_dynamic_group_alloc_major(coresys):
+    """Test dynamic cgroup generation based on minor."""
+    for device in (
+        Device(
+            "gpio16",
+            Path("/dev/gpio16"),
+            Path("/sys/bus/usb/001"),
+            "gpio",
+            [],
+            {"MAJOR": "254", "MINOR": "10"},
+        ),
+        Device(
+            "gpiomem",
+            Path("/dev/gpiomem"),
+            Path("/sys/bus/usb/000"),
+            "gpiomem",
+            [Path("/dev/ttyS1"), Path("/dev/serial/by-id/xyx")],
+            {"MAJOR": "239", "MINOR": "10"},
+        ),
+        Device(
+            "ttyS0",
+            Path("/dev/ttyS0"),
+            Path("/sys/bus/usb/002"),
+            "tty",
+            [],
+            {"MAJOR": "4", "MINOR": "65"},
+        ),
+        Device(
+            "video1",
+            Path("/dev/video1"),
+            Path("/sys/bus/usb/003"),
+            "misc",
+            [],
+            {"MAJOR": "38", "MINOR": "10"},
+        ),
+    ):
+        coresys.hardware.update_device(device)
+
+    assert coresys.hardware.policy.get_cgroups_rules(PolicyGroup.GPIO) == [
+        "c 254:* rwm",
+        "c 239:* rwm",
+    ]