Move device-related exceptions to the exc module

qubes/device_protocol.py

@@ -38,20 +38,14 @@
 from typing import TYPE_CHECKING
 
 import qubes.utils
-from qubes.exc import ProtocolError, QubesValueError
+from qubes.exc import ProtocolError, QubesValueError, UnexpectedDeviceProperty
 
 if TYPE_CHECKING:
     from qubes.vm.qubesvm import QubesVM
 else:
     QubesVM = "qubes.vm.qubesvm.QubesVM"
 
 
-class UnexpectedDeviceProperty(qubes.exc.QubesException, ValueError):
-    """
-    Device has unexpected property such as backend_domain, devclass etc.
-    """
-
-
 def qbool(value):
     return qubes.property.bool(None, None, value)
 

qubes/devices.py

@@ -71,31 +71,12 @@
     VirtualDevice,
     AssignmentMode,
 )
-from qubes.exc import ProtocolError
-
-
-class DeviceNotAssigned(qubes.exc.QubesException, KeyError):
-    """
-    Trying to unassign not assigned device.
-    """
-
-
-class DeviceAlreadyAttached(qubes.exc.QubesException, KeyError):
-    """
-    Trying to attach already attached device.
-    """
-
-
-class DeviceAlreadyAssigned(qubes.exc.QubesException, KeyError):
-    """
-    Trying to assign already assigned device.
-    """
-
-
-class UnrecognizedDevice(qubes.exc.QubesException, ValueError):
-    """
-    Device identity is not as expected.
-    """
+from qubes.exc import (
+    ProtocolError,
+    DeviceNotAssigned,
+    DeviceAlreadyAttached,
+    DeviceAlreadyAssigned,
+)
 
 
 class DeviceCollection:

qubes/exc.py

@@ -269,3 +269,37 @@ class ProtocolError(AssertionError):
 
 class PermissionDenied(Exception):
     """Raised deliberately by handlers when we decide not to cooperate"""
+
+
+class DeviceNotAssigned(QubesException, KeyError):
+    """
+    Trying to unassign not assigned device.
+    """
+
+
+class DeviceAlreadyAttached(QubesException, KeyError):
+    """
+    Trying to attach already attached device.
+    """
+
+
+class DeviceAlreadyAssigned(QubesException, KeyError):
+    """
+    Trying to assign already assigned device.
+    """
+
+
+class UnrecognizedDevice(QubesException, ValueError):
+    """
+    Device identity is not as expected.
+    """
+
+
+class UnexpectedDeviceProperty(QubesException, ValueError):
+    """
+    Device has unexpected property such as backend_domain, devclass etc.
+    """
+
+
+class StoragePoolException(QubesException):
+    """A general storage exception"""

qubes/ext/block.py

@@ -30,6 +30,7 @@
 
 import qubes.device_protocol
 import qubes.devices
+import qubes.exc
 import qubes.ext
 from qubes.ext import utils
 from qubes.storage import Storage
@@ -514,7 +515,7 @@
                 "not available, skipping.",
                 file=sys.stderr,
             )
-            raise qubes.devices.UnrecognizedDevice()
+            raise qubes.exc.UnrecognizedDevice()
 
         # validate options
         for option, value in options.items():
@@ -561,7 +562,7 @@
             return
 
         if device.attachment and device.attachment != expected_attachment:
-            raise qubes.devices.DeviceAlreadyAttached(
+            raise qubes.exc.DeviceAlreadyAttached(
                 "Device {!s} already attached to {!s}".format(
                     device, device.attachment
                 )

qubes/storage/__init__.py

@@ -38,6 +38,7 @@
 import qubes
 import qubes.exc
 import qubes.utils
+from qubes.exc import StoragePoolException
 
 STORAGE_ENTRY_POINT = "qubes.storage"
 _am_root = os.getuid() == 0
@@ -46,10 +47,6 @@
 _big_buffer = b"\0" * BYTES_TO_ZERO
 
 
-class StoragePoolException(qubes.exc.QubesException):
-    """A general storage exception"""
-
-
 class BlockDevice:
     """Represents a storage block device."""
 

qubes/storage/callback.py

@@ -25,12 +25,14 @@
 import asyncio
 import locale
 from shlex import quote
+
+import qubes.exc
 from qubes.utils import coro_maybe
 
 import qubes.storage
 
 
-class UnhandledSignalException(qubes.storage.StoragePoolException):
+class UnhandledSignalException(qubes.exc.StoragePoolException):
     def __init__(self, pool, signal):
         super().__init__(
             "The pool %s failed to handle the signal %s, likely because it was run from synchronous code."
@@ -204,7 +206,7 @@
             "qubes.storage.callback"
         )  #: Logger instance.
         if not isinstance(conf_id, str):
-            raise qubes.storage.StoragePoolException(
+            raise qubes.exc.StoragePoolException(
                 "conf_id is no String. VM attack?!"
             )
         self._cb_conf_id = (
@@ -215,7 +217,7 @@
         with open(config_path, encoding="utf-8") as json_file:
             conf_all = json.load(json_file)
         if not isinstance(conf_all, dict):
-            raise qubes.storage.StoragePoolException(
+            raise qubes.exc.StoragePoolException(
                 "The file %s is supposed to define a dict." % config_path
             )
 
@@ -225,15 +227,15 @@
             ]  #: Dictionary holding all configuration for the given _cb_conf_id.
         except KeyError:
             # we cannot throw KeyErrors as we'll otherwise generate incorrect error messages @qubes.app._get_pool()
-            raise qubes.storage.StoragePoolException(
+            raise qubes.exc.StoragePoolException(
                 "The specified conf_id %s could not be found inside %s."
                 % (self._cb_conf_id, config_path)
             )
 
         try:
             bdriver = self._cb_conf["bdriver"]
         except KeyError:
-            raise qubes.storage.StoragePoolException(
+            raise qubes.exc.StoragePoolException(
                 "Missing bdriver for the conf_id %s inside %s."
                 % (self._cb_conf_id, config_path)
             )
@@ -247,12 +249,12 @@
                 qubes.storage.STORAGE_ENTRY_POINT, bdriver
             )
         except KeyError:
-            raise qubes.storage.StoragePoolException(
+            raise qubes.exc.StoragePoolException(
                 "The driver %s was not found on your system." % bdriver
             )
 
         if not issubclass(cls, qubes.storage.Pool):
-            raise qubes.storage.StoragePoolException(
+            raise qubes.exc.StoragePoolException(
                 "The class %s must be a subclass of qubes.storage.Pool." % cls
             )
 

qubes/storage/file.py

@@ -30,6 +30,7 @@
 import subprocess
 from contextlib import suppress
 
+import qubes.exc
 import qubes.storage
 import qubes.utils
 
@@ -350,7 +351,7 @@
         """
         if not self.rw:
             msg = "Can not resize readonly volume {!s}".format(self)
-            raise qubes.storage.StoragePoolException(msg)
+            raise qubes.exc.StoragePoolException(msg)
 
         if self.snap_on_start:
             # this theoretically could be supported, but it's unusual
@@ -359,10 +360,10 @@
                 "Cannot resize volume based on a template - resize"
                 "the template instead"
             )
-            raise qubes.storage.StoragePoolException(msg)
+            raise qubes.exc.StoragePoolException(msg)
 
         if size < self.size:
-            raise qubes.storage.StoragePoolException(
+            raise qubes.exc.StoragePoolException(
                 "For your own safety, shrinking of %s is"
                 " disabled. If you really know what you"
                 " are doing, use `truncate` on %s manually."
@@ -407,11 +408,11 @@
             assert (
                 self._export_lock is FileVolume._marker_running
             ), "nested calls to export()"
-            raise qubes.storage.StoragePoolException(
+            raise qubes.exc.StoragePoolException(
                 "file pool cannot export running volumes"
             )
         if self.is_dirty():
-            raise qubes.storage.StoragePoolException(
+            raise qubes.exc.StoragePoolException(
                 "file pool cannot export dirty volumes"
             )
         self._export_lock = FileVolume._marker_exported
@@ -425,7 +426,7 @@
 
     async def import_volume(self, src_volume):
         if src_volume.snap_on_start:
-            raise qubes.storage.StoragePoolException(
+            raise qubes.exc.StoragePoolException(
                 "Can not import snapshot volume {!s} in to pool {!s} ".format(
                     src_volume, self
                 )
@@ -441,7 +442,7 @@
 
     def import_data(self, size):  # pylint: disable=invalid-overridden-method
         if not self.save_on_stop:
-            raise qubes.storage.StoragePoolException(
+            raise qubes.exc.StoragePoolException(
                 "Can not import into save_on_stop=False volume {!s}".format(
                     self
                 )
@@ -476,7 +477,7 @@
             assert (
                 self._export_lock is FileVolume._marker_exported
             ), "nested calls to start()"
-            raise qubes.storage.StoragePoolException(
+            raise qubes.exc.StoragePoolException(
                 "file pool cannot start a VM with an exported volume"
             )
         self._export_lock = FileVolume._marker_running
@@ -559,7 +560,7 @@
             self.snap_on_start or self.save_on_stop
         ):
             msg = "Missing image file: {!s}.".format(self.path)
-            raise qubes.storage.StoragePoolException(msg)
+            raise qubes.exc.StoragePoolException(msg)
         return True
 
     @property
@@ -722,4 +723,4 @@
     """Raise an StoragePoolException if ``path`` does not exist"""
     if not os.path.exists(path):
         msg = "Missing image file: %s" % path
-        raise qubes.storage.StoragePoolException(msg)
+        raise qubes.exc.StoragePoolException(msg)

qubes/storage/kernels.py

@@ -25,7 +25,8 @@
 
 import qubes.exc
 import qubes.storage
-from qubes.storage import Pool, StoragePoolException, Volume
+from qubes.storage import Pool, Volume
+from qubes.exc import StoragePoolException
 
 
 class LinuxModules(Volume):