diff --git a/doc/qubes-devices.rst b/doc/qubes-devices.rst index e64811a3e..be6d705fc 100644 --- a/doc/qubes-devices.rst +++ b/doc/qubes-devices.rst @@ -100,7 +100,7 @@ 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 diff --git a/qubes/api/admin.py b/qubes/api/admin.py index 65e039f3d..fff39a710 100644 --- a/qubes/api/admin.py +++ b/qubes/api/admin.py @@ -45,7 +45,7 @@ import qubes.vm import qubes.vm.adminvm import qubes.vm.qubesvm -from qubes.device_protocol import Device +from qubes.device_protocol import Port class QubesMgmtEventsDispatcher: @@ -1309,7 +1309,7 @@ async def vm_device_assign(self, endpoint, untrusted_payload): dev = self.app.domains[backend_domain].devices[devclass][ident] assignment = qubes.device_protocol.DeviceAssignment.deserialize( - untrusted_payload, expected_device=dev + untrusted_payload, expected_port=dev ) self.fire_event_for_permission( @@ -1364,7 +1364,7 @@ async def vm_device_attach(self, endpoint, untrusted_payload): dev = self.app.domains[backend_domain].devices[devclass][ident] assignment = qubes.device_protocol.DeviceAssignment.deserialize( - untrusted_payload, expected_device=dev + untrusted_payload, expected_port=dev ) self.fire_event_for_permission( @@ -1425,7 +1425,7 @@ async def vm_device_set_required(self, endpoint, untrusted_payload): # 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 = Port(backend_domain, ident, devclass) self.fire_event_for_permission(device=dev, assignment=assignment) diff --git a/qubes/device_protocol.py b/qubes/device_protocol.py index fc6defdb8..ac365f63a 100644 --- a/qubes/device_protocol.py +++ b/qubes/device_protocol.py @@ -5,10 +5,10 @@ # Copyright (C) 2010-2016 Joanna Rutkowska # Copyright (C) 2015-2016 Wojtek Porczyk # Copyright (C) 2016 Bahtiar `kalkin-` Gadimov -# Copyright (C) 2017 Marek Marczykowski-Górecki +# Copyright (C) 2017 Marek Marczykowski-Górecki # # Copyright (C) 2024 Piotr Bartman-Szwarc -# +# # # This library is free software; you can redistribute it and/or # modify it under the terms of the GNU Lesser General Public @@ -53,101 +53,82 @@ def qbool(value): return qubes.property.bool(None, None, value) -class Device: +class Port: """ - Basic class of a *bus* device with *ident* exposed by a *backend domain*. + Class of a *bus* device port with *ident* exposed by a *backend domain*. Attributes: backend_domain (QubesVM): The domain which exposes devices, e.g.`sys-usb`. - ident (str): A unique identifier for the device within - the backend domain. - devclass (str, optional): The class of the device (e.g., 'usb', 'pci'). + ident (str): A unique identifier for the port within the backend domain. + devclass (str): The class of the port (e.g., 'usb', 'pci'). """ ALLOWED_CHARS_KEY = set( string.digits + string.ascii_letters + r"!#$%&()*+,-./:;<>?@[\]^_{|}~") ALLOWED_CHARS_PARAM = ALLOWED_CHARS_KEY.union(set(string.punctuation + ' ')) - def __init__(self, backend_domain, ident, devclass=None): + def __init__(self, backend_domain, ident, devclass): self.__backend_domain = backend_domain self.__ident = ident - self.__bus = devclass + self.__devclass = devclass def __hash__(self): - return hash((str(self.backend_domain), self.ident)) + return hash((self.backend_domain.name, self.ident, self.devclass)) def __eq__(self, other): - if isinstance(other, Device): + if isinstance(other, Port): return ( self.backend_domain == other.backend_domain and - self.ident == other.ident + self.ident == other.ident and + self.devclass == other.devclass ) - raise TypeError(f"Comparing instances of 'Device' and '{type(other)}' " + raise TypeError(f"Comparing instances of 'Port' and '{type(other)}' " "is not supported") def __lt__(self, other): - if isinstance(other, Device): - return (self.backend_domain.name, self.ident) < \ - (other.backend_domain.name, other.ident) - raise TypeError(f"Comparing instances of 'Device' and '{type(other)}' " + if isinstance(other, Port): + return (self.backend_domain.name, self.devclass, self.ident) < \ + (other.backend_domain.name, other.devclass, other.ident) + raise TypeError(f"Comparing instances of 'Port' and '{type(other)}' " "is not supported") def __repr__(self): - return "[%s]:%s" % (self.backend_domain, self.ident) + return f"[{self.backend_domain.name}]:{self.devclass}:{self.ident}" def __str__(self): - return '{!s}:{!s}'.format(self.backend_domain, self.ident) + return f"{self.backend_domain.name}:{self.ident}" @property def ident(self) -> str: """ - Immutable device identifier. + Immutable port identifier. - Unique for given domain and device type. + Unique for given domain and devclass. """ return self.__ident @property def backend_domain(self) -> QubesVM: - """ Which domain provides this device. (immutable)""" + """ Which domain exposed this port. (immutable)""" return self.__backend_domain @property def devclass(self) -> str: - """ Immutable* Device class such like: 'usb', 'pci' etc. + """ Immutable port class such like: 'usb', 'pci' etc. - For unknown devices "peripheral" is returned. - - *see `@devclass.setter` + For unknown classes "peripheral" is returned. """ - if self.__bus: - return self.__bus + if self.__devclass: + return self.__devclass return "peripheral" - @property - def devclass_is_set(self) -> bool: - """ - Returns true if devclass is already initialised. - """ - return bool(self.__bus) - - @devclass.setter - def devclass(self, devclass: str): - """ Once a value is set, it should not be overridden. - - However, if it has not been set, i.e., the value is `None`, - we can override it.""" - if self.__bus is not None: - raise TypeError("Attribute devclass is immutable") - self.__bus = devclass - @classmethod def unpack_properties( cls, untrusted_serialization: bytes ) -> Tuple[Dict, Dict]: """ - Unpacks basic device properties from a serialized encoded string. + Unpacks basic port properties from a serialized encoded string. Returns: tuple: A tuple containing two dictionaries, properties and options, @@ -215,9 +196,9 @@ def pack_property(cls, key: str, value: str): @staticmethod def check_device_properties( - expected_device: 'Device', properties: Dict[str, Any]): + expected_port: 'Port', properties: Dict[str, Any]): """ - Validates properties against an expected device configuration. + Validates properties against an expected port configuration. Modifies `properties`. @@ -225,7 +206,7 @@ def check_device_properties( UnexpectedDeviceProperty: If any property does not match the expected values. """ - expected = expected_device + expected = expected_port exp_vm_name = expected.backend_domain.name if properties.get('backend_domain', exp_vm_name) != exp_vm_name: raise UnexpectedDeviceProperty( @@ -239,13 +220,11 @@ def check_device_properties( f"when expected id: {expected.ident}.") properties['ident'] = expected.ident - if expected.devclass_is_set: - if (properties.get('devclass', expected.devclass) - != expected.devclass): - raise UnexpectedDeviceProperty( - f"Got {properties['devclass']} device " - f"when expected {expected.devclass}.") - properties['devclass'] = expected.devclass + if properties.get('devclass', expected.devclass) != expected.devclass: + raise UnexpectedDeviceProperty( + f"Got {properties['devclass']} device " + f"when expected {expected.devclass}.") + properties['devclass'] = expected.devclass class DeviceCategory(Enum): @@ -428,27 +407,24 @@ def _load_classes(bus: str): return result -class DeviceInfo(Device): +class DeviceInfo(Port): """ Holds all information about a device """ def __init__( self, - backend_domain: QubesVM, - ident: str, - *, - devclass: Optional[str] = None, + port: Port, vendor: Optional[str] = None, product: Optional[str] = None, manufacturer: Optional[str] = None, name: Optional[str] = None, serial: Optional[str] = None, interfaces: Optional[List[DeviceInterface]] = None, - parent: Optional[Device] = None, + parent: Optional[Port] = None, attachment: Optional[QubesVM] = None, self_identity: Optional[str] = None, **kwargs ): - super().__init__(backend_domain, ident, devclass) + super().__init__(port.backend_domain, port.ident, port.devclass) self._vendor = vendor self._product = product @@ -462,6 +438,13 @@ def __init__( self.data = kwargs + @property + def port(self) -> Port: + """ + Device port visible in Qubes. + """ + return Port(self.backend_domain, self.ident, self.devclass) + @property def vendor(self) -> str: """ @@ -570,7 +553,7 @@ def interfaces(self) -> List[DeviceInterface]: return self._interfaces @property - def parent_device(self) -> Optional[Device]: + def parent_device(self) -> Optional[Port]: """ The parent device, if any. @@ -663,7 +646,7 @@ def deserialize( def _deserialize( cls, untrusted_serialization: bytes, - expected_device: Device + expected_port: Port ) -> 'DeviceInfo': """ Actually deserializes the object. @@ -671,20 +654,19 @@ def _deserialize( properties, options = cls.unpack_properties(untrusted_serialization) properties.update(options) - cls.check_device_properties(expected_device, properties) + cls.check_device_properties(expected_port, properties) if 'attachment' not in properties or not properties['attachment']: properties['attachment'] = None else: - app = expected_device.backend_domain.app + app = expected_port.backend_domain.app properties['attachment'] = app.domains.get_blind( properties['attachment']) - if (expected_device.devclass_is_set - and properties['devclass'] != expected_device.devclass): + if properties['devclass'] != expected_port.devclass: raise UnexpectedDeviceProperty( f"Got {properties['devclass']} device " - f"when expected {expected_device.devclass}.") + f"when expected {expected_port.devclass}.") if 'interfaces' in properties: interfaces = properties['interfaces'] @@ -694,15 +676,23 @@ def _deserialize( properties['interfaces'] = interfaces if 'parent_ident' in properties: - properties['parent'] = Device( - backend_domain=expected_device.backend_domain, + properties['parent'] = Port( + backend_domain=expected_port.backend_domain, ident=properties['parent_ident'], devclass=properties['parent_devclass'], ) del properties['parent_ident'] del properties['parent_devclass'] - return cls(**properties) + port = Port( + properties['backend_domain'], + properties['ident'], + properties['devclass']) + del properties['backend_domain'] + del properties['ident'] + del properties['devclass'] + + return cls(port, **properties) @property def self_identity(self) -> str: @@ -770,10 +760,11 @@ class UnknownDevice(DeviceInfo): """Unknown device - for example, exposed by domain not running currently""" def __init__(self, backend_domain, ident, *, devclass, **kwargs): - super().__init__(backend_domain, ident, devclass=devclass, **kwargs) + port = Port(backend_domain, ident, devclass) + super().__init__(port, **kwargs) -class DeviceAssignment(Device): +class DeviceAssignment(Port): """ Maps a device to a frontend_domain. There are 3 flags `attached`, `automatically_attached` and `required`. @@ -821,7 +812,7 @@ def clone(self, **kwargs): return self.__class__(**attr) @classmethod - def from_device(cls, device: Device, **kwargs) -> 'DeviceAssignment': + def from_device(cls, device: Port, **kwargs) -> 'DeviceAssignment': """ Get assignment of the device. """ @@ -923,13 +914,13 @@ def serialize(self) -> bytes: def deserialize( cls, serialization: bytes, - expected_device: Device, + expected_port: Port, ) -> 'DeviceAssignment': """ Recovers a serialized object, see: :py:meth:`serialize`. """ try: - result = cls._deserialize(serialization, expected_device) + result = cls._deserialize(serialization, expected_port) except Exception as exc: raise ProtocolError() from exc return result @@ -938,7 +929,7 @@ def deserialize( def _deserialize( cls, untrusted_serialization: bytes, - expected_device: Device, + expected_port: Port, ) -> 'DeviceAssignment': """ Actually deserializes the object. @@ -946,7 +937,7 @@ def _deserialize( properties, options = cls.unpack_properties(untrusted_serialization) properties['options'] = options - cls.check_device_properties(expected_device, properties) + cls.check_device_properties(expected_port, properties) properties['attach_automatically'] = qbool( properties.get('attach_automatically', 'no')) diff --git a/qubes/devices.py b/qubes/devices.py index c073ef759..058dc9ae5 100644 --- a/qubes/devices.py +++ b/qubes/devices.py @@ -63,7 +63,7 @@ import qubes.exc import qubes.utils -from qubes.device_protocol import (Device, DeviceInfo, UnknownDevice, +from qubes.device_protocol import (Port, DeviceInfo, UnknownDevice, DeviceAssignment) @@ -193,9 +193,7 @@ async def attach(self, assignment: DeviceAssignment): Attach device to domain. """ - if not assignment.devclass_is_set: - assignment.devclass = self._bus - elif assignment.devclass != self._bus: + if assignment.devclass != self._bus: raise ValueError( f'Trying to attach {assignment.devclass} device ' f'when {self._bus} device expected.') @@ -223,9 +221,7 @@ async def assign(self, assignment: DeviceAssignment): """ Assign device to domain. """ - if not assignment.devclass_is_set: - assignment.devclass = self._bus - elif assignment.devclass != self._bus: + if assignment.devclass != self._bus: raise ValueError( f'Trying to attach {assignment.devclass} device ' f'when {self._bus} device expected.') @@ -250,10 +246,9 @@ def load_assignment(self, device_assignment: DeviceAssignment): """ assert not self._vm.events_enabled assert device_assignment.attach_automatically - device_assignment.devclass = self._bus self._set.add(device_assignment) - async def update_required(self, device: Device, required: bool): + async def update_required(self, device: Port, required: bool): """ Update `required` flag of an already attached device. @@ -284,7 +279,7 @@ async def update_required(self, device: Device, required: bool): await self._vm.fire_event_async( 'device-assignment-changed:' + self._bus, device=device) - async def detach(self, device: Device): + async def detach(self, device: Port): """ Detach device from domain. """ @@ -316,15 +311,17 @@ async def unassign(self, device_assignment: DeviceAssignment): """ Unassign device from domain. """ + all_ass = [] for assignment in self.get_assigned_devices(): + all_ass.append(assignment.devclass) if device_assignment == assignment: # load all options device_assignment = assignment break else: raise DeviceNotAssigned( - f'device {device_assignment.ident!s} of class {self._bus} not ' - f'assigned to {self._vm!s}') + f'{self._bus} device at port {device_assignment}' + f'not assigned to {self._vm!s} | {all_ass} vs {device_assignment.devclass}') self._set.discard(assignment) @@ -353,9 +350,9 @@ def get_attached_devices(self) -> Iterable[DeviceAssignment]: yield DeviceAssignment( backend_domain=dev.backend_domain, ident=dev.ident, - options=options, - frontend_domain=self._vm, devclass=dev.devclass, + frontend_domain=self._vm, + options=options, attach_automatically=False, required=False, ) diff --git a/qubes/ext/block.py b/qubes/ext/block.py index 268e6ddcd..34db054a4 100644 --- a/qubes/ext/block.py +++ b/qubes/ext/block.py @@ -48,8 +48,9 @@ class BlockDevice(qubes.device_protocol.DeviceInfo): def __init__(self, backend_domain, ident): - super().__init__( + port = qubes.device_protocol.Port( backend_domain=backend_domain, ident=ident, devclass="block") + super().__init__(port) # lazy loading self._mode = None @@ -161,7 +162,7 @@ def interfaces(self) -> List[qubes.device_protocol.DeviceInterface]: return [qubes.device_protocol.DeviceInterface("******", "block")] @property - def parent_device(self) -> Optional[qubes.device_protocol.Device]: + def parent_device(self) -> Optional[qubes.device_protocol.Port]: """ The parent device, if any. diff --git a/qubes/ext/pci.py b/qubes/ext/pci.py index 1535a7f9d..3ca8c0ba6 100644 --- a/qubes/ext/pci.py +++ b/qubes/ext/pci.py @@ -168,8 +168,9 @@ def __init__(self, backend_domain, ident, libvirt_name=None): raise UnsupportedDevice(libvirt_name) ident = '{bus}_{device}.{function}'.format(**dev_match.groupdict()) - super().__init__( + port = qubes.device_protocol.Port( backend_domain=backend_domain, ident=ident, devclass="pci") + super().__init__(port) dev_match = self.regex.match(ident) if not dev_match: diff --git a/qubes/tests/api_admin.py b/qubes/tests/api_admin.py index 357104109..ebc54eee1 100644 --- a/qubes/tests/api_admin.py +++ b/qubes/tests/api_admin.py @@ -1730,11 +1730,11 @@ def device_list_testclass(self, vm, event): if vm is not self.vm: return dev = qubes.device_protocol.DeviceInfo( - self.vm, '1234', product='Some device') + Port(self.vm, '1234', 'testclass'), product='Some device') dev.extra_prop = 'xx' yield dev dev = qubes.device_protocol.DeviceInfo( - self.vm, '4321', product='Some other device') + Port(self.vm, '4321', 'testclass'), product='Some other device') yield dev def assertSerializedEqual(self, actual, expected): @@ -1785,7 +1785,8 @@ def test_462_vm_device_available_invalid(self): self.assertFalse(self.app.save.called) def test_470_vm_device_list_assigned(self): - assignment = qubes.device_protocol.DeviceAssignment(self.vm, '1234', + assignment = qubes.device_protocol.DeviceAssignment( + self.vm, '1234', 'test', attach_automatically=True, required=True) self.loop.run_until_complete( self.vm.devices['testclass'].assign(assignment)) @@ -1797,11 +1798,13 @@ def test_470_vm_device_list_assigned(self): self.assertFalse(self.app.save.called) def test_471_vm_device_list_assigned_options(self): - assignment = qubes.device_protocol.DeviceAssignment(self.vm, '1234', + assignment = qubes.device_protocol.DeviceAssignment( + self.vm, '1234', 'test', attach_automatically=True, required=True, options={'opt1': 'value'}) self.loop.run_until_complete( self.vm.devices['testclass'].assign(assignment)) - assignment = qubes.device_protocol.DeviceAssignment(self.vm, '4321', + assignment = qubes.device_protocol.DeviceAssignment( + self.vm, '4321', 'test', attach_automatically=True, required=True) self.loop.run_until_complete( self.vm.devices['testclass'].assign(assignment)) @@ -1818,7 +1821,7 @@ def test_471_vm_device_list_assigned_options(self): def device_list_single_attached_testclass(self, vm, event, **kwargs): if vm is not self.vm: return - dev = qubes.device_protocol.DeviceInfo(self.vm, '1234', devclass='testclass') + dev = qubes.device_protocol.DeviceInfo(Port(self.vm, '1234', devclass='testclass')) yield (dev, {'attach_opt': 'value'}) def test_472_vm_device_list_attached(self): @@ -1834,11 +1837,11 @@ def test_472_vm_device_list_attached(self): def test_473_vm_device_list_assigned_specific(self): assignment = qubes.device_protocol.DeviceAssignment( - self.vm, '1234', attach_automatically=True, required=True) + self.vm, '1234', 'test', attach_automatically=True, required=True) self.loop.run_until_complete( self.vm.devices['testclass'].assign(assignment)) assignment = qubes.device_protocol.DeviceAssignment( - self.vm, '4321', attach_automatically=True, required=True) + self.vm, '4321', 'test', attach_automatically=True, required=True) self.loop.run_until_complete( self.vm.devices['testclass'].assign(assignment)) value = self.call_mgmt_func(b'admin.vm.device.testclass.Assigned', @@ -1851,9 +1854,9 @@ def test_473_vm_device_list_assigned_specific(self): def device_list_multiple_attached_testclass(self, vm, event, **kwargs): if vm is not self.vm: return - dev = qubes.device_protocol.DeviceInfo(self.vm, '1234', devclass='testclass') + dev = qubes.device_protocol.DeviceInfo(Port(self.vm, '1234', devclass='testclass')) yield (dev, {'attach_opt': 'value'}) - dev = qubes.device_protocol.DeviceInfo(self.vm, '4321', devclass='testclass') + dev = qubes.device_protocol.DeviceInfo(Port(self.vm, '4321', devclass='testclass')) yield (dev, {'attach_opt': 'value'}) def test_474_vm_device_list_attached_specific(self): @@ -2021,7 +2024,7 @@ def test_488_vm_device_assign_options(self): def test_490_vm_device_unassign_from_running(self): assignment = qubes.device_protocol.DeviceAssignment( - self.vm, '1234', attach_automatically=True, required=False, + self.vm, '1234', 'test', attach_automatically=True, required=False, options={'opt1': 'value'}) self.loop.run_until_complete( self.vm.devices['testclass'].assign(assignment)) @@ -2041,7 +2044,7 @@ def test_490_vm_device_unassign_from_running(self): def test_491_vm_device_unassign_required_from_running(self): assignment = qubes.device_protocol.DeviceAssignment( - self.vm, '1234', attach_automatically=True, required=True, + self.vm, '1234', 'test', attach_automatically=True, required=True, options={'opt1': 'value'}) self.loop.run_until_complete( self.vm.devices['testclass'].assign(assignment)) @@ -2062,7 +2065,7 @@ def test_491_vm_device_unassign_required_from_running(self): def test_492_vm_device_unassign_from_halted(self): assignment = qubes.device_protocol.DeviceAssignment( - self.vm, '1234', attach_automatically=True, required=False, + self.vm, '1234', 'test', attach_automatically=True, required=False, options={'opt1': 'value'}) self.loop.run_until_complete( self.vm.devices['testclass'].assign(assignment)) @@ -2080,7 +2083,7 @@ def test_492_vm_device_unassign_from_halted(self): def test_493_vm_device_unassign_required_from_halted(self): assignment = qubes.device_protocol.DeviceAssignment( - self.vm, '1234', attach_automatically=True, required=True, + self.vm, '1234', 'test', attach_automatically=True, required=True, options={'opt1': 'value'}) self.loop.run_until_complete( self.vm.devices['testclass'].assign(assignment)) @@ -2101,7 +2104,7 @@ def test_494_vm_device_unassign_attached(self): self.vm.add_handler('device-list-attached:testclass', self.device_list_single_attached_testclass) assignment = qubes.device_protocol.DeviceAssignment( - self.vm, '1234', attach_automatically=True, required=False, + self.vm, '1234', 'test', attach_automatically=True, required=False, options={'opt1': 'value'}) self.loop.run_until_complete( self.vm.devices['testclass'].assign(assignment)) @@ -2188,7 +2191,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.device_protocol.DeviceAssignment( - self.vm, '1234', attach_automatically=True, required=True) + self.vm, '1234', 'test', attach_automatically=True, required=True) self.loop.run_until_complete( self.vm2.devices['testclass'].assign(assignment)) @@ -2907,7 +2910,7 @@ def test_642_vm_create_disposable_not_allowed(self, storage_mock): def test_650_vm_device_set_required_true(self): assignment = qubes.device_protocol.DeviceAssignment( - self.vm, '1234', attach_automatically=True, required=False, + self.vm, '1234', 'test', attach_automatically=True, required=False, options={'opt1': 'value'}) self.loop.run_until_complete( self.vm.devices['testclass'].assign(assignment)) @@ -2923,7 +2926,7 @@ def test_650_vm_device_set_required_true(self): b'test-vm1', b'test-vm1+1234', b'True') self.assertIsNone(value) - dev = qubes.device_protocol.DeviceInfo(self.vm, '1234') + dev = qubes.device_protocol.DeviceInfo(Port(self.vm, '1234', 'testclass')) required = self.vm.devices['testclass'].get_assigned_devices( required_only=True) self.assertIn(dev, required) @@ -2937,7 +2940,7 @@ def test_650_vm_device_set_required_true(self): def test_651_vm_device_set_required_false(self): assignment = qubes.device_protocol.DeviceAssignment( - self.vm, '1234', attach_automatically=True, required=True, + self.vm, '1234', 'test', attach_automatically=True, required=True, options={'opt1': 'value'}) self.loop.run_until_complete( self.vm.devices['testclass'].assign(assignment)) @@ -2953,7 +2956,7 @@ def test_651_vm_device_set_required_false(self): b'test-vm1', b'test-vm1+1234', b'False') self.assertIsNone(value) - dev = qubes.device_protocol.DeviceInfo(self.vm, '1234') + dev = qubes.device_protocol.DeviceInfo(Port(self.vm, '1234', 'testclass')) required = self.vm.devices['testclass'].get_assigned_devices( required_only=True) self.assertNotIn(dev, required) @@ -2967,7 +2970,7 @@ def test_651_vm_device_set_required_false(self): def test_652_vm_device_set_required_true_unchanged(self): assignment = qubes.device_protocol.DeviceAssignment( - self.vm, '1234', attach_automatically=True, required=True, + self.vm, '1234', 'test', attach_automatically=True, required=True, options={'opt1': 'value'}) self.loop.run_until_complete( self.vm.devices['testclass'].assign(assignment)) @@ -2977,7 +2980,7 @@ def test_652_vm_device_set_required_true_unchanged(self): b'admin.vm.device.testclass.Set.required', b'test-vm1', b'test-vm1+1234', b'True') self.assertIsNone(value) - dev = qubes.device_protocol.DeviceInfo(self.vm, '1234') + dev = qubes.device_protocol.DeviceInfo(Port(self.vm, '1234', 'testclass')) required = self.vm.devices['testclass'].get_assigned_devices( required_only=True) self.assertIn(dev, required) @@ -2985,7 +2988,7 @@ def test_652_vm_device_set_required_true_unchanged(self): def test_653_vm_device_set_required_false_unchanged(self): assignment = qubes.device_protocol.DeviceAssignment( - self.vm, '1234', attach_automatically=True, required=False, + self.vm, '1234', 'test', attach_automatically=True, required=False, options={'opt1': 'value'}) self.loop.run_until_complete( self.vm.devices['testclass'].assign(assignment)) @@ -2995,7 +2998,7 @@ def test_653_vm_device_set_required_false_unchanged(self): b'admin.vm.device.testclass.Set.required', b'test-vm1', b'test-vm1+1234', b'False') self.assertIsNone(value) - dev = qubes.device_protocol.DeviceInfo(self.vm, '1234') + dev = qubes.device_protocol.DeviceInfo(Port(self.vm, '1234', 'testclass')) required = self.vm.devices['testclass'].get_assigned_devices( required_only=True) self.assertNotIn(dev, required) @@ -3010,7 +3013,7 @@ def test_654_vm_device_set_persistent_not_assigned(self): self.call_mgmt_func( b'admin.vm.device.testclass.Set.required', b'test-vm1', b'test-vm1+1234', b'True') - dev = qubes.device_protocol.DeviceInfo(self.vm, '1234') + dev = qubes.device_protocol.DeviceInfo(Port(self.vm, '1234', 'testclass')) self.assertNotIn( dev, self.vm.devices['testclass'].get_assigned_devices()) self.assertFalse(self.app.save.called) @@ -3024,7 +3027,7 @@ def test_655_vm_device_set_persistent_invalid_value(self): self.call_mgmt_func( b'admin.vm.device.testclass.Set.required', b'test-vm1', b'test-vm1+1234', b'maybe') - dev = qubes.device_protocol.DeviceInfo(self.vm, '1234') + dev = qubes.device_protocol.DeviceInfo(Port(self.vm, '1234', 'testclass')) self.assertNotIn(dev, self.vm.devices['testclass'].get_assigned_devices()) self.assertFalse(self.app.save.called) diff --git a/qubes/tests/devices.py b/qubes/tests/devices.py index e7c81b9ba..aaec7c754 100644 --- a/qubes/tests/devices.py +++ b/qubes/tests/devices.py @@ -21,7 +21,7 @@ # import qubes.devices -from qubes.device_protocol import (Device, DeviceInfo, DeviceAssignment, +from qubes.device_protocol import (Port, DeviceInfo, DeviceAssignment, DeviceInterface, UnknownDevice) import qubes.tests @@ -90,9 +90,8 @@ def setUp(self): self.app.domains['vm'] = self.emitter self.device = self.emitter.device self.collection = self.emitter.devices['testclass'] - self.assignment = DeviceAssignment( - backend_domain=self.device.backend_domain, - ident=self.device.ident, + self.assignment = DeviceAssignment.from_device( + self.device, attach_automatically=True, required=True, ) @@ -340,10 +339,8 @@ def test_000_init(self): def test_001_missing(self): device = TestDevice(self.emitter.app.domains['vm'], 'testdev') - assignment = DeviceAssignment( - backend_domain=device.backend_domain, - ident=device.ident, - attach_automatically=True, required=True) + assignment = DeviceAssignment.from_device( + device, attach_automatically=True, required=True) self.loop.run_until_complete( self.manager['testclass'].assign(assignment)) self.assertEqual( @@ -358,9 +355,9 @@ def setUp(self): def test_010_serialize(self): device = DeviceInfo( - backend_domain=self.vm, - ident="1-1.1.1", - devclass="bus", + Port(backend_domain=self.vm, + ident="1-1.1.1", + devclass="bus"), vendor="ITL", product="Qubes", manufacturer="", @@ -386,9 +383,9 @@ def test_010_serialize(self): def test_011_serialize_with_parent(self): device = DeviceInfo( - backend_domain=self.vm, - ident="1-1.1.1", - devclass="bus", + Port(backend_domain=self.vm, + ident="1-1.1.1", + devclass="bus"), vendor="ITL", product="Qubes", manufacturer="", @@ -398,7 +395,7 @@ def test_011_serialize_with_parent(self): DeviceInterface("u03**01")], additional_info="", date="06.12.23", - parent=Device(self.vm, '1-1.1', 'pci') + parent=Port(self.vm, '1-1.1', 'pci') ) actual = device.serialize() expected = ( @@ -416,9 +413,9 @@ def test_011_serialize_with_parent(self): def test_012_invalid_serialize(self): device = DeviceInfo( - backend_domain=self.vm, - ident="1-1.1.1", - devclass="bus?", + Port(backend_domain=self.vm, + ident="1-1.1.1", + devclass="bus?"), vendor="malicious", product="suspicious", manufacturer="", @@ -438,9 +435,9 @@ def test_020_deserialize(self): b"parent_ident='1-1.1' parent_devclass='None'") actual = DeviceInfo.deserialize(serialized, self.vm) expected = DeviceInfo( - backend_domain=self.vm, - ident="1-1.1.1", - devclass="bus", + Port(backend_domain=self.vm, + ident="1-1.1.1", + devclass="bus"), vendor="ITL", product="Qubes", manufacturer="unknown", @@ -481,9 +478,9 @@ def test_021_invalid_deserialize(self): def test_030_serialize_and_deserialize(self): device = DeviceInfo( - backend_domain=self.vm, - ident="1-1.1.1", - devclass="bus?", + Port(backend_domain=self.vm, + ident="1-1.1.1", + devclass="bus?"), vendor="malicious", product="suspicious", manufacturer="", @@ -589,7 +586,7 @@ def test_020_deserialize(self): b"ident='1-1.1.1' frontend_domain='vm' devclass='bus' " b"backend_domain='vm' required='no' attach_automatically='yes' " b"_read-only='yes'") - expected_device = Device(self.vm, '1-1.1.1', 'bus') + expected_device = Port(self.vm, '1-1.1.1', 'bus') actual = DeviceAssignment.deserialize(serialized, expected_device) expected = DeviceAssignment( backend_domain=self.vm, @@ -614,7 +611,7 @@ def test_021_invalid_deserialize(self): b"ident='1-1.1.1' frontend_domain='vm' devclass='bus' " b"backend_domain='vm' required='no' attach_automatically='yes' " b"_read'only='yes'") - expected_device = Device(self.vm, '1-1.1.1', 'bus') + expected_device = Port(self.vm, '1-1.1.1', 'bus') with self.assertRaises(qubes.exc.ProtocolError): _ = DeviceAssignment.deserialize(serialized, expected_device) @@ -623,7 +620,7 @@ def test_022_invalid_deserialize_2(self): b"ident='1-1.1.1' frontend_domain='vm' devclass='bus' " b"backend_domain='vm' required='no' attach_automatically='yes' " b"read-only='yes'") - expected_device = Device(self.vm, '1-1.1.1', 'bus') + expected_device = Port(self.vm, '1-1.1.1', 'bus') with self.assertRaises(qubes.exc.ProtocolError): _ = DeviceAssignment.deserialize(serialized, expected_device) @@ -638,7 +635,7 @@ def test_030_serialize_and_deserialize(self): options={'read-only': 'yes'}, ) serialized = expected.serialize() - expected_device = Device(self.vm, '1-1.1.1', 'bus') + expected_device = Port(self.vm, '1-1.1.1', 'bus') actual = DeviceAssignment.deserialize(serialized, expected_device) self.assertEqual(actual.backend_domain, expected.backend_domain) self.assertEqual(actual.ident, expected.ident) diff --git a/qubes/tests/devices_block.py b/qubes/tests/devices_block.py index 0d19eebb4..1f5744680 100644 --- a/qubes/tests/devices_block.py +++ b/qubes/tests/devices_block.py @@ -24,7 +24,7 @@ import qubes.tests import qubes.ext.block -from qubes.device_protocol import DeviceInterface, Device, DeviceInfo, \ +from qubes.device_protocol import DeviceInterface, Port, DeviceInfo, \ DeviceAssignment modules_disk = ''' @@ -187,7 +187,7 @@ def test_000_device_get(self): '/qubes-block-devices/sda/mode': b'w', '/qubes-block-devices/sda/parent': b'1-1.1:1.0', }, domain_xml=domain_xml_template.format("")) - parent = DeviceInfo(vm, '1-1.1', devclass='usb') + parent = DeviceInfo(Port(vm, '1-1.1', devclass='usb')) vm.devices['usb'] = TestDeviceCollection(backend_vm=vm, devclass='usb') vm.devices['usb']._exposed.append(parent) vm.is_running = lambda: True @@ -228,7 +228,7 @@ def test_000_device_get(self): self.assertEqual(device_info.interfaces, [DeviceInterface("b******")]) self.assertEqual(device_info.parent_device, - Device(vm, '1-1.1', devclass='usb')) + Port(vm, '1-1.1', devclass='usb')) self.assertEqual(device_info.attachment, front) self.assertEqual(device_info.self_identity, '1-1.1:0000:0000::?******:1.0') @@ -664,7 +664,7 @@ def test_060_on_qdb_change_added(self): '/qubes-block-devices/sda/size': b'1024000', '/qubes-block-devices/sda/mode': b'r', }, domain_xml=domain_xml_template.format("")) - exp_dev = Device(back_vm, 'sda', 'block') + exp_dev = Port(back_vm, 'sda', 'block') self.ext.on_qdb_change(back_vm, None, None) @@ -680,7 +680,7 @@ def test_061_on_qdb_change_auto_attached(self): '/qubes-block-devices/sda/size': b'1024000', '/qubes-block-devices/sda/mode': b'r', }, domain_xml=domain_xml_template.format("")) - exp_dev = Device(back_vm, 'sda', 'block') + exp_dev = Port(back_vm, 'sda', 'block') front = TestVM({}, domain_xml=domain_xml_template.format(""), name='front-vm') dom0 = TestVM({}, name='dom0', @@ -725,7 +725,7 @@ def test_062_on_qdb_change_attached(self): '/qubes-block-devices/sda/size': b'1024000', '/qubes-block-devices/sda/mode': b'r', }, domain_xml=domain_xml_template.format("")) - exp_dev = Device(back_vm, 'sda', 'block') + exp_dev = Port(back_vm, 'sda', 'block') self.ext.devices_cache = {'sys-usb': {'sda': None}} @@ -774,7 +774,7 @@ def test_063_on_qdb_change_changed(self): '/qubes-block-devices/sda/size': b'1024000', '/qubes-block-devices/sda/mode': b'r', }, domain_xml=domain_xml_template.format("")) - exp_dev = Device(back_vm, 'sda', 'block') + exp_dev = Port(back_vm, 'sda', 'block') front = TestVM({}, name='front-vm') dom0 = TestVM({}, name='dom0', @@ -841,7 +841,7 @@ def test_064_on_qdb_change_removed_attached(self): }, domain_xml=domain_xml_template.format("")) dom0 = TestVM({}, name='dom0', domain_xml=domain_xml_template.format("")) - exp_dev = Device(back_vm, 'sda', 'block') + exp_dev = Port(back_vm, 'sda', 'block') disk = ''' diff --git a/qubes/tests/integ/devices_pci.py b/qubes/tests/integ/devices_pci.py index 7b1ea5579..f3136e9b8 100644 --- a/qubes/tests/integ/devices_pci.py +++ b/qubes/tests/integ/devices_pci.py @@ -28,6 +28,7 @@ import qubes.devices import qubes.ext.pci import qubes.tests +from qubes.device_protocol import DeviceAssignment @qubes.tests.skipUnlessEnv('QUBES_TEST_PCIDEV') @@ -37,14 +38,11 @@ def setUp(self): if self._testMethodName not in ['test_000_list']: pcidev = os.environ['QUBES_TEST_PCIDEV'] self.dev = self.app.domains[0].devices['pci'][pcidev] - self.assignment = qubes.device_protocol.DeviceAssignment( - backend_domain=self.dev.backend_domain, - ident=self.dev.ident, - attach_automatically=True, + self.assignment = DeviceAssignment.from_device( + self.dev, attach_automatically=True ) - self.required_assignment = qubes.device_protocol.DeviceAssignment( - backend_domain=self.dev.backend_domain, - ident=self.dev.ident, + self.required_assignment = DeviceAssignment.from_device( + self.dev, attach_automatically=True, required=True, ) diff --git a/qubes/vm/__init__.py b/qubes/vm/__init__.py index ad52fe735..8c152d850 100644 --- a/qubes/vm/__init__.py +++ b/qubes/vm/__init__.py @@ -287,6 +287,7 @@ def load_extras(self): device_assignment = qubes.device_protocol.DeviceAssignment( self.app.domains[node.get('backend-domain')], node.get('id'), + devclass=devclass, options=options, attach_automatically=True, # backward compatibility: persistent~>required=True