q-dev: refactor device_protocol.py

doc/qubes-devices.rst

@@ -6,8 +6,8 @@ devices, which can be attached to other domains (frontend). Devices can be of
 different buses (like 'pci', 'usb', etc.). Each device bus is implemented by
 an extension (see :py:mod:`qubes.ext`).
 
-Devices are identified by pair of (backend domain, `ident`), where `ident` is
-:py:class:`str` and can contain only characters from `[a-zA-Z0-9._-]` set.
+Devices are identified by pair of (backend domain, `port_id`), where `port_id`
+is :py:class:`str` and can contain only characters from `[a-zA-Z0-9._-]` set.
 
 
 Device Assignment vs Attachment
@@ -106,11 +106,11 @@ is connected. Therefore, when assigning a device to a VM, such as
 `sys-usb:1-1.1`, the port `1-1.1` is actually assigned, and thus
 *every* devices connected to it will be automatically attached.
 Similarly, when assigning `vm:sda`, every block device with the name `sda`
-will be automatically attached. We can limit this using :py:meth:`qubes.device_protocol.DeviceInfo.self_identity`, which returns a string containing information
+will be automatically attached. We can limit this using :py:meth:`qubes.device_protocol.DeviceInfo.device_id`, which returns a string containing information
 presented by the device, such as, `vendor_id`, `product_id`, `serial_number`,
-and encoded interfaces. In the case of block devices, `self_identity`
+and encoded interfaces. In the case of block devices, `device_id`
 consists of the parent port to which the device is connected (if any),
-the parent's `self_identity`, and the interface/partition number.
+the parent's `device_id`, and the interface/partition number.
 In practice, this means that, a partition on a USB drive will only be
 automatically attached to a frontend domain if the parent presents
 the correct serial number etc., and is connected to a specific port.

qubes/api/admin.py

@@ -45,7 +45,7 @@
 import qubes.vm
 import qubes.vm.adminvm
 import qubes.vm.qubesvm
-from qubes.device_protocol import Port
+from qubes.device_protocol import Port, Device, DeviceInfo
 
 
 class QubesMgmtEventsDispatcher:
@@ -1218,15 +1218,15 @@ async def vm_device_available(self, endpoint):
                 raise qubes.exc.QubesException("qubesd shutdown in progress")
             raise
         if self.arg:
-            devices = [dev for dev in devices if dev.ident == self.arg]
+            devices = [dev for dev in devices if dev.port_id == self.arg]
             # no duplicated devices, but device may not exist, in which case
             #  the list is empty
             self.enforce(len(devices) <= 1)
         devices = self.fire_event_for_filter(devices, devclass=devclass)
-        dev_info = {f'{dev.ident}:{dev.self_identity}':
+        dev_info = {f'{dev.port_id}:{dev.device_id}':
                         dev.serialize().decode() for dev in devices}
-        return ''.join('{} {}\n'.format(ident, dev_info[ident])
-                       for ident in sorted(dev_info))
+        return ''.join('{} {}\n'.format(port_id, dev_info[port_id])
+                       for port_id in sorted(dev_info))
 
     @qubes.api.method('admin.vm.device.{endpoint}.Assigned', endpoints=(ep.name
             for ep in importlib.metadata.entry_points(group='qubes.devices')),
@@ -1245,7 +1245,7 @@ async def vm_device_list(self, endpoint):
         if self.arg:
             select_backend, select_ident = self.arg.split('+', 1)
             device_assignments = [dev for dev in device_assignments
-                if (str(dev.backend_domain), dev.ident)
+                if (str(dev.backend_domain), dev.port_id)
                    == (select_backend, select_ident)]
             # no duplicated devices, but device may not exist, in which case
             #  the list is empty
@@ -1255,12 +1255,12 @@ async def vm_device_list(self, endpoint):
 
         dev_info = {
             (f'{assignment.backend_domain}'
-             f'+{assignment.ident}:{assignment.device_identity}'):
+             f'+{assignment.port_id}:{assignment.device_id}'):
                 assignment.serialize().decode('ascii', errors="ignore")
             for assignment in device_assignments}
 
-        return ''.join('{} {}\n'.format(ident, dev_info[ident])
-            for ident in sorted(dev_info))
+        return ''.join('{} {}\n'.format(port_id, dev_info[port_id])
+            for port_id in sorted(dev_info))
 
     @qubes.api.method(
         'admin.vm.device.{endpoint}.Attached',
@@ -1281,7 +1281,7 @@ async def vm_device_attached(self, endpoint):
         if self.arg:
             select_backend, select_ident = self.arg.split('+', 1)
             device_assignments = [dev for dev in device_assignments
-                                  if (str(dev.backend_domain), dev.ident)
+                                  if (str(dev.backend_domain), dev.port_id)
                                   == (select_backend, select_ident)]
             # no duplicated devices, but device may not exist, in which case
             #  the list is empty
@@ -1291,12 +1291,12 @@ async def vm_device_attached(self, endpoint):
 
         dev_info = {
             (f'{assignment.backend_domain}'
-             f'+{assignment.ident}:{assignment.device_identity}'):
+             f'+{assignment.port_id}:{assignment.device_id}'):
                 assignment.serialize().decode('ascii', errors="ignore")
             for assignment in device_assignments}
 
-        return ''.join('{} {}\n'.format(ident, dev_info[ident])
-                       for ident in sorted(dev_info))
+        return ''.join('{} {}\n'.format(port_id, dev_info[port_id])
+                       for port_id in sorted(dev_info))
 
     # Assign/Unassign action can modify only persistent state of running VM.
     # For this reason, write=True
@@ -1305,14 +1305,10 @@ async def vm_device_attached(self, endpoint):
         scope='local', write=True)
     async def vm_device_assign(self, endpoint, untrusted_payload):
         devclass = endpoint
-
-        # qrexec already verified that no strange characters are in self.arg
-        backend_domain, ident = self.arg.split('+', 1)
-        # may raise KeyError, either on domain or ident
-        dev = self.app.domains[backend_domain].devices[devclass][ident]
+        dev = self.load_device_info(devclass)
 
         assignment = qubes.device_protocol.DeviceAssignment.deserialize(
-            untrusted_payload, expected_port=dev
+            untrusted_payload, expected_device=dev
         )
 
         self.fire_event_for_permission(
@@ -1325,6 +1321,13 @@ async def vm_device_assign(self, endpoint, untrusted_payload):
         await self.dest.devices[devclass].assign(assignment)
         self.app.save()
 
+    def load_device_info(self, devclass) -> DeviceInfo:
+        # qrexec already verified that no strange characters are in self.arg
+        _dev = Device.from_qarg(self.arg, devclass, self.app.domains)
+        # load all info, may raise KeyError, either on domain or port_id
+        return self.app.domains[
+            _dev.backend_domain].devices[devclass][_dev.port_id]
+
     # Assign/Unassign action can modify only persistent state of running VM.
     # For this reason, write=True
     @qubes.api.method(
@@ -1335,18 +1338,11 @@ async def vm_device_assign(self, endpoint, untrusted_payload):
             no_payload=True, scope='local', write=True)
     async def vm_device_unassign(self, endpoint):
         devclass = endpoint
-
-        # qrexec already verified that no strange characters are in self.arg
-        backend_domain, ident = self.arg.split('+', 1)
-        # may raise KeyError; if a device isn't found, it will be UnknownDevice
-        #  instance - but allow it, otherwise it will be impossible to unassign
-        #  an already removed device
-        dev = self.app.domains[backend_domain].devices[devclass][ident]
+        dev = self.load_device_info(devclass)
 
         self.fire_event_for_permission(device=dev, devclass=devclass)
 
-        assignment = qubes.device_protocol.DeviceAssignment(
-            qubes.device_protocol.Port(dev.backend_domain, dev.ident, devclass))
+        assignment = qubes.device_protocol.DeviceAssignment(dev)
         await self.dest.devices[devclass].unassign(assignment)
         self.app.save()
 
@@ -1360,14 +1356,10 @@ async def vm_device_unassign(self, endpoint):
         scope='local', execute=True)
     async def vm_device_attach(self, endpoint, untrusted_payload):
         devclass = endpoint
-
-        # qrexec already verified that no strange characters are in self.arg
-        backend_domain, ident = self.arg.split('+', 1)
-        # may raise KeyError, either on domain or ident
-        dev = self.app.domains[backend_domain].devices[devclass][ident]
+        dev = self.load_device_info(devclass)
 
         assignment = qubes.device_protocol.DeviceAssignment.deserialize(
-            untrusted_payload, expected_port=dev
+            untrusted_payload, expected_device=dev
         )
 
         self.fire_event_for_permission(
@@ -1389,18 +1381,11 @@ async def vm_device_attach(self, endpoint, untrusted_payload):
         no_payload=True, scope='local', execute=True)
     async def vm_device_detach(self, endpoint):
         devclass = endpoint
-
-        # qrexec already verified that no strange characters are in self.arg
-        backend_domain, ident = self.arg.split('+', 1)
-        # may raise KeyError; if device isn't found, it will be UnknownDevice
-        #  instance - but allow it, otherwise it will be impossible to detach
-        #  already removed device
-        dev = self.app.domains[backend_domain].devices[devclass][ident]
+        dev = self.load_device_info(devclass)
 
         self.fire_event_for_permission(device=dev, devclass=devclass)
 
-        assignment = qubes.device_protocol.DeviceAssignment(
-            qubes.device_protocol.Port(dev.backend_domain, dev.ident, devclass))
+        assignment = qubes.device_protocol.DeviceAssignment(dev)
         await self.dest.devices[devclass].detach(assignment)
 
     # Assign/Unassign action can modify only a persistent state of running VM.
@@ -1425,10 +1410,7 @@ async def vm_device_set_required(self, endpoint, untrusted_payload):
         assignment = eval(untrusted_payload)
         del untrusted_payload
 
-        # qrexec already verified that no strange characters are in self.arg
-        backend_domain_name, ident = self.arg.split('+', 1)
-        backend_domain = self.app.domains[backend_domain_name]
-        dev = Port(backend_domain, ident, devclass)
+        dev = Device.from_qarg(self.arg, devclass, self.app.domains)
 
         self.fire_event_for_permission(device=dev, assignment=assignment)
 

qubes/app.py

@@ -1542,7 +1542,7 @@ def on_domain_pre_deleted(self, event, vm):
         assignments = vm.get_provided_assignments()
         if assignments:
             desc = ', '.join(
-                assignment.ident for assignment in assignments)
+                assignment.port_id for assignment in assignments)
             raise qubes.exc.QubesVMInUseError(
                 vm,
                 'VM has devices assigned to other VMs: ' + desc)