Use Port for device and make mypy happy

Related to work being done in QubesOS/qubes-core-admin#627
QubesOS · Nov 15, 2024 · f83d6b6 · f83d6b6
1 parent 6013161
commit f83d6b6
Show file tree

Hide file tree

Showing 4 changed files with 56 additions and 35 deletions.
diff --git a/.gitlab-ci.yml b/.gitlab-ci.yml
@@ -11,3 +11,16 @@ include:
   project: QubesOS/qubes-continuous-integration
 - file: /r4.2/gitlab-vm.yml
   project: QubesOS/qubes-continuous-integration
+
+mypy:
+  stage: checks
+  image: fedora:40
+  tags:
+    - docker
+  before_script:
+    - sudo dnf install -y python3-mypy python3-pip
+  script:
+    - mypy --install-types --non-interactive --ignore-missing-imports --junit-xml mypy.xml qubesusbproxy
+  artifacts:
+    reports:
+      junit: mypy.xml
diff --git a/qubesusbproxy/core3ext.py b/qubesusbproxy/core3ext.py
@@ -58,7 +58,7 @@ class DescriptionOverrider:
         def description(self):
             return self.name
 
-    class DeviceInfo(DescriptionOverrider, LegacyDeviceInfo):
+    class DeviceInfo(DescriptionOverrider, LegacyDeviceInfo):  # type: ignore
         def __init__(self, port):
             # not supported options in legacy code
             self.safe_chars = self.safe_chars.replace(" ", "")
@@ -78,12 +78,12 @@ def frontend_domain(self):
             return self.attachment
 
     @dataclasses.dataclass
-    class Port:
+    class Port:  # type: ignore
         backend_domain: Any
         port_id: Any
         devclass: Any
 
-    class DeviceInterface:
+    class DeviceInterface:  # type: ignore
         # pylint: disable=too-few-public-methods
         pass
 
@@ -105,20 +105,24 @@ def get_assigned_devices(devices):
 
 class USBDevice(DeviceInfo):
     # pylint: disable=too-few-public-methods
-    def __init__(self, backend_domain, port_id):
+    def __init__(self, port: qubes.device_protocol.Port):
+        if port.devclass != "usb":
+            raise qubes.exc.QubesValueError(
+                f"Incompatible device class for input port: {port.devclass}"
+            )
+
         # the superclass can restrict the allowed characters
         self.safe_chars = (
             string.ascii_letters + string.digits + string.punctuation + " "
         )
-        port = Port(
-            backend_domain=backend_domain, port_id=port_id, devclass="usb"
-        )
+
+        # init parent class
         super().__init__(port)
 
-        self._qdb_ident = port_id.replace(".", "_")
+        self._qdb_ident = port.port_id.replace(".", "_")
         self._qdb_path = "/qubes-usb-devices/" + self._qdb_ident
-        self._vendor_id = None
-        self._product_id = None
+        self._vendor_id: Optional[str] = None
+        self._product_id: Optional[str] = None
 
     @property
     def vendor(self) -> str:
@@ -410,10 +414,12 @@ def _load_usb_known_devices() -> Dict[str, Dict[str, Tuple[str, str]]]:
         # C class  class_name
         #       subclass  subclass_name         <-- single tab
         #               prog-if  prog-if_name   <-- two tabs
-        result = {}
+        result: Dict[str, Dict] = {}
         with open(
             HWDATA_PATH + "/usb.ids", encoding="utf-8", errors="ignore"
         ) as usb_ids:
+            vendor_id: Optional[str] = None
+            vendor_name: Optional[str] = None
             for line in usb_ids.readlines():
                 line = line.rstrip()
                 if line.startswith("#"):
@@ -431,6 +437,8 @@ def _load_usb_known_devices() -> Dict[str, Dict[str, Tuple[str, str]]]:
                 if line.startswith("\t"):
                     # save vendor, device pair
                     device_id, _, device_name = line[1:].split(" ", 2)
+                    if vendor_id is None or vendor_name is None:
+                        continue
                     result[vendor_id][device_id] = vendor_name, device_name
                 else:
                     # new vendor
@@ -589,7 +597,7 @@ def on_device_list_usb(self, vm, event):
                 vm.log.warning("Invalid USB device name detected")
                 continue
             port_id = untrusted_qdb_ident.replace("_", ".")
-            yield USBDevice(vm, port_id)
+            yield USBDevice(Port(vm, port_id, "usb"))
 
     @qubes.ext.handler("device-get:usb")
     def on_device_get_usb(self, vm, event, port_id):
@@ -600,12 +608,12 @@ def on_device_get_usb(self, vm, event, port_id):
         if vm.untrusted_qdb.list(
             "/qubes-usb-devices/" + port_id.replace(".", "_")
         ):
-            yield USBDevice(vm, port_id)
+            yield USBDevice(Port(vm, port_id, "usb"))
 
     @staticmethod
     def get_all_devices(app):
         for vm in app.domains:
-            if not vm.is_running():
+            if not vm.is_running() or not hasattr(vm, "devices"):
                 continue
 
             for dev in vm.devices["usb"]:

diff --git a/qubesusbproxy/tests.py b/qubesusbproxy/tests.py
@@ -884,7 +884,7 @@ def added_assign_setup(attachment=None):
     def test_010_on_qdb_change_multiple_assignments_including_full(self):
         back, front = self.added_assign_setup()
 
-        exp_dev = qubesusbproxy.core3ext.USBDevice(back, "1-1")
+        exp_dev = qubesusbproxy.core3ext.USBDevice(Port(back, "1-1", "usb"))
         full_assig = DeviceAssignment(
             VirtualDevice(exp_dev.port, exp_dev.device_id),
             mode="auto-attach",
@@ -907,7 +907,7 @@ def test_010_on_qdb_change_multiple_assignments_including_full(self):
         front.devices["usb"]._assigned.append(port_assign)
         front.devices["usb"]._assigned.append(full_assig)
         back.devices["usb"]._exposed.append(
-            qubesusbproxy.core3ext.USBDevice(back, "1-1")
+            qubesusbproxy.core3ext.USBDevice(Port(back, "1-1", "usb"))
         )
 
         self.ext.attach_and_notify = Mock()
@@ -922,7 +922,7 @@ def test_010_on_qdb_change_multiple_assignments_including_full(self):
     def test_011_on_qdb_change_multiple_assignments_port_vs_dev(self):
         back, front = self.added_assign_setup()
 
-        exp_dev = qubesusbproxy.core3ext.USBDevice(back, "1-1")
+        exp_dev = qubesusbproxy.core3ext.USBDevice(Port(back, "1-1", "usb"))
         port_assign = DeviceAssignment(
             VirtualDevice(exp_dev.port, "*"),
             mode="auto-attach",
@@ -939,7 +939,7 @@ def test_011_on_qdb_change_multiple_assignments_port_vs_dev(self):
         front.devices["usb"]._assigned.append(dev_assign)
         front.devices["usb"]._assigned.append(port_assign)
         back.devices["usb"]._exposed.append(
-            qubesusbproxy.core3ext.USBDevice(back, "1-1")
+            qubesusbproxy.core3ext.USBDevice(Port(back, "1-1", "usb"))
         )
 
         self.ext.attach_and_notify = Mock()
@@ -954,7 +954,7 @@ def test_011_on_qdb_change_multiple_assignments_port_vs_dev(self):
     def test_012_on_qdb_change_multiple_assignments_dev(self):
         back, front = self.added_assign_setup()
 
-        exp_dev = qubesusbproxy.core3ext.USBDevice(back, "1-1")
+        exp_dev = qubesusbproxy.core3ext.USBDevice(Port(back, "1-1", "usb"))
         port_assign = DeviceAssignment(
             VirtualDevice(Port(exp_dev.backend_domain, "1-2", "usb"), "*"),
             mode="auto-attach",
@@ -971,10 +971,10 @@ def test_012_on_qdb_change_multiple_assignments_dev(self):
         front.devices["usb"]._assigned.append(dev_assign)
         front.devices["usb"]._assigned.append(port_assign)
         back.devices["usb"]._exposed.append(
-            qubesusbproxy.core3ext.USBDevice(back, "1-1")
+            qubesusbproxy.core3ext.USBDevice(Port(back, "1-1", "usb"))
         )
         back.devices["usb"]._exposed.append(
-            qubesusbproxy.core3ext.USBDevice(back, "1-2")
+            qubesusbproxy.core3ext.USBDevice(Port(back, "1-2", "usb"))
         )
 
         self.ext.attach_and_notify = Mock()
@@ -989,7 +989,7 @@ def test_012_on_qdb_change_multiple_assignments_dev(self):
     def test_013_on_qdb_change_two_fronts(self):
         back, front = self.added_assign_setup()
 
-        exp_dev = qubesusbproxy.core3ext.USBDevice(back, "1-1")
+        exp_dev = qubesusbproxy.core3ext.USBDevice(Port(back, "1-1", "usb"))
         assmnt = DeviceAssignment(exp_dev, mode="auto-attach")
 
         front.devices["usb"]._assigned.append(assmnt)
@@ -1010,7 +1010,7 @@ def test_013_on_qdb_change_two_fronts(self):
     def test_014_failed_confirmation(self, socket):
         back, front = self.added_assign_setup()
 
-        exp_dev = qubesusbproxy.core3ext.USBDevice(back, "1-1")
+        exp_dev = qubesusbproxy.core3ext.USBDevice(Port(back, "1-1", "usb"))
         assmnt = DeviceAssignment(exp_dev, mode="auto-attach")
 
         front.devices["usb"]._assigned.append(assmnt)
@@ -1034,7 +1034,7 @@ def test_014_failed_confirmation(self, socket):
     def test_015_successful_confirmation(self, socket):
         back, front = self.added_assign_setup()
 
-        exp_dev = qubesusbproxy.core3ext.USBDevice(back, "1-1")
+        exp_dev = qubesusbproxy.core3ext.USBDevice(Port(back, "1-1", "usb"))
         assmnt = DeviceAssignment(exp_dev, mode="auto-attach")
 
         front.devices["usb"]._assigned.append(assmnt)
@@ -1055,7 +1055,7 @@ def test_015_successful_confirmation(self, socket):
     def test_016_on_qdb_change_ask(self):
         back, front = self.added_assign_setup()
 
-        exp_dev = qubesusbproxy.core3ext.USBDevice(back, "1-1")
+        exp_dev = qubesusbproxy.core3ext.USBDevice(Port(back, "1-1", "usb"))
         assmnt = DeviceAssignment(exp_dev, mode="ask-to-attach")
 
         front.devices["usb"]._assigned.append(assmnt)
@@ -1070,7 +1070,7 @@ def test_016_on_qdb_change_ask(self):
     def test_020_on_startup_multiple_assignments_including_full(self):
         back, front = self.added_assign_setup()
 
-        exp_dev = qubesusbproxy.core3ext.USBDevice(back, "1-1")
+        exp_dev = qubesusbproxy.core3ext.USBDevice(Port(back, "1-1", "usb"))
         full_assig = DeviceAssignment(
             VirtualDevice(exp_dev.port, exp_dev.device_id),
             mode="auto-attach",
@@ -1093,7 +1093,7 @@ def test_020_on_startup_multiple_assignments_including_full(self):
         front.devices["usb"]._assigned.append(port_assign)
         front.devices["usb"]._assigned.append(full_assig)
         back.devices["usb"]._exposed.append(
-            qubesusbproxy.core3ext.USBDevice(back, "1-1")
+            qubesusbproxy.core3ext.USBDevice(Port(back, "1-1", "usb"))
         )
 
         self.ext.attach_and_notify = AsyncMock()
@@ -1106,7 +1106,7 @@ def test_020_on_startup_multiple_assignments_including_full(self):
     def test_021_on_startup_multiple_assignments_port_vs_dev(self):
         back, front = self.added_assign_setup()
 
-        exp_dev = qubesusbproxy.core3ext.USBDevice(back, "1-1")
+        exp_dev = qubesusbproxy.core3ext.USBDevice(Port(back, "1-1", "usb"))
         port_assign = DeviceAssignment(
             VirtualDevice(exp_dev.port, "*"),
             mode="auto-attach",
@@ -1123,7 +1123,7 @@ def test_021_on_startup_multiple_assignments_port_vs_dev(self):
         front.devices["usb"]._assigned.append(dev_assign)
         front.devices["usb"]._assigned.append(port_assign)
         back.devices["usb"]._exposed.append(
-            qubesusbproxy.core3ext.USBDevice(back, "1-1")
+            qubesusbproxy.core3ext.USBDevice(Port(back, "1-1", "usb"))
         )
 
         loop = asyncio.get_event_loop()
@@ -1136,7 +1136,7 @@ def test_021_on_startup_multiple_assignments_port_vs_dev(self):
     def test_022_on_startup_multiple_assignments_dev(self):
         back, front = self.added_assign_setup()
 
-        exp_dev = qubesusbproxy.core3ext.USBDevice(back, "1-1")
+        exp_dev = qubesusbproxy.core3ext.USBDevice(Port(back, "1-1", "usb"))
         port_assign = DeviceAssignment(
             VirtualDevice(Port(exp_dev.backend_domain, "1-2", "usb"), "*"),
             mode="auto-attach",
@@ -1153,10 +1153,10 @@ def test_022_on_startup_multiple_assignments_dev(self):
         front.devices["usb"]._assigned.append(dev_assign)
         front.devices["usb"]._assigned.append(port_assign)
         back.devices["usb"]._exposed.append(
-            qubesusbproxy.core3ext.USBDevice(back, "1-1")
+            qubesusbproxy.core3ext.USBDevice(Port(back, "1-1", "usb"))
         )
         back.devices["usb"]._exposed.append(
-            qubesusbproxy.core3ext.USBDevice(back, "1-2")
+            qubesusbproxy.core3ext.USBDevice(Port(back, "1-2", "usb"))
         )
 
         self.ext.attach_and_notify = AsyncMock()
@@ -1169,7 +1169,7 @@ def test_022_on_startup_multiple_assignments_dev(self):
     def test_023_on_startup_already_attached(self):
         back, front = self.added_assign_setup(attachment="sys-usb")
 
-        exp_dev = qubesusbproxy.core3ext.USBDevice(back, "1-1")
+        exp_dev = qubesusbproxy.core3ext.USBDevice(Port(back, "1-1", "usb"))
         assmnt = DeviceAssignment(
             VirtualDevice(exp_dev.port, exp_dev.device_id), mode="auto-attach"
         )

diff --git a/qubesusbproxy/utils.py b/qubesusbproxy/utils.py
@@ -23,7 +23,7 @@
 
 import qubes
 
-from typing import Type
+from typing import Type, Dict, Any
 
 from qubes import device_protocol
 from qubes.device_protocol import VirtualDevice
@@ -75,7 +75,7 @@ def device_list_change(
 
     ext.devices_cache[vm.name] = current_devices
 
-    to_attach = {}
+    to_attach: Dict[str, Dict] = {}
     for front_vm in vm.app.domains:
         if not front_vm.is_running():
             continue