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)