q-dev: @marmarek comments

qubes/api/__init__.py

@@ -30,13 +30,7 @@
 import traceback
 
 import qubes.exc
-
-class ProtocolError(AssertionError):
-    '''Raised when something is wrong with data received'''
-
-
-class PermissionDenied(Exception):
-    '''Raised deliberately by handlers when we decide not to cooperate'''
+from qubes.exc import ProtocolError, PermissionDenied
 
 
 def method(name, *, no_payload=False, endpoints=None, **classifiers):
@@ -213,11 +207,11 @@ def enforce(predicate):
     def validate_size(self, untrusted_size: bytes) -> int:
         self.enforce(isinstance(untrusted_size, bytes))
         if not untrusted_size.isdigit():
-            raise qubes.api.ProtocolError('Size must be ASCII digits (only)')
+            raise qubes.exc.ProtocolError('Size must be ASCII digits (only)')
         if len(untrusted_size) >= 20:
-            raise qubes.api.ProtocolError('Sizes limited to 19 decimal digits')
+            raise qubes.exc.ProtocolError('Sizes limited to 19 decimal digits')
         if untrusted_size[0] == 48 and untrusted_size != b'0':
-            raise qubes.api.ProtocolError('Spurious leading zeros not allowed')
+            raise qubes.exc.ProtocolError('Spurious leading zeros not allowed')
         return int(untrusted_size)
 
 class QubesDaemonProtocol(asyncio.Protocol):

qubes/api/admin.py

@@ -37,6 +37,7 @@
 import qubes.backup
 import qubes.config
 import qubes.devices
+import qubes.exc
 import qubes.ext
 import qubes.firewall
 import qubes.storage
@@ -501,7 +502,7 @@ async def vm_volume_set_rw(self, untrusted_payload):
             newvalue = qubes.property.bool(None, None,
                 untrusted_payload.decode('ascii'))
         except (UnicodeDecodeError, ValueError):
-            raise qubes.api.ProtocolError('Invalid value')
+            raise qubes.exc.ProtocolError('Invalid value')
         del untrusted_payload
 
         self.fire_event_for_permission(newvalue=newvalue)
@@ -520,7 +521,7 @@ async def vm_volume_set_ephemeral(self, untrusted_payload):
             newvalue = qubes.property.bool(None, None,
                 untrusted_payload.decode('ascii'))
         except (UnicodeDecodeError, ValueError):
-            raise qubes.api.ProtocolError('Invalid value')
+            raise qubes.exc.ProtocolError('Invalid value')
         del untrusted_payload
 
         self.fire_event_for_permission(newvalue=newvalue)
@@ -751,7 +752,7 @@ async def pool_set_ephemeral(self, untrusted_payload):
             newvalue = qubes.property.bool(None, None,
                 untrusted_payload.decode('ascii', 'strict'))
         except (UnicodeDecodeError, ValueError):
-            raise qubes.api.ProtocolError('Invalid value')
+            raise qubes.exc.ProtocolError('Invalid value')
         del untrusted_payload
 
         self.fire_event_for_permission(newvalue=newvalue)
@@ -1076,7 +1077,7 @@ async def _vm_create(self, vm_type, allow_pool=False,
                 errors='strict').split(' '):
             untrusted_key, untrusted_value = untrusted_param.split('=', 1)
             if untrusted_key in kwargs:
-                raise qubes.api.ProtocolError('duplicated parameters')
+                raise qubes.exc.ProtocolError('duplicated parameters')
 
             if untrusted_key == 'name':
                 qubes.vm.validate_name(None, None, untrusted_value)
@@ -1094,7 +1095,7 @@ async def _vm_create(self, vm_type, allow_pool=False,
 
             elif untrusted_key == 'pool' and allow_pool:
                 if pool is not None:
-                    raise qubes.api.ProtocolError('duplicated pool parameter')
+                    raise qubes.exc.ProtocolError('duplicated pool parameter')
                 pool = self.app.get_pool(untrusted_value)
             elif untrusted_key.startswith('pool:') and allow_pool:
                 untrusted_volume = untrusted_key.split(':', 1)[1]
@@ -1104,19 +1105,19 @@ async def _vm_create(self, vm_type, allow_pool=False,
                     'root', 'private', 'volatile', 'kernel'])
                 volume = untrusted_volume
                 if volume in pools:
-                    raise qubes.api.ProtocolError(
+                    raise qubes.exc.ProtocolError(
                         'duplicated pool:{} parameter'.format(volume))
                 pools[volume] = self.app.get_pool(untrusted_value)
 
             else:
-                raise qubes.api.ProtocolError('Invalid param name')
+                raise qubes.exc.ProtocolError('Invalid param name')
         del untrusted_payload
 
         if 'name' not in kwargs or 'label' not in kwargs:
-            raise qubes.api.ProtocolError('Missing name or label')
+            raise qubes.exc.ProtocolError('Missing name or label')
 
         if pool and pools:
-            raise qubes.api.ProtocolError(
+            raise qubes.exc.ProtocolError(
                 'Only one of \'pool=\' and \'pool:volume=\' can be used')
 
         if kwargs['name'] in self.app.domains:
@@ -1550,7 +1551,7 @@ async def backup_execute(self):
         profile_path = os.path.join(qubes.config.backup_profile_dir,
             self.arg + '.conf')
         if not os.path.exists(profile_path):
-            raise qubes.api.PermissionDenied(
+            raise qubes.exc.PermissionDenied(
                 'Backup profile {} does not exist'.format(self.arg))
 
         if not hasattr(self.app, 'api_admin_running_backups'):
@@ -1602,7 +1603,7 @@ async def backup_info(self):
         profile_path = os.path.join(qubes.config.backup_profile_dir,
             self.arg + '.conf')
         if not os.path.exists(profile_path):
-            raise qubes.api.PermissionDenied(
+            raise qubes.exc.PermissionDenied(
                 'Backup profile {} does not exist'.format(self.arg))
 
         backup = await self._load_backup_profile(self.arg,

qubes/device_protocol.py

@@ -36,8 +36,7 @@
 
 import qubes.utils
 
-from qubes.api import ProtocolError
-
+from qubes.exc import ProtocolError
 
 QubesVM = 'qubes.vm.BaseVM'
 

qubes/devices.py

@@ -61,8 +61,9 @@
 import itertools
 from typing import Optional, Iterable
 
+import qubes.exc
 import qubes.utils
-from qubes.api import ProtocolError
+from qubes.exc import ProtocolError
 from qubes.device_protocol import (Device, DeviceInfo, UnknownDevice,
                                    DeviceAssignment)
 
@@ -118,7 +119,7 @@ def sanitize_str(
     """
     if replace_char is None:
         if any(x not in allowed_chars for x in untrusted_value):
-            raise qubes.api.ProtocolError(error_message)
+            raise qubes.exc.ProtocolError(error_message)
         return untrusted_value
     result = ""
     for char in untrusted_value:

qubes/exc.py

@@ -219,3 +219,11 @@ def __init__(self, label):
     def __str__(self):
         # KeyError overrides __str__ method
         return QubesException.__str__(self)
+
+
+class ProtocolError(AssertionError):
+    '''Raised when something is wrong with data received'''
+
+
+class PermissionDenied(Exception):
+    '''Raised deliberately by handlers when we decide not to cooperate'''

qubes/ext/admin.py

@@ -19,6 +19,7 @@
 
 import qubes.api
 import qubes.api.internal
+import qubes.exc
 import qubes.ext
 import qubes.vm.adminvm
 from qrexec.policy import utils, parser
@@ -52,7 +53,7 @@ def on_tag_set_or_remove(self, vm, event, arg, **kwargs):
         # pylint: disable=unused-argument
         if arg.startswith('created-by-') and \
                 not isinstance(vm, qubes.vm.adminvm.AdminVM):
-            raise qubes.api.PermissionDenied(
+            raise qubes.exc.PermissionDenied(
                 'changing this tag is prohibited by {}.{}'.format(
                     __name__, type(self).__name__))
 

qubes/ext/pci.py

@@ -171,15 +171,13 @@ def __init__(self, backend_domain, ident, libvirt_name=None):
         super().__init__(
             backend_domain=backend_domain, ident=ident, devclass="pci")
 
-        if hasattr(self, 'regex'):
-            # pylint: disable=no-member
-            dev_match = self.regex.match(ident)
-            if not dev_match:
-                raise ValueError('Invalid device identifier: {!r}'.format(
-                    ident))
+        dev_match = self.regex.match(ident)
+        if not dev_match:
+            raise ValueError('Invalid device identifier: {!r}'.format(
+                ident))
 
-            for group in self.regex.groupindex:
-                setattr(self, group, dev_match.group(group))
+        for group in self.regex.groupindex:
+            setattr(self, group, dev_match.group(group))
 
         # lazy loading
         self._description = None

qubes/tests/api.py

@@ -23,6 +23,7 @@
 import unittest.mock
 
 import qubes.api
+import qubes.exc
 import qubes.tests
 
 
@@ -43,7 +44,7 @@ def __init__(self, app, src, method, dest, arg, send_event=None):
                 'mgmt.event': self.event,
             }[self.method.decode()]
         except KeyError:
-            raise qubes.api.ProtocolError('Invalid method')
+            raise qubes.exc.ProtocolError('Invalid method')
 
     def execute(self, untrusted_payload):
         self.task = asyncio.Task(self.function(