From 00dd883d02468ca53a49b0cfd59958061571295c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marek=20Marczykowski-G=C3=B3recki?= Date: Tue, 21 Oct 2025 12:31:57 +0200 Subject: [PATCH] Fix unassigning non-existing devices When device got removed from the system, do not crash when trying to unassign it. Fixes QubesOS/qubes-issues#10302 --- qubes/exc.py | 6 +++ qubes/ext/pci.py | 12 +++++- qubes/tests/devices_pci.py | 77 ++++++++++++++++++++++++++++++++++++-- qubes/utils.py | 6 ++- 4 files changed, 94 insertions(+), 7 deletions(-) diff --git a/qubes/exc.py b/qubes/exc.py index 93dc2b5e7..2a577e44a 100644 --- a/qubes/exc.py +++ b/qubes/exc.py @@ -277,6 +277,12 @@ class DeviceNotAssigned(QubesException, KeyError): """ +class DeviceNotFound(QubesException, KeyError): + """ + Non-existing device. + """ + + class DeviceAlreadyAttached(QubesException, KeyError): """ Trying to attach already attached device. diff --git a/qubes/ext/pci.py b/qubes/ext/pci.py index 0571da337..507640c53 100644 --- a/qubes/ext/pci.py +++ b/qubes/ext/pci.py @@ -34,7 +34,8 @@ import qubes.device_protocol import qubes.devices import qubes.ext -from qubes.device_protocol import Port +from qubes.device_protocol import Port, UnknownDevice +from qubes.exc import DeviceNotFound from qubes.utils import sbdf_to_path, path_to_sbdf, is_pci_path #: cache of PCI device classes @@ -165,6 +166,8 @@ def __init__(self, port: Port, libvirt_name=None): sbdf = path_to_sbdf(port.port_id) else: sbdf = port.port_id + if sbdf is None: + raise DeviceNotFound(port.port_id) dev_match = self.regex.match(sbdf) if not dev_match: raise ValueError( @@ -542,7 +545,12 @@ def on_app_close(self, app, event): _cache_get.cache_clear() +# FIXME: needs to be smarter when pci-hotplug arrives @functools.lru_cache(maxsize=None) def _cache_get(vm, port_id): """Caching wrapper around `PCIDevice(vm, port_id)`.""" - return PCIDevice(Port(vm, port_id, "pci")) + try: + dev = PCIDevice(Port(vm, port_id, "pci")) + except DeviceNotFound: + dev = UnknownDevice(Port(vm, port_id, "pci")) + return dev diff --git a/qubes/tests/devices_pci.py b/qubes/tests/devices_pci.py index 18d783fa8..dab1cd719 100644 --- a/qubes/tests/devices_pci.py +++ b/qubes/tests/devices_pci.py @@ -17,30 +17,39 @@ # # You should have received a copy of the GNU Lesser General Public # License along with this library; if not, see . +import functools import os.path import unittest from unittest import mock import qubes.tests import qubes.ext.pci -from qubes.device_protocol import DeviceInterface +import qubes.devices +from qubes.device_protocol import ( + DeviceInterface, + DeviceAssignment, + VirtualDevice, +) from qubes.utils import sbdf_to_path, path_to_sbdf, is_pci_path orig_open = open class TestVM(object): - def __init__(self, running=True, name="dom0", qid=0): + def __init__(self, running=True, name="dom0", qid=0, app=None): self.name = name self.qid = qid self.is_running = lambda: running self.log = mock.Mock() - self.app = mock.Mock() + self.app = app or mock.Mock() def __eq__(self, other): if isinstance(other, TestVM): return self.name == other.name + def __hash__(self): + return hash(self.name) + PCI_XML = """ pci_{}_00_14_0 @@ -157,6 +166,10 @@ def test_011_path_to_sbdf2(self): path = path_to_sbdf("0000_00_18.4") self.assertEqual(path, "0000:00:18.4") + def test_012_path_to_sbdf_missing(self): + path = path_to_sbdf("0000_c0_03.7-00_00.0-00_00.0") + self.assertEqual(path, None) + def test_020_is_pci_path(self): self.assertTrue(is_pci_path("0000_00_18.4")) @@ -171,6 +184,9 @@ def test_022_is_pci_path_non_00_bus(self): class TC_10_PCI(qubes.tests.QubesTestCase): def setUp(self): super().setUp() + self.dom0 = TestVM(name="dom0", qid=0) + self.app = self.dom0.app + self.app.domains = {"dom0": self.dom0, self.dom0: self.dom0} self.ext = qubes.ext.pci.PCIDeviceExtension() @mock.patch("builtins.open", new=mock_file_open) @@ -218,3 +234,58 @@ def test_000_unsupported_device(self): "Chipset Family USB xHCI Controller", ) self.assertEqual(devices[0].device_id, "0x8086:0x8cb1::p0c0330") + + def _mock_fire_event(self, vm, event, pre_event=False, **kwargs): + if event == "device-get:pci": + return list(self.ext.on_device_get_pci(vm, event, **kwargs)) + elif event.startswith("admin-permission:"): + pass + else: + assert False + + def test_010_unassign_missing(self): + vm = TestVM(name="testvm", qid=1, app=self.app) + vm.app.vmm.offline_mode = False + vm.events_enabled = False + vm.fire_event_async = mock.AsyncMock() + vm.fire_event = functools.partial(self._mock_fire_event, vm) + pci_devices = qubes.devices.DeviceCollection(vm, "pci") + vm.devices = {"pci": pci_devices} + self.dom0.fire_event = functools.partial( + self._mock_fire_event, self.dom0 + ) + dom0_pci_devices = qubes.devices.DeviceCollection(self.dom0, "pci") + self.dom0.devices = {"pci": dom0_pci_devices} + self.app.domains["testvm"] = vm + self.app.domains[vm] = vm + missing_port_id = "0000_c0_03.7-00_00.0-00_00.0" + missing_device_id = "0x8086:0x8cb1::p0c0330" + device_assignment = qubes.device_protocol.DeviceAssignment( + qubes.device_protocol.VirtualDevice( + qubes.device_protocol.Port( + backend_domain=self.dom0, + port_id=missing_port_id, + devclass="pci", + ), + device_id=missing_device_id, + ), + frontend_domain=vm, + options={}, + mode=qubes.device_protocol.AssignmentMode.REQUIRED, + ) + pci_devices.load_assignment(device_assignment) + + vm.events_enabled = True + mgmt_obj = qubes.api.admin.QubesAdminAPI( + self.app, + b"dom0", + b"admin.vm.device.pci.Unassign", + b"testvm", + ("dom0+" + missing_port_id + ":" + missing_device_id).encode(), + ) + + response = self.loop.run_until_complete( + mgmt_obj.execute(untrusted_payload=b"") + ) + self.assertEqual(response, None) + self.assertListEqual(list(pci_devices.get_assigned_devices()), []) diff --git a/qubes/utils.py b/qubes/utils.py index e88d10fef..49e0d0e89 100644 --- a/qubes/utils.py +++ b/qubes/utils.py @@ -365,7 +365,7 @@ def sbdf_to_path(device_id: str): :param device_id: sbdf, for example 0000:02:03.0; accepts also libvirt format like 0000_02_03_0 - :return: converted identifier of None if device is not found + :return: converted identifier or None if device is not found """ regex = re.compile( r"\A(?:pci_)?((?P[0-9a-f]{4})[_:])?(?P[0-9a-f]{2})[_:]" @@ -455,7 +455,9 @@ def path_to_sbdf(path: str): bus_offset = 0 current_dev = "" for path_part in path.split("-"): - assert bus_offset != -1 + if bus_offset == -1: + # no such bus + return None # first part may include segment if bus_offset == 0: segment_match = segment_re.match(path_part)