qubes/ext/utils: provide Port for device_class

Author: fepitre

qubes/device_protocol.py

@@ -261,7 +261,7 @@ class Port:
     def __init__(
         self,
         backend_domain: Optional[QubesVM],
-        port_id: str,
+        port_id: Optional[str],
         devclass: Optional[str],
     ):
         self.__backend_domain = backend_domain
@@ -1302,7 +1302,7 @@ def devices(self) -> List[DeviceInfo]:
             dev = self.backend_domain.devices[self.devclass][self.port_id]
             if (
                 isinstance(dev, UnknownDevice)
-                or self.device_id in (dev.device_id, "*")
+                or (dev and self.device_id in (dev.device_id, "*"))
             ):
                 return [dev]
         if self.device_id == "0000:0000::?******":

qubes/ext/block.py

@@ -34,6 +34,7 @@
 from qubes.ext import utils
 from qubes.storage import Storage
 from qubes.vm.qubesvm import QubesVM
+from qubes.devices import Port
 
 name_re = re.compile(r"\A[a-z0-9-]{1,12}\Z")
 device_re = re.compile(r"\A[a-z0-9/-]{1,64}\Z")
@@ -47,16 +48,20 @@
 
 
 class BlockDevice(qubes.device_protocol.DeviceInfo):
-    def __init__(self, backend_domain, port_id):
-        port = qubes.device_protocol.Port(
-            backend_domain=backend_domain, port_id=port_id, devclass="block"
-        )
+
+    def __init__(self, port: qubes.device_protocol.Port):
+        if port.devclass != "block":
+            raise qubes.exc.QubesValueError(
+                f"Incompatible device class for input port: {port.devclass}"
+            )
+
+        # init parent class
         super().__init__(port)
 
         # lazy loading
-        self._mode = None
-        self._size = None
-        self._interface_num = None
+        self._mode: Optional[str] = None
+        self._size: Optional[int] = None
+        self._interface_num: Optional[str] = None
 
     @property
     def name(self):
@@ -192,8 +197,9 @@ def parent_device(self) -> Optional[qubes.device_protocol.DeviceInfo]:
             if not parent_ident:
                 return None
             try:
-                self._parent = self.backend_domain.devices[
-                    devclass][parent_ident]
+                self._parent = self.backend_domain.devices[devclass][
+                    parent_ident
+                ]
             except KeyError:
                 self._parent = qubes.device_protocol.UnknownDevice(
                     qubes.device_protocol.Port(
@@ -250,7 +256,7 @@ def device_id(self) -> str:
             # device interface number (not partition)
             self_id = self._interface_num
         else:
-            self_id = self._get_possible_partition_number()
+            self_id = str(self._get_possible_partition_number())
         return f"{parent_identity}:{self_id}"
 
     def _get_possible_partition_number(self) -> Optional[int]:
@@ -372,7 +378,9 @@ def device_get(vm, port_id):
         )
         if not untrusted_qubes_device_attrs:
             return None
-        return BlockDevice(vm, port_id)
+        return BlockDevice(
+            Port(backend_domain=vm, port_id=port_id, devclass="block")
+        )
 
     @qubes.ext.handler("device-list:block")
     def on_device_list_block(self, vm, event):
@@ -462,7 +470,7 @@ def on_device_list_attached(self, vm, event, **kwargs):
 
             port_id = port_id.replace("/", "_")
 
-            yield (BlockDevice(backend_domain, port_id), options)
+            yield BlockDevice(Port(backend_domain, port_id, "block")), options
 
     @staticmethod
     def find_unused_frontend(vm, devtype="disk"):
@@ -638,13 +646,13 @@ async def on_domain_shutdown(self, vm, event, **_kwargs):
                 for dev_id, front_vm in self.devices_cache[domain.name].items():
                     if front_vm is None:
                         continue
-                    dev = BlockDevice(vm, dev_id)
+                    dev = BlockDevice(Port(vm, dev_id, "block"))
                     vm.fire_event("device-removed:block", port=dev.port)
                     await self.detach_and_notify(front_vm, dev.port)
                 continue
             for dev_id, front_vm in self.devices_cache[domain.name].items():
                 if front_vm == vm:
-                    dev = BlockDevice(vm, dev_id)
+                    dev = BlockDevice(Port(vm, dev_id, "block"))
                     asyncio.ensure_future(
                         front_vm.fire_event_async(
                             "device-detach:block", port=dev.port

qubes/ext/pci.py

@@ -34,6 +34,7 @@
 import qubes.device_protocol
 import qubes.devices
 import qubes.ext
+from qubes.device_protocol import Port
 
 #: cache of PCI device classes
 pci_classes = None
@@ -161,31 +162,35 @@ class PCIDevice(qubes.device_protocol.DeviceInfo):
         r"(?P<function>[0-9a-f]+)\Z"
     )
 
-    def __init__(self, backend_domain, port_id, libvirt_name=None):
+    def __init__(self, port: Port, libvirt_name=None):
         if libvirt_name:
             dev_match = self._libvirt_regex.match(libvirt_name)
             if not dev_match:
                 raise UnsupportedDevice(libvirt_name)
             port_id = "{bus}_{device}.{function}".format(
                 **dev_match.groupdict()
             )
+            port = Port(
+                backend_domain=port.backend_domain,
+                port_id=port_id,
+                devclass="pci",
+            )
 
-        port = qubes.device_protocol.Port(
-            backend_domain=backend_domain, port_id=port_id, devclass="pci"
-        )
         super().__init__(port)
 
-        dev_match = self.regex.match(port_id)
+        dev_match = self.regex.match(port.port_id)
         if not dev_match:
-            raise ValueError("Invalid device identifier: {!r}".format(port_id))
+            raise ValueError(
+                "Invalid device identifier: {!r}".format(port.port_id)
+            )
 
         for group in self.regex.groupindex:
             setattr(self, group, dev_match.group(group))
 
         # lazy loading
-        self._description = None
-        self._vendor_id = None
-        self._product_id = None
+        self._description: Optional[str] = None
+        self._vendor_id: Optional[str] = None
+        self._product_id: Optional[str] = None
 
     @property
     def vendor(self) -> str:
@@ -247,7 +252,6 @@ def interfaces(self) -> List[qubes.device_protocol.DeviceInterface]:
             ]
         return self._interfaces or []
 
-
     @property
     def parent_device(self) -> Optional[qubes.device_protocol.DeviceInfo]:
         """
@@ -300,12 +304,12 @@ def _load_desc(self) -> Dict[str, str]:
         unknown = "unknown"
         result = {
             "vendor": unknown,
-              "vendor ID": "0000",
-              "product": unknown,
-              "product ID": "0000",
-              "manufacturer": unknown,
-              "name": unknown,
-              "serial": unknown
+            "vendor ID": "0000",
+            "product": unknown,
+            "product ID": "0000",
+            "manufacturer": unknown,
+            "name": unknown,
+            "serial": unknown,
         }
 
         if (
@@ -324,11 +328,12 @@ def _load_desc(self) -> Dict[str, str]:
         # Data successfully loaded, cache these values
         hostdev_xml = lxml.etree.fromstring(hostdev_details.XMLDesc())
 
-
-        self._vendor = result["vendor"] = hostdev_xml.findtext(
-            "capability/vendor") or unknown
-        self._product = result["product"] = hostdev_xml.findtext(
-            "capability/product") or unknown
+        self._vendor = result["vendor"] = (
+            hostdev_xml.findtext("capability/vendor") or unknown
+        )
+        self._product = result["product"] = (
+            hostdev_xml.findtext("capability/product") or unknown
+        )
 
         vendor = hostdev_xml.xpath("//vendor/@id") or []
         if vendor and isinstance(vendor, List):
@@ -365,7 +370,10 @@ def on_device_list_pci(self, vm, event):
             xml_desc = lxml.etree.fromstring(dev.XMLDesc())
             libvirt_name = xml_desc.findtext("name")
             try:
-                yield PCIDevice(vm, None, libvirt_name=libvirt_name)
+                yield PCIDevice(
+                    Port(backend_domain=vm, port_id=None, devclass="pci"),
+                    libvirt_name=libvirt_name,
+                )
             except UnsupportedDevice:
                 if libvirt_name not in unsupported_devices_warned:
                     vm.log.warning("Unsupported device: %s", libvirt_name)
@@ -397,7 +405,13 @@ def on_device_list_attached(self, vm, event, **kwargs):
                 device=device,
                 function=function,
             )
-            yield (PCIDevice(vm.app.domains[0], port_id), {})
+            yield PCIDevice(
+                Port(
+                    backend_domain=vm.app.domains[0],
+                    port_id=port_id,
+                    devclass="pci",
+                )
+            ), {}
 
     @qubes.ext.handler("device-pre-attach:pci")
     def on_device_pre_attached_pci(self, vm, event, device, options):
@@ -538,4 +552,4 @@ def on_app_close(self, app, event):
 @functools.lru_cache(maxsize=None)
 def _cache_get(vm, port_id):
     """Caching wrapper around `PCIDevice(vm, port_id)`."""
-    return PCIDevice(vm, port_id)
+    return PCIDevice(Port(vm, port_id, "pci"))

qubes/ext/utils.py

@@ -25,7 +25,7 @@
 import qubes.ext
 from qrexec.server import call_socket_service
 from qubes import device_protocol
-from qubes.device_protocol import VirtualDevice
+from qubes.device_protocol import VirtualDevice, Port
 
 SOCKET_PATH = "/var/run/qubes"
 
@@ -48,25 +48,33 @@ def device_list_change(
 
     # send events about devices detached/attached outside by themselves
     for port_id, front_vm in detached.items():
-        dev = device_class(vm, port_id)
-        ext.ensure_detach(front_vm, dev.port)
+        device = device_class(
+            Port(backend_domain=vm, port_id=port_id, devclass=devclass)
+        )
+        ext.ensure_detach(front_vm, device.port)
         asyncio.ensure_future(
             front_vm.fire_event_async(
-                f"device-detach:{devclass}", port=dev.port
+                f"device-detach:{devclass}", port=device.port
             )
         )
     for port_id in removed:
-        device = device_class(vm, port_id)
+        device = device_class(
+            Port(backend_domain=vm, port_id=port_id, devclass=devclass)
+        )
         vm.fire_event(f"device-removed:{devclass}", port=device.port)
     for port_id in added:
-        device = device_class(vm, port_id)
+        device = device_class(
+            Port(backend_domain=vm, port_id=port_id, devclass=devclass)
+        )
         vm.fire_event(f"device-added:{devclass}", device=device)
     for port_id, front_vm in attached.items():
-        dev = device_class(vm, port_id)
+        device = device_class(
+            Port(backend_domain=vm, port_id=port_id, devclass=devclass)
+        )
         # options are unknown, device already attached
         asyncio.ensure_future(
             front_vm.fire_event_async(
-                f"device-attach:{devclass}", device=dev, options={}
+                f"device-attach:{devclass}", device=device, options={}
             )
         )
 

qubes/tests/app.py

@@ -70,6 +70,7 @@ def setUp(self):
         self.app = TestApp()
         self.app.vmm = mock.Mock()
         self.qubes_host = qubes.app.QubesHost(self.app)
+        self.maxDiff = None
 
     def test_000_get_vm_stats_single(self):
         self.app.vmm.configure_mock(**{