Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

update to new device API #184

Merged
merged 4 commits into from
Jun 17, 2024
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 5 additions & 3 deletions qubes_config/global_config/usb_devices.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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

Expand All @@ -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(
Expand Down
14 changes: 14 additions & 0 deletions qubes_config/tests/conftest.py
Original file line number Diff line number Diff line change
Expand Up @@ -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


Expand Down Expand Up @@ -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


Expand Down Expand Up @@ -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


Expand Down
42 changes: 30 additions & 12 deletions qubes_config/tests/test_usb_devices.py
Original file line number Diff line number Diff line change
Expand Up @@ -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)

Expand All @@ -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)

Expand Down
10 changes: 4 additions & 6 deletions qui/devices/backend.py
Original file line number Diff line number Diff line change
Expand Up @@ -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(
Expand Down Expand Up @@ -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(
Expand Down
5 changes: 3 additions & 2 deletions qui/devices/device_widget.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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)
Expand Down