Make mypy happy and add CI job

Author: fepitre

.gitlab-ci.yml

@@ -6,6 +6,7 @@ checks:pylint:
   script:
   - PYTHONPATH=test-packages:~/qubes-core-qrexec python3 -m pylint qubes
   stage: checks
+
 checks:tests:
   after_script:
   - ci/codecov-wrapper -F unittests
@@ -37,6 +38,21 @@ checks:tests:
   stage: checks
   tags:
   - vm-kvm
+
+mypy:
+  stage: checks
+  image: fedora:40
+  tags:
+    - docker
+  before_script:
+    - sudo dnf install -y python3-mypy python3-pip
+    - sudo python3 -m pip install lxml-stubs types-docutils types-pywin32
+  script:
+    - mypy --install-types --non-interactive --ignore-missing-imports --exclude tests/ --junit-xml mypy.xml qubes
+  artifacts:
+    reports:
+      junit: mypy.xml
+
 include:
 - file: /r4.3/gitlab-base.yml
   project: QubesOS/qubes-continuous-integration

qubes/api/__init__.py

@@ -135,7 +135,7 @@ class AbstractQubesAPI:
     '''
 
     #: the preferred socket location (to be overridden in child's class)
-    SOCKNAME = None
+    SOCKNAME = ""
 
     app: qubes.Qubes
     src: qubes.vm.qubesvm.QubesVM
@@ -144,7 +144,7 @@ def __init__(self,
                  src: bytes,
                  method_name: bytes,
                  dest: bytes,
-                 arg: qubes.Qubes,
+                 arg: bytes,
                  send_event: Any = None) -> None:
         #: :py:class:`qubes.Qubes` object
         self.app = app

qubes/config.py

@@ -28,7 +28,26 @@
 
 import os.path
 
+from typing import TypedDict, Dict
+
+class PoolConfig(TypedDict, total=False):
+    dir_path: str
+    name: str
+    driver: str
+
+class Defaults(TypedDict):
+    libvirt_uri: str
+    memory: int
+    hvm_memory: int
+    kernelopts: str
+    kernelopts_pcidevs: str
+    kernelopts_common: str
+    private_img_size: int
+    root_img_size: int
+    pool_configs: Dict[str, PoolConfig]
+
 qubes_base_dir = "/var/lib/qubes"
+
 system_path = {
     'qrexec_daemon_path': '/usr/sbin/qrexec-daemon',
     'qrexec_client_path': '/usr/bin/qrexec-client',
@@ -48,7 +67,7 @@
     'dom0_services_dir': '/var/run/qubes-service',
 }
 
-defaults = {
+defaults: Defaults = {
     'libvirt_uri': 'xen:///',
     'memory': 400,
     'hvm_memory': 400,
@@ -67,8 +86,9 @@
             'name': 'varlibqubes'
         },
         'linux-kernel': {
-            'dir_path': os.path.join(qubes_base_dir,
-                                     system_path['qubes_kernels_base_dir']),
+            'dir_path': os.path.join(
+                qubes_base_dir, system_path['qubes_kernels_base_dir']
+            ),
             'driver': 'linux-kernel',
             'name': 'linux-kernel'
         }

qubes/device_protocol.py

@@ -35,12 +35,15 @@
 import sys
 from enum import Enum
 from typing import Optional, Dict, Any, List, Union, Tuple, Callable
+from typing import TYPE_CHECKING
 
 import qubes.utils
-
 from qubes.exc import ProtocolError
 
-QubesVM = "qubes.vm.BaseVM"
+if TYPE_CHECKING:
+    from qubes.vm.qubesvm import QubesVM
+else:
+    QubesVM = "qubes.vm.qubesvm.QubesVM"
 
 
 class UnexpectedDeviceProperty(qubes.exc.QubesException, ValueError):
@@ -82,8 +85,8 @@ def unpack_properties(
             "ascii", errors="strict"
         ).strip()
 
-        properties = {}
-        options = {}
+        properties: Dict[str, str] = {}
+        options: Dict[str, str] = {}
 
         if not ut_decoded:
             return properties, options
@@ -221,7 +224,7 @@ def deserialize_str(value: str) -> str:
     def sanitize_str(
         untrusted_value: str,
         allowed_chars: set,
-        replace_char: str = None,
+        replace_char: Optional[str] = None,
         error_message: str = "",
     ) -> str:
         """
@@ -256,7 +259,10 @@ class Port:
     """
 
     def __init__(
-        self, backend_domain: Optional[QubesVM], port_id: str, devclass: str
+        self,
+        backend_domain: Optional[QubesVM],
+        port_id: str,
+        devclass: Optional[str],
     ):
         self.__backend_domain = backend_domain
         self.__port_id = port_id
@@ -370,6 +376,7 @@ def has_devclass(self):
 
 class AnyPort(Port):
     """Represents any port in virtual devices ("*")"""
+
     def __init__(self, devclass: str):
         super().__init__(None, "*", devclass)
 
@@ -395,14 +402,14 @@ def __init__(
         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,
         }
@@ -429,7 +436,7 @@ def port(self, value: Union[Port, str, None]):
     @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 "*"
 
@@ -467,7 +474,7 @@ def description(self) -> str:
         """
         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
 
@@ -581,12 +588,11 @@ def _parse(
                 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:
@@ -849,7 +855,7 @@ def __init__(
         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,
@@ -1002,6 +1008,8 @@ def subdevices(self) -> List[VirtualDevice]:
         If the device has subdevices (e.g., partitions of a USB stick),
         the subdevices id should be here.
         """
+        if not self.backend_domain:
+            return []
         return [
             dev
             for devclass in self.backend_domain.devices.keys()
@@ -1103,7 +1111,7 @@ def _deserialize(
 
         if "attachment" not in properties or not properties["attachment"]:
             properties["attachment"] = None
-        else:
+        elif expected_device.backend_domain:
             app = expected_device.backend_domain.app
             properties["attachment"] = app.domains.get_blind(
                 properties["attachment"]
@@ -1260,7 +1268,7 @@ def __lt__(self, other):
         )
 
     @property
-    def backend_domain(self) -> QubesVM:
+    def backend_domain(self) -> Optional[QubesVM]:
         # pylint: disable=missing-function-docstring
         return self.virtual_device.backend_domain
 
@@ -1287,14 +1295,16 @@ def device_id(self) -> str:
     @property
     def devices(self) -> List[DeviceInfo]:
         """Get DeviceInfo objects corresponding to this DeviceAssignment"""
+        result: List[DeviceInfo] = []
+        if not self.backend_domain:
+            return result
         if self.port_id != "*":
             dev = self.backend_domain.devices[self.devclass][self.port_id]
             if (
                 isinstance(dev, UnknownDevice)
                 or 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]:
@@ -1334,8 +1344,13 @@ def frontend_domain(self) -> Optional[QubesVM]:
     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")
+            self.__frontend_domain: Optional[QubesVM] = (
+                self.backend_domain.app.domains[frontend_domain]
+            )
+        else:
+            self.__frontend_domain = frontend_domain
 
     @property
     def attached(self) -> bool:

qubes/dochelpers.py

@@ -51,7 +51,7 @@
 try:
     log = logging.getLogger(__name__)
 except AttributeError:
-    log = None
+    log = None  # type: ignore
 
 class GithubTicket:
     # pylint: disable=too-few-public-methods
@@ -89,7 +89,8 @@ def ticket(name, rawtext, text, lineno, inliner, options=None, content=None):
         that called this function
     :param options: Directive options for customisation
     :param content: The directive content for customisation
-    """  # pylint: disable=unused-argument,too-many-positional-arguments
+    """
+    # pylint: disable=unused-argument,too-many-positional-arguments
 
     if options is None:
         options = {}

qubes/ext/__init__.py

@@ -24,7 +24,7 @@
 some systems. They may be OS- or architecture-dependent or custom-developed for
 particular customer.
 '''
-
+import collections
 import importlib.metadata
 import qubes.events
 
@@ -53,6 +53,13 @@ def __new__(cls):
 
         return cls._instance
 
+    def __init__(self):
+        #: This is to be implemented in extension handling devices
+        self.devices_cache = collections.defaultdict(dict)
+
+    #: This is to be implemented in extension handling devices
+    def ensure_detach(self, vm, port):
+        pass
 
 def get_extensions():
     return set(ext.load()()

qubes/ext/admin.py

@@ -169,7 +169,7 @@ def on_tag_add(self, vm, event, tag, **kwargs):
     @qubes.ext.handler(*(f'admin-permission:admin.vm.device.{ep.name}.Attach'
         for ep in importlib.metadata.entry_points(group='qubes.devices')))
     def on_device_attach(
-        self, vm, event, dest, arg, device, mode, options, **kwargs
+        self, vm, event, dest, arg, device, mode, **kwargs
     ):
         # pylint: disable=unused-argument,too-many-positional-arguments
         # ignore auto-attachment
@@ -210,9 +210,8 @@ def _load_deny_list(deny: dict, path: str) -> None:
 
                     if line:
                         name, *values = line.split()
-
                         values = ' '.join(values).replace(',', ' ').split()
-                        values = {v for v in values if len(v) > 0}
+                        values = [v for v in values if len(v) > 0]
 
                         deny[name] = deny.get(name, set()).union(set(values))
         except IOError: