From 4389dd0b2b8dbb4f974fe12bfeae855d481c9de5 Mon Sep 17 00:00:00 2001 From: Piotr Bartman-Szwarc Date: Thu, 15 Aug 2024 15:33:30 +0200 Subject: [PATCH] q-dev: virtual device --- qubesadmin/backup/restore.py | 4 ++-- qubesadmin/device_protocol.py | 42 +++++++++++++++++----------------- qubesadmin/devices.py | 8 +++---- qubesadmin/events/__init__.py | 4 ++-- qubesadmin/tests/devices.py | 18 +++++++-------- qubesadmin/tools/qvm_device.py | 5 ++-- qubesadmin/tools/qvm_start.py | 2 +- 7 files changed, 41 insertions(+), 42 deletions(-) diff --git a/qubesadmin/backup/restore.py b/qubesadmin/backup/restore.py index cce03530..27ac9ef4 100644 --- a/qubesadmin/backup/restore.py +++ b/qubesadmin/backup/restore.py @@ -50,7 +50,7 @@ from qubesadmin.backup import BackupVM from qubesadmin.backup.core2 import Core2Qubes from qubesadmin.backup.core3 import Core3Qubes -from qubesadmin.device_protocol import DeviceAssignment, Port, Device +from qubesadmin.device_protocol import DeviceAssignment, Port, VirtualDevice from qubesadmin.exc import QubesException from qubesadmin.utils import size_to_human @@ -2094,7 +2094,7 @@ def _restore_vms_metadata(self, restore_info): else: required = False assignment = DeviceAssignment( - Device(Port( + VirtualDevice(Port( backend_domain=self.app.domains[backend_domain], port_id=port_id, devclass=bus, diff --git a/qubesadmin/device_protocol.py b/qubesadmin/device_protocol.py index 4532a334..cc5fae73 100644 --- a/qubesadmin/device_protocol.py +++ b/qubesadmin/device_protocol.py @@ -154,7 +154,7 @@ def pack_property(cls, key: str, value: Optional[str]): @staticmethod def parse_basic_device_properties( - expected_device: 'Device', properties: Dict[str, Any]): + expected_device: 'VirtualDevice', properties: Dict[str, Any]): """ Validates properties against an expected port configuration. @@ -338,7 +338,7 @@ def devclass(self) -> str: return "peripheral" -class Device: +class VirtualDevice: def __init__( self, port: Optional[Port] = None, @@ -396,7 +396,7 @@ def __hash__(self): return hash((self.port, self.device_id)) def __eq__(self, other): - if isinstance(other, (Device, DeviceAssignment)): + if isinstance(other, (VirtualDevice, DeviceAssignment)): result = ( self.port == other.port and self.device_id == other.device_id @@ -418,7 +418,7 @@ def __lt__(self, other): 3. *: 4. *:* """ - if isinstance(other, (Device, DeviceAssignment)): + if isinstance(other, (VirtualDevice, DeviceAssignment)): if self.port == '*' and other.port != '*': return True if self.port != '*' and other.port == '*': @@ -429,7 +429,7 @@ def __lt__(self, other): reprs[obj].append(obj.device_id) return reprs[self] < reprs[other] elif isinstance(other, Port): - _other = Device(other, '*') + _other = VirtualDevice(other, '*') return self < _other else: raise TypeError( @@ -450,7 +450,7 @@ def from_qarg( domains, blind=False, backend=None, - ) -> 'Device': + ) -> 'VirtualDevice': if backend is None: if blind: get_domain = domains.get_blind @@ -464,7 +464,7 @@ def from_qarg( def from_str( cls, representation: str, devclass: Optional[str], domains, blind=False, backend=None - ) -> 'Device': + ) -> 'VirtualDevice': if backend is None: if blind: get_domain = domains.get_blind @@ -482,7 +482,7 @@ def _parse( get_domain: Callable, backend, sep: str - ) -> 'Device': + ) -> 'VirtualDevice': if backend is None: backend_name, identity = representation.split(sep, 1) backend = get_domain(backend_name) @@ -693,7 +693,7 @@ def _load_classes(bus: str): return result -class DeviceInfo(Device): +class DeviceInfo(VirtualDevice): """ Holds all information about a device """ def __init__( @@ -833,7 +833,7 @@ def interfaces(self) -> List[DeviceInterface]: return self._interfaces @property - def parent_device(self) -> Optional[Device]: + def parent_device(self) -> Optional[VirtualDevice]: """ The parent device, if any. @@ -843,7 +843,7 @@ def parent_device(self) -> Optional[Device]: return self._parent @property - def subdevices(self) -> List[Device]: + def subdevices(self) -> List[VirtualDevice]: """ The list of children devices if any. @@ -864,7 +864,7 @@ def serialize(self) -> bytes: """ Serialize an object to be transmitted via Qubes API. """ - properties = Device.serialize(self) + properties = VirtualDevice.serialize(self) # 'attachment', 'interfaces', 'data', 'parent_device' # are not string, so they need special treatment default = DeviceInfo(self.port) @@ -904,7 +904,7 @@ def deserialize( Recovers a serialized object, see: :py:meth:`serialize`. """ head, _, rest = serialization.partition(b' ') - device = Device.from_str( + device = VirtualDevice.from_str( head.decode('ascii', errors='ignore'), expected_devclass, domains=None, backend=expected_backend_domain) @@ -920,7 +920,7 @@ def deserialize( def _deserialize( cls, untrusted_serialization: bytes, - expected_device: Device + expected_device: VirtualDevice ) -> 'DeviceInfo': """ Actually deserializes the object. @@ -1020,13 +1020,13 @@ class DeviceAssignment: def __init__( self, - device: Device, + device: VirtualDevice, frontend_domain=None, options=None, mode: Union[str, AssignmentMode] = "manual", ): if isinstance(device, DeviceInfo): - device = Device(device.port, device.device_id) + device = VirtualDevice(device.port, device.device_id) self._device_ident = device self.__options = options or {} if isinstance(mode, AssignmentMode): @@ -1040,7 +1040,7 @@ def clone(self, **kwargs): Clone object and substitute attributes with explicitly given. """ kwargs["device"] = kwargs.get( - "device", Device( + "device", VirtualDevice( Port(self.backend_domain, self.port_id, self.devclass), self.device_id )) @@ -1062,7 +1062,7 @@ def __hash__(self): return hash(self._device_ident) def __eq__(self, other): - if isinstance(other, (Device, DeviceAssignment)): + if isinstance(other, (VirtualDevice, DeviceAssignment)): result = ( self.port == other.port and self.device_id == other.device_id @@ -1073,7 +1073,7 @@ def __eq__(self, other): def __lt__(self, other): if isinstance(other, DeviceAssignment): return self._device_ident < other._device_ident - if isinstance(other, Device): + if isinstance(other, VirtualDevice): return self._device_ident < other raise TypeError( f"Comparing instances of {type(self)} and '{type(other)}' " @@ -1182,7 +1182,7 @@ def serialize(self) -> bytes: def deserialize( cls, serialization: bytes, - expected_device: Device, + expected_device: VirtualDevice, ) -> 'DeviceAssignment': """ Recovers a serialized object, see: :py:meth:`serialize`. @@ -1197,7 +1197,7 @@ def deserialize( def _deserialize( cls, untrusted_serialization: bytes, - expected_device: Device, + expected_device: VirtualDevice, ) -> 'DeviceAssignment': """ Actually deserializes the object. diff --git a/qubesadmin/devices.py b/qubesadmin/devices.py index 55a8d184..c0c5e0ea 100644 --- a/qubesadmin/devices.py +++ b/qubesadmin/devices.py @@ -36,7 +36,7 @@ class is implemented by an extension. import qubesadmin.exc from qubesadmin.device_protocol import (Port, DeviceInfo, UnknownDevice, - DeviceAssignment, Device) + DeviceAssignment, VirtualDevice) class DeviceCollection: @@ -153,7 +153,7 @@ def get_attached_devices(self) -> Iterable[DeviceAssignment]: None, 'admin.vm.device.{}.Attached'.format(self._class)).decode() for assignment_str in assignments_str.splitlines(): head, _, untrusted_rest = assignment_str.partition(' ') - device = Device.from_qarg( + device = VirtualDevice.from_qarg( head, self._class, self._vm.app.domains, blind=True) yield DeviceAssignment.deserialize( @@ -171,7 +171,7 @@ def get_assigned_devices( None, 'admin.vm.device.{}.Assigned'.format(self._class)).decode() for assignment_str in assignments_str.splitlines(): head, _, untrusted_rest = assignment_str.partition(' ') - device = Device.from_qarg( + device = VirtualDevice.from_qarg( head, self._class, self._vm.app.domains, blind=True) assignment = DeviceAssignment.deserialize( @@ -196,7 +196,7 @@ def update_assignment(self, device: Port, required: Optional[bool]): """ Update assignment of already attached device. - :param Device device: device for which change required flag + :param VirtualDevice device: device for which change required flag :param bool required: new assignment: `None` -> unassign device from qube `False` -> device will be auto-attached to qube diff --git a/qubesadmin/events/__init__.py b/qubesadmin/events/__init__.py index 00c00f28..49feb806 100644 --- a/qubesadmin/events/__init__.py +++ b/qubesadmin/events/__init__.py @@ -26,7 +26,7 @@ import qubesadmin.config import qubesadmin.exc -from qubesadmin.device_protocol import Device, Port +from qubesadmin.device_protocol import VirtualDevice, Port class EventsDispatcher(object): @@ -234,7 +234,7 @@ def handle(self, subject, event, **kwargs): try: if 'device' in kwargs: devclass = event.split(':', 1)[1] - device = Device.from_str( + device = VirtualDevice.from_str( kwargs['device'], devclass, self.app.domains, diff --git a/qubesadmin/tests/devices.py b/qubesadmin/tests/devices.py index 2965866b..53b108f7 100644 --- a/qubesadmin/tests/devices.py +++ b/qubesadmin/tests/devices.py @@ -21,7 +21,7 @@ import qubesadmin.tests import qubesadmin.device_protocol -from qubesadmin.device_protocol import (DeviceAssignment, Port, Device, +from qubesadmin.device_protocol import (DeviceAssignment, Port, VirtualDevice, DeviceInfo, UnknownDevice) @@ -119,7 +119,7 @@ def test_020_attach(self): b"frontend_domain='test-vm'")] = \ b'0\0' assign = DeviceAssignment( - Device(Port( + VirtualDevice(Port( self.app.domains['test-vm2'], 'dev1', devclass='test',))) self.vm.devices['test'].attach(assign) self.assertAllCalled() @@ -132,7 +132,7 @@ def test_021_attach_options(self): b"frontend_domain='test-vm' _ro='True' " b"_something='value'")] = b'0\0' assign = DeviceAssignment( - Device(Port( + VirtualDevice(Port( self.app.domains['test-vm2'], 'dev1', devclass='test'))) assign.options['ro'] = True assign.options['something'] = 'value' @@ -146,7 +146,7 @@ def test_022_attach_required(self): b"backend_domain='test-vm2' mode='required' " b"frontend_domain='test-vm'")] = b'0\0' assign = DeviceAssignment( - Device(Port( + VirtualDevice(Port( self.app.domains['test-vm2'], 'dev1', devclass='test')), mode='required') self.vm.devices['test'].attach(assign) @@ -159,7 +159,7 @@ def test_023_attach_required_options(self): b"backend_domain='test-vm2' mode='required' " b"frontend_domain='test-vm' _ro='True'")] = b'0\0' assign = DeviceAssignment( - Device(Port( + VirtualDevice(Port( self.app.domains['test-vm2'], 'dev1', devclass='test')), mode='required') assign.options['ro'] = True @@ -171,7 +171,7 @@ def test_030_detach(self): ('test-vm', 'admin.vm.device.test.Detach', 'test-vm2+dev1:*', None)] = b'0\0' assign = DeviceAssignment( - Device(Port( + VirtualDevice(Port( self.app.domains['test-vm2'], 'dev1', devclass='test'))) self.vm.devices['test'].detach(assign) self.assertAllCalled() @@ -293,7 +293,7 @@ def test_070_update_assignment(self): ('test-vm', 'admin.vm.device.test.Set.assignment', 'test-vm2+dev1:*', b'True')] = b'0\0' dev = DeviceAssignment( - Device( + VirtualDevice( Port( self.app.domains['test-vm2'], devclass='test', @@ -307,7 +307,7 @@ def test_071_update_assignment_false(self): ('test-vm', 'admin.vm.device.test.Set.assignment', 'test-vm2+dev1:*', b'False')] = b'0\0' dev = DeviceAssignment( - Device( + VirtualDevice( Port( self.app.domains['test-vm2'], devclass='test', @@ -321,7 +321,7 @@ def test_072_update_assignment_none(self): ('test-vm', 'admin.vm.device.test.Set.assignment', 'test-vm2+dev1:*', b'None')] = b'0\0' dev = DeviceAssignment( - Device( + VirtualDevice( Port( self.app.domains['test-vm2'], devclass='test', diff --git a/qubesadmin/tools/qvm_device.py b/qubesadmin/tools/qvm_device.py index 31536ace..e10868fe 100644 --- a/qubesadmin/tools/qvm_device.py +++ b/qubesadmin/tools/qvm_device.py @@ -263,9 +263,8 @@ def info_device(args): vm = args.domains[0] if args.device: device = args.device - assignment = DeviceAssignment(device) - print("description:", assignment.device.description) - print("data:", assignment.device.data) + print("description:", device.description) + print("data:", device.data) else: for device_assignment in ( vm.devices[args.devclass].get_dedicated_devices()): diff --git a/qubesadmin/tools/qvm_start.py b/qubesadmin/tools/qvm_start.py index 485a8b32..f708b98a 100644 --- a/qubesadmin/tools/qvm_start.py +++ b/qubesadmin/tools/qvm_start.py @@ -158,7 +158,7 @@ def get_drive_assignment(app, drive_str): 'read-only': devtype == 'cdrom' } assignment = qubesadmin.device_protocol.DeviceAssignment( - qubesadmin.device_protocol.Device(qubesadmin.device_protocol.Port( + qubesadmin.device_protocol.VirtualDevice(qubesadmin.device_protocol.Port( backend_domain=backend_domain, port_id=port_id, devclass='block',