From 93c89e26b64f14dc30c72f75da230efc0fbc5d2a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Fr=C3=A9d=C3=A9ric=20Pierret=20=28fepitre=29?= Date: Tue, 5 Nov 2024 15:47:11 +0100 Subject: [PATCH] Use Port for device and make mypy happy Related to work being done in https://github.com/QubesOS/qubes-core-admin/pull/627 --- .gitlab-ci.yml | 13 +++++++++++++ qubesusbproxy/core3ext.py | 16 ++++++++++------ qubesusbproxy/tests.py | 38 +++++++++++++++++++------------------- qubesusbproxy/utils.py | 4 ++-- 4 files changed, 44 insertions(+), 27 deletions(-) diff --git a/.gitlab-ci.yml b/.gitlab-ci.yml index 55e8a44..d03739d 100644 --- 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 \ No newline at end of file diff --git a/qubesusbproxy/core3ext.py b/qubesusbproxy/core3ext.py index cd66188..48a6f2b 100644 --- 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 @@ -410,10 +410,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 +433,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 +593,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,7 +604,7 @@ 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): diff --git a/qubesusbproxy/tests.py b/qubesusbproxy/tests.py index c0922aa..26a5211 100644 --- 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 index b83c900..03a701c 100644 --- 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