Skip to content
Open
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
49 changes: 49 additions & 0 deletions qubesadmin/device_protocol.py
Original file line number Diff line number Diff line change
Expand Up @@ -45,6 +45,29 @@
from qubesadmin.app import VMCollection


# duplicated from qubesadmin.utils to avoid an import cycle
def qbool(value):
"""
Property setter for boolean properties.

It accepts (case-insensitive) ``'0'``, ``'no'`` and ``false`` as
:py:obj:`False` and ``'1'``, ``'yes'`` and ``'true'`` as
:py:obj:`True`.
"""

if isinstance(value, str):
lcvalue = value.lower()
if lcvalue in ("0", "no", "false", "off"):
return False
if lcvalue in ("1", "yes", "true", "on"):
return True
raise QubesValueError(
"Invalid literal for boolean property: {!r}".format(value)
)

return bool(value)


class ProtocolError(AssertionError):
"""
Raised when something is wrong with data received.
Expand Down Expand Up @@ -908,6 +931,7 @@ def __init__(
parent: DeviceInfo | None = None,
attachment: QubesVM | None = None,
device_id: str | None = None,
busy: bool | None = None,
**kwargs,
):
super().__init__(port, device_id)
Expand All @@ -920,6 +944,7 @@ def __init__(
self._interfaces = interfaces
self._parent = parent
self._attachment = attachment
self._busy = busy

self.data = kwargs

Expand Down Expand Up @@ -1074,6 +1099,17 @@ def attachment(self) -> QubesVM | None:
"""
return self._attachment

@property
def busy(self) -> bool:
"""
Is the device currently busy (unavailable for attachment)?

True when a child device (partition or sub-device) is currently
passed through to a VM. The device remains visible but cannot be
attached until all children are detached.
"""
return bool(self._busy)

def serialize(self) -> bytes:
"""
Serialize an object to be transmitted via Qubes API.
Expand All @@ -1097,6 +1133,11 @@ def serialize(self) -> bytes:
"attachment", self.attachment.name
)

if self.busy:
properties += b" " + DeviceSerializer.pack_property(
"busy", "True"
)

properties += b" " + DeviceSerializer.pack_property(
"interfaces", "".join(repr(ifc) for ifc in self.interfaces)
)
Expand Down Expand Up @@ -1180,6 +1221,14 @@ def _deserialize(
del properties["parent_port_id"]
del properties["parent_devclass"]

# A missing property means free; otherwise parse strictly. An
# invalid value raises QubesValueError, which the caller
# (deserialize) turns into an UnknownDevice.
untrusted_busy = properties.pop("busy", None)
properties["busy"] = (
qbool(untrusted_busy) if untrusted_busy is not None else False
)

return cls(**properties)

@property
Expand Down