Merge remote-tracking branch 'origin/pr/643'

* origin/pr/643:
  device interface denied list: reformat code
  device interface denied list: make add/remove idempotent
  device interface denied list: special value "all" for removing all items
  device interface denied list: allow hexdigits
  device interface denied list: update no-payload/no-argument tests
  device interface denied list: test renumbering
  device interface denied list: device denied as qube property

Pull request description:

Delete the file and drop-in folder for the denied device interfaces list and move the list to `qvm-prefs`.

Solves QubesOS/qubes-issues/issues/9674

Author: marmarek

Makefile

@@ -92,6 +92,9 @@ ADMIN_API_METHODS_SIMPLE = \
 	admin.vm.device.mic.Detach \
 	admin.vm.device.mic.Set.assignment \
 	admin.vm.device.mic.Unassign \
+	admin.vm.device.denied.List \
+	admin.vm.device.denied.Add \
+	admin.vm.device.denied.Remove \
 	admin.vm.feature.CheckWithNetvm \
 	admin.vm.feature.CheckWithTemplate \
 	admin.vm.feature.CheckWithAdminVM \

qubes/api/admin.py

@@ -50,6 +50,7 @@
     UnknownDevice,
     DeviceAssignment,
     AssignmentMode,
+    DeviceInterface,
 )
 
 
@@ -1630,6 +1631,83 @@ async def vm_device_set_required(self, endpoint, untrusted_payload):
         await self.dest.devices[devclass].update_assignment(dev, mode)
         self.app.save()
 
+    @qubes.api.method(
+        "admin.vm.device.denied.List", no_payload=True, scope="local", read=True
+    )
+    async def vm_device_denied_list(self):
+        """
+        List all denied device interfaces for the VM.
+
+        Returns a newline-separated string.
+        """
+        self.enforce(not self.arg)
+
+        self.fire_event_for_permission()
+
+        denied = self.dest.devices_denied
+        return "\n".join(map(repr, DeviceInterface.from_str_bulk(denied)))
+
+    @qubes.api.method("admin.vm.device.denied.Add", scope="local", write=True)
+    async def vm_device_denied_add(self, untrusted_payload):
+        """
+        Add device interface(s) to the denied list for the VM.
+
+        Payload:
+            Encoded device interface (can be repeated without any separator).
+        """
+        payload = untrusted_payload.decode("ascii", errors="strict")
+        to_add = DeviceInterface.from_str_bulk(payload)
+
+        if len(set(to_add)) != len(to_add):
+            raise qubes.exc.QubesValueError(
+                "Duplicated device interfaces in payload."
+            )
+
+        self.fire_event_for_permission(interfaces=to_add)
+
+        to_add_enc = "".join(map(repr, to_add))
+
+        prev = self.dest.devices_denied
+        # "auto" ignoring of duplications
+        self.dest.devices_denied = self.dest.devices_denied + to_add_enc
+        # do not save if nothing changed
+        if prev != self.dest.devices_denied:
+            self.app.save()
+
+    @qubes.api.method(
+        "admin.vm.device.denied.Remove", scope="local", write=True
+    )
+    async def vm_device_denied_remove(self, untrusted_payload):
+        """
+        Remove device interface(s) from the denied list for the VM.
+
+        Payload:
+            Encoded device interface (can be repeated without any separator).
+            If payload is "all", all interfaces are removed.
+        """
+        denied = DeviceInterface.from_str_bulk(self.dest.devices_denied)
+
+        payload = untrusted_payload.decode("ascii", errors="strict")
+        if payload != "all":
+            to_remove = DeviceInterface.from_str_bulk(payload)
+        else:
+            to_remove = denied.copy()
+
+        if len(set(to_remove)) != len(to_remove):
+            raise qubes.exc.QubesValueError(
+                "Duplicated device interfaces in payload."
+            )
+
+        # may contain missing values
+        self.fire_event_for_permission(interfaces=to_remove)
+
+        # ignore missing values
+        new_denied = "".join(repr(i) for i in denied if i not in to_remove)
+
+        if new_denied != self.dest.devices_denied:
+            self.dest.devices_denied = new_denied
+            self.app.save()
+
     @qubes.api.method(
         "admin.vm.firewall.Get", no_payload=True, scope="local", read=True
     )

qubes/device_protocol.py

@@ -38,7 +38,7 @@
 from typing import TYPE_CHECKING
 
 import qubes.utils
-from qubes.exc import ProtocolError
+from qubes.exc import ProtocolError, QubesValueError
 
 if TYPE_CHECKING:
     from qubes.vm.qubesvm import QubesVM
@@ -693,7 +693,12 @@ def __init__(self, interface_encoding: str, devclass: Optional[str] = None):
                     f"for given {devclass=}",
                     file=sys.stderr,
                 )
-            ifc_full = devclass[0] + ifc_padded
+            if not all(c in string.hexdigits + "*" for c in ifc_padded):
+                raise ProtocolError("Invalid characters in interface encoding")
+            devclass_code = devclass[0].lower()
+            if devclass_code not in string.ascii_lowercase:
+                raise ProtocolError("Invalid characters in devclass encoding")
+            ifc_full = devclass_code + ifc_padded
         else:
             known_devclasses = {
                 "p": "pci",
@@ -733,6 +738,19 @@ def unknown(cls) -> "DeviceInterface":
         """Value for unknown device interface."""
         return cls("?******")
 
+    @staticmethod
+    def from_str_bulk(interfaces: Optional[str]) -> List["DeviceInterface"]:
+        interfaces = interfaces or ""
+        if len(interfaces) % 7 != 0:
+            raise QubesValueError(
+                f"Invalid length of {interfaces=} "
+                f"(is {len(interfaces)}, expected multiple of 7)",
+            )
+        return [
+            DeviceInterface(interfaces[i : i + 7])
+            for i in range(0, len(interfaces), 7)
+        ]
+
     def __repr__(self):
         return self._interface_encoding
 
@@ -1120,11 +1138,7 @@ def _deserialize(
 
         if "interfaces" in properties:
             interfaces = properties["interfaces"]
-            interfaces = [
-                DeviceInterface(interfaces[i : i + 7])
-                for i in range(0, len(interfaces), 7)
-            ]
-            properties["interfaces"] = interfaces
+            properties["interfaces"] = DeviceInterface.from_str_bulk(interfaces)
 
         if "parent_ident" in properties:
             properties["parent"] = Port(

qubes/devices.py

@@ -73,8 +73,6 @@
 )
 from qubes.exc import ProtocolError
 
-DEVICE_DENY_LIST = "/etc/qubes/device-deny.list"
-
 
 class DeviceNotAssigned(qubes.exc.QubesException, KeyError):
     """

qubes/ext/admin.py

@@ -17,7 +17,6 @@
 # You should have received a copy of the GNU Lesser General Public
 # License along with this library; if not, see <https://www.gnu.org/licenses/>.
 import importlib
-import os
 
 import qubes.api
 import qubes.api.internal
@@ -27,7 +26,6 @@
 from qrexec.policy import utils, parser
 
 from qubes.device_protocol import DeviceInterface
-from qubes.devices import DEVICE_DENY_LIST
 
 
 class JustEvaluateAskResolution(parser.AskResolution):
@@ -184,44 +182,11 @@ def on_device_attach(self, vm, event, dest, arg, device, mode, **kwargs):
         if mode != "manual":
             return
 
-        # load device deny list
-        deny = {}
-        AdminExtension._load_deny_list(deny, DEVICE_DENY_LIST)
-
-        # load drop ins
-        drop_in_path = DEVICE_DENY_LIST + ".d"
-        if os.path.isdir(drop_in_path):
-            for deny_list_name in os.listdir(drop_in_path):
-                deny_list_path = os.path.join(drop_in_path, deny_list_name)
-
-                if os.path.isfile(deny_list_path):
-                    AdminExtension._load_deny_list(deny, deny_list_path)
-
-        # check if any presented interface is on deny list
-        for interface in deny.get(dest.name, set()):
-            pattern = DeviceInterface(interface)
+        # check if any presented interface is on denied list
+        denied = set(DeviceInterface.from_str_bulk(dest.devices_denied))
+        for pattern in denied:
             for devint in device.interfaces:
                 if pattern.matches(devint):
                     raise qubes.exc.PermissionDenied(
                         f"Device exposes a banned interface: {devint}"
                     )
-
-    @staticmethod
-    def _load_deny_list(deny: dict, path: str) -> None:
-        try:
-            with open(path, "r", encoding="utf-8") as file:
-                for line in file:
-                    line = line.strip()
-
-                    # skip comments
-                    if line.startswith("#"):
-                        continue
-
-                    if line:
-                        name, *values = line.split()
-                        values = " ".join(values).replace(",", " ").split()
-                        values = [v for v in values if len(v) > 0]
-
-                        deny[name] = deny.get(name, set()).union(set(values))
-        except IOError:
-            pass