device identity

Makefile

@@ -66,31 +66,31 @@ ADMIN_API_METHODS_SIMPLE = \
 	admin.vm.device.pci.Attached \
 	admin.vm.device.pci.Available \
 	admin.vm.device.pci.Detach \
-	admin.vm.device.pci.Set.required \
+	admin.vm.device.pci.Set.assignment \
 	admin.vm.device.pci.Unassign \
 	admin.vm.device.block.Assign \
 	admin.vm.device.block.Assigned \
 	admin.vm.device.block.Attach \
 	admin.vm.device.block.Attached \
 	admin.vm.device.block.Available \
 	admin.vm.device.block.Detach \
-	admin.vm.device.block.Set.required \
+	admin.vm.device.block.Set.assignment \
 	admin.vm.device.block.Unassign \
 	admin.vm.device.usb.Assign \
 	admin.vm.device.usb.Assigned \
 	admin.vm.device.usb.Attach \
 	admin.vm.device.usb.Attached \
 	admin.vm.device.usb.Available \
 	admin.vm.device.usb.Detach \
-	admin.vm.device.usb.Set.required \
+	admin.vm.device.usb.Set.assignment \
 	admin.vm.device.usb.Unassign \
 	admin.vm.device.mic.Assign \
 	admin.vm.device.mic.Assigned \
 	admin.vm.device.mic.Attach \
 	admin.vm.device.mic.Attached \
 	admin.vm.device.mic.Available \
 	admin.vm.device.mic.Detach \
-	admin.vm.device.mic.Set.required \
+	admin.vm.device.mic.Set.assignment \
 	admin.vm.device.mic.Unassign \
 	admin.vm.feature.CheckWithNetvm \
 	admin.vm.feature.CheckWithTemplate \
@@ -227,7 +227,7 @@ endif
 				  admin.vm.device.testclass.Unassign \
 				  admin.vm.device.testclass.Attached \
 				  admin.vm.device.testclass.Assigned \
-				  admin.vm.device.testclass.Set.required \
+				  admin.vm.device.testclass.Set.assignment \
 				  admin.vm.device.testclass.Available
 	install -d $(DESTDIR)/etc/qubes/policy.d/include
 	install -m 0644 qubes-rpc-policy/admin-local-ro \

doc/qubes-devices.rst

@@ -6,8 +6,8 @@ devices, which can be attached to other domains (frontend). Devices can be of
 different buses (like 'pci', 'usb', etc.). Each device bus is implemented by
 an extension (see :py:mod:`qubes.ext`).
 
-Devices are identified by pair of (backend domain, `ident`), where `ident` is
-:py:class:`str` and can contain only characters from `[a-zA-Z0-9._-]` set.
+Devices are identified by pair of (backend domain, `port_id`), where `port_id`
+is :py:class:`str` and can contain only characters from `[a-zA-Z0-9._-]` set.
 
 
 Device Assignment vs Attachment
@@ -100,17 +100,17 @@ The microphone cannot be assigned (potentially) to any VM (attempting to attach
 Understanding Device Self Identity
 ----------------------------------
 
-It is important to understand that :py:class:`qubes.device_protocol.Device` does not
+It is important to understand that :py:class:`qubes.device_protocol.Port` does not
 correspond to the device itself, but rather to the *port* to which the device
 is connected. Therefore, when assigning a device to a VM, such as
 `sys-usb:1-1.1`, the port `1-1.1` is actually assigned, and thus
 *every* devices connected to it will be automatically attached.
 Similarly, when assigning `vm:sda`, every block device with the name `sda`
-will be automatically attached. We can limit this using :py:meth:`qubes.device_protocol.DeviceInfo.self_identity`, which returns a string containing information
+will be automatically attached. We can limit this using :py:meth:`qubes.device_protocol.DeviceInfo.device_id`, which returns a string containing information
 presented by the device, such as, `vendor_id`, `product_id`, `serial_number`,
-and encoded interfaces. In the case of block devices, `self_identity`
+and encoded interfaces. In the case of block devices, `device_id`
 consists of the parent port to which the device is connected (if any),
-the parent's `self_identity`, and the interface/partition number.
+the parent's `device_id`, and the interface/partition number.
 In practice, this means that, a partition on a USB drive will only be
 automatically attached to a frontend domain if the parent presents
 the correct serial number etc., and is connected to a specific port.

qubes-rpc-policy/90-admin-default.policy.header

@@ -26,14 +26,14 @@
 !include-service admin.vm.device.mic.Attached * include/admin-local-ro
 !include-service admin.vm.device.mic.Available * include/admin-local-ro
 !include-service admin.vm.device.mic.Detach * include/admin-local-rwx
-!include-service admin.vm.device.mic.Set.required * include/admin-local-rwx
+!include-service admin.vm.device.mic.Set.assignment * include/admin-local-rwx
 !include-service admin.vm.device.mic.Unassign * include/admin-local-rwx
 !include-service admin.vm.device.usb.Assign * include/admin-local-rwx
 !include-service admin.vm.device.usb.Assigned * include/admin-local-ro
 !include-service admin.vm.device.usb.Attach * include/admin-local-rwx
 !include-service admin.vm.device.usb.Attached * include/admin-local-ro
 !include-service admin.vm.device.usb.Available * include/admin-local-ro
 !include-service admin.vm.device.usb.Detach * include/admin-local-rwx
-!include-service admin.vm.device.usb.Set.required * include/admin-local-rwx
+!include-service admin.vm.device.usb.Set.assignment * include/admin-local-rwx
 !include-service admin.vm.device.usb.Unassign * include/admin-local-rwx
 

qubes/api/admin.py

@@ -45,7 +45,8 @@
 import qubes.vm
 import qubes.vm.adminvm
 import qubes.vm.qubesvm
-from qubes.device_protocol import Device
+from qubes.device_protocol import (
+    VirtualDevice, UnknownDevice, DeviceAssignment, AssignmentMode)
 
 
 class QubesMgmtEventsDispatcher:
@@ -1211,14 +1212,15 @@
                 raise qubes.exc.QubesException("qubesd shutdown in progress")
             raise
         if self.arg:
-            devices = [dev for dev in devices if dev.ident == self.arg]
+            devices = [dev for dev in devices if dev.port_id == self.arg]
             # no duplicated devices, but device may not exist, in which case
             #  the list is empty
             self.enforce(len(devices) <= 1)
         devices = self.fire_event_for_filter(devices, devclass=devclass)
-        dev_info = {dev.ident: dev.serialize().decode() for dev in devices}
-        return ''.join('{} {}\n'.format(ident, dev_info[ident])
-                       for ident in sorted(dev_info))
+        dev_info = {f'{dev.port_id}:{dev.device_id}':
+                        dev.serialize().decode() for dev in devices}
+        return ''.join(f'{port_id} {dev_info[port_id]}\n'
+                       for port_id in sorted(dev_info))
 
     @qubes.api.method('admin.vm.device.{endpoint}.Assigned', endpoints=(ep.name
             for ep in importlib.metadata.entry_points(group='qubes.devices')),
@@ -1237,7 +1239,7 @@
         if self.arg:
             select_backend, select_ident = self.arg.split('+', 1)
             device_assignments = [dev for dev in device_assignments
-                if (str(dev.backend_domain), dev.ident)
+                if (str(dev.backend_domain), dev.port_id)
                    == (select_backend, select_ident)]
             # no duplicated devices, but device may not exist, in which case
             #  the list is empty
@@ -1246,12 +1248,13 @@
             device_assignments, devclass=devclass)
 
         dev_info = {
-            f'{assignment.backend_domain}+{assignment.ident}':
+            (f'{assignment.backend_domain}'
+             f'+{assignment.port_id}:{assignment.device_id}'):
                 assignment.serialize().decode('ascii', errors="ignore")
             for assignment in device_assignments}
 
-        return ''.join('{} {}\n'.format(ident, dev_info[ident])
-            for ident in sorted(dev_info))
+        return ''.join('{} {}\n'.format(port_id, dev_info[port_id])
+            for port_id in sorted(dev_info))
 
     @qubes.api.method(
         'admin.vm.device.{endpoint}.Attached',
@@ -1272,21 +1275,24 @@
         if self.arg:
             select_backend, select_ident = self.arg.split('+', 1)
             device_assignments = [dev for dev in device_assignments
-                                  if (str(dev.backend_domain), dev.ident)
+                                  if (str(dev.backend_domain), dev.port_id)
                                   == (select_backend, select_ident)]
             # no duplicated devices, but device may not exist, in which case
             #  the list is empty
             self.enforce(len(device_assignments) <= 1)
-        device_assignments = self.fire_event_for_filter(device_assignments,
-                                                        devclass=devclass)
+        device_assignments = [
+            a.clone(device=self.app.domains[a.backend_name
+            ].devices[devclass][a.port_id]) for a in device_assignments]
+        device_assignments = self.fire_event_for_filter(
+            device_assignments, devclass=devclass)
 
         dev_info = {
-            f'{assignment.backend_domain}+{assignment.ident}':
+            (f'{assignment.backend_domain}'
+             f'+{assignment.port_id}:{assignment.device_id}'):
                 assignment.serialize().decode('ascii', errors="ignore")
             for assignment in device_assignments}
-
-        return ''.join('{} {}\n'.format(ident, dev_info[ident])
-                       for ident in sorted(dev_info))
+        return ''.join('{} {}\n'.format(port_id, dev_info[port_id])
+                       for port_id in sorted(dev_info))
 
     # Assign/Unassign action can modify only persistent state of running VM.
     # For this reason, write=True
@@ -1295,11 +1301,7 @@
         scope='local', write=True)
     async def vm_device_assign(self, endpoint, untrusted_payload):
         devclass = endpoint
-
-        # qrexec already verified that no strange characters are in self.arg
-        backend_domain, ident = self.arg.split('+', 1)
-        # may raise KeyError, either on domain or ident
-        dev = self.app.domains[backend_domain].devices[devclass][ident]
+        dev = self.load_device_info(devclass)
 
         assignment = qubes.device_protocol.DeviceAssignment.deserialize(
             untrusted_payload, expected_device=dev
@@ -1315,28 +1317,36 @@
         await self.dest.devices[devclass].assign(assignment)
         self.app.save()
 
+    def load_device_info(self, devclass) -> VirtualDevice:
+        # qrexec already verified that no strange characters are in self.arg
+        _dev = VirtualDevice.from_qarg(self.arg, devclass, self.app.domains)
+        if _dev.port_id == '*' or _dev.device_id == '*':
+            return _dev
+        # load all info, may raise KeyError, either on domain or port_id
+        try:
+            dev = self.app.domains[
+                _dev.backend_domain].devices[devclass][_dev.port_id]
+            if isinstance(dev, UnknownDevice):
+                return _dev
+            return dev
+        except KeyError:
+            return _dev
+
     # Assign/Unassign action can modify only persistent state of running VM.
     # For this reason, write=True
     @qubes.api.method(
         'admin.vm.device.{endpoint}.Unassign',
         endpoints=(
             ep.name
             for ep in importlib.metadata.entry_points(group='qubes.devices')),
-            no_payload=True, scope='local', write=True)
+        no_payload=True, scope='local', write=True)
     async def vm_device_unassign(self, endpoint):
         devclass = endpoint
-
-        # qrexec already verified that no strange characters are in self.arg
-        backend_domain, ident = self.arg.split('+', 1)
-        # may raise KeyError; if a device isn't found, it will be UnknownDevice
-        #  instance - but allow it, otherwise it will be impossible to unassign
-        #  an already removed device
-        dev = self.app.domains[backend_domain].devices[devclass][ident]
+        dev = self.load_device_info(devclass)
+        assignment = DeviceAssignment(dev)
 
         self.fire_event_for_permission(device=dev, devclass=devclass)
 
-        assignment = qubes.device_protocol.DeviceAssignment(
-            dev.backend_domain, dev.ident, devclass)
         await self.dest.devices[devclass].unassign(assignment)
         self.app.save()
 
@@ -1350,20 +1360,13 @@
         scope='local', execute=True)
     async def vm_device_attach(self, endpoint, untrusted_payload):
         devclass = endpoint
-
-        # qrexec already verified that no strange characters are in self.arg
-        backend_domain, ident = self.arg.split('+', 1)
-        # may raise KeyError, either on domain or ident
-        dev = self.app.domains[backend_domain].devices[devclass][ident]
-
-        assignment = qubes.device_protocol.DeviceAssignment.deserialize(
-            untrusted_payload, expected_device=dev
-        )
+        dev = self.load_device_info(devclass)
+        assignment = DeviceAssignment.deserialize(
+            untrusted_payload, expected_device=dev)
 
         self.fire_event_for_permission(
-            device=dev, devclass=devclass,
-            required=assignment.required,
-            attach_automatically=assignment.attach_automatically,
+            device=dev,
+            mode=assignment.mode.value,
             options=assignment.options
         )
 
@@ -1379,23 +1382,16 @@
         no_payload=True, scope='local', execute=True)
     async def vm_device_detach(self, endpoint):
         devclass = endpoint
-
-        # qrexec already verified that no strange characters are in self.arg
-        backend_domain, ident = self.arg.split('+', 1)
-        # may raise KeyError; if device isn't found, it will be UnknownDevice
-        #  instance - but allow it, otherwise it will be impossible to detach
-        #  already removed device
-        dev = self.app.domains[backend_domain].devices[devclass][ident]
+        dev = self.load_device_info(devclass)
 
         self.fire_event_for_permission(device=dev, devclass=devclass)
 
-        assignment = qubes.device_protocol.DeviceAssignment(
-            dev.backend_domain, dev.ident, devclass)
+        assignment = qubes.device_protocol.DeviceAssignment(dev)
         await self.dest.devices[devclass].detach(assignment)
 
     # Assign/Unassign action can modify only a persistent state of running VM.
     # For this reason, write=True
-    @qubes.api.method('admin.vm.device.{endpoint}.Set.required',
+    @qubes.api.method('admin.vm.device.{endpoint}.Set.assignment',
         endpoints=(ep.name
             for ep in importlib.metadata.entry_points(group='qubes.devices')),
         scope='local', write=True)
@@ -1409,20 +1405,20 @@
         """
         devclass = endpoint
 
-        self.enforce(untrusted_payload in (b'True', b'False'))
-        # now is safe to eval, since the value of untrusted_payload is trusted
-        # pylint: disable=eval-used
-        assignment = eval(untrusted_payload)
-        del untrusted_payload
+        allowed_values = {
+            b'required': AssignmentMode.REQUIRED,
+            b'ask-to-attach': AssignmentMode.ASK,
+            b'auto-attach': AssignmentMode.AUTO}
+        try:
+            mode = allowed_values[untrusted_payload]
+        except KeyError:
+            raise qubes.exc.PermissionDenied()
 
-        # qrexec already verified that no strange characters are in self.arg
-        backend_domain_name, ident = self.arg.split('+', 1)
-        backend_domain = self.app.domains[backend_domain_name]
-        dev = Device(backend_domain, ident, devclass)
+        dev = VirtualDevice.from_qarg(self.arg, devclass, self.app.domains)
 
-        self.fire_event_for_permission(device=dev, assignment=assignment)
+        self.fire_event_for_permission(device=dev, mode=mode)
 
-        await self.dest.devices[devclass].update_required(dev, assignment)
+        await self.dest.devices[devclass].update_assignment(dev, mode)
         self.app.save()
 
     @qubes.api.method('admin.vm.firewall.Get', no_payload=True,

qubes/app.py

@@ -1521,7 +1521,7 @@ def on_domain_pre_deleted(self, event, vm):
         assignments = vm.get_provided_assignments()
         if assignments:
             desc = ', '.join(
-                assignment.ident for assignment in assignments)
+                assignment.port_id for assignment in assignments)
             raise qubes.exc.QubesVMInUseError(
                 vm,
                 'VM has devices assigned to other VMs: ' + desc)