q-dev: device_protocol

qubes/api/admin.py

@@ -45,7 +45,8 @@
 import qubes.vm
 import qubes.vm.adminvm
 import qubes.vm.qubesvm
-from qubes.device_protocol import Port, VirtualDevice, DeviceInfo
+from qubes.device_protocol import (Port, VirtualDevice, UnknownDevice,
+                                   DeviceAssignment)
 
 
 class QubesMgmtEventsDispatcher:
@@ -1324,10 +1325,15 @@ async def vm_device_assign(self, endpoint, untrusted_payload):
     def load_device_info(self, devclass) -> VirtualDevice:
         # qrexec already verified that no strange characters are in self.arg
         _dev = VirtualDevice.from_qarg(self.arg, devclass, self.app.domains)
+        if _dev.port_id == '*' or _dev.device_id == '*':
+            return _dev
         # load all info, may raise KeyError, either on domain or port_id
         try:
-            return self.app.domains[
+            dev = self.app.domains[
                 _dev.backend_domain].devices[devclass][_dev.port_id]
+            if isinstance(dev, UnknownDevice):
+                return _dev
+            return dev
         except KeyError:
             return _dev
 
@@ -1338,14 +1344,15 @@ def load_device_info(self, devclass) -> VirtualDevice:
         endpoints=(
             ep.name
             for ep in importlib.metadata.entry_points(group='qubes.devices')),
-            no_payload=True, scope='local', write=True)
-    async def vm_device_unassign(self, endpoint):
+        scope='local', write=True)
+    async def vm_device_unassign(self, endpoint, untrusted_payload):
         devclass = endpoint
         dev = self.load_device_info(devclass)
+        assignment = DeviceAssignment.deserialize(
+            untrusted_payload, expected_device=dev)
 
         self.fire_event_for_permission(device=dev, devclass=devclass)
 
-        assignment = qubes.device_protocol.DeviceAssignment(dev)
         await self.dest.devices[devclass].unassign(assignment)
         self.app.save()
 
@@ -1356,14 +1363,12 @@ async def vm_device_unassign(self, endpoint):
         endpoints=(
             ep.name
             for ep in importlib.metadata.entry_points(group='qubes.devices')),
-        scope='local', execute=True)
-    async def vm_device_attach(self, endpoint, untrusted_payload):
+        no_payload=True, scope='local', execute=True)
+    async def vm_device_attach(self, endpoint):
         devclass = endpoint
         dev = self.load_device_info(devclass)
 
-        assignment = qubes.device_protocol.DeviceAssignment.deserialize(
-            untrusted_payload, expected_device=dev
-        )
+        assignment = DeviceAssignment(dev)
 
         self.fire_event_for_permission(
             device=dev, devclass=devclass,

qubes/device_protocol.py

@@ -145,7 +145,7 @@ def parse_basic_device_properties(
             the expected values.
         """
         expected = expected_device.port
-        exp_vm_name = expected.backend_domain.name
+        exp_vm_name = expected.backend_name
         if properties.get('backend_domain', exp_vm_name) != exp_vm_name:
             raise UnexpectedDeviceProperty(
                 f"Got device exposed by {properties['backend_domain']}"
@@ -264,7 +264,7 @@ def __str__(self):
 
     @property
     def backend_name(self) -> str:
-        if self.backend_domain is not None:
+        if self.backend_domain not in (None, "*"):
             return self.backend_domain.name
         return "*"
 
@@ -307,7 +307,9 @@ def port_id(self) -> str:
 
         Unique for given domain and devclass.
         """
-        return self.__port_id
+        if self.__port_id is not None:
+            return self.__port_id
+        return '*'
 
     @property
     def backend_domain(self) -> Optional[QubesVM]:
@@ -338,7 +340,7 @@ def __init__(
             port: Optional[Port] = None,
             device_id: Optional[str] = None,
     ):
-        # TODO! one of them cannot be None
+        assert port is not None or device_id is not None
         self.port: Optional[Port] = port
         self._device_id = device_id
 
@@ -379,7 +381,6 @@ def backend_name(self):
             return self.port.backend_name
         return '*'
 
-
     @property
     def port_id(self):
         if self.port != '*' and self.port.port_id is not None:
@@ -496,8 +497,8 @@ def _parse(
         else:
             identity = representation
         port_id, _, devid = identity.partition(':')
-        if devid in ('', '*'):
-            devid = '*'
+        if devid == '':
+            devid = None
         return cls(
             Port(backend_domain=backend, port_id=port_id, devclass=devclass),
             device_id=devid
@@ -515,7 +516,7 @@ def serialize(self) -> bytes:
                 ('devclass', self.devclass)))
 
         properties += b' ' + DeviceSerializer.pack_property(
-            'backend_domain', self.backend_domain.name)
+            'backend_domain', self.backend_name)
 
         return properties
 
@@ -1143,7 +1144,7 @@ def attached(self) -> bool:
         Returns False if device is attached to different domain
         """
         for device in self.devices:
-            if device.attachment == self.frontend_domain:
+            if device.attachment and device.attachment == self.frontend_domain:
                 return True
         return False
 
@@ -1233,11 +1234,11 @@ def _deserialize(
         return cls(**properties)
 
     def matches(self, device: VirtualDevice) -> bool:
-        if self.backend_domain != '*' and self.backend_domain != device.backend_domain:
+        if self.devclass != device.devclass:
             return False
-        if self.port_id != '*' and self.port_id != device.port_id:
+        if self.backend_domain != device.backend_domain:
             return False
-        if self.devclass != '*' and self.devclass != device.devclass:
+        if self.port_id != '*' and self.port_id != device.port_id:
             return False
         if self.device_id != '*' and self.device_id != device.device_id:
             return False

qubes/devices.py

@@ -319,16 +319,14 @@ async def unassign(self, assignment: DeviceAssignment):
         """
         all_ass = []
         for assign in self.get_assigned_devices():
-            all_ass.append(assign.devclass)
+            all_ass.append(assign)
             if assignment == assign:
                 # load all options
                 assignment = assign
                 break
         else:
             raise DeviceNotAssigned(
-                f'{self._bus} device at port {assignment}'
-                f'not assigned to {self._vm!s} '
-                f'| {all_ass} vs {assignment.devclass}')
+                f'{self._bus} device {assignment} not assigned to {self._vm!s}')
 
         self._set.discard(assignment)