q-dev: matches

qubes/device_protocol.py

@@ -1213,3 +1213,14 @@ def _deserialize(
         properties['device'] = expected_device
 
         return cls(**properties)
+
+    def matches(self, device: VirtualDevice) -> bool:
+        if self.backend_domain != '*' and self.backend_domain != device.backend_domain:
+            return False
+        if self.port_id != '*' and self.port_id != device.port_id:
+            return False
+        if self.devclass != '*' and self.devclass != device.devclass:
+            return False
+        if self.device_id != '*' and self.device_id != device.device_id:
+            return False
+        return True

qubes/ext/block.py

@@ -545,9 +545,8 @@ async def on_domain_start(self, vm, _event, **_kwargs):
             self.notify_auto_attached(vm, assignment)
 
     def notify_auto_attached(self, vm, assignment):
-        identity = assignment.device_id
         for device in assignment.devices:
-            if identity not in ('*', device.device_id):
+            if not assignment.matches(device):
                 print("Unrecognized identity, skipping attachment of device "
                       f"from the port {assignment}", file=sys.stderr)
                 continue

qubes/ext/utils.py

@@ -18,7 +18,6 @@
 # along with this program; if not, write to the Free Software
 # Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301,
 # USA.
-import importlib
 import asyncio
 import subprocess
 
@@ -67,8 +66,7 @@ def device_list_change(
             continue
         for assignment in front_vm.devices[devclass].get_assigned_devices():
             for device in assignment.devices:
-                if (device.backend_domain == vm
-                        and assignment.device_id == device.device_id
+                if (assignment.matches(device)
                         and device.port_id in added
                         and device.port_id not in attached
                 ):

qubes/tests/devices_block.py

@@ -731,7 +731,7 @@ def test_062_on_qdb_change_attached(self):
             '/qubes-block-devices/sda/size': b'1024000',
             '/qubes-block-devices/sda/mode': b'r',
         }, domain_xml=domain_xml_template.format(""))
-        exp_dev =  qubes.ext.block.BlockDevice(back_vm, 'sda')
+        exp_dev = qubes.ext.block.BlockDevice(back_vm, 'sda')
 
         self.ext.devices_cache = {'sys-usb': {'sda': None}}