q-dev: check identity

Author: piotrbartman

qubes/device_protocol.py

@@ -857,14 +857,15 @@ def clone(self, **kwargs):
     def device(self) -> DeviceInfo:
         """Get DeviceInfo object corresponding to this DeviceAssignment"""
         dev = self.backend_domain.devices[self.devclass][self.ident]
-        if (self.device_identity is not None
-                and self.device_identity != dev.self_identity):
+        # TODO: device identity could not match
+        # if (self.device_identity is not None
+        #         and self.device_identity != dev.self_identity):
             # raise ProtocolError(
             #     "Device identity does not match, expected "
             #     f"'{self.device_identity}' got '{dev.self_identity}'")
             # TODO
-            return UnknownDevice(
-                self.backend_domain, self.ident, devclass=self.devclass)
+            # return UnknownDevice(
+            #     self.backend_domain, self.ident, devclass=self.devclass)
         return dev
 
     @property

qubes/ext/block.py

@@ -502,15 +502,6 @@ def pre_attachment_internal(
                 raise qubes.exc.QubesValueError(
                     'Unsupported option {}'.format(option))
 
-        # identity = value  # TODO!
-        # if identity not in ('any', device.self_identity):
-        #     print("Unrecognized identity, skipping attachment of"
-        #           f" {device}", file=sys.stderr)
-        #     raise qubes.devices.UnrecognizedDevice(
-        #         f"Device presented identity {device.self_identity} "
-        #         f"does not match expected {identity}"
-        #     )
-
         if 'read-only' not in options:
             options['read-only'] = 'yes' if device.mode == 'r' else 'no'
         if options.get('read-only', 'no') == 'no' and device.mode == 'r':
@@ -550,19 +541,34 @@ def pre_attachment_internal(
     async def on_domain_start(self, vm, _event, **_kwargs):
         # pylint: disable=unused-argument
         for assignment in vm.devices['block'].get_assigned_devices():
-            self.notify_auto_attached(
-                vm, assignment.device, assignment.options)
+            if assignment.mode == qubes.device_protocol.AssignmentMode.ASK:
+                pass
+            self.notify_auto_attached(vm, assignment)
+
+    def notify_auto_attached(self, vm, assignment):
+        identity = assignment.device_ientity
+        device = assignment.device
+        if identity not in ('any', device.self_identity):
+            print("Unrecognized identity, skipping attachment of device in port"
+                  f" {assignment}", file=sys.stderr)
+            raise qubes.devices.UnrecognizedDevice(
+                f"Device presented identity {device.self_identity} "
+                f"does not match expected {identity}"
+            )
 
-    def notify_auto_attached(self, vm, device, options):
         self.pre_attachment_internal(
-            vm, device, options, expected_attachment=vm)
+            vm, device, assignment.options, expected_attachment=vm)
+
         asyncio.ensure_future(vm.fire_event_async(
-            'device-attach:block', device=device, options=options))
+            'device-attach:block',
+            device=device,
+            options=assignment.options,
+        ))
 
-    async def attach_and_notify(self, vm, device, options):
+    async def attach_and_notify(self, vm, assignment):
         # bypass DeviceCollection logic preventing double attach
         # we expected that these devices are already attached to this vm
-        self.notify_auto_attached(vm, device, options)
+        self.notify_auto_attached(vm, assignment)
 
     @qubes.ext.handler('domain-shutdown')
     async def on_domain_shutdown(self, vm, event, **_kwargs):

qubes/ext/utils.py

@@ -100,8 +100,7 @@ def device_list_change(
         else:
             target = tuple(frontends.keys())[0]
         assignment = frontends[target]
-        asyncio.ensure_future(ext.attach_and_notify(
-            target, assignment.device, assignment.options))
+        asyncio.ensure_future(ext.attach_and_notify(target, assignment))
 
 
 def compare_device_cache(vm, devices_cache, current_devices):

qubes/vm/__init__.py

@@ -285,31 +285,39 @@ def load_extras(self):
                     options[option.get('name')] = str(option.text)
 
                 try:
+                    import sys  # TODO
+                    print(f'{self.name=}', file=sys.stderr)  # TODO
                     # backward compatibility: persistent~>required=True
                     legacy_required = node.get('required', 'absent')
+                    print(f'{legacy_required=}', file=sys.stderr)  # TODO
                     if legacy_required == 'absent':
                         mode_str = node.get('mode', 'required')
                         try:
                             mode = (qubes.device_protocol.
                                     AssignmentMode(mode_str))
+                            print(f'{mode=}', file=sys.stderr)  # TODO
                         except ValueError:
                             self.log.error(
                                 "Unrecognized assignment mode, ignoring.")
                             continue
                     else:
                         required = qubes.property.bool(
                             None, None, legacy_required)
+                        print(f'{required=}', file=sys.stderr)  # TODO
                         if required:
                             mode = (qubes.device_protocol.
                                     AssignmentMode.REQUIRED)
+                            print(f'{mode=}', file=sys.stderr)  # TODO
                         else:
                             mode = (qubes.device_protocol.
                                     AssignmentMode.AUTO)
+                            print(f'{mode=}', file=sys.stderr)  # TODO
                     if 'identity' in options:
                         identity = options.get('identity')
                         del options['identity']
                     else:
                         identity = node.get('identity', 'any')
+                    print(mode.value, file=sys.stderr)  # TODO
                     device_assignment = qubes.device_protocol.DeviceAssignment(
                         qubes.device_protocol.Port(
                             backend_domain=self.app.domains[