q-dev: fixes

doc/manpages/qvm-device.rst

@@ -116,11 +116,11 @@ Assign the device with *DEVICE_ID* from *BACKEND_DOMAIN* to the domain *VMNAME*
 
    Assign device but always ask before auto-attachment.
 
-.. option:: --port
+.. option:: --port, --only-port
 
    Ignore device presented identity and attach any device connected to the given port number.
 
-.. option:: --device
+.. option:: --device, --only-device
 
    Ignore current port identity and attach this device connected to any port.
 
@@ -134,6 +134,14 @@ unassign
 Remove assignment of device with *BACKEND_DOMAIN:DEVICE_ID* from domain *VMNAME*.
 If no device is given, remove assignments of all *DEVICE_CLASS* devices.
 
+.. option:: --port, --only-port
+
+   Remove port assignment.
+
+.. option:: --device, --only-device
+
+   Remove device identity based assignment.
+
 aliases: u
 
 info

qubesadmin/device_protocol.py

@@ -165,7 +165,7 @@ def parse_basic_device_properties(
             the expected values.
         """
         expected = expected_device.port
-        exp_vm_name = expected.backend_domain.name
+        exp_vm_name = expected.backend_name
         if properties.get('backend_domain', exp_vm_name) != exp_vm_name:
             raise UnexpectedDeviceProperty(
                 f"Got device exposed by {properties['backend_domain']}"
@@ -284,7 +284,7 @@ def __str__(self):
 
     @property
     def backend_name(self) -> str:
-        if self.backend_domain is not None:
+        if self.backend_domain not in (None, "*"):
             return self.backend_domain.name
         return "*"
 
@@ -360,7 +360,7 @@ def __init__(
             port: Optional[Port] = None,
             device_id: Optional[str] = None,
     ):
-        # TODO! one of them cannot be None
+        assert port is not None or device_id is not None
         self.port: Optional[Port] = port
         self._device_id = device_id
 
@@ -517,8 +517,8 @@ def _parse(
         else:
             identity = representation
         port_id, _, devid = identity.partition(':')
-        if devid in ('', '*'):
-            devid = '*'
+        if devid == '':
+            devid = None
         return cls(
             Port(backend_domain=backend, port_id=port_id, devclass=devclass),
             device_id=devid
@@ -1054,7 +1054,7 @@ def __init__(
             mode: Union[str, AssignmentMode] = "manual",
     ):
         if isinstance(device, DeviceInfo):
-            device = VirtualDevice(device.port, device._device_id)
+            device = VirtualDevice(device.port, device.device_id)
         self.virtual_device = device
         self.__options = options or {}
         if isinstance(mode, AssignmentMode):
@@ -1164,7 +1164,7 @@ def attached(self) -> bool:
         Returns False if device is attached to different domain
         """
         for device in self.devices:
-            if device.attachment == self.frontend_domain:
+            if device.attachment and device.attachment == self.frontend_domain:
                 return True
         return False
 
@@ -1256,7 +1256,7 @@ def _deserialize(
     def matches(self, device: VirtualDevice) -> bool:
         if self.devclass != device.devclass:
             return False
-        if self.backend_domain != '*' and self.backend_domain != device.backend_domain:
+        if self.backend_domain != device.backend_domain:
             return False
         if self.port_id != '*' and self.port_id != device.port_id:
             return False

qubesadmin/tests/tools/qvm_device.py

@@ -199,7 +199,8 @@ def test_012_attach_invalid(self):
                 qubesadmin.tools.qvm_device.main(
                     ['testclass', 'attach', '-p', 'test-vm2', 'dev1'],
                     app=self.app)
-            self.assertIn('expected a backend vm & device id',
+            self.assertIn(
+                'expected a backend vm, port id and [optional] device id',
                 stderr.getvalue())
         self.assertAllCalled()
 
@@ -221,7 +222,7 @@ def test_014_attach_invalid_backend(self):
                 qubesadmin.tools.qvm_device.main(
                     ['testclass', 'attach', '-p', 'test-vm2', 'no-such-vm:dev3'],
                     app=self.app)
-            self.assertIn('no backend vm',
+            self.assertIn('no such backend vm!',
                 stderr.getvalue())
         self.assertAllCalled()
 
@@ -311,7 +312,8 @@ def test_033_assign_invalid(self):
                 qubesadmin.tools.qvm_device.main(
                     ['testclass', 'assign', 'test-vm2', 'dev1'],
                     app=self.app)
-            self.assertIn('expected a backend vm & device id',
+            self.assertIn(
+                'expected a backend vm, port id and [optional] device id',
                 stderr.getvalue())
         self.assertAllCalled()
 
@@ -332,14 +334,14 @@ def test_035_assign_invalid_backend(self):
                 qubesadmin.tools.qvm_device.main(
                     ['testclass', 'assign', 'test-vm2', 'no-such-vm:dev3'],
                     app=self.app)
-            self.assertIn('no backend vm', stderr.getvalue())
+            self.assertIn('no such backend vm!', stderr.getvalue())
         self.assertAllCalled()
 
     def test_040_unassign(self):
         """ Test unassign action """
         self.app.expected_calls[
             ('test-vm2', 'admin.vm.device.testclass.Unassign',
-             'test-vm1+dev1:*', None)] = b'0\0'
+             'test-vm1+dev1:dead:beef:babe:u0123456', None)] = b'0\0'
         qubesadmin.tools.qvm_device.main(
             ['testclass', 'unassign', 'test-vm2', 'test-vm1:dev1'], app=self.app)
         self.assertAllCalled()
@@ -350,7 +352,7 @@ def test_041_unassign_unknown(self):
             ('test-vm2', 'admin.vm.device.testclass.Unassign',
              'test-vm1+dev7:*', None)] = b'0\0'
         qubesadmin.tools.qvm_device.main(
-            ['testclass', 'unassign', 'test-vm2', 'test-vm1:dev7'], app=self.app)
+            ['testclass', 'unassign', 'test-vm2', 'test-vm1:dev7', '--port'], app=self.app)
         self.assertAllCalled()
 
     def test_042_unassign_all(self):

qubesadmin/tools/qvm_device.py

@@ -33,7 +33,7 @@
 import qubesadmin.tools
 import qubesadmin.device_protocol
 from qubesadmin.device_protocol import (Port, DeviceInfo, UnknownDevice,
-                                        DeviceAssignment)
+                                        DeviceAssignment, VirtualDevice)
 
 
 def prepare_table(dev_list):
@@ -206,7 +206,7 @@ def detach_device(args):
     if args.device:
         device = args.device
         # ignore device id, detach any device
-        device.device_id = None
+        device.device_id = '*'
         assignment = DeviceAssignment(device)
         vm.devices[args.devclass].detach(assignment)
     else:
@@ -221,9 +221,10 @@ def assign_device(args):
     vm = args.domains[0]
     device = args.device
     if args.only_port:
-        device.device_id = None
+        device = device.clone(device_id="*")
     if args.only_device:
-        device.port = Port(None, None, device.devclass)
+        device = device.clone(
+            port=Port(device.backend_domain, "*", device.devclass))
     options = dict(opt.split('=', 1) for opt in args.option or [])
     if args.ro:
         options['read-only'] = 'yes'
@@ -233,12 +234,10 @@ def assign_device(args):
         mode = 'required'
     if args.ask:
         mode = 'ask-to-attach'
-    assignment = DeviceAssignment(
-        device,
-        mode=mode,
-        options=options
-    )
+    assignment = DeviceAssignment(device, mode=mode, options=options)
     vm.devices[args.devclass].assign(assignment)
+    # retrieve current port info
+    assignment = DeviceAssignment(args.device, mode=mode, options=options)
     if vm.is_running() and not assignment.attached and not args.quiet:
         print("Assigned. To attach you can now restart domain or run: \n"
               f"\tqvm-{assignment.devclass} attach {vm} "
@@ -252,10 +251,12 @@ def unassign_device(args):
     vm = args.domains[0]
     if args.device:
         device = args.device
-        # ignore device id, detach any device
-        device.device_id = None
-        assignment = DeviceAssignment(
-            device, frontend_domain=vm)
+        if args.only_port:
+            device = device.clone(device_id="*")
+        if args.only_device:
+            device = device.clone(
+                port=Port(device.backend_domain, '*', device.devclass))
+        assignment = DeviceAssignment(device, frontend_domain=vm)
         _unassign_and_show_message(assignment, vm, args)
     else:
         for assignment in vm.devices[args.devclass].get_assigned_devices():
@@ -306,11 +307,11 @@ def init_list_parser(sub_parsers):
 
 class DeviceAction(qubesadmin.tools.QubesAction):
     """ Action for argument parser that gets the
-        :py:class:``qubesadmin.device.Device`` from a
-        BACKEND:DEVICE_ID string.
+        :py:class:``qubesadmin.device_protocol.VirtualDevice`` from a
+        BACKEND:PORT_ID:DEVICE_ID string.
     """  # pylint: disable=too-few-public-methods
 
-    def __init__(self, help='A backend & device id combination',
+    def __init__(self, help='A backend, port & device id combination',
                  required=True, allow_unknown=False, **kwargs):
         # pylint: disable=redefined-builtin
         self.allow_unknown = allow_unknown
@@ -322,34 +323,36 @@ def __call__(self, parser, namespace, values, option_string=None):
 
     def parse_qubes_app(self, parser, namespace):
         app = namespace.app
-        backend_device_id = getattr(namespace, self.dest)
+        representation = getattr(namespace, self.dest)
         devclass = namespace.devclass
-        if backend_device_id is None:
+        if representation is None:
             return
 
         try:
-            vmname, port_id = backend_device_id.split(':', 1)
-            vm = None
             try:
-                vm = app.domains[vmname]
+                dev = VirtualDevice.from_str(
+                    representation, devclass, app.domains)
             except KeyError:
-                parser.error_runtime("no backend vm {!r}".format(vmname))
+                parser.error_runtime("no such backend vm!")
+                return
 
             try:
-                dev = vm.devices[devclass][port_id]
-                if not self.allow_unknown and \
-                        isinstance(dev, UnknownDevice):
-                    raise KeyError(port_id)
+                # load device info
+                _dev = dev.backend_domain.devices[devclass][dev.port_id]
+                if dev._device_id is None or dev.device_id == _dev.device_id:
+                    dev = _dev
+                if not self.allow_unknown and isinstance(dev, UnknownDevice):
+                    raise KeyError(dev.port_id)
             except KeyError:
                 parser.error_runtime(
-                    f"backend vm {vmname!r} doesn't expose "
-                    f"{devclass} device {port_id!r}")
-                dev = UnknownDevice(Port(vm, port_id, devclass))
+                    f"backend vm {dev.backend_name} doesn't expose "
+                    f"{devclass} device {dev.port_id!r}")
+                dev = UnknownDevice.from_device(dev)
             setattr(namespace, self.dest, dev)
         except ValueError:
             parser.error(
-                'expected a backend vm & device id combination like foo:bar '
-                'got %s' % backend_device_id)
+                'expected a backend vm, port id and [optional] device id '
+                f'combination like foo:bar[:baz] got {representation}')
 
 
 def get_parser(device_class=None):
@@ -455,20 +458,18 @@ def get_parser(device_class=None):
                                   "be required to the qube's startup and then"
                                   " automatically attached)")
 
-    id_parser = assign_parser.add_mutually_exclusive_group()
-    id_parser.add_argument('--port',
-                           dest='only_port',
-                           action='store_true',
-                           default=False,
-                           help="Ignore device presented identity and "
-                                "attach later any device connected "
-                                "to the given port")
-    id_parser.add_argument('--device',
-                           dest='only_device',
-                           action='store_true',
-                           default=False,
-                           help="Ignore current port identity and "
-                                "attach this device connected to any port")
+    for pars in (assign_parser, unassign_parser):
+        id_parser = pars.add_mutually_exclusive_group()
+        id_parser.add_argument('--port', '--only-port',
+                               dest='only_port',
+                               action='store_true',
+                               default=False,
+                               help="Ignore device presented identity")
+        id_parser.add_argument('--device', '--only-device',
+                               dest='only_device',
+                               action='store_true',
+                               default=False,
+                               help="Ignore current port identity")
     attach_parser.set_defaults(func=attach_device)
     detach_parser.set_defaults(func=detach_device)
     assign_parser.set_defaults(func=assign_device)