q-dev: virtual device

Author: piotrbartman

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, Device, DeviceInfo
+from qubes.device_protocol import Port, VirtualDevice, DeviceInfo
 
 
 class QubesMgmtEventsDispatcher:
@@ -1323,7 +1323,7 @@ async def vm_device_assign(self, endpoint, untrusted_payload):
 
     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)
+        _dev = VirtualDevice.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]
@@ -1410,7 +1410,7 @@ async def vm_device_set_required(self, endpoint, untrusted_payload):
         assignment = eval(untrusted_payload)
         del untrusted_payload
 
-        dev = Device.from_qarg(self.arg, devclass, self.app.domains)
+        dev = VirtualDevice.from_qarg(self.arg, devclass, self.app.domains)
 
         self.fire_event_for_permission(device=dev, assignment=assignment)
 

qubes/device_protocol.py

@@ -134,7 +134,7 @@ def pack_property(cls, key: str, value: Optional[str]):
 
     @staticmethod
     def parse_basic_device_properties(
-            expected_device: 'Device', properties: Dict[str, Any]):
+            expected_device: 'VirtualDevice', properties: Dict[str, Any]):
         """
         Validates properties against an expected port configuration.
 
@@ -318,7 +318,7 @@ def devclass(self) -> str:
         return "peripheral"
 
 
-class Device:
+class VirtualDevice:
     def __init__(
             self,
             port: Optional[Port] = None,
@@ -376,7 +376,7 @@ def __hash__(self):
         return hash((self.port, self.device_id))
 
     def __eq__(self, other):
-        if isinstance(other, (Device, DeviceAssignment)):
+        if isinstance(other, (VirtualDevice, DeviceAssignment)):
             result = (
                     self.port == other.port
                     and self.device_id == other.device_id
@@ -398,7 +398,7 @@ def __lt__(self, other):
         3. *:<devid>
         4. *:*
         """
-        if isinstance(other, (Device, DeviceAssignment)):
+        if isinstance(other, (VirtualDevice, DeviceAssignment)):
             if self.port == '*' and other.port != '*':
                 return True
             if self.port != '*' and other.port == '*':
@@ -409,7 +409,7 @@ def __lt__(self, other):
                     reprs[obj].append(obj.device_id)
             return reprs[self] < reprs[other]
         elif isinstance(other, Port):
-            _other = Device(other, '*')
+            _other = VirtualDevice(other, '*')
             return self < _other
         else:
             raise TypeError(
@@ -430,7 +430,7 @@ def from_qarg(
             domains,
             blind=False,
             backend=None,
-    ) -> 'Device':
+    ) -> 'VirtualDevice':
         if backend is None:
             if blind:
                 get_domain = domains.get_blind
@@ -444,7 +444,7 @@ def from_qarg(
     def from_str(
             cls, representation: str, devclass: Optional[str], domains,
             blind=False, backend=None
-    ) -> 'Device':
+    ) -> 'VirtualDevice':
         if backend is None:
             if blind:
                 get_domain = domains.get_blind
@@ -462,7 +462,7 @@ def _parse(
             get_domain: Callable,
             backend,
             sep: str
-    ) -> 'Device':
+    ) -> 'VirtualDevice':
         if backend is None:
             backend_name, identity = representation.split(sep, 1)
             backend = get_domain(backend_name)
@@ -673,7 +673,7 @@ def _load_classes(bus: str):
         return result
 
 
-class DeviceInfo(Device):
+class DeviceInfo(VirtualDevice):
     """ Holds all information about a device """
 
     def __init__(
@@ -813,7 +813,7 @@ def interfaces(self) -> List[DeviceInterface]:
         return self._interfaces
 
     @property
-    def parent_device(self) -> Optional[Device]:
+    def parent_device(self) -> Optional[VirtualDevice]:
         """
         The parent device, if any.
 
@@ -823,7 +823,7 @@ def parent_device(self) -> Optional[Device]:
         return self._parent
 
     @property
-    def subdevices(self) -> List[Device]:
+    def subdevices(self) -> List[VirtualDevice]:
         """
         The list of children devices if any.
 
@@ -844,7 +844,7 @@ def serialize(self) -> bytes:
         """
         Serialize an object to be transmitted via Qubes API.
         """
-        properties = Device.serialize(self)
+        properties = VirtualDevice.serialize(self)
         # 'attachment', 'interfaces', 'data', 'parent_device'
         # are not string, so they need special treatment
         default = DeviceInfo(self.port)
@@ -884,7 +884,7 @@ def deserialize(
         Recovers a serialized object, see: :py:meth:`serialize`.
         """
         head, _, rest = serialization.partition(b' ')
-        device = Device.from_str(
+        device = VirtualDevice.from_str(
             head.decode('ascii', errors='ignore'), expected_devclass,
             domains=None, backend=expected_backend_domain)
 
@@ -900,7 +900,7 @@ def deserialize(
     def _deserialize(
             cls,
             untrusted_serialization: bytes,
-            expected_device: Device
+            expected_device: VirtualDevice
     ) -> 'DeviceInfo':
         """
         Actually deserializes the object.
@@ -1000,13 +1000,13 @@ class DeviceAssignment:
 
     def __init__(
             self,
-            device: Device,
+            device: VirtualDevice,
             frontend_domain=None,
             options=None,
             mode: Union[str, AssignmentMode] = "manual",
     ):
         if isinstance(device, DeviceInfo):
-            device = Device(device.port, device.device_id)
+            device = VirtualDevice(device.port, device.device_id)
         self._device_ident = device
         self.__options = options or {}
         if isinstance(mode, AssignmentMode):
@@ -1020,7 +1020,7 @@ def clone(self, **kwargs):
         Clone object and substitute attributes with explicitly given.
         """
         kwargs["device"] = kwargs.get(
-            "device", Device(
+            "device", VirtualDevice(
                 Port(self.backend_domain, self.port_id, self.devclass),
                 self.device_id
             ))
@@ -1042,7 +1042,7 @@ def __hash__(self):
         return hash(self._device_ident)
 
     def __eq__(self, other):
-        if isinstance(other, (Device, DeviceAssignment)):
+        if isinstance(other, (VirtualDevice, DeviceAssignment)):
             result = (
                     self.port == other.port
                     and self.device_id == other.device_id
@@ -1053,7 +1053,7 @@ def __eq__(self, other):
     def __lt__(self, other):
         if isinstance(other, DeviceAssignment):
             return self._device_ident < other._device_ident
-        if isinstance(other, Device):
+        if isinstance(other, VirtualDevice):
             return self._device_ident < other
         raise TypeError(
             f"Comparing instances of {type(self)} and '{type(other)}' "
@@ -1078,7 +1078,8 @@ def device_id(self):
     @property
     def device(self) -> DeviceInfo:
         """Get DeviceInfo object corresponding to this DeviceAssignment"""
-        dev = self.backend_domain.devices[self.devclass][self.port_id]
+        if self.port_id:
+            dev = self.backend_domain.devices[self.devclass][self.port_id]
         # TODO: device identity could not match
         return dev
 
@@ -1162,7 +1163,7 @@ def serialize(self) -> bytes:
     def deserialize(
             cls,
             serialization: bytes,
-            expected_device: Device,
+            expected_device: VirtualDevice,
     ) -> 'DeviceAssignment':
         """
         Recovers a serialized object, see: :py:meth:`serialize`.
@@ -1177,7 +1178,7 @@ def deserialize(
     def _deserialize(
             cls,
             untrusted_serialization: bytes,
-            expected_device: Device,
+            expected_device: VirtualDevice,
     ) -> 'DeviceAssignment':
         """
         Actually deserializes the object.

qubes/devices.py

@@ -65,7 +65,7 @@
 import qubes.exc
 import qubes.utils
 from qubes.device_protocol import (Port, DeviceInfo, UnknownDevice,
-                                   DeviceAssignment, Device)
+                                   DeviceAssignment, VirtualDevice)
 
 
 class DeviceNotAssigned(qubes.exc.QubesException, KeyError):
@@ -249,11 +249,11 @@ def load_assignment(self, device_assignment: DeviceAssignment):
         assert device_assignment.attach_automatically
         self._set.add(device_assignment)
 
-    async def update_required(self, device: Device, required: bool):
+    async def update_required(self, device: VirtualDevice, required: bool):
         """
         Update `required` flag of an already attached device.
 
-        :param Device device: device for which change required flag
+        :param VirtualDevice device: device for which change required flag
         :param bool required: new assignment:
                               `False` -> device will be auto-attached to qube
                               `True` -> device is required to start qube

qubes/tests/devices.py

@@ -23,7 +23,7 @@
 
 import qubes.devices
 from qubes.device_protocol import (Port, DeviceInfo, DeviceAssignment,
-                                   DeviceInterface, UnknownDevice, Device)
+                                   DeviceInterface, UnknownDevice, VirtualDevice)
 
 import qubes.tests
 
@@ -515,7 +515,7 @@ def setUp(self):
         self.vm = TestVM(self.app, 'vm')
 
     def test_010_serialize(self):
-        assignment = DeviceAssignment(Device(Port(
+        assignment = DeviceAssignment(VirtualDevice(Port(
             backend_domain=self.vm,
             port_id="1-1.1.1",
             devclass="bus",
@@ -530,7 +530,7 @@ def test_010_serialize(self):
 
     def test_011_serialize_required(self):
         assignment = DeviceAssignment(
-            Device(Port(
+            VirtualDevice(Port(
                 backend_domain=self.vm,
                 port_id="1-1.1.1",
                 devclass="bus",
@@ -547,7 +547,7 @@ def test_011_serialize_required(self):
 
     def test_012_serialize_fronted(self):
         assignment = DeviceAssignment(
-            Device(Port(
+            VirtualDevice(Port(
                 backend_domain=self.vm,
                 port_id="1-1.1.1",
                 devclass="bus",
@@ -564,7 +564,7 @@ def test_012_serialize_fronted(self):
 
     def test_013_serialize_options(self):
         assignment = DeviceAssignment(
-            Device(Port(
+            VirtualDevice(Port(
                 backend_domain=self.vm,
                 port_id="1-1.1.1",
                 devclass="bus",
@@ -581,7 +581,7 @@ def test_013_serialize_options(self):
 
     def test_014_invalid_serialize(self):
         assignment = DeviceAssignment(
-            Device(Port(
+            VirtualDevice(Port(
                 backend_domain=self.vm,
                 port_id="1-1.1.1",
                 devclass="bus",
@@ -596,10 +596,10 @@ def test_020_deserialize(self):
             b"device_id='*' port_id='1-1.1.1' frontend_domain='vm' "
             b"devclass='bus' backend_domain='vm' mode='auto-attach' "
             b"_read-only='yes'")
-        expected_device = Device(Port(self.vm, '1-1.1.1', 'bus'))
+        expected_device = VirtualDevice(Port(self.vm, '1-1.1.1', 'bus'))
         actual = DeviceAssignment.deserialize(serialized, expected_device)
         expected = DeviceAssignment(
-            Device(Port(
+            VirtualDevice(Port(
                 backend_domain=self.vm,
                 port_id="1-1.1.1",
                 devclass="bus",
@@ -622,7 +622,7 @@ def test_021_invalid_deserialize(self):
             b"device_id='*' port_id='1-1.1.1' frontend_domain='vm' "
             b"devclass='bus' backend_domain='vm' mode='auto-attach' "
             b"_read'only='yes'")
-        expected_device = Device(Port(self.vm, '1-1.1.1', 'bus'))
+        expected_device = VirtualDevice(Port(self.vm, '1-1.1.1', 'bus'))
         with self.assertRaises(qubes.exc.ProtocolError):
             _ = DeviceAssignment.deserialize(serialized, expected_device)
 
@@ -631,12 +631,12 @@ def test_022_invalid_deserialize_2(self):
             b"device_id='*' port_id='1-1.1.1' frontend_domain='vm' "
             b"devclass='bus' backend_domain='vm' mode='auto-attach' "
             b"read-only='yes'")
-        expected_device = Device(Port(self.vm, '1-1.1.1', 'bus'))
+        expected_device = VirtualDevice(Port(self.vm, '1-1.1.1', 'bus'))
         with self.assertRaises(qubes.exc.ProtocolError):
             _ = DeviceAssignment.deserialize(serialized, expected_device)
 
     def test_030_serialize_and_deserialize(self):
-        expected_device = Device(Port(self.vm, '1-1.1.1', 'bus'))
+        expected_device = VirtualDevice(Port(self.vm, '1-1.1.1', 'bus'))
         expected = DeviceAssignment(
             expected_device,
             frontend_domain=self.vm,

qubes/tests/devices_block.py

@@ -26,7 +26,7 @@
 import qubes.tests
 import qubes.ext.block
 from qubes.device_protocol import DeviceInterface, Port, DeviceInfo, \
-    DeviceAssignment, Device
+    DeviceAssignment
 
 modules_disk = '''
     <disk type='block' device='disk'>

qubes/tests/integ/audio.py

@@ -272,7 +272,7 @@ async def _check_audio_input_status(vm, status):
 
     def attach_mic(self):
         deva = qubes.device_protocol.DeviceAssignment(
-            qubes.device_protocol.Device(
+            qubes.device_protocol.VirtualDevice(
                 qubes.device_protocol.Port(self.app.domains[0], 'mic', 'mic')))
         self.loop.run_until_complete(
             self.testvm1.devices['mic'].attach(deva)
@@ -281,7 +281,7 @@ def attach_mic(self):
 
     def detach_mic(self):
         deva = qubes.device_protocol.DeviceAssignment(
-            qubes.device_protocol.Device(
+            qubes.device_protocol.VirtualDevice(
                 qubes.device_protocol.Port(self.app.domains[0], 'mic', 'mic')))
         self.loop.run_until_complete(
             self.testvm1.devices['mic'].detach(deva)

qubes/tests/integ/devices_block.py

@@ -328,7 +328,7 @@ def setUp(self):
 
     def test_000_attach_reattach(self):
         ass = qubes.device_protocol.DeviceAssignment(
-            qubes.device_protocol.Device(
+            qubes.device_protocol.VirtualDevice(
                 qubes.device_protocol.Port(
                     self.backend, self.device_ident, 'block')
             ))