q-dev: add device_identity to device assignment

Author: piotrbartman

qubesadmin/device_protocol.py

@@ -842,7 +842,7 @@ class DeviceAssignment(Port):
     def __init__(
             self,
             port: Port,
-            device_id=None,
+            device_identity=None,
             frontend_domain=None,
             options=None,
             mode: Union[str, AssignmentMode] = "manual",
@@ -854,25 +854,33 @@ def __init__(
         else:
             self.mode = AssignmentMode(mode)
         self.frontend_domain = frontend_domain
+        self.device_identity = device_identity
 
     def clone(self, **kwargs):
         """
         Clone object and substitute attributes with explicitly given.
         """
+        port = kwargs.get(
+            "port", Port(self.backend_domain, self.ident, self.devclass))
         attr = {
             "options": self.options,
-            "required": self.required,
-            "attach_automatically": self.attach_automatically,
+            "mode": self.mode,
+            "device_identity": self.device_identity,
             "frontend_domain": self.frontend_domain,
         }
         attr.update(kwargs)
-        return self.__class__(
-            Port(self.backend_domain, self.ident, self.devclass), **attr)
+        return self.__class__(port, **attr)
 
     @property
     def device(self) -> DeviceInfo:
         """Get DeviceInfo object corresponding to this DeviceAssignment"""
-        return self.backend_domain.devices[self.devclass][self.ident]
+        dev = self.backend_domain.devices[self.devclass][self.ident]
+        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}'")
+        return dev
 
     @property
     def frontend_domain(self) -> Optional[QubesVM]:
@@ -917,6 +925,7 @@ def attach_automatically(self) -> bool:
         """
         return self.mode in (
             AssignmentMode.AUTO,
+            AssignmentMode.ASK,
             AssignmentMode.REQUIRED
         )
 
@@ -942,6 +951,7 @@ def serialize(self) -> bytes:
             self.pack_property(key, value)
             for key, value in (
                 ('mode', self.mode.value),
+                ('device_identity', self.device_identity),
                 ('ident', self.ident),
                 ('devclass', self.devclass)))
 

qubesadmin/tests/tools/qvm_device.py

@@ -262,7 +262,7 @@ def test_030_assign(self):
             'test-vm2', 'admin.vm.device.testclass.Assign', 'test-vm1+dev1',
             b"required='no' attach_automatically='yes' ident='dev1' "
             b"devclass='testclass' backend_domain='test-vm1' "
-            b"frontend_domain='test-vm2' _identity='0000:0000::?******'"
+            b"frontend_domain='test-vm2' device_identity='0000:0000::?******'"
         )] = b'0\0'
         qubesadmin.tools.qvm_device.main(
             ['testclass', 'assign', 'test-vm2', 'test-vm1:dev1'], app=self.app)
@@ -275,7 +275,7 @@ def test_031_assign_required(self):
             'test-vm2', 'admin.vm.device.testclass.Assign', 'test-vm1+dev1',
             b"required='yes' attach_automatically='yes' ident='dev1' "
             b"devclass='testclass' backend_domain='test-vm1' "
-            b"frontend_domain='test-vm2' _identity='0000:0000::?******'"
+            b"frontend_domain='test-vm2' device_identity='0000:0000::?******'"
         )] = b'0\0'
         qubesadmin.tools.qvm_device.main(
             ['testclass', 'assign', '--required', 'test-vm2', 'test-vm1:dev1'], app=self.app)
@@ -289,7 +289,7 @@ def test_032_assign_options(self):
             b"required='no' attach_automatically='yes' ident='dev1' "
             b"devclass='testclass' backend_domain='test-vm1' "
             b"frontend_domain='test-vm2' _read-only='yes' "
-            b"_identity='0000:0000::?******'")] = b'0\0'
+            b"device_identity='0000:0000::?******'")] = b'0\0'
         with qubesadmin.tests.tools.StdoutBuffer() as buf:
             qubesadmin.tools.qvm_device.main(
                 ['testclass', 'assign', '--ro', 'test-vm2', 'test-vm1:dev1'],

qubesadmin/tools/qvm_device.py

@@ -207,16 +207,17 @@ def assign_device(args):
     """
     vm = args.domains[0]
     device = args.device
-    assignment = DeviceAssignment(
-        device, mode='required' if args.required else 'auto-attach')
+    identity = device.self_identity if not args.port else None
     options = dict(opt.split('=', 1) for opt in args.option or [])
     if args.ro:
         options['read-only'] = 'yes'
     parse_ro_option_as_read_only(options)
-    options['identity'] = device.self_identity
-    if args.port:
-        options['identity'] = 'any'
-    assignment.options = options
+    assignment = DeviceAssignment(
+        device,
+        device_identity=identity,
+        mode='required' if args.required else 'auto-attach',
+        options=options
+    )
     vm.devices[args.devclass].assign(assignment)
     if vm.is_running() and not assignment.attached and not args.quiet:
         print("Assigned. To attach you can now restart domain or run: \n"