Redefine DeviceCategories and improvements in MockDevices

- changed DeviceCategory listings to more closely align with
desired GUI list of blockable devices
- improved mocks to include better MockDevices

Author: marmarta

qubesadmin/device_protocol.py

@@ -641,29 +641,31 @@ class DeviceCategory(Enum):
     """
 
     # pylint: disable=invalid-name
-    Other = "*******"
+    Other = "*******"  # also matches all devices, if used to block attachment
 
-    Communication = ("u02****", "p07****")  # eg. modems
+    # The following devices are used in GUI for blocks; take note when changing
+
+    # modems, WiFi and Ethernet adapters
+    Network = ("u02****", "p07****", "p02****")
     Input = ("u03****", "p09****")  # HID etc.
     Keyboard = ("u03**01", "p0900**")
     Mouse = ("u03**02", "p0902**")
     Printer = ("u07****",)
-    Scanner = ("p0903**",)
-    Microphone = ("m******",)
+    Image_Input = ("p0903**", "u06****", "u0e****") # cameras and scanners
+
     # Multimedia = Audio, Video, Displays etc.
-    Multimedia = (
-        "u06****",
-        "u10****",
-        "p03****",
-        "p04****",
-    )
-    Audio = ("p0403**", "p0401**", "u01****")
-    Display = ("p0300**", "p0380**")
-    Video = ("p0400**", "u0e****")
-    Wireless = ("ue0****", "p0d****")
-    Bluetooth = ("ue00101", "p0d11**")
+    Multimedia_Output = ("u10****", "p03****", "p04****")
+    Audio_Output = ("p0403**", "p0401**", "u01****")
+    Audio = ("p0403**", "p0401**", "u01****", "m******")
+    Microphone = ("m******",)
+    USB_Storage = ("u08****")
+    Block_Storage = ("b******")
     Storage = ("b******", "u08****", "p01****")
-    Network = ("p02****",)
+    Bluetooth = ("ue00101", "p0d11**")
+    Smart_Card_Readers = ("u0b****")
+
+    Display = ("p0300**", "p0380**")  # PCI screens?
+    Wireless = ("ue0****", "p0d****")
     Memory = ("p05****",)
     PCI_Bridge = ("p06****",)
     Docking_Station = ("p0a****",)

qubesadmin/tests/mock_app.py

@@ -63,7 +63,7 @@ def main():
 from unittest import mock
 from copy import deepcopy
 
-from typing import List, Optional, Dict, Tuple
+from typing import List, Optional, Dict, Tuple, Any
 
 import qubesadmin.events
 from qubesadmin.tests import QubesTest
@@ -116,6 +116,7 @@ def default_string(self):
     "debug": Property("False", "bool", True),
     "default_dispvm": Property("default-dvm", "vm", False),
     "default_user": Property("user", "str", True),
+    "devices_denied": Property("", "str", True),
     "dns": Property("10.139.1.1 10.139.1.2", "str", True),
     "gateway": Property("", "str", True),
     "gateway6": Property("", "str", True),
@@ -389,9 +390,9 @@ def update_calls(self):
 
         # create all propertyget calls
         for prop, value in self.properties.items():
-            if ((prop == 'template') and \
+            if ((prop == 'template') and
                     self.klass in ("TemplateVM", "StandaloneVM")) \
-                    or ((prop == 'appvm_default_bootmode') and \
+                    or ((prop == 'appvm_default_bootmode') and
                     self.klass in ("AppVM",)):
                 self.qapp.expected_calls[
                     (self.name, "admin.vm.property.Get", prop, None)] = \
@@ -576,59 +577,73 @@ def __init__(self, qapp):
                 continue
 
 
+ # pylint: disable=too-many-positional-arguments
 class MockDevice:
     """helper for adding a device to a qubes test instance"""
-    def __init__(self, qapp: QubesTest, dev_class: str,
-                 description: str, dev_id: str, backend_vm: str,
-                 attached: Optional[str] = None):
+    def __init__(self,
+                 qapp: QubesTest,
+                 dev_class: str,
+                 device_id: str,
+                 backend_vm: str,
+                 port: str,
+                 product: str,
+                 vendor: str,
+                 attached: Optional[str] = None,
+                 assigned: Optional[List[Tuple[str, str, list[Any] | None]]] =
+                 None):
         """
         :param qapp: QubesTest object
         :param dev_class: block / mic / usb
-        :param description: device description
-        :param dev_id: dev id (such as sda, 2-1, mic)
+        :param device_id: device_id
+        :param port: port
         :param backend_vm: name of the vm providing this device
-        :param attached: name of the qube to which the device is attached,
-        if any
+        :param attached: name of the qube to which the device is
+        currently attached,
+        :param assigned: list of the qubes to which the device is currently
+        assigned, tupled with mode ('ask-to-attach' or 'auto-attach')
         """
         self.qapp = qapp
         self.dev_class = dev_class
-        self.description = description
-        self.dev_id = dev_id
+        self.device_id = device_id
+        self.port = port
         self.backend_vm = backend_vm
         self.attached = attached
+        self.assigned = assigned
+        self.product = product
+        self.vendor = vendor
+
+        self.interface = self.device_id.split(":")[-1]
 
         self.update_calls()
 
     def device_string(self):
-        if self.dev_class == 'block':
-            port, d_id = self.dev_id.split(":", 1)
-            return f"{self.dev_id} device_id='{d_id}' port_id='{port}' " \
-                f"devclass='block' backend_domain='{self.backend_vm}' " \
-                f"serial='root/test.img' manufacturer='{self.description}'  " \
-                "interfaces='b******'\n"
-        if self.dev_class == 'pci':
-            _, d_id = self.dev_id.split(":")
-            port = self.dev_id.replace(':', '_')
-            return f"{port} device_id='{d_id}' " \
-                f"port_id='{port}' devclass='pci' " \
-                f"product='{self.description}' vendor='{self.description}' " \
-                f"interfaces='p0c0500' backend_domain='{self.backend_vm}'\n"
-        return f"{self.dev_id} description='{self.description}'\n"
+        return f"{self.port} device_id='{self.device_id}' " \
+               f"port_id='{self.port}' devclass='{self.dev_class}' " \
+               f"product='{self.product}' vendor='{self.vendor}' " \
+               f"interfaces='{self.interface}' " \
+               f"backend_domain='{self.backend_vm}'\n"
 
     def attachment_string(self):
-        if ":" in self.dev_id:
-            port, _ = self.dev_id.split(":")
-        elif self.dev_class == 'mic':
-            port = "mic"
-        else:
-            port = "00"
-        string = (f"{self.backend_vm}+{self.dev_id} "
-                  f"port_id='{port}' devclass='{self.dev_class}' "
-                  f"backend_domain='{self.backend_vm}' mode='required' "
+        string = (f"{self.backend_vm}+{self.port} "
+                  f"port_id='{self.port}' devclass='{self.dev_class}' "
+                  f"backend_domain='{self.backend_vm}' mode='manual' "
                   f"frontend_domain='{self.attached}'")
         string += "\n"
         return string
 
+    def assignment_string(self, vm, mode, opts):
+        string = (f"{self.backend_vm}+{self.port} device_id='"
+                  f"{self.device_id}' "
+                  f"port_id='{self.port}' devclass='{self.dev_class}' "
+                  f"backend_domain='{self.backend_vm}' mode='{mode}' "
+                  f"frontend_domain='{vm}'")
+        if opts:
+            for opt in opts:
+                string += f" _{opt}='True'"
+        string += "\n"
+        return string
+
+
     def update_calls(self):
         # modify call
         current_response = self.qapp.expected_calls[
@@ -641,13 +656,23 @@ def update_calls(self):
 
         if self.attached:
             current_response = self.qapp.expected_calls[
-                (self.attached, f"admin.vm.device.{self.dev_class}.Assigned",
+                (self.attached, f"admin.vm.device.{self.dev_class}.Attached",
                  None, None)]
             self.qapp.expected_calls[
-                (self.attached, f"admin.vm.device.{self.dev_class}.Assigned",
+                (self.attached, f"admin.vm.device.{self.dev_class}.Attached",
                  None, None)] = current_response + \
                 self.attachment_string().encode()
 
+        if self.assigned:
+            for vm, mode, opts in self.assigned:
+                current_response = self.qapp.expected_calls[
+                    (vm, f"admin.vm.device.{self.dev_class}.Assigned",
+                     None, None)]
+                self.qapp.expected_calls[
+                    (vm, f"admin.vm.device.{self.dev_class}.Assigned",
+                     None, None)] = current_response + \
+                    self.assignment_string(vm, mode, opts).encode()
+
 
 class QubesTestWrapper(QubesTest):
     def __init__(self):
@@ -822,16 +847,40 @@ def __init__(self):
 
         # also add a bunch of devices
         self._devices = [
-            MockDevice(self, 'mic', 'Internal Microphone', 'mic', 'dom0',
+            MockDevice(self, dev_class='mic', device_id='dom0:mic::m000000',
+                       backend_vm='dom0', port='mic',
+                       product='Internal Mic', vendor='ACME',
                        attached='test-blue'),
             # the usb stick appears as multiple devices, as they are wont to
-            MockDevice(self, 'usb', 'My USB Drive', '2-1', 'sys-usb'),
-            MockDevice(self, 'block', ' ()', 'sda::0', 'sys-usb'),
-            MockDevice(self, 'block', '(USB DISK)', 'sda1::1', 'sys-usb'),
-            MockDevice(self, 'usb', 'Internal Camera', '2-10', 'sys-usb'),
-            MockDevice(self, 'pci', 'Host bridge', '00:00.0', 'dom0'),
-            MockDevice(self, 'pci', 'PCI Bridge', '00:02.2', 'dom0'),
-            MockDevice(self, 'pci', 'USB Controller', '00:03.2', 'dom0'),
+            MockDevice(self, dev_class='usb',
+                       device_id='1d6b:0104:CAFEBABE:u030101u300000',
+                       backend_vm='sys-usb', port='2-1',
+                       product='My USB Drive', vendor='SCP Foundation'),
+            MockDevice(self, dev_class='block',
+                       device_id='1d6b:0104:CAFEBABE:b123456',
+                       backend_vm='sys-usb', port='sda',
+                       product='(USB)', vendor='SCP Foundation'),
+            MockDevice(self, dev_class='block',
+                       device_id='1d6b:0104:CAFEBABE:b123456',
+                       backend_vm='sys-usb', port='sda',
+                       product='()', vendor='SCP Foundation'),
+            MockDevice(self, dev_class='usb',
+                       device_id='04f2:b684:01.00.00:u0e0101u0e0201',
+                       backend_vm='sys-usb', port='2-7',
+                       product='Internal Camera', vendor='Saruman Industries'),
+
+            MockDevice(self, dev_class='pci',
+                       device_id='0x8086:0x4621::p060000',
+                       backend_vm='dom0', port='00_00.0',
+                       product='PCI Bridge', vendor='Unnamed Industries'),
+            MockDevice(self, dev_class='pci',
+                       device_id='0x8086:0x51e9::p0c8000',
+                       backend_vm='dom0', port='00_02.0',
+                       product='I2C Controller', vendor='Unnamed Industries'),
+            MockDevice(self, dev_class='pci',
+                       device_id='0x8086:0x461e::p0c0330',
+                       backend_vm='dom0', port='00_03.0',
+                       product='I2C Controller', vendor='Unnamed Industries'),
         ]
 
         self.update_vm_calls()

qubesadmin/tests/tools/qvm_device.py

@@ -80,7 +80,7 @@ def test_000_list_all(self):
                 ['testclass', 'list'], app=self.app)
             self.assertEqual(
                 [x.rstrip() for x in buf.getvalue().splitlines()],
-                ['test-vm1:dev1  Audio: itl test-device',
+                ['test-vm1:dev1  Audio_Output: itl test-device',
                  'test-vm2:dev2  ?******: ? ` test-device']
             )
 
@@ -155,7 +155,7 @@ def test_002_list_attach(self):
                 ['testclass', 'list', 'test-vm3'], app=self.app)
             self.assertEqual(
                 buf.getvalue(),
-                'test-vm1:dev1  Audio: itl test-device  '
+                'test-vm1:dev1  Audio_Output: itl test-device  '
                 'test-vm3 (attached)\n'
             )
 
@@ -573,7 +573,7 @@ def test_060_device_info(self):
             qubesadmin.tools.qvm_device.main(
                 ['testclass', 'info', 'test-vm1:dev1'],
                 app=self.app)
-            self.assertIn('Audio: itl test-device\n'
+            self.assertIn('Audio_Output: itl test-device\n'
                           'device ID: dead:beef:babe:u012345',
                           buf.getvalue())
         self.assertAllCalled()