Skip to content

Commit 34d5e8d

Browse files
committed
Sync device_protocol with core-admin
Update mostly type hints.
1 parent d90bd1f commit 34d5e8d

File tree

1 file changed

+35
-22
lines changed

1 file changed

+35
-22
lines changed

qubesadmin/device_protocol.py

Lines changed: 35 additions & 22 deletions
Original file line numberDiff line numberDiff line change
@@ -5,10 +5,10 @@
55
# Copyright (C) 2010-2016 Joanna Rutkowska <[email protected]>
66
# Copyright (C) 2015-2016 Wojtek Porczyk <[email protected]>
77
# Copyright (C) 2016 Bahtiar `kalkin-` Gadimov <[email protected]>
8-
# Copyright (C) 2017 Marek Marczykowski-Górecki
8+
# Copyright (C) 2017 Marek Marczykowski-Górecki
99
1010
# Copyright (C) 2024 Piotr Bartman-Szwarc
11-
11+
1212
#
1313
# This library is free software; you can redistribute it and/or
1414
# modify it under the terms of the GNU Lesser General Public
@@ -29,6 +29,8 @@
2929
The same in `qubes-core-admin` and `qubes-core-admin-client`,
3030
should be moved to one place.
3131
"""
32+
33+
3234
import string
3335
import sys
3436
from enum import Enum
@@ -102,8 +104,8 @@ def unpack_properties(
102104
"ascii", errors="strict"
103105
).strip()
104106

105-
properties = {}
106-
options = {}
107+
properties: Dict[str, str] = {}
108+
options: Dict[str, str] = {}
107109

108110
if not ut_decoded:
109111
return properties, options
@@ -241,7 +243,7 @@ def deserialize_str(value: str) -> str:
241243
def sanitize_str(
242244
untrusted_value: str,
243245
allowed_chars: set,
244-
replace_char: str = None,
246+
replace_char: Optional[str] = None,
245247
error_message: str = "",
246248
) -> str:
247249
"""
@@ -276,7 +278,10 @@ class Port:
276278
"""
277279

278280
def __init__(
279-
self, backend_domain: Optional[QubesVM], port_id: str, devclass: str
281+
self,
282+
backend_domain: Optional[QubesVM],
283+
port_id: Optional[str],
284+
devclass: Optional[str],
280285
):
281286
self.__backend_domain = backend_domain
282287
self.__port_id = port_id
@@ -390,6 +395,7 @@ def has_devclass(self):
390395

391396
class AnyPort(Port):
392397
"""Represents any port in virtual devices ("*")"""
398+
393399
def __init__(self, devclass: str):
394400
super().__init__(None, "*", devclass)
395401

@@ -415,14 +421,14 @@ def __init__(
415421
device_id: Optional[str] = None,
416422
):
417423
assert not isinstance(port, AnyPort) or device_id is not None
418-
self.port: Optional[Port] = port
424+
self.port: Optional[Port] = port # type: ignore
419425
self._device_id = device_id
420426

421427
def clone(self, **kwargs) -> "VirtualDevice":
422428
"""
423429
Clone object and substitute attributes with explicitly given.
424430
"""
425-
attr = {
431+
attr: Dict[str, Any] = {
426432
"port": self.port,
427433
"device_id": self.device_id,
428434
}
@@ -449,7 +455,7 @@ def port(self, value: Union[Port, str, None]):
449455
@property
450456
def device_id(self) -> str:
451457
# pylint: disable=missing-function-docstring
452-
if self.is_device_id_set:
458+
if self._device_id is not None and self.is_device_id_set:
453459
return self._device_id
454460
return "*"
455461

@@ -487,7 +493,7 @@ def description(self) -> str:
487493
"""
488494
Return human-readable description of the device identity.
489495
"""
490-
if self.device_id == "*":
496+
if not self.device_id or self.device_id == "*":
491497
return "any device"
492498
return self.device_id
493499

@@ -601,12 +607,11 @@ def _parse(
601607
backend = get_domain(backend_name)
602608
else:
603609
identity = representation
610+
604611
port_id, _, devid = identity.partition(":")
605-
if devid == "":
606-
devid = None
607612
return cls(
608613
Port(backend_domain=backend, port_id=port_id, devclass=devclass),
609-
device_id=devid,
614+
device_id=devid or None,
610615
)
611616

612617
def serialize(self) -> bytes:
@@ -870,7 +875,7 @@ def __init__(
870875
name: Optional[str] = None,
871876
serial: Optional[str] = None,
872877
interfaces: Optional[List[DeviceInterface]] = None,
873-
parent: Optional[Port] = None,
878+
parent: Optional["DeviceInfo"] = None,
874879
attachment: Optional[QubesVM] = None,
875880
device_id: Optional[str] = None,
876881
**kwargs,
@@ -1023,6 +1028,8 @@ def subdevices(self) -> List[VirtualDevice]:
10231028
If the device has subdevices (e.g., partitions of a USB stick),
10241029
the subdevices id should be here.
10251030
"""
1031+
if not self.backend_domain:
1032+
return []
10261033
return [
10271034
dev
10281035
for devclass in self.backend_domain.devices.keys()
@@ -1124,7 +1131,7 @@ def _deserialize(
11241131

11251132
if "attachment" not in properties or not properties["attachment"]:
11261133
properties["attachment"] = None
1127-
else:
1134+
elif expected_device.backend_domain:
11281135
app = expected_device.backend_domain.app
11291136
properties["attachment"] = app.domains.get_blind(
11301137
properties["attachment"]
@@ -1281,7 +1288,7 @@ def __lt__(self, other):
12811288
)
12821289

12831290
@property
1284-
def backend_domain(self) -> QubesVM:
1291+
def backend_domain(self) -> Optional[QubesVM]:
12851292
# pylint: disable=missing-function-docstring
12861293
return self.virtual_device.backend_domain
12871294

@@ -1308,14 +1315,15 @@ def device_id(self) -> str:
13081315
@property
13091316
def devices(self) -> List[DeviceInfo]:
13101317
"""Get DeviceInfo objects corresponding to this DeviceAssignment"""
1318+
result: List[DeviceInfo] = []
1319+
if not self.backend_domain:
1320+
return result
13111321
if self.port_id != "*":
13121322
dev = self.backend_domain.devices[self.devclass][self.port_id]
1313-
if (
1314-
isinstance(dev, UnknownDevice)
1315-
or dev.device_id == self.device_id
1323+
if isinstance(dev, UnknownDevice) or (
1324+
dev and self.device_id in (dev.device_id, "*")
13161325
):
13171326
return [dev]
1318-
result = []
13191327
if self.device_id == "0000:0000::?******":
13201328
return result
13211329
for dev in self.backend_domain.devices[self.devclass]:
@@ -1355,8 +1363,13 @@ def frontend_domain(self) -> Optional[QubesVM]:
13551363
def frontend_domain(self, frontend_domain: Optional[Union[str, QubesVM]]):
13561364
"""Which domain the device is attached/assigned to."""
13571365
if isinstance(frontend_domain, str):
1358-
frontend_domain = self.backend_domain.app.domains[frontend_domain]
1359-
self.__frontend_domain = frontend_domain
1366+
if not self.backend_domain:
1367+
raise ProtocolError("Cannot determine backend domain")
1368+
self.__frontend_domain: Optional[QubesVM] = (
1369+
self.backend_domain.app.domains[frontend_domain]
1370+
)
1371+
else:
1372+
self.__frontend_domain = frontend_domain
13601373

13611374
@property
13621375
def attached(self) -> bool:

0 commit comments

Comments
 (0)