Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Make "multimedia" devices more detailed #322

Merged
merged 2 commits into from
Dec 17, 2024
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
62 changes: 38 additions & 24 deletions qubesadmin/device_protocol.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,10 +5,10 @@
# Copyright (C) 2010-2016 Joanna Rutkowska <[email protected]>
# Copyright (C) 2015-2016 Wojtek Porczyk <[email protected]>
# Copyright (C) 2016 Bahtiar `kalkin-` Gadimov <[email protected]>
# Copyright (C) 2017 Marek Marczykowski-Górecki
# Copyright (C) 2017 Marek Marczykowski-Górecki
# <[email protected]>
# Copyright (C) 2024 Piotr Bartman-Szwarc
# <[email protected]>
# <[email protected]>
#
# This library is free software; you can redistribute it and/or
# modify it under the terms of the GNU Lesser General Public
Expand All @@ -29,6 +29,8 @@
The same in `qubes-core-admin` and `qubes-core-admin-client`,
should be moved to one place.
"""


import string
import sys
from enum import Enum
Expand Down Expand Up @@ -102,8 +104,8 @@
"ascii", errors="strict"
).strip()

properties = {}
options = {}
properties: Dict[str, str] = {}
options: Dict[str, str] = {}

if not ut_decoded:
return properties, options
Expand Down Expand Up @@ -241,7 +243,7 @@
def sanitize_str(
untrusted_value: str,
allowed_chars: set,
replace_char: str = None,
replace_char: Optional[str] = None,
error_message: str = "",
) -> str:
"""
Expand Down Expand Up @@ -276,7 +278,10 @@
"""

def __init__(
self, backend_domain: Optional[QubesVM], port_id: str, devclass: str
self,
backend_domain: Optional[QubesVM],
port_id: Optional[str],
devclass: Optional[str],
):
self.__backend_domain = backend_domain
self.__port_id = port_id
Expand Down Expand Up @@ -390,6 +395,7 @@

class AnyPort(Port):
"""Represents any port in virtual devices ("*")"""

def __init__(self, devclass: str):
super().__init__(None, "*", devclass)

Expand All @@ -415,14 +421,14 @@
device_id: Optional[str] = None,
):
assert not isinstance(port, AnyPort) or device_id is not None
self.port: Optional[Port] = port
self.port: Optional[Port] = port # type: ignore
self._device_id = device_id

def clone(self, **kwargs) -> "VirtualDevice":
"""
Clone object and substitute attributes with explicitly given.
"""
attr = {
attr: Dict[str, Any] = {
"port": self.port,
"device_id": self.device_id,
}
Expand All @@ -449,7 +455,7 @@
@property
def device_id(self) -> str:
# pylint: disable=missing-function-docstring
if self.is_device_id_set:
if self._device_id is not None and self.is_device_id_set:
return self._device_id
return "*"

Expand Down Expand Up @@ -487,7 +493,7 @@
"""
Return human-readable description of the device identity.
"""
if self.device_id == "*":
if not self.device_id or self.device_id == "*":
return "any device"
return self.device_id

Expand Down Expand Up @@ -601,12 +607,11 @@
backend = get_domain(backend_name)
else:
identity = representation

port_id, _, devid = identity.partition(":")
if devid == "":
devid = None
return cls(
Port(backend_domain=backend, port_id=port_id, devclass=devclass),
device_id=devid,
device_id=devid or None,
)

def serialize(self) -> bytes:
Expand Down Expand Up @@ -646,13 +651,14 @@
Microphone = ("m******",)
# Multimedia = Audio, Video, Displays etc.
Multimedia = (
"u01****",
"u0e****",
"u06****",
"u10****",
"p03****",
"p04****",
)
Audio = ("p0403**", "u01****")
Display = ("p0300**", "p0380**")
Video = ("p0400**", "u0e****")
Wireless = ("ue0****", "p0d****")
Bluetooth = ("ue00101", "p0d11**")
Storage = ("b******", "u08****", "p01****")
Expand Down Expand Up @@ -869,7 +875,7 @@
name: Optional[str] = None,
serial: Optional[str] = None,
interfaces: Optional[List[DeviceInterface]] = None,
parent: Optional[Port] = None,
parent: Optional["DeviceInfo"] = None,
attachment: Optional[QubesVM] = None,
device_id: Optional[str] = None,
**kwargs,
Expand Down Expand Up @@ -1022,6 +1028,8 @@
If the device has subdevices (e.g., partitions of a USB stick),
the subdevices id should be here.
"""
if not self.backend_domain:
return []

Check warning on line 1032 in qubesadmin/device_protocol.py

View check run for this annotation

Codecov / codecov/patch

qubesadmin/device_protocol.py#L1031-L1032

Added lines #L1031 - L1032 were not covered by tests
return [
dev
for devclass in self.backend_domain.devices.keys()
Expand Down Expand Up @@ -1123,7 +1131,7 @@

if "attachment" not in properties or not properties["attachment"]:
properties["attachment"] = None
else:
elif expected_device.backend_domain:

Check warning on line 1134 in qubesadmin/device_protocol.py

View check run for this annotation

Codecov / codecov/patch

qubesadmin/device_protocol.py#L1134

Added line #L1134 was not covered by tests
app = expected_device.backend_domain.app
properties["attachment"] = app.domains.get_blind(
properties["attachment"]
Expand Down Expand Up @@ -1280,7 +1288,7 @@
)

@property
def backend_domain(self) -> QubesVM:
def backend_domain(self) -> Optional[QubesVM]:
# pylint: disable=missing-function-docstring
return self.virtual_device.backend_domain

Expand All @@ -1307,14 +1315,15 @@
@property
def devices(self) -> List[DeviceInfo]:
"""Get DeviceInfo objects corresponding to this DeviceAssignment"""
result: List[DeviceInfo] = []
if not self.backend_domain:
return result

Check warning on line 1320 in qubesadmin/device_protocol.py

View check run for this annotation

Codecov / codecov/patch

qubesadmin/device_protocol.py#L1320

Added line #L1320 was not covered by tests
if self.port_id != "*":
dev = self.backend_domain.devices[self.devclass][self.port_id]
if (
isinstance(dev, UnknownDevice)
or dev.device_id == self.device_id
if isinstance(dev, UnknownDevice) or (
dev and self.device_id in (dev.device_id, "*")
):
return [dev]
result = []
if self.device_id == "0000:0000::?******":
return result
for dev in self.backend_domain.devices[self.devclass]:
Expand Down Expand Up @@ -1354,8 +1363,13 @@
def frontend_domain(self, frontend_domain: Optional[Union[str, QubesVM]]):
"""Which domain the device is attached/assigned to."""
if isinstance(frontend_domain, str):
frontend_domain = self.backend_domain.app.domains[frontend_domain]
self.__frontend_domain = frontend_domain
if not self.backend_domain:
raise ProtocolError("Cannot determine backend domain")

Check warning on line 1367 in qubesadmin/device_protocol.py

View check run for this annotation

Codecov / codecov/patch

qubesadmin/device_protocol.py#L1367

Added line #L1367 was not covered by tests
self.__frontend_domain: Optional[QubesVM] = (
self.backend_domain.app.domains[frontend_domain]
)
else:
self.__frontend_domain = frontend_domain

@property
def attached(self) -> bool:
Expand Down
6 changes: 3 additions & 3 deletions qubesadmin/tests/tools/qvm_device.py
Original file line number Diff line number Diff line change
Expand Up @@ -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 Multimedia: itl test-device',
['test-vm1:dev1 Audio: itl test-device',
'test-vm2:dev2 ?******: ? ` test-device']
)

Expand Down Expand Up @@ -155,7 +155,7 @@ def test_002_list_attach(self):
['testclass', 'list', 'test-vm3'], app=self.app)
self.assertEqual(
buf.getvalue(),
'test-vm1:dev1 Multimedia: itl test-device '
'test-vm1:dev1 Audio: itl test-device '
'test-vm3 (attached)\n'
)

Expand Down Expand Up @@ -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('Multimedia: itl test-device\n'
self.assertIn('Audio: itl test-device\n'
'device ID: dead:beef:babe:u012345',
buf.getvalue())
self.assertAllCalled()