Skip to content

Commit

Permalink
q-dev: port
Browse files Browse the repository at this point in the history
  • Loading branch information
piotrbartman committed Oct 15, 2024
1 parent 58ae845 commit 0005d08
Show file tree
Hide file tree
Showing 10 changed files with 140 additions and 91 deletions.
4 changes: 2 additions & 2 deletions qubes/api/admin.py
Original file line number Diff line number Diff line change
Expand Up @@ -1343,7 +1343,7 @@ async def vm_device_unassign(self, endpoint):
self.fire_event_for_permission(device=dev, devclass=devclass)

assignment = qubes.device_protocol.DeviceAssignment(
dev.backend_domain, dev.ident, devclass=devclass)
qubes.device_protocol.Port(dev.backend_domain, dev.ident, devclass))
await self.dest.devices[devclass].unassign(assignment)
self.app.save()

Expand Down Expand Up @@ -1397,7 +1397,7 @@ async def vm_device_detach(self, endpoint):
self.fire_event_for_permission(device=dev, devclass=devclass)

assignment = qubes.device_protocol.DeviceAssignment(
dev.backend_domain, dev.ident, devclass=devclass)
qubes.device_protocol.Port(dev.backend_domain, dev.ident, devclass))
await self.dest.devices[devclass].detach(assignment)

# Assign/Unassign action can modify only a persistent state of running VM.
Expand Down
70 changes: 39 additions & 31 deletions qubes/device_protocol.py
Original file line number Diff line number Diff line change
Expand Up @@ -243,8 +243,8 @@ class DeviceCategory(Enum):
Mouse = ("u03**02", "p0902**")
Printer = ("u07****",)
Scanner = ("p0903**",)
# Multimedia = Audio, Video, Displays etc.
Microphone = ("m******",)
# Multimedia = Audio, Video, Displays etc.
Multimedia = ("u01****", "u0e****", "u06****", "u10****", "p03****",
"p04****")
Wireless = ("ue0****", "p0d****")
Expand Down Expand Up @@ -349,7 +349,7 @@ def __eq__(self, other):

def __str__(self):
if self.devclass == "block":
return "Block device"
return "Block Device"
if self.devclass in ("usb", "pci"):
# try subclass first as in `lspci`
result = self._load_classes(self.devclass).get(
Expand Down Expand Up @@ -538,8 +538,10 @@ def description(self) -> str:
else:
vendor = "unknown vendor"

main_interface = str(self.interfaces[0])
return f"{main_interface}: {vendor} {prod}"
cat = self.interfaces[0].category.name
if cat == "Other":
cat = str(self.interfaces[0])
return f"{cat}: {vendor} {prod}"

@property
def interfaces(self) -> List[DeviceInterface]:
Expand Down Expand Up @@ -784,44 +786,44 @@ class DeviceAssignment(Port):
and required to start domain.
"""

def __init__(self, backend_domain, ident, *, options=None,
frontend_domain=None, devclass=None,
required=False, attach_automatically=False):
super().__init__(backend_domain, ident, devclass)
class AssignmentType(Enum):
MANUAL = 0
ASK = 1
AUTO = 2
REQUIRED = 3

def __init__(
self,
port: Port,
frontend_domain=None,
options=None,
required=False,
attach_automatically=False
):
super().__init__(port.backend_domain, port.ident, port.devclass)
self.__options = options or {}
if required:
assert attach_automatically
self.__required = required
self.__attach_automatically = attach_automatically
self.type = DeviceAssignment.AssignmentType.REQUIRED
elif attach_automatically:
self.type = DeviceAssignment.AssignmentType.AUTO
else:
self.type = DeviceAssignment.AssignmentType.MANUAL
self.frontend_domain = frontend_domain

def clone(self, **kwargs):
"""
Clone object and substitute attributes with explicitly given.
"""
attr = {
"backend_domain": self.backend_domain,
"ident": self.ident,
"options": self.options,
"required": self.required,
"attach_automatically": self.attach_automatically,
"frontend_domain": self.frontend_domain,
"devclass": self.devclass,
}
attr.update(kwargs)
return self.__class__(**attr)

@classmethod
def from_device(cls, device: Port, **kwargs) -> 'DeviceAssignment':
"""
Get assignment of the device.
"""
return cls(
backend_domain=device.backend_domain,
ident=device.ident,
devclass=device.devclass,
**kwargs
)
return self.__class__(
Port(self.backend_domain, self.ident, self.devclass), **attr)

@property
def device(self) -> DeviceInfo:
Expand Down Expand Up @@ -857,23 +859,26 @@ def required(self) -> bool:
Is the presence of this device required for the domain to start? If yes,
it will be attached automatically.
"""
return self.__required
return self.type == DeviceAssignment.AssignmentType.REQUIRED

@required.setter
def required(self, required: bool):
self.__required = required
self.type = DeviceAssignment.AssignmentType.REQUIRED

@property
def attach_automatically(self) -> bool:
"""
Should this device automatically connect to the frontend domain when
available and not connected to other qubes?
"""
return self.__attach_automatically
return self.type in (
DeviceAssignment.AssignmentType.AUTO,
DeviceAssignment.AssignmentType.REQUIRED
)

@attach_automatically.setter
def attach_automatically(self, attach_automatically: bool):
self.__attach_automatically = attach_automatically
self.type = DeviceAssignment.AssignmentType.AUTO

@property
def options(self) -> Dict[str, Any]:
Expand Down Expand Up @@ -938,9 +943,12 @@ def _deserialize(
properties['options'] = options

cls.check_device_properties(expected_port, properties)
del properties['backend_domain']
del properties['ident']
del properties['devclass']

properties['attach_automatically'] = qbool(
properties.get('attach_automatically', 'no'))
properties['required'] = qbool(properties.get('required', 'no'))

return cls(**properties)
return cls(expected_port, **properties)
8 changes: 5 additions & 3 deletions qubes/devices.py
Original file line number Diff line number Diff line change
Expand Up @@ -348,9 +348,11 @@ def get_attached_devices(self) -> Iterable[DeviceAssignment]:
break
else:
yield DeviceAssignment(
backend_domain=dev.backend_domain,
ident=dev.ident,
devclass=dev.devclass,
Port(
backend_domain=dev.backend_domain,
ident=dev.ident,
devclass=dev.devclass,
),
frontend_domain=self._vm,
options=options,
attach_automatically=False,
Expand Down
40 changes: 26 additions & 14 deletions qubes/tests/api_admin.py
Original file line number Diff line number Diff line change
Expand Up @@ -1786,7 +1786,7 @@ def test_462_vm_device_available_invalid(self):

def test_470_vm_device_list_assigned(self):
assignment = qubes.device_protocol.DeviceAssignment(
self.vm, '1234', 'test',
qubes.device_protocol.Port(self.vm, '1234', 'test'),
attach_automatically=True, required=True)
self.loop.run_until_complete(
self.vm.devices['testclass'].assign(assignment))
Expand All @@ -1799,7 +1799,7 @@ def test_470_vm_device_list_assigned(self):

def test_471_vm_device_list_assigned_options(self):
assignment = qubes.device_protocol.DeviceAssignment(
self.vm, '1234', 'test',
qubes.device_protocol.Port(self.vm, '1234', 'test'),
attach_automatically=True, required=True, options={'opt1': 'value'})
self.loop.run_until_complete(
self.vm.devices['testclass'].assign(assignment))
Expand Down Expand Up @@ -1837,11 +1837,13 @@ 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', 'test', attach_automatically=True, required=True)
qubes.device_protocol.Port(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', 'test', attach_automatically=True, required=True)
qubes.device_protocol.Port(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',
Expand Down Expand Up @@ -2024,7 +2026,8 @@ 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', 'test', attach_automatically=True, required=False,
qubes.device_protocol.Port(self.vm, '1234', 'test'),
attach_automatically=True, required=False,
options={'opt1': 'value'})
self.loop.run_until_complete(
self.vm.devices['testclass'].assign(assignment))
Expand All @@ -2044,7 +2047,8 @@ 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', 'test', attach_automatically=True, required=True,
qubes.device_protocol.Port(self.vm, '1234', 'test'),
attach_automatically=True, required=True,
options={'opt1': 'value'})
self.loop.run_until_complete(
self.vm.devices['testclass'].assign(assignment))
Expand All @@ -2065,7 +2069,8 @@ 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', 'test', attach_automatically=True, required=False,
qubes.device_protocol.Port(self.vm, '1234', 'test'),
attach_automatically=True, required=False,
options={'opt1': 'value'})
self.loop.run_until_complete(
self.vm.devices['testclass'].assign(assignment))
Expand All @@ -2083,7 +2088,8 @@ 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', 'test', attach_automatically=True, required=True,
qubes.device_protocol.Port(self.vm, '1234', 'test'),
attach_automatically=True, required=True,
options={'opt1': 'value'})
self.loop.run_until_complete(
self.vm.devices['testclass'].assign(assignment))
Expand All @@ -2104,7 +2110,8 @@ 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', 'test', attach_automatically=True, required=False,
qubes.device_protocol.Port(self.vm, '1234', 'test'),
attach_automatically=True, required=False,
options={'opt1': 'value'})
self.loop.run_until_complete(
self.vm.devices['testclass'].assign(assignment))
Expand Down Expand Up @@ -2191,7 +2198,8 @@ 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', 'test', attach_automatically=True, required=True)
qubes.device_protocol.Port(self.vm, '1234', 'test'),
attach_automatically=True, required=True)
self.loop.run_until_complete(
self.vm2.devices['testclass'].assign(assignment))

Expand Down Expand Up @@ -2910,7 +2918,8 @@ 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', 'test', attach_automatically=True, required=False,
qubes.device_protocol.Port(self.vm, '1234', 'test'),
attach_automatically=True, required=False,
options={'opt1': 'value'})
self.loop.run_until_complete(
self.vm.devices['testclass'].assign(assignment))
Expand Down Expand Up @@ -2940,7 +2949,8 @@ 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', 'test', attach_automatically=True, required=True,
qubes.device_protocol.Port(self.vm, '1234', 'test'),
attach_automatically=True, required=True,
options={'opt1': 'value'})
self.loop.run_until_complete(
self.vm.devices['testclass'].assign(assignment))
Expand Down Expand Up @@ -2970,7 +2980,8 @@ 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', 'test', attach_automatically=True, required=True,
qubes.device_protocol.Port(self.vm, '1234', 'test'),
attach_automatically=True, required=True,
options={'opt1': 'value'})
self.loop.run_until_complete(
self.vm.devices['testclass'].assign(assignment))
Expand All @@ -2988,7 +2999,8 @@ 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', 'test', attach_automatically=True, required=False,
qubes.device_protocol.Port(self.vm, '1234', 'test'),
attach_automatically=True, required=False,
options={'opt1': 'value'})
self.loop.run_until_complete(
self.vm.devices['testclass'].assign(assignment))
Expand Down
Loading

0 comments on commit 0005d08

Please sign in to comment.