q-dev: block devices

port assignment
better block device description
fix extracting partition number
serialize parent devclass
block auto-attach

qubes/devices.py

@@ -434,8 +434,6 @@ def description(self) -> str:
             prod = self.name
         elif self.serial and self.serial != "unknown":
             prod = self.serial
-        elif self.parent_device is not None:
-            return f"sub-device of {self.parent_device}"
         else:
             prod = f"unknown {self.devclass if self.devclass else ''} device"
 
@@ -517,9 +515,12 @@ def serialize(self) -> bytes:
         properties += b' ' + interfaces_prop
 
         if self.parent_device is not None:
-            parent_ident = serialize_str(self.parent_device.ident)
-            parent_prop = (b'parent=' + parent_ident.encode('ascii'))
-            properties += b' ' + parent_prop
+            ident = serialize_str(self.parent_device.ident)
+            ident_prop = (b'parent_ident=' + ident.encode('ascii'))
+            properties += b' ' + ident_prop
+            devclass = serialize_str(self.parent_device.devclass)
+            devclass_prop = (b'parent_devclass=' + devclass.encode('ascii'))
+            properties += b' ' + devclass_prop
 
         data = b' '.join(
             f'_{prop}={serialize_str(value)}'.encode('ascii')
@@ -609,11 +610,14 @@ def _deserialize(
             for i in range(0, len(interfaces), 7)]
         properties['interfaces'] = interfaces
 
-        if 'parent' in properties:
+        if 'parent_ident' in properties:
             properties['parent'] = Device(
                 backend_domain=expected_backend_domain,
-                ident=properties['parent']
+                ident=properties['parent_ident'],
+                devclass=properties['parent_devclass'],
             )
+            del properties['parent_ident']
+            del properties['parent_devclass']
 
         return cls(**properties)
 
@@ -704,6 +708,8 @@ def __init__(self, backend_domain, ident, options=None,
                  required=False, attach_automatically=False):
         super().__init__(backend_domain, ident, devclass)
         self.__options = options or {}
+        if required:
+            assert attach_automatically
         self.__required = required
         self.__attach_automatically = attach_automatically
         self.frontend_domain = frontend_domain
@@ -720,6 +726,18 @@ def clone(self):
             devclass=self.devclass,
         )
 
+    @classmethod
+    def from_device(cls, device: Device, **kwargs) -> 'DeviceAssignment':
+        """
+        Get assignment of the device.
+        """
+        return cls(
+            backend_domain=device.backend_domain,
+            ident=device.ident,
+            devclass=device.devclass,
+            **kwargs
+        )
+
     @property
     def device(self) -> DeviceInfo:
         """Get DeviceInfo object corresponding to this DeviceAssignment"""

qubes/ext/block.py

@@ -23,6 +23,7 @@
 import collections
 import re
 import string
+import sys
 from typing import Optional, List
 
 import lxml.etree
@@ -54,25 +55,61 @@ def __init__(self, backend_domain, ident):
         # lazy loading
         self._mode = None
         self._size = None
-        self._description = None
         self._interface_num = None
 
     @property
-    def description(self):
-        """Human readable device description"""
-        if self._description is None:
-            if not self.backend_domain.is_running():
-                return self.ident
-            safe_set = {ord(c) for c in
-                string.ascii_letters + string.digits + '()+,-.:=_/ '}
-            untrusted_desc = self.backend_domain.untrusted_qdb.read(
-                '/qubes-block-devices/{}/desc'.format(self.ident))
-            if not untrusted_desc:
-                return ''
-            desc = ''.join((chr(c) if c in safe_set else '_')
-                for c in untrusted_desc)
-            self._description = desc
-        return self._description
+    def name(self):
+        """
+        The name of the device it introduced itself with.
+
+        Could be empty string or "unknown".
+        """
+        if self._name is None:
+            name, _ = self._load_lazily_name_and_serial()
+            return name
+        return self._name
+
+    @property
+    def serial(self) -> str:
+        """
+        The serial number of the device it introduced itself with.
+
+        Could be empty string or "unknown".
+
+        Override this method to return proper name directly from device itself.
+        """
+        if self._serial is None:
+            _, serial = self._load_lazily_name_and_serial()
+            return serial
+        return self._serial
+
+    def _load_lazily_name_and_serial(self):
+        if not self.backend_domain.is_running():
+            return "unknown", "unknown"
+        untrusted_desc = self.backend_domain.untrusted_qdb.read(
+            f'/qubes-block-devices/{self.ident}/desc')
+        if not untrusted_desc:
+            return "unknown", "unknown"
+        desc = BlockDevice._sanitize(
+            untrusted_desc,
+            string.ascii_letters + string.digits + '()+,-.:=_/ ')
+        model, _, label = desc.partition(' ')
+        if model:
+            serial = self._serial = model.replace('_', ' ').strip()
+        else:
+            serial = "unknown"
+        # label: '(EXAMPLE)' or '()'
+        if label[1:-1]:
+            name = self._name = label.replace('_', ' ')[1:-1].strip()
+        else:
+            name = "unknown"
+        return name, serial
+
+    @property
+    def manufacturer(self) -> str:
+        if self.parent_device:
+            return f"sub-device of {self.parent_device}"
+        return f"hosted by {self.backend_domain!s}"
 
     @property
     def mode(self):
@@ -142,10 +179,13 @@ def parent_device(self) -> Optional[qubes.devices.Device]:
             else:
                 # '4-4.1:1.0' -> parent_ident='4-4.1', interface_num='1.0'
                 # 'sda' -> parent_ident='sda', interface_num=''
-                parent_ident, _, interface_num = self._sanitize(
+                parent_ident, sep, interface_num = self._sanitize(
                     untrusted_parent_info).partition(":")
+                devclass = 'usb' if sep == ':' else 'block'
+                if not parent_ident:
+                    return None
                 self._parent = qubes.devices.Device(
-                    self.backend_domain, parent_ident)
+                    self.backend_domain, parent_ident, devclass=devclass)
                 self._interface_num = interface_num
         return self._parent
 
@@ -154,8 +194,29 @@ def self_identity(self) -> str:
         """
         Get identification of device not related to port.
         """
-        parent_ident = self.parent_device.ident if self._parent else ''
-        return f'{parent_ident}:{self._interface_num}'
+        parent_identity = ''
+        p = self.parent_device
+        if p is not None:
+            p_info = p.backend_domain.devices[p.devclass][p.ident]
+            parent_identity = p_info.self_identity
+            if p.devclass == 'usb':
+                parent_identity = f'{p.ident}:{parent_identity}'
+        if self._interface_num:
+            # device interface number (not partition)
+            self_id = self._interface_num
+        else:
+            self_id = self._get_possible_partition_number()
+        return f'{parent_identity}:{self_id}'
+
+    def _get_possible_partition_number(self) -> Optional[int]:
+        """
+        If the device is partition return partition number.
+
+        The behavior is undefined for the rest block devices.
+        """
+        # partition number: 'xxxxx12' -> '12' (partition)
+        numbers = re.findall(r'\d+$', self.ident)
+        return int(numbers[-1]) if numbers else None
 
     @staticmethod
     def _sanitize(
@@ -251,6 +312,17 @@ def on_qdb_change(self, vm, event, path):
 
         self.devices_cache[vm.name] = current_devices
 
+        for front_vm in vm.app.domains:
+            if not front_vm.is_running():
+                continue
+            for assignment in front_vm.devices['block'].get_assigned_devices():
+                if (assignment.backend_domain == vm
+                        and assignment.ident in added
+                        and assignment.ident not in attached
+                ):
+                    asyncio.ensure_future(self._attach_and_notify(
+                        front_vm, assignment.device, assignment.options))
+
     def device_get(self, vm, ident):
         '''Read information about device from QubesDB
 
@@ -371,6 +443,10 @@ def find_unused_frontend(self, vm, devtype='disk'):
     @qubes.ext.handler('device-pre-attach:block')
     def on_device_pre_attached_block(self, vm, event, device, options):
         # pylint: disable=unused-argument
+        if isinstance(device, qubes.devices.UnknownDevice):
+            print(f'{device.devclass.capitalize()} device {device} '
+                  'not available, skipping.', file=sys.stderr)
+            return
 
         # validate options
         for option, value in options.items():
@@ -386,6 +462,15 @@ def on_device_pre_attached_block(self, vm, event, device, options):
                     raise qubes.exc.QubesValueError(
                         'devtype option can only have '
                         '\'disk\' or \'cdrom\' value')
+            elif option == 'identity':
+                identity = value
+                if identity != 'any' and device.self_identity != 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}"
+                    )
             else:
                 raise qubes.exc.QubesValueError(
                     'Unsupported option {}'.format(option))
@@ -412,6 +497,23 @@ def on_device_pre_attached_block(self, vm, event, device, options):
             vm.app.env.get_template('libvirt/devices/block.xml').render(
                 device=device, vm=vm, options=options))
 
+    @qubes.ext.handler('domain-start')
+    async def on_domain_start(self, vm, _event, **_kwargs):
+        # pylint: disable=unused-argument
+        for assignment in vm.devices['block'].get_assigned_devices():
+            asyncio.ensure_future(self._attach_and_notify(
+                vm, assignment.device, assignment.options))
+
+    async def _attach_and_notify(self, vm, device, options):
+        # bypass DeviceCollection logic preventing double attach
+        try:
+            self.on_device_pre_attached_block(
+                vm, 'device-pre-attach:block', device, options)
+        except qubes.devices.UnrecognizedDevice:
+            return
+        await vm.fire_event_async(
+            'device-attach:block', device=device, options=options)
+
     @qubes.ext.handler('device-pre-detach:block')
     def on_device_pre_detached_block(self, vm, event, device):
         # pylint: disable=unused-argument

qubes/tests/api_admin.py

@@ -1722,7 +1722,7 @@ def test_462_vm_device_available_invalid(self):
 
     def test_470_vm_device_list_persistent(self):
         assignment = qubes.devices.DeviceAssignment(self.vm, '1234',
-            persistent=True)
+            attach_automatically=True, required=True)
         self.loop.run_until_complete(
             self.vm.devices['testclass'].attach(assignment))
         value = self.call_mgmt_func(b'admin.vm.device.testclass.List',
@@ -1733,11 +1733,11 @@ def test_470_vm_device_list_persistent(self):
 
     def test_471_vm_device_list_persistent_options(self):
         assignment = qubes.devices.DeviceAssignment(self.vm, '1234',
-            persistent=True, options={'opt1': 'value'})
+            attach_automatically=True, required=True, options={'opt1': 'value'})
         self.loop.run_until_complete(
             self.vm.devices['testclass'].attach(assignment))
         assignment = qubes.devices.DeviceAssignment(self.vm, '4321',
-            persistent=True)
+            attach_automatically=True, required=True)
         self.loop.run_until_complete(
             self.vm.devices['testclass'].attach(assignment))
         value = self.call_mgmt_func(b'admin.vm.device.testclass.List',
@@ -1766,7 +1766,7 @@ def test_473_vm_device_list_mixed(self):
         self.vm.add_handler('device-list-attached:testclass',
             self.device_list_attached_testclass)
         assignment = qubes.devices.DeviceAssignment(self.vm, '4321',
-            persistent=True)
+            attach_automatically=True, required=True)
         self.loop.run_until_complete(
             self.vm.devices['testclass'].attach(assignment))
         value = self.call_mgmt_func(b'admin.vm.device.testclass.List',
@@ -1780,7 +1780,7 @@ def test_474_vm_device_list_specific(self):
         self.vm.add_handler('device-list-attached:testclass',
             self.device_list_attached_testclass)
         assignment = qubes.devices.DeviceAssignment(self.vm, '4321',
-            persistent=True)
+            attach_automatically=True, required=True)
         self.loop.run_until_complete(
             self.vm.devices['testclass'].attach(assignment))
         value = self.call_mgmt_func(b'admin.vm.device.testclass.List',
@@ -1944,7 +1944,7 @@ def test_501_vm_remove_running(self, mock_rmtree, mock_remove):
     def test_502_vm_remove_attached(self, mock_rmtree, mock_remove):
         self.setup_for_clone()
         assignment = qubes.devices.DeviceAssignment(
-            self.vm, '1234', persistent=True)
+            self.vm, '1234', attach_automatically=True, required=True)
         self.loop.run_until_complete(
             self.vm2.devices['testclass'].attach(assignment))
 
@@ -2665,7 +2665,7 @@ def test_652_vm_device_set_persistent_false(self):
         self.vm.add_handler('device-list:testclass',
             self.device_list_testclass)
         assignment = qubes.devices.DeviceAssignment(self.vm, '1234', {},
-            True)
+            attach_automatically=True, required=True)
         self.loop.run_until_complete(
             self.vm.devices['testclass'].attach(assignment))
         self.vm.add_handler('device-list-attached:testclass',
@@ -2675,7 +2675,7 @@ def test_652_vm_device_set_persistent_false(self):
         with unittest.mock.patch.object(qubes.vm.qubesvm.QubesVM,
                 'is_halted', lambda _: False):
             value = self.call_mgmt_func(
-                b'admin.vm.device.testclass.Set.persistent',
+                b'admin.vm.device.testclass.Set.persistent',  # TODO
                 b'test-vm1', b'test-vm1+1234', b'False')
         self.assertIsNone(value)
         self.assertNotIn(dev, self.vm.devices['testclass'].get_assigned_devices())
@@ -2686,15 +2686,15 @@ def test_653_vm_device_set_persistent_true_unchanged(self):
         self.vm.add_handler('device-list:testclass',
             self.device_list_testclass)
         assignment = qubes.devices.DeviceAssignment(self.vm, '1234', {},
-            True)
+            attach_automatically=True, required=True)
         self.loop.run_until_complete(
             self.vm.devices['testclass'].attach(assignment))
         self.vm.add_handler('device-list-attached:testclass',
             self.device_list_attached_testclass)
         with unittest.mock.patch.object(qubes.vm.qubesvm.QubesVM,
                 'is_halted', lambda _: False):
             value = self.call_mgmt_func(
-                b'admin.vm.device.testclass.Set.persistent',
+                b'admin.vm.device.testclass.Set.persistent',  # TODO
                 b'test-vm1', b'test-vm1+1234', b'True')
         self.assertIsNone(value)
         dev = qubes.devices.DeviceInfo(self.vm, '1234')

qubes/tests/devices.py

@@ -91,7 +91,8 @@ def setUp(self):
         self.assignment = qubes.devices.DeviceAssignment(
             backend_domain=self.device.backend_domain,
             ident=self.device.ident,
-            persistent=True
+            attach_automatically=True,
+            required=True,
         )
 
     def test_000_init(self):
@@ -220,7 +221,7 @@ def test_001_missing(self):
         assignment = qubes.devices.DeviceAssignment(
             backend_domain=device.backend_domain,
             ident=device.ident,
-            persistent=True)
+            attach_automatically=True, required=True)
         self.loop.run_until_complete(
             self.manager['testclass'].attach(assignment))
         self.assertEventFired(self.emitter, 'device-attach:testclass')