diff --git a/qubes_config/global_config/usb_devices.py b/qubes_config/global_config/usb_devices.py index 23a5257a..57c1722a 100644 --- a/qubes_config/global_config/usb_devices.py +++ b/qubes_config/global_config/usb_devices.py @@ -24,6 +24,7 @@ from typing import List, Union, Optional, Dict, Tuple, Set from qrexec.policy.parser import Allow +from qubesadmin.device_protocol import DeviceCategory from ..widgets.gtk_widgets import TokenName, TextModeler, VMListModeler from ..widgets.utils import get_feature, apply_feature_change @@ -628,7 +629,7 @@ def __init__(self, qapp: qubesadmin.Qubes, policy_manager: PolicyManager, gtk_builder: Gtk.Builder - ): + ): self.qapp = qapp self.policy_manager = policy_manager @@ -637,8 +638,9 @@ def __init__(self, usb_qubes: Set[qubesadmin.vm.QubesVM] = set() for vm in self.qapp.domains: - for device in vm.devices['pci'].attached(): - if device.description.startswith('USB controller'): + for assignment in vm.devices['pci'].get_attached_devices(): + cats = [infc.category for infc in assignment.device.interfaces] + if DeviceCategory.PCI_USB in cats: usb_qubes.add(vm) self.input_handler = InputDeviceHandler( diff --git a/qubes_config/tests/conftest.py b/qubes_config/tests/conftest.py index 85ae48b4..103e077c 100644 --- a/qubes_config/tests/conftest.py +++ b/qubes_config/tests/conftest.py @@ -307,6 +307,10 @@ def test_qapp_impl(): add_feature_to_all(qapp, 'service.qubes-u2f-proxy', ['test-vm']) + for vm in qapp.domains: + qapp.expected_calls[ + (vm.name, 'admin.vm.device.pci.Attached', None, None)] = b'0\x00' + return qapp @@ -380,6 +384,10 @@ def test_qapp_simple(): # pylint: disable=redefined-outer-name {"netvm": ("vm", False, '')}, {'service.qubes-update-check': None}, []) + for vm in qapp.domains: + qapp.expected_calls[ + (vm.name, 'admin.vm.device.pci.Attached', None, None)] = b'0\x00' + return qapp @@ -421,6 +429,12 @@ def test_qapp_broken(): # pylint: disable=redefined-outer-name 'gui-default-secure-copy-sequence': None, 'gui-default-secure-paste-sequence': None }, []) + + # + for vm in qapp.domains: + qapp.expected_calls[ + (vm.name, 'admin.vm.device.pci.Attached', None, None)] = b'0\x00' + return qapp diff --git a/qubes_config/tests/test_usb_devices.py b/qubes_config/tests/test_usb_devices.py index ac7667d1..c766200c 100644 --- a/qubes_config/tests/test_usb_devices.py +++ b/qubes_config/tests/test_usb_devices.py @@ -843,13 +843,19 @@ def test_u2f_handler_add_without_service(test_qapp, def test_devices_handler_unsaved(test_qapp, test_policy_manager, real_builder): - test_qapp.expected_calls[('sys-usb', "admin.vm.device.pci.List", + test_qapp.expected_calls[('sys-usb', "admin.vm.device.pci.Attached", None, None)] = \ - b"0\x00dom0+00_0d.0 no-strict-reset=True persistent=yes\n" + b"0\x00dom0+00_0d.0 ident='00_0d.0' devclass='pci' " \ + b"backend_domain='dom0' required='yes' attach_automatically='yes' " \ + b"_no-strict-reset='yes'\n" test_qapp.expected_calls[('dom0', "admin.vm.device.pci.Available", None, None)] = \ - b"0\x0000_0d.0 function=0 bus=00 libvirt_name=pci_0000_00_0d_0 " \ - b"device=0d description=USB controller: Intel Corporation \n" + b"0\x0000_0d.0 ident='00_0d.0' devclass='pci' backend_domain='dom0' " \ + b"serial='unknown' manufacturer='unknown' " \ + b"self_identity='0000:0000::p0c0300' vendor='unknown' " \ + b"product='unknown' name='unknown' interfaces='p0c0300' " \ + b"_function='0' _bus='00' _libvirt_name='pci_0000_00_0d_0' " \ + b"_device='0d'\n" handler = DevicesHandler(test_qapp, test_policy_manager, real_builder) @@ -870,18 +876,30 @@ def test_devices_handler_unsaved(test_qapp, test_policy_manager, real_builder): def test_devices_handler_detect_usbvms(test_qapp, test_policy_manager, real_builder): - test_qapp.expected_calls[('sys-usb', "admin.vm.device.pci.List", + test_qapp.expected_calls[('sys-usb', "admin.vm.device.pci.Attached", None, None)] = \ - b"0\x00dom0+00_0d.0 no-strict-reset=True persistent=yes\n" - test_qapp.expected_calls[('test-standalone', "admin.vm.device.pci.List", + b"0\x00dom0+00_0d.0 ident='00_0d.0' devclass='pci' " \ + b"backend_domain='dom0' required='yes' attach_automatically='yes' " \ + b"_no-strict-reset='yes'\n" + test_qapp.expected_calls[('test-standalone', "admin.vm.device.pci.Attached", None, None)] = \ - b"0\x00dom0+00_0f.0 no-strict-reset=True persistent=yes\n" + b"0\x00dom0+00_0f.0 ident='00_0f.0' devclass='pci' " \ + b"backend_domain='dom0' required='yes' attach_automatically='yes' " \ + b"_no-strict-reset='yes'\n" test_qapp.expected_calls[('dom0', "admin.vm.device.pci.Available", None, None)] = \ - b"0\x0000_0d.0 function=0 bus=00 libvirt_name=pci_0000_00_0d_0 " \ - b"device=0d description=USB controller: Intel Corporation \n" \ - b"00_0f.0 function=0 bus=00 libvirt_name=pci_0000_00_0d_0 " \ - b"device=0d description=USB controller: Intel Corporation \n" + b"0\x0000_0f.0 ident='00_0f.0' devclass='pci' backend_domain='dom0' " \ + b"serial='unknown' manufacturer='unknown' " \ + b"self_identity='0000:0000::p0c0300' vendor='unknown' " \ + b"product='unknown' name='unknown' interfaces='p0c0300' " \ + b"_function='0' _bus='00' _libvirt_name='pci_0000_00_0f_0' " \ + b"_device='0f'\n" \ + b"00_0d.0 ident='00_0d.0' devclass='pci' backend_domain='dom0' " \ + b"serial='unknown' manufacturer='unknown' " \ + b"self_identity='0000:0000::p0c0300' vendor='unknown' " \ + b"product='unknown' name='unknown' interfaces='p0c0300' " \ + b"_function='0' _bus='00' _libvirt_name='pci_0000_00_0d_0' " \ + b"_device='0d'\n" handler = DevicesHandler(test_qapp, test_policy_manager, real_builder) diff --git a/qui/devices/backend.py b/qui/devices/backend.py index d424edce..1dfaacd9 100644 --- a/qui/devices/backend.py +++ b/qui/devices/backend.py @@ -229,9 +229,8 @@ def attach_to_vm(self, vm: VM): Perform attachment to provided VM. """ try: - assignment = qubesadmin.devices.DeviceAssignment( - self.backend_domain, self.id_string, - persistent=False) + assignment = qubesadmin.device_protocol.DeviceAssignment( + self.backend_domain, self.id_string) vm.vm_object.devices[self.device_class].attach(assignment) self.gtk_app.emit_notification( @@ -261,9 +260,8 @@ def detach_from_vm(self, vm: VM): Gio.NotificationPriority.NORMAL, notification_id=self.notification_id) try: - assignment = qubesadmin.devices.DeviceAssignment( - self.backend_domain, self._ident, - persistent=False) + assignment = qubesadmin.device_protocol.DeviceAssignment( + self.backend_domain, self._ident) vm.vm_object.devices[self.device_class].detach(assignment) except qubesadmin.exc.QubesException as ex: self.gtk_app.emit_notification( diff --git a/qui/devices/device_widget.py b/qui/devices/device_widget.py index 4fcc66f9..386b282d 100644 --- a/qui/devices/device_widget.py +++ b/qui/devices/device_widget.py @@ -179,7 +179,8 @@ def initialize_dev_data(self): for domain in self.qapp.domains: for devclass in DEV_TYPES: try: - for device in domain.devices[devclass].attached(): + for device in domain.devices[devclass + ].get_attached_devices(): dev = str(device) if dev in self.devices: # occassionally ghost UnknownDevices appear when a @@ -227,7 +228,7 @@ def vm_start(self, vm, _event, **_kwargs): for devclass in DEV_TYPES: try: - for device in vm.devices[devclass].attached(): + for device in vm.devices[devclass].get_attached_devices(): dev = str(device) if dev in self.devices: self.devices[dev].attachments.add(wrapped_vm)