diff --git a/doc/qubes-exc.rst b/doc/qubes-exc.rst
index f3a644b14..0b1c45e40 100644
--- a/doc/qubes-exc.rst
+++ b/doc/qubes-exc.rst
@@ -43,7 +43,7 @@ call for effect). Instead consider writing in negative form, implying expected
 state: "Domain is not running" instead of "Domain is paused" (yeah, what's wrong
 with that?).
 
-Also avoid implying the personhood of the computer, including adressing user in
+Also avoid implying the personhood of the computer, including addressing user in
 second person. For example, write "Sending message failed" instead of "I failed
 to send the message".
 
diff --git a/qubes/api/__init__.py b/qubes/api/__init__.py
index 764c98679..420710715 100644
--- a/qubes/api/__init__.py
+++ b/qubes/api/__init__.py
@@ -27,6 +27,7 @@
 import re
 import shutil
 import socket
+import string
 import struct
 import traceback
 from typing import Union, Any
@@ -189,6 +190,11 @@ def __init__(
         self._handler = candidates[0]
         self._running_handler = None
 
+        # pylint: disable=invalid-name
+        self.EXC_ARG_NOT_IN_DEST_VOLUMES = "{} volumes".format(self.dest.name)
+        # pylint: disable=invalid-name
+        self.EXC_ARG_NOT_IN_POOLS = "pools"
+
     @classmethod
     def list_methods(cls, select_method=None):
         for attr in dir(cls):
@@ -239,26 +245,76 @@ def fire_event_for_filter(self, iterable, **kwargs):
         """Fire an event on the source qube to filter for permission"""
         return apply_filters(iterable, self.fire_event_for_permission(**kwargs))
 
+    def enforce_arg(
+        self,
+        wants: Union[None, bool, str, list, tuple] = None,
+        short_reason: str = "",
+    ) -> None:
+        """Enforce argument to be absent or be in iterable."""
+        if wants is None:
+            self.enforce(not self.arg, reason="Argument is present")
+        elif wants is True:
+            self.enforce(self.arg, reason="Argument is empty")
+        else:
+            assert isinstance(self.arg, str)
+            assert short_reason
+            if wants is None or isinstance(wants, bool):
+                assertion = self.arg is wants
+            elif isinstance(wants, str):
+                assertion = self.arg == wants
+            else:
+                assertion = self.arg in wants
+            self.enforce(
+                assertion,
+                reason="Argument not in {!r}".format(short_reason),
+            )
+
+    def enforce_dest_dom0(self, wants: bool = True) -> None:
+        """Enforce destination to be or not to be dom0."""
+        if wants:
+            self.enforce(
+                self.dest.name == "dom0", reason="Destination is not dom0"
+            )
+        else:
+            self.enforce(self.dest.name != "dom0", reason="Destination is dom0")
+
     @staticmethod
-    def enforce(predicate):
-        """An assert replacement, but works even with optimisations."""
+    def enforce(predicate, reason: str = "") -> None:
+        """If predicate is false, raise an exception to terminate handling
+        the request.
+
+        This will raise :py:class:`ProtocolError` if the predicate is false.
+        See the documentation of that class for details.
+
+        :param str reason: Exception motive.
+        """
         if not predicate:
-            raise PermissionDenied()
+            raise ProtocolError(reason)
 
     def validate_size(
-        self, untrusted_size: bytes, allow_negative: bool = False
+        self, untrusted_size: bytes, name: str, allow_negative: bool = False
     ) -> int:
-        self.enforce(isinstance(untrusted_size, bytes))
+        allowed_chars = string.ascii_letters + string.digits + "-_. "
+        assert all(c in allowed_chars for c in name)
+        if not isinstance(untrusted_size, bytes):
+            raise ProtocolError(
+                "Expected {!r} to be of type bytes, got {!r}".format(
+                    name, type(untrusted_size).__name__
+                )
+            )
         coefficient = 1
         if allow_negative and untrusted_size.startswith(b"-"):
             coefficient = -1
             untrusted_size = untrusted_size[1:]
+        name_cap = name.capitalize()
         if not untrusted_size.isdigit():
-            raise qubes.exc.ProtocolError("Size must be ASCII digits (only)")
+            raise ProtocolError(name_cap + " size contains non ASCII digits")
         if len(untrusted_size) >= 20:
-            raise qubes.exc.ProtocolError("Sizes limited to 19 decimal digits")
+            raise ProtocolError(
+                name_cap + " size is bigger than 19 decimal digits"
+            )
         if untrusted_size[0] == 48 and untrusted_size != b"0":
-            raise qubes.exc.ProtocolError("Spurious leading zeros not allowed")
+            raise ProtocolError(name_cap + " contains spurious leading zeros")
         return int(untrusted_size) * coefficient
 
 
@@ -345,21 +401,15 @@ async def respond(self, src, meth, dest, arg, *, untrusted_payload):
 
         # except clauses will fall through to transport.abort() below
 
-        except PermissionDenied:
-            self.app.log.warning(
-                "permission denied for call %s+%s (%s → %s) "
-                "with payload of %d bytes",
-                meth,
-                arg,
-                src,
-                dest,
-                len(untrusted_payload),
-            )
-
-        except ProtocolError:
+        except (PermissionDenied, ProtocolError) as exc:
+            log_prefix = None
+            if isinstance(exc, PermissionDenied):
+                log_prefix = "permission denied"
+            elif isinstance(exc, ProtocolError):
+                log_prefix = "protocol error"
             self.app.log.warning(
-                "protocol error for call %s+%s (%s → %s) "
-                "with payload of %d bytes",
+                "%s for call %s+%s (%s → %s) with payload of %d bytes",
+                log_prefix,
                 meth,
                 arg,
                 src,
diff --git a/qubes/api/admin.py b/qubes/api/admin.py
index ff5a1eb8f..27a8435b2 100644
--- a/qubes/api/admin.py
+++ b/qubes/api/admin.py
@@ -116,8 +116,8 @@ class QubesAdminAPI(qubes.api.AbstractQubesAPI):
     )
     async def vmclass_list(self):
         """List all VM classes"""
-        self.enforce(not self.arg)
-        self.enforce(self.dest.name == "dom0")
+        self.enforce_arg(wants=None)
+        self.enforce_dest_dom0(wants=True)
 
         entrypoints = self.fire_event_for_filter(
             importlib.metadata.entry_points(group=qubes.vm.VM_ENTRY_POINT)
@@ -130,7 +130,7 @@ async def vmclass_list(self):
     )
     async def vm_list(self):
         """List all the domains"""
-        self.enforce(not self.arg)
+        self.enforce_arg(wants=None)
 
         if self.dest.name == "dom0":
             domains = self.fire_event_for_filter(self.app.domains)
@@ -156,11 +156,11 @@ async def vm_property_list(self):
     )
     async def property_list(self):
         """List all global properties"""
-        self.enforce(self.dest.name == "dom0")
+        self.enforce_dest_dom0(wants=True)
         return self._property_list(self.app)
 
     def _property_list(self, dest):
-        self.enforce(not self.arg)
+        self.enforce_arg(wants=None)
 
         properties = self.fire_event_for_filter(dest.property_list())
 
@@ -178,7 +178,7 @@ async def vm_property_get(self):
     )
     async def property_get(self):
         """Get a value of one global property"""
-        self.enforce(self.dest.name == "dom0")
+        self.enforce_dest_dom0(wants=True)
         return self._property_get(self.app)
 
     def _property_get(self, dest):
@@ -226,11 +226,11 @@ async def vm_property_get_all(self):
     )
     async def property_get_all(self):
         """Get value all global properties"""
-        self.enforce(self.dest.name == "dom0")
+        self.enforce_dest_dom0(wants=True)
         return self._property_get_all(self.app)
 
     def _property_get_all(self, dest):
-        self.enforce(not self.arg)
+        self.enforce_arg(wants=None)
 
         properties = dest.property_list()
 
@@ -263,7 +263,7 @@ async def vm_property_get_default(self):
     async def property_get_default(self):
         """Get default value of a global property (what value will be set if
         this property is set to default)."""
-        self.enforce(self.dest.name == "dom0")
+        self.enforce_dest_dom0(wants=True)
         return self._property_get_default(self.app)
 
     def _property_get_default(self, dest):
@@ -303,7 +303,7 @@ async def vm_property_set(self, untrusted_payload):
     @qubes.api.method("admin.property.Set", scope="global", write=True)
     async def property_set(self, untrusted_payload):
         """Set property value"""
-        self.enforce(self.dest.name == "dom0")
+        self.enforce_dest_dom0(wants=True)
         return self._property_set(self.app, untrusted_payload=untrusted_payload)
 
     def _property_set(self, dest, untrusted_payload):
@@ -330,7 +330,7 @@ async def vm_property_help(self):
     )
     async def property_help(self):
         """Get help for one property"""
-        self.enforce(self.dest.name == "dom0")
+        self.enforce_dest_dom0(wants=True)
         return self._property_help(self.app)
 
     def _property_help(self, dest):
@@ -358,7 +358,7 @@ async def vm_property_reset(self):
     )
     async def property_reset(self):
         """Reset a property to a default value"""
-        self.enforce(self.dest.name == "dom0")
+        self.enforce_dest_dom0(wants=True)
         return self._property_reset(self.app)
 
     def _property_reset(self, dest):
@@ -374,7 +374,7 @@ def _property_reset(self, dest):
         "admin.vm.volume.List", no_payload=True, scope="local", read=True
     )
     async def vm_volume_list(self):
-        self.enforce(not self.arg)
+        self.enforce_arg(wants=None)
 
         volume_names = (
             self.fire_event_for_filter(self.dest.volumes.keys())
@@ -387,7 +387,10 @@ async def vm_volume_list(self):
         "admin.vm.volume.Info", no_payload=True, scope="local", read=True
     )
     async def vm_volume_info(self):
-        self.enforce(self.arg in self.dest.volumes.keys())
+        self.enforce_arg(
+            wants=self.dest.volumes.keys(),
+            short_reason=self.EXC_ARG_NOT_IN_DEST_VOLUMES,
+        )
 
         self.fire_event_for_permission()
 
@@ -431,7 +434,10 @@ def _serialize(value):
         read=True,
     )
     async def vm_volume_listsnapshots(self):
-        self.enforce(self.arg in self.dest.volumes.keys())
+        self.enforce_arg(
+            wants=self.dest.volumes.keys(),
+            short_reason=self.EXC_ARG_NOT_IN_DEST_VOLUMES,
+        )
 
         volume = self.dest.volumes[self.arg]
         id_to_timestamp = volume.revisions
@@ -442,13 +448,18 @@ async def vm_volume_listsnapshots(self):
 
     @qubes.api.method("admin.vm.volume.Revert", scope="local", write=True)
     async def vm_volume_revert(self, untrusted_payload):
-        self.enforce(self.arg in self.dest.volumes.keys())
+        self.enforce_arg(
+            wants=self.dest.volumes.keys(),
+            short_reason=self.EXC_ARG_NOT_IN_DEST_VOLUMES,
+        )
         untrusted_revision = untrusted_payload.decode("ascii").strip()
         del untrusted_payload
 
         volume = self.dest.volumes[self.arg]
         snapshots = volume.revisions
-        self.enforce(untrusted_revision in snapshots)
+        # TODO: ben: info-leak: revision existence
+        if untrusted_revision not in snapshots:
+            raise qubes.exc.QubesVolumeRevisionNotFoundError()
         revision = untrusted_revision
 
         self.fire_event_for_permission(volume=volume, revision=revision)
@@ -461,7 +472,10 @@ async def vm_volume_revert(self, untrusted_payload):
         "admin.vm.volume.CloneFrom", no_payload=True, scope="local", write=True
     )
     async def vm_volume_clone_from(self):
-        self.enforce(self.arg in self.dest.volumes.keys())
+        self.enforce_arg(
+            wants=self.dest.volumes.keys(),
+            short_reason=self.EXC_ARG_NOT_IN_DEST_VOLUMES,
+        )
 
         volume = self.dest.volumes[self.arg]
 
@@ -473,19 +487,31 @@ async def vm_volume_clone_from(self):
             self.app.api_admin_pending_clone = {}
         # don't handle collisions any better - if someone is so much out of
         # luck, can try again anyway
-        self.enforce(token not in self.app.api_admin_pending_clone)
+        if token in self.app.api_admin_pending_clone:
+            raise qubes.exc.QubesVolumeCopyInUseError(
+                vm=self.dest, volume=self.arg
+            )
 
         self.app.api_admin_pending_clone[token] = volume
         return token
 
     @qubes.api.method("admin.vm.volume.CloneTo", scope="local", write=True)
     async def vm_volume_clone_to(self, untrusted_payload):
-        self.enforce(self.arg in self.dest.volumes.keys())
-        untrusted_token = untrusted_payload.decode("ascii").strip()
-        del untrusted_payload
-        self.enforce(
-            untrusted_token in getattr(self.app, "api_admin_pending_clone", {})
+        self.enforce_arg(
+            wants=self.dest.volumes.keys(),
+            short_reason=self.EXC_ARG_NOT_IN_DEST_VOLUMES,
         )
+        try:
+            untrusted_token = untrusted_payload.decode("ascii").strip()
+        except UnicodeDecodeError:
+            raise qubes.exc.ProtocolError("Token contains non-ASCII characters")
+
+        del untrusted_payload
+        if untrusted_token not in getattr(
+            self.app, "api_admin_pending_clone", {}
+        ):
+            raise qubes.exc.QubesVolumeCopyTokenNotFoundError()
+
         token = untrusted_token
         del untrusted_token
 
@@ -493,8 +519,12 @@ async def vm_volume_clone_to(self, untrusted_payload):
         del self.app.api_admin_pending_clone[token]
 
         # make sure the volume still exists, but invalidate token anyway
-        self.enforce(str(src_volume.pool) in self.app.pools)
-        self.enforce(src_volume in self.app.pools[str(src_volume.pool)].volumes)
+        # TODO: ben: info-leak: pool existence
+        if str(src_volume.pool) not in self.app.pools:
+            raise qubes.exc.QubesPoolNotFoundError()
+        # TODO: ben: info-leak: volume existence
+        if src_volume not in self.app.pools[str(src_volume.pool)].volumes:
+            raise qubes.exc.QubesVolumeNotFoundError()
 
         dst_volume = self.dest.volumes[self.arg]
 
@@ -508,8 +538,11 @@ async def vm_volume_clone_to(self, untrusted_payload):
 
     @qubes.api.method("admin.vm.volume.Resize", scope="local", write=True)
     async def vm_volume_resize(self, untrusted_payload):
-        self.enforce(self.arg in self.dest.volumes.keys())
-        size = self.validate_size(untrusted_payload.strip())
+        self.enforce_arg(
+            wants=self.dest.volumes.keys(),
+            short_reason=self.EXC_ARG_NOT_IN_DEST_VOLUMES,
+        )
+        size = self.validate_size(untrusted_payload.strip(), name="volume size")
         self.fire_event_for_permission(size=size)
         try:
             await self.dest.storage.resize(self.arg, size)
@@ -520,7 +553,10 @@ async def vm_volume_resize(self, untrusted_payload):
         "admin.vm.volume.Clear", no_payload=True, scope="local", write=True
     )
     async def vm_volume_clear(self):
-        self.enforce(self.arg in self.dest.volumes.keys())
+        self.enforce_arg(
+            wants=self.dest.volumes.keys(),
+            short_reason=self.EXC_ARG_NOT_IN_DEST_VOLUMES,
+        )
 
         self.fire_event_for_permission()
 
@@ -549,15 +585,23 @@ async def vm_volume_clear(self):
         "admin.vm.volume.Set.revisions_to_keep", scope="local", write=True
     )
     async def vm_volume_set_revisions_to_keep(self, untrusted_payload):
-        self.enforce(self.arg in self.dest.volumes.keys())
-        newvalue = self.validate_size(untrusted_payload, allow_negative=True)
+        self.enforce_arg(
+            wants=self.dest.volumes.keys(),
+            short_reason=self.EXC_ARG_NOT_IN_DEST_VOLUMES,
+        )
+        newvalue = self.validate_size(
+            untrusted_payload, name="revisions to keep", allow_negative=True
+        )
         self.fire_event_for_permission(newvalue=newvalue)
         self.dest.storage.set_revisions_to_keep(self.arg, newvalue)
         self.app.save()
 
     @qubes.api.method("admin.vm.volume.Set.rw", scope="local", write=True)
     async def vm_volume_set_rw(self, untrusted_payload):
-        self.enforce(self.arg in self.dest.volumes.keys())
+        self.enforce_arg(
+            wants=self.dest.volumes.keys(),
+            short_reason=self.EXC_ARG_NOT_IN_DEST_VOLUMES,
+        )
         try:
             newvalue = qubes.property.bool(
                 None, None, untrusted_payload.decode("ascii")
@@ -578,7 +622,10 @@ async def vm_volume_set_rw(self, untrusted_payload):
         "admin.vm.volume.Set.ephemeral", scope="local", write=True
     )
     async def vm_volume_set_ephemeral(self, untrusted_payload):
-        self.enforce(self.arg in self.dest.volumes.keys())
+        self.enforce_arg(
+            wants=self.dest.volumes.keys(),
+            short_reason=self.EXC_ARG_NOT_IN_DEST_VOLUMES,
+        )
         try:
             newvalue = qubes.property.bool(
                 None, None, untrusted_payload.decode("ascii")
@@ -599,7 +646,7 @@ async def vm_volume_set_ephemeral(self, untrusted_payload):
         "admin.vm.tag.List", no_payload=True, scope="local", read=True
     )
     async def vm_tag_list(self):
-        self.enforce(not self.arg)
+        self.enforce_arg(wants=None)
 
         tags = self.dest.tags
 
@@ -646,7 +693,7 @@ async def vm_tag_remove(self):
         "admin.vm.Console", no_payload=True, scope="local", write=True
     )
     async def vm_console(self):
-        self.enforce(not self.arg)
+        self.enforce_arg(wants=None)
 
         self.fire_event_for_permission()
 
@@ -663,8 +710,8 @@ async def vm_console(self):
         "admin.pool.List", no_payload=True, scope="global", read=True
     )
     async def pool_list(self):
-        self.enforce(not self.arg)
-        self.enforce(self.dest.name == "dom0")
+        self.enforce_arg(wants=None)
+        self.enforce_dest_dom0(wants=True)
 
         pools = self.fire_event_for_filter(self.app.pools)
 
@@ -674,8 +721,8 @@ async def pool_list(self):
         "admin.pool.ListDrivers", no_payload=True, scope="global", read=True
     )
     async def pool_listdrivers(self):
-        self.enforce(self.dest.name == "dom0")
-        self.enforce(not self.arg)
+        self.enforce_dest_dom0(wants=True)
+        self.enforce_arg(wants=None)
 
         drivers = self.fire_event_for_filter(qubes.storage.pool_drivers())
 
@@ -690,8 +737,10 @@ async def pool_listdrivers(self):
         "admin.pool.Info", no_payload=True, scope="global", read=True
     )
     async def pool_info(self):
-        self.enforce(self.dest.name == "dom0")
-        self.enforce(self.arg in self.app.pools.keys())
+        self.enforce_dest_dom0(wants=True)
+        self.enforce_arg(
+            wants=self.app.pools.keys(), short_reason=self.EXC_ARG_NOT_IN_POOLS
+        )
 
         pool = self.app.pools[self.arg]
 
@@ -726,8 +775,10 @@ async def pool_info(self):
         "admin.pool.UsageDetails", no_payload=True, scope="global", read=True
     )
     async def pool_usage(self):
-        self.enforce(self.dest.name == "dom0")
-        self.enforce(self.arg in self.app.pools.keys())
+        self.enforce_dest_dom0(wants=True)
+        self.enforce_arg(
+            wants=self.app.pools.keys(), short_reason=self.EXC_ARG_NOT_IN_POOLS
+        )
 
         pool = self.app.pools[self.arg]
 
@@ -744,15 +795,25 @@ async def pool_usage(self):
 
     @qubes.api.method("admin.pool.Add", scope="global", write=True)
     async def pool_add(self, untrusted_payload):
-        self.enforce(self.dest.name == "dom0")
-        if self.arg not in qubes.storage.pool_drivers():
-            raise qubes.exc.QubesException(
-                "unexpected driver name: " + self.arg
-            )
+        self.enforce_dest_dom0(wants=True)
+        # TODO: ben: info-leak: pool existence
+        self.enforce_arg(
+            wants=qubes.storage.pool_drivers(), short_reason="available pools"
+        )
 
-        untrusted_pool_config = untrusted_payload.decode("ascii").splitlines()
+        try:
+            untrusted_pool_config = untrusted_payload.decode(
+                "ascii"
+            ).splitlines()
+        except UnicodeDecodeError:
+            raise qubes.exc.ProtocolError(
+                "Pool parameters contains non-ASCII characters"
+            )
         del untrusted_payload
-        self.enforce(all(("=" in line) for line in untrusted_pool_config))
+        self.enforce(
+            all(("=" in line) for line in untrusted_pool_config),
+            reason="Options must be separated by equal sign",
+        )
         # pairs of (option, value)
         untrusted_pool_config = [
             line.split("=", 1) for line in untrusted_pool_config
@@ -760,25 +821,35 @@ async def pool_add(self, untrusted_payload):
         # reject duplicated options
         self.enforce(
             len(set(x[0] for x in untrusted_pool_config))
-            == len([x[0] for x in untrusted_pool_config])
+            == len([x[0] for x in untrusted_pool_config]),
+            reason="Options must be unique",
         )
         # and convert to dict
         untrusted_pool_config = dict(untrusted_pool_config)
 
-        self.enforce("name" in untrusted_pool_config)
+        self.enforce(
+            "name" in untrusted_pool_config,
+            reason="Pool configuration must contain key: name",
+        )
         untrusted_pool_name = untrusted_pool_config.pop("name")
         allowed_chars = string.ascii_letters + string.digits + "-_."
-        self.enforce(all(c in allowed_chars for c in untrusted_pool_name))
+        self.enforce(
+            all(c in allowed_chars for c in untrusted_pool_name),
+            reason="Pool name must be in safe set: " + allowed_chars,
+        )
         pool_name = untrusted_pool_name
-        self.enforce(pool_name not in self.app.pools)
+        # TODO: ben: info-leak: pool existence
+        if pool_name in self.app.pools:
+            raise qubes.exc.QubesPoolInUseError(pool_name=pool_name)
 
         driver_parameters = qubes.storage.driver_parameters(self.arg)
         unexpected_parameters = [
             key for key in untrusted_pool_config if key not in driver_parameters
         ]
+        # TODO: ben: dangerous logging any ASCII value?
         if unexpected_parameters:
-            raise qubes.exc.QubesException(
-                "unexpected driver options: " + " ".join(unexpected_parameters)
+            raise qubes.exc.ProtocolError(
+                "Unexpected driver options: " + " ".join(unexpected_parameters)
             )
         pool_config = untrusted_pool_config
 
@@ -791,8 +862,10 @@ async def pool_add(self, untrusted_payload):
         "admin.pool.Remove", no_payload=True, scope="global", write=True
     )
     async def pool_remove(self):
-        self.enforce(self.dest.name == "dom0")
-        self.enforce(self.arg in self.app.pools.keys())
+        self.enforce_dest_dom0(wants=True)
+        self.enforce_arg(
+            wants=self.app.pools.keys(), short_reason=self.EXC_ARG_NOT_IN_POOLS
+        )
 
         self.fire_event_for_permission()
 
@@ -803,8 +876,10 @@ async def pool_remove(self):
         "admin.pool.volume.List", no_payload=True, scope="global", read=True
     )
     async def pool_volume_list(self):
-        self.enforce(self.dest.name == "dom0")
-        self.enforce(self.arg in self.app.pools.keys())
+        self.enforce_dest_dom0(wants=True)
+        self.enforce_arg(
+            wants=self.app.pools.keys(), short_reason=self.EXC_ARG_NOT_IN_POOLS
+        )
 
         pool = self.app.pools[self.arg]
 
@@ -815,13 +890,16 @@ async def pool_volume_list(self):
         "admin.pool.Set.revisions_to_keep", scope="global", write=True
     )
     async def pool_set_revisions_to_keep(self, untrusted_payload):
-        self.enforce(self.dest.name == "dom0")
-        self.enforce(self.arg in self.app.pools.keys())
+        self.enforce_dest_dom0(wants=True)
+        self.enforce_arg(
+            wants=self.app.pools.keys(), short_reason=self.EXC_ARG_NOT_IN_POOLS
+        )
         pool = self.app.pools[self.arg]
-        newvalue = self.validate_size(untrusted_payload, allow_negative=True)
+        newvalue = self.validate_size(
+            untrusted_payload, name="revisions to keep", allow_negative=True
+        )
 
-        if newvalue < -1:
-            raise qubes.api.ProtocolError("Invalid value for revisions_to_keep")
+        self.enforce(newvalue >= -1, reason="Value can't be lower than -1")
 
         self.fire_event_for_permission(newvalue=newvalue)
 
@@ -832,8 +910,10 @@ async def pool_set_revisions_to_keep(self, untrusted_payload):
         "admin.pool.Set.ephemeral_volatile", scope="global", write=True
     )
     async def pool_set_ephemeral(self, untrusted_payload):
-        self.enforce(self.dest.name == "dom0")
-        self.enforce(self.arg in self.app.pools.keys())
+        self.enforce_dest_dom0(wants=True)
+        self.enforce_arg(
+            wants=self.app.pools.keys(), short_reason=self.EXC_ARG_NOT_IN_POOLS
+        )
         pool = self.app.pools[self.arg]
         try:
             newvalue = qubes.property.bool(
@@ -852,8 +932,8 @@ async def pool_set_ephemeral(self, untrusted_payload):
         "admin.label.List", no_payload=True, scope="global", read=True
     )
     async def label_list(self):
-        self.enforce(self.dest.name == "dom0")
-        self.enforce(not self.arg)
+        self.enforce_dest_dom0(wants=True)
+        self.enforce_arg(wants=None)
 
         labels = self.fire_event_for_filter(self.app.labels.values())
 
@@ -863,12 +943,11 @@ async def label_list(self):
         "admin.label.Get", no_payload=True, scope="global", read=True
     )
     async def label_get(self):
-        self.enforce(self.dest.name == "dom0")
+        self.enforce_dest_dom0(wants=True)
+        qubes.utils.validate_label(untrusted_label=self.arg)
 
-        try:
-            label = self.app.get_label(self.arg)
-        except KeyError:
-            raise qubes.exc.QubesValueError
+        # TODO: ben: info-leak: label existence
+        label = self.app.get_label(self.arg)
 
         self.fire_event_for_permission(label=label)
 
@@ -878,12 +957,11 @@ async def label_get(self):
         "admin.label.Index", no_payload=True, scope="global", read=True
     )
     async def label_index(self):
-        self.enforce(self.dest.name == "dom0")
+        self.enforce_dest_dom0(wants=True)
+        qubes.utils.validate_label(untrusted_label=self.arg)
 
-        try:
-            label = self.app.get_label(self.arg)
-        except KeyError:
-            raise qubes.exc.QubesValueError
+        # TODO: ben: info-leak: label existence
+        label = self.app.get_label(self.arg)
 
         self.fire_event_for_permission(label=label)
 
@@ -891,28 +969,25 @@ async def label_index(self):
 
     @qubes.api.method("admin.label.Create", scope="global", write=True)
     async def label_create(self, untrusted_payload):
-        self.enforce(self.dest.name == "dom0")
+        self.enforce_dest_dom0(wants=True)
+        qubes.utils.validate_label(untrusted_label=self.arg)
 
-        # don't confuse label name with label index
-        self.enforce(not self.arg.isdigit())
-        allowed_chars = string.ascii_letters + string.digits + "-_."
-        self.enforce(all(c in allowed_chars for c in self.arg))
+        # TODO: ben: info-leak: label existence
         try:
             self.app.get_label(self.arg)
         except KeyError:
             # ok, no such label yet
             pass
         else:
-            raise qubes.exc.QubesValueError("label already exists")
+            raise qubes.exc.QubesInvalidLabelError("Label value must be new")
 
-        untrusted_payload = untrusted_payload.decode("ascii", "strict").strip()
-        self.enforce(len(untrusted_payload) == 8)
-        self.enforce(untrusted_payload.startswith("0x"))
-        # besides prefix, only hex digits are allowed
-        self.enforce(all(x in string.hexdigits for x in untrusted_payload[2:]))
+        qubes.utils.validate_label_value(
+            untrusted_label_value=untrusted_payload
+        )
 
-        # SEE: #2732
+        # TODO: avoid creating too-similar qube labels: #2732
         color = untrusted_payload
+        del untrusted_payload
 
         self.fire_event_for_permission(color=color)
 
@@ -929,21 +1004,22 @@ async def label_create(self, untrusted_payload):
         "admin.label.Remove", no_payload=True, scope="global", write=True
     )
     async def label_remove(self):
-        self.enforce(self.dest.name == "dom0")
+        self.enforce_dest_dom0(wants=True)
+        qubes.utils.validate_label(untrusted_label=self.arg)
 
-        try:
-            label = self.app.get_label(self.arg)
-        except KeyError:
-            raise qubes.exc.QubesValueError
-        # don't allow removing default labels
-        self.enforce(label.index > qubes.config.max_default_label)
+        # TODO: ben: info-leak: label existence
+        label = self.app.get_label(self.arg)
+        self.enforce(
+            label.index > qubes.config.max_default_label,
+            reason="Can only remove custom labels",
+        )
+
+        self.fire_event_for_permission(label=label)
 
         # FIXME: this should be in app.add_label()
         for vm in self.app.domains:
             if vm.label == label:
-                raise qubes.exc.QubesException("label still in use")
-
-        self.fire_event_for_permission(label=label)
+                raise qubes.exc.QubesLabelInUseError(label=label)
 
         del self.app.labels[label.index]
         self.app.save()
@@ -952,7 +1028,7 @@ async def label_remove(self):
         "admin.vm.Start", no_payload=True, scope="local", execute=True
     )
     async def vm_start(self):
-        self.enforce(not self.arg)
+        self.enforce_arg(wants=None)
         self.fire_event_for_permission()
         try:
             await self.dest.start()
@@ -972,7 +1048,11 @@ async def vm_shutdown(self):
             args = self.arg.split("+")
         else:
             args = []
-        self.enforce(all(arg in ("force", "wait") for arg in args))
+        allowed_args = ("force", "wait")
+        self.enforce(
+            all(arg in allowed_args for arg in args),
+            reason="Argument must match: " + ", ".join(allowed_args),
+        )
         force = "force" in args
         wait = "wait" in args
         self.fire_event_for_permission(force=force, wait=wait)
@@ -982,7 +1062,7 @@ async def vm_shutdown(self):
         "admin.vm.Pause", no_payload=True, scope="local", execute=True
     )
     async def vm_pause(self):
-        self.enforce(not self.arg)
+        self.enforce_arg(wants=None)
         self.fire_event_for_permission()
         await self.dest.pause()
 
@@ -990,7 +1070,7 @@ async def vm_pause(self):
         "admin.vm.Unpause", no_payload=True, scope="local", execute=True
     )
     async def vm_unpause(self):
-        self.enforce(not self.arg)
+        self.enforce_arg(wants=None)
         self.fire_event_for_permission()
         await self.dest.unpause()
 
@@ -998,7 +1078,7 @@ async def vm_unpause(self):
         "admin.vm.Suspend", no_payload=True, scope="local", execute=True
     )
     async def vm_suspend(self):
-        self.enforce(not self.arg)
+        self.enforce_arg(wants=None)
         self.fire_event_for_permission()
         await self.dest.suspend()
 
@@ -1006,7 +1086,7 @@ async def vm_suspend(self):
         "admin.vm.Resume", no_payload=True, scope="local", execute=True
     )
     async def vm_resume(self):
-        self.enforce(not self.arg)
+        self.enforce_arg(wants=None)
         self.fire_event_for_permission()
         await self.dest.resume()
 
@@ -1014,7 +1094,7 @@ async def vm_resume(self):
         "admin.vm.Kill", no_payload=True, scope="local", execute=True
     )
     async def vm_kill(self):
-        self.enforce(not self.arg)
+        self.enforce_arg(wants=None)
         self.fire_event_for_permission()
         await self.dest.kill()
 
@@ -1022,7 +1102,7 @@ async def vm_kill(self):
         "admin.Events", no_payload=True, scope="global", read=True
     )
     async def events(self):
-        self.enforce(not self.arg)
+        self.enforce_arg(wants=None)
 
         # run until client connection is terminated
         self.cancellable = True
@@ -1065,7 +1145,7 @@ async def events(self):
         "admin.vm.feature.List", no_payload=True, scope="local", read=True
     )
     async def vm_feature_list(self):
-        self.enforce(not self.arg)
+        self.enforce_arg(wants=None)
         features = self.fire_event_for_filter(self.dest.features.keys())
         return "".join("{}\n".format(feature) for feature in features)
 
@@ -1213,7 +1293,7 @@ def vm_create_in_pool(self, endpoint, untrusted_payload=None):
     async def _vm_create(
         self, vm_type, allow_pool=False, untrusted_payload=None
     ):
-        self.enforce(self.dest.name == "dom0")
+        self.enforce_dest_dom0(wants=True)
 
         kwargs = {}
         pool = None
@@ -1228,6 +1308,7 @@ async def _vm_create(
         # when given VM class do need a template
         if self.arg:
             if hasattr(vm_class, "template"):
+                # TODO: ben: info-leak: template existence
                 if self.arg not in self.app.domains:
                     raise qubes.exc.QubesVMNotFoundError(self.arg)
                 kwargs["template"] = self.app.domains[self.arg]
@@ -1240,8 +1321,9 @@ async def _vm_create(
             "ascii", errors="strict"
         ).split(" "):
             untrusted_key, untrusted_value = untrusted_param.split("=", 1)
-            if untrusted_key in kwargs:
-                raise qubes.exc.ProtocolError("duplicated parameters")
+            self.enforce(
+                untrusted_key not in kwargs, reason="Options must be unique"
+            )
 
             if untrusted_key == "name":
                 qubes.vm.validate_name(None, None, untrusted_value)
@@ -1249,46 +1331,70 @@ async def _vm_create(
 
             elif untrusted_key == "label":
                 # don't confuse label name with label index
-                self.enforce(not untrusted_value.isdigit())
+                self.enforce(
+                    not untrusted_value.isdigit(),
+                    reason="Label must not be a digit",
+                )
                 allowed_chars = string.ascii_letters + string.digits + "-_."
-                self.enforce(all(c in allowed_chars for c in untrusted_value))
-                # this may raise QubesLabelNotFoundError
+                self.enforce(
+                    all(c in allowed_chars for c in untrusted_value),
+                    reason="Label must be in safe set: " + allowed_chars,
+                )
+                # TODO: ben: info-leak: label existence
+                qubes.utils.validate_label(untrusted_label=untrusted_value)
                 kwargs["label"] = self.app.get_label(untrusted_value)
 
             elif untrusted_key == "pool" and allow_pool:
-                if pool is not None:
-                    raise qubes.exc.ProtocolError("duplicated pool parameter")
+                self.enforce(
+                    pool is None, reason="Option 'pool' must be unique"
+                )
                 pool = self.app.get_pool(untrusted_value)
+
             elif untrusted_key.startswith("pool:") and allow_pool:
                 untrusted_volume = untrusted_key.split(":", 1)[1]
                 # kind of ugly, but actual list of volumes is available only
                 # after creating a VM
+                allowed_volumes = ["root", "private", "volatile", "kernel"]
                 self.enforce(
-                    untrusted_volume
-                    in ["root", "private", "volatile", "kernel"]
+                    untrusted_volume in allowed_volumes,
+                    reason="Volume must be one of: "
+                    + ", ".join(allowed_volumes),
                 )
                 volume = untrusted_volume
-                if volume in pools:
-                    raise qubes.exc.ProtocolError(
-                        "duplicated pool:{} parameter".format(volume)
-                    )
+                self.enforce(
+                    volume not in pools,
+                    reason="Option 'pool:{}' must be unique".format(volume),
+                )
                 pools[volume] = self.app.get_pool(untrusted_value)
 
             else:
-                raise qubes.exc.ProtocolError("Invalid param name")
+                if not allow_pool and (
+                    untrusted_key == "pool" or untrusted_key.startswith("pool:")
+                ):
+                    raise qubes.exc.ProtocolError(
+                        "Pool option specified but 'allow_pool' is False"
+                    )
+                allowed_options = ["name", "label", "pool", "pool:"]
+                raise qubes.exc.ProtocolError(
+                    "Options does not match: " + ", ".join(allowed_options)
+                )
         del untrusted_payload
 
-        if "name" not in kwargs or "label" not in kwargs:
-            raise qubes.exc.ProtocolError("Missing name or label")
-
-        if pool and pools:
-            raise qubes.exc.ProtocolError(
-                "Only one of 'pool=' and 'pool:volume=' can be used"
+        for required_key in ["name", "label"]:
+            self.enforce(
+                required_key in kwargs,
+                reason="Creation requested without key: " + required_key,
             )
 
+        self.enforce(
+            not (pool and pools),
+            reason="Conflicting options specified: 'pool=', 'pool:volume='",
+        )
+
+        # TODO: ben: info-leak: qube existence
         if kwargs["name"] in self.app.domains:
             raise qubes.exc.QubesValueError(
-                "VM {} already exists".format(kwargs["name"])
+                "Qube already exists: {}".format(kwargs["name"])
             )
 
         self.fire_event_for_permission(pool=pool, pools=pools, **kwargs)
@@ -1311,7 +1417,10 @@ async def create_disposable(self, untrusted_payload):
         Create a disposable. If the RPC argument is ``preload-autostart``,
         cleanse the preload list and start preloading fresh disposables.
         """
-        self.enforce(self.arg in [None, "", "preload-autostart"])
+        self.enforce_arg(
+            [None, "", "preload-autostart"],
+            short_reason="'', 'preload-autostart'",
+        )
         preload_autostart = False
         if self.arg == "preload-autostart":
             preload_autostart = True
@@ -1342,7 +1451,7 @@ async def create_disposable(self, untrusted_payload):
         "admin.vm.Remove", no_payload=True, scope="global", write=True
     )
     async def vm_remove(self):
-        self.enforce(not self.arg)
+        self.enforce_arg(wants=None)
 
         self.fire_event_for_permission()
 
@@ -1376,8 +1485,8 @@ async def vm_remove(self):
     )
     async def deviceclass_list(self):
         """List all DEVICES classes"""
-        self.enforce(not self.arg)
-        self.enforce(self.dest.name == "dom0")
+        self.enforce_arg(wants=None)
+        self.enforce_dest_dom0(wants=True)
 
         entrypoints = self.fire_event_for_filter(
             importlib.metadata.entry_points(group="qubes.devices")
@@ -1408,7 +1517,8 @@ async def vm_device_available(self, endpoint):
             devices = [dev for dev in devices if dev.port_id == self.arg]
             # no duplicated devices, but device may not exist, in which case
             #  the list is empty
-            self.enforce(len(devices) <= 1)
+            if len(devices) > 1:
+                raise qubes.exc.DeviceNotFound()
         devices = self.fire_event_for_filter(devices, devclass=devclass)
         dev_info = {
             f"{dev.port_id}:{dev.device_id}": dev.serialize().decode()
@@ -1449,7 +1559,8 @@ async def vm_device_list(self, endpoint):
             ]
             # no duplicated devices, but device may not exist, in which case
             #  the list is empty
-            self.enforce(len(device_assignments) <= 1)
+            if len(device_assignments) > 1:
+                raise qubes.exc.DeviceNotFound()
         device_assignments = self.fire_event_for_filter(
             device_assignments, devclass=devclass
         )
@@ -1498,7 +1609,8 @@ async def vm_device_attached(self, endpoint):
             ]
             # no duplicated devices, but device may not exist, in which case
             #  the list is empty
-            self.enforce(len(device_assignments) <= 1)
+            if len(device_assignments) > 1:
+                raise qubes.exc.DeviceNotFound()
         device_assignments = [
             a.clone(
                 device=self.app.domains[a.backend_name].devices[devclass][
@@ -1665,7 +1777,7 @@ async def vm_device_set_required(self, endpoint, untrusted_payload):
         try:
             mode = allowed_values[untrusted_payload]
         except KeyError:
-            raise qubes.exc.PermissionDenied()
+            raise qubes.exc.ProtocolError("Invalid assignment mode")
 
         dev = VirtualDevice.from_qarg(self.arg, devclass, self.app.domains)
 
@@ -1686,7 +1798,7 @@ async def vm_device_denied_list(self):
 
         Returns a newline-separated string.
         """
-        self.enforce(not self.arg)
+        self.enforce_arg(wants=None)
 
         self.fire_event_for_permission()
 
@@ -1701,11 +1813,16 @@ async def vm_device_denied_add(self, untrusted_payload):
         Payload:
             Encoded device interface (can be repeated without any separator).
         """
-        payload = untrusted_payload.decode("ascii", errors="strict")
+        try:
+            payload = untrusted_payload.decode("ascii", errors="strict")
+        except UnicodeDecodeError:
+            raise qubes.exc.ProtocolError(
+                "Device interface contains non-ASCII characters"
+            )
         to_add = DeviceInterface.from_str_bulk(payload)
 
         if len(set(to_add)) != len(to_add):
-            raise qubes.exc.QubesValueError(
+            raise qubes.exc.ProtocolError(
                 "Duplicated device interfaces in payload."
             )
 
@@ -1733,14 +1850,19 @@ async def vm_device_denied_remove(self, untrusted_payload):
         """
         denied = DeviceInterface.from_str_bulk(self.dest.devices_denied)
 
-        payload = untrusted_payload.decode("ascii", errors="strict")
+        try:
+            payload = untrusted_payload.decode("ascii", errors="strict")
+        except UnicodeDecodeError:
+            raise qubes.exc.ProtocolError(
+                "Device interface contains non-ASCII characters"
+            )
         if payload != "all":
             to_remove = DeviceInterface.from_str_bulk(payload)
         else:
             to_remove = denied.copy()
 
         if len(set(to_remove)) != len(to_remove):
-            raise qubes.exc.QubesValueError(
+            raise qubes.exc.ProtocolError(
                 "Duplicated device interfaces in payload."
             )
 
@@ -1758,7 +1880,7 @@ async def vm_device_denied_remove(self, untrusted_payload):
         "admin.vm.firewall.Get", no_payload=True, scope="local", read=True
     )
     async def vm_firewall_get(self):
-        self.enforce(not self.arg)
+        self.enforce_arg(wants=None)
 
         self.fire_event_for_permission()
 
@@ -1770,15 +1892,18 @@ async def vm_firewall_get(self):
 
     @qubes.api.method("admin.vm.firewall.Set", scope="local", write=True)
     async def vm_firewall_set(self, untrusted_payload):
-        self.enforce(not self.arg)
+        self.enforce_arg(wants=None)
         rules = []
-        for untrusted_line in untrusted_payload.decode(
-            "ascii", errors="strict"
-        ).splitlines():
-            rule = qubes.firewall.Rule.from_api_string(
-                untrusted_rule=untrusted_line
-            )
-            rules.append(rule)
+        try:
+            for untrusted_line in untrusted_payload.decode(
+                "ascii", errors="strict"
+            ).splitlines():
+                rule = qubes.firewall.Rule.from_api_string(
+                    untrusted_rule=untrusted_line
+                )
+                rules.append(rule)
+        except UnicodeDecodeError:
+            raise qubes.exc.ProtocolError("Rules contains non-ASCII characters")
 
         self.fire_event_for_permission(rules=rules)
 
@@ -1792,7 +1917,7 @@ async def vm_firewall_set(self, untrusted_payload):
         execute=True,
     )
     async def vm_firewall_reload(self):
-        self.enforce(not self.arg)
+        self.enforce_arg(wants=None)
 
         self.fire_event_for_permission()
 
@@ -1857,7 +1982,10 @@ async def _load_backup_profile(self, profile_name, skip_passphrase=False):
                 )
                 # make it foolproof against "echo passphrase" implementation
                 passphrase = passphrase.strip()
-                self.enforce(b"\n" not in passphrase)
+                self.enforce(
+                    b"\n" not in passphrase,
+                    reason="Passphrase must not contain newline character",
+                )
             except subprocess.CalledProcessError:
                 raise qubes.exc.QubesException(
                     "Failed to retrieve passphrase from '{}' VM".format(
@@ -1919,9 +2047,12 @@ def _backup_progress_callback(self, profile_name, progress):
         execute=True,
     )
     async def backup_execute(self):
-        self.enforce(self.dest.name == "dom0")
-        self.enforce(self.arg)
-        self.enforce("/" not in self.arg)
+        self.enforce_dest_dom0(wants=True)
+        self.enforce_arg(wants=True)
+        self.enforce(
+            "/" not in self.arg,
+            reason="Backup profile ID must not contain forward slash character",
+        )
 
         self.fire_event_for_permission()
 
@@ -1929,8 +2060,8 @@ async def backup_execute(self):
             qubes.config.backup_profile_dir, self.arg + ".conf"
         )
         if not os.path.exists(profile_path):
-            raise qubes.exc.PermissionDenied(
-                "Backup profile {} does not exist".format(self.arg)
+            raise qubes.exc.ProtocolError(
+                "Backup profile does not exist: {}".format(self.arg)
             )
 
         if not hasattr(self.app, "api_admin_running_backups"):
@@ -1958,9 +2089,12 @@ async def backup_execute(self):
         "admin.backup.Cancel", no_payload=True, scope="global", execute=True
     )
     async def backup_cancel(self):
-        self.enforce(self.dest.name == "dom0")
-        self.enforce(self.arg)
-        self.enforce("/" not in self.arg)
+        self.enforce_dest_dom0(wants=True)
+        self.enforce_arg(wants=True)
+        self.enforce(
+            "/" not in self.arg,
+            reason="Backup profile ID must not contain forward slash character",
+        )
 
         self.fire_event_for_permission()
 
@@ -1976,9 +2110,12 @@ async def backup_cancel(self):
         "admin.backup.Info", no_payload=True, scope="local", read=True
     )
     async def backup_info(self):
-        self.enforce(self.dest.name == "dom0")
-        self.enforce(self.arg)
-        self.enforce("/" not in self.arg)
+        self.enforce_dest_dom0(wants=True)
+        self.enforce_arg(wants=True)
+        self.enforce(
+            "/" not in self.arg,
+            reason="Backup profile ID must not contain forward slash character",
+        )
 
         self.fire_event_for_permission()
 
@@ -1986,8 +2123,8 @@ async def backup_info(self):
             qubes.config.backup_profile_dir, self.arg + ".conf"
         )
         if not os.path.exists(profile_path):
-            raise qubes.exc.PermissionDenied(
-                "Backup profile {} does not exist".format(self.arg)
+            raise qubes.exc.ProtocolError(
+                "Backup profile does not exist: {}".format(self.arg)
             )
 
         backup = await self._load_backup_profile(self.arg, skip_passphrase=True)
@@ -2046,7 +2183,7 @@ def _send_stats_single(
         "admin.vm.Stats", no_payload=True, scope="global", read=True
     )
     async def vm_stats(self):
-        self.enforce(not self.arg)
+        self.enforce_arg(wants=None)
 
         # run until client connection is terminated
         self.cancellable = True
@@ -2077,7 +2214,7 @@ async def vm_stats(self):
         "admin.vm.CurrentState", no_payload=True, scope="local", read=True
     )
     async def vm_current_state(self):
-        self.enforce(not self.arg)
+        self.enforce_arg(wants=None)
         self.fire_event_for_permission()
 
         state = {
@@ -2093,7 +2230,7 @@ async def vm_current_state(self):
     )
     async def vm_notes_get(self):
         """Get qube notes"""
-        self.enforce(self.dest.name != "dom0")
+        self.enforce_dest_dom0(wants=False)
         self.fire_event_for_permission()
         notes = self.dest.get_notes()
         return notes
@@ -2101,7 +2238,7 @@ async def vm_notes_get(self):
     @qubes.api.method("admin.vm.notes.Set", scope="local", write=True)
     async def vm_notes_set(self, untrusted_payload):
         """Set qube notes"""
-        self.enforce(self.dest.name != "dom0")
+        self.enforce_dest_dom0(wants=False)
         self.fire_event_for_permission()
         if len(untrusted_payload) > 256000:
             raise qubes.exc.ProtocolError(
diff --git a/qubes/api/internal.py b/qubes/api/internal.py
index 4abfcc2e5..b516be49c 100644
--- a/qubes/api/internal.py
+++ b/qubes/api/internal.py
@@ -165,8 +165,8 @@ class QubesInternalAPI(qubes.api.AbstractQubesAPI):
 
     @qubes.api.method("internal.GetSystemInfo", no_payload=True)
     async def getsysteminfo(self):
-        self.enforce(self.dest.name == "dom0")
-        self.enforce(not self.arg)
+        self.enforce_dest_dom0(wants=True)
+        self.enforce_arg(wants=None)
 
         system_info = SystemInfoCache.get_system_info(self.app)
 
@@ -188,7 +188,10 @@ async def vm_volume_import(self, untrusted_payload):
         payload) and response from that call will be actually send to the
         caller.
         """
-        self.enforce(self.arg in self.dest.volumes.keys())
+        self.enforce_arg(
+            wants=self.dest.volumes.keys(),
+            short_reason=self.EXC_ARG_NOT_IN_DEST_VOLUMES,
+        )
 
         if untrusted_payload:
             original_method = "admin.vm.volume.ImportWithSize"
@@ -205,12 +208,14 @@ async def vm_volume_import(self, untrusted_payload):
             raise qubes.exc.QubesVMNotHaltedError(self.dest)
 
         requested_size = (
-            self.validate_size(untrusted_payload) if untrusted_payload else None
+            self.validate_size(untrusted_payload, name="size of new data")
+            if untrusted_payload
+            else None
         )
         del untrusted_payload
 
         path = await self.dest.storage.import_data(self.arg, requested_size)
-        self.enforce(" " not in path)
+        self.enforce(" " not in path, reason="Path contains whitespace")
         if requested_size is None:
             size = self.dest.volumes[self.arg].size
         else:
@@ -233,7 +238,10 @@ async def vm_volume_import_end(self, untrusted_payload):
 
         The payload is either 'ok', or 'fail\n<error message>'.
         """
-        self.enforce(self.arg in self.dest.volumes.keys())
+        self.enforce_arg(
+            wants=self.dest.volumes.keys(),
+            short_reason=self.EXC_ARG_NOT_IN_DEST_VOLUMES,
+        )
         success = untrusted_payload == b"ok"
 
         try:
diff --git a/qubes/api/misc.py b/qubes/api/misc.py
index d296af935..6396b94ef 100644
--- a/qubes/api/misc.py
+++ b/qubes/api/misc.py
@@ -18,8 +18,8 @@
 # You should have received a copy of the GNU Lesser General Public
 # License along with this library; if not, see <https://www.gnu.org/licenses/>.
 
-""" Interface for methods not being part of Admin API, but still handled by
-qubesd. """
+"""Interface for methods not being part of Admin API, but still handled by
+qubesd."""
 
 import string
 from datetime import datetime
@@ -45,8 +45,8 @@ async def qubes_features_request(self):
         appropriate extensions. Requests not explicitly handled by some
         extension are ignored.
         """
-        self.enforce(self.dest.name == "dom0")
-        self.enforce(not self.arg)
+        self.enforce_dest_dom0(wants=True)
+        self.enforce_arg(wants=None)
 
         prefix = "/features-request/"
 
@@ -61,7 +61,10 @@ async def qubes_features_request(self):
         safe_set = string.ascii_letters + string.digits + "-.,_= "
         for untrusted_key in untrusted_features:
             untrusted_value = untrusted_features[untrusted_key]
-            self.enforce(all((c in safe_set) for c in untrusted_value))
+            self.enforce(
+                all((c in safe_set) for c in untrusted_value),
+                reason="Value outside safe set: " + safe_set,
+            )
 
         await self.src.fire_event_async(
             "features-request", untrusted_features=untrusted_features
@@ -73,8 +76,8 @@ async def qubes_notify_tools(self):
         """
         Legacy version of qubes.FeaturesRequest, used by Qubes Windows Tools
         """
-        self.enforce(self.dest.name == "dom0")
-        self.enforce(not self.arg)
+        self.enforce_dest_dom0(wants=True)
+        self.enforce_arg(wants=None)
 
         untrusted_features = {}
         safe_set = string.ascii_letters + string.digits
@@ -93,7 +96,10 @@ async def qubes_notify_tools(self):
                 untrusted_value = untrusted_value.decode(
                     "ascii", errors="strict"
                 )
-                self.enforce(all((c in safe_set) for c in untrusted_value))
+                self.enforce(
+                    all((c in safe_set) for c in untrusted_value),
+                    reason="Value outside safe set: " + safe_set,
+                )
                 untrusted_features[feature] = untrusted_value
             del untrusted_value
 
@@ -112,7 +118,10 @@ async def qubes_notify_updates(self, untrusted_payload):
         """
 
         untrusted_update_count = untrusted_payload.strip()
-        self.enforce(untrusted_update_count.isdigit())
+        self.enforce(
+            untrusted_update_count.isdigit(),
+            reason="Update count is not a digit",
+        )
         # now sanitized
         update_count = int(untrusted_update_count)
         del untrusted_update_count
diff --git a/qubes/exc.py b/qubes/exc.py
index 2a577e44a..247aff1ea 100644
--- a/qubes/exc.py
+++ b/qubes/exc.py
@@ -27,6 +27,40 @@ class QubesException(Exception):
     """Exception that can be shown to the user"""
 
 
+class ProtocolError(AssertionError):
+    """Raised deliberately by handlers to indicate a malformed client request.
+
+    Client programming errors. The client made a request that it should have
+    known better than to send in the first place. Includes things like passing
+    an argument or payload to a service documented to not take one. The only
+    way that a correctly behaving client can get ProtocolError is qubesd is
+    either buggy or too old. In HTTP, this would be 400 Bad Request.
+
+    This does not provide any useful information to the client making the
+    request. Therefore, it should only be raised if there is a client
+    *programming* error, such as passing an argument to a request that does not
+    take an argument. It should not be used to reject requests that are valid,
+    but which qubesd is refusing to process. Instead, raise a subclass of
+    :py:class:`QubesException` with a useful error message.
+    """
+
+
+class PermissionDenied(PermissionError):
+    """Raised deliberately by handlers to inform the request is prohibited.
+
+    The request is valid, but the client does not have permission to perform
+    the operation. Clients in dom0 should usually not get this error. In HTTP,
+    this would be 403 Forbidden.
+
+    This does not provide any useful information to the client making the
+    request. Therefore, it should only be raised if there is a client
+    *programming* error, such as passing an argument to a request that does not
+    take an argument. It should not be used to reject requests that are valid,
+    but which qubesd is refusing to process. Instead, raise a subclass of
+    :py:class:`QubesException` with a useful error message.
+    """
+
+
 class QubesVMNotFoundError(QubesException, KeyError):
     """Domain cannot be found in the system"""
 
@@ -154,13 +188,17 @@ def __init__(self, vm, msg=None):
 class QubesPoolInUseError(QubesException):
     """VM is in use, cannot remove."""
 
-    def __init__(self, pool, msg=None):
+    def __init__(self, pool_name, msg=None):
         super().__init__(
-            msg or "Storage pool is in use: {!r}".format(pool.name)
+            msg or "Storage pool is in use: {!r}".format(pool_name)
         )
 
 
-class QubesValueError(QubesException, ValueError):
+class QubesTypeError(ProtocolError, TypeError):
+    """Cannot set some value, because it has incorrect type."""
+
+
+class QubesValueError(ProtocolError, ValueError):
     """Cannot set some value, because it is invalid, out of bounds, etc."""
 
 
@@ -238,7 +276,7 @@ def __str__(self):
         return QubesException.__str__(self)
 
 
-class QubesTagNotFoundError(QubesException, KeyError):
+class QubesTagNotFoundError(QubesException, ProtocolError, KeyError):
     """Tag not set for a given domain"""
 
     def __init__(self, domain, tag):
@@ -251,7 +289,7 @@ def __str__(self):
         return QubesException.__str__(self)
 
 
-class QubesLabelNotFoundError(QubesException, KeyError):
+class QubesLabelNotFoundError(QubesException, ProtocolError, KeyError):
     """Label does not exists"""
 
     def __init__(self, label):
@@ -263,14 +301,6 @@ def __str__(self):
         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"""
-
-
 class DeviceNotAssigned(QubesException, KeyError):
     """
     Trying to unassign not assigned device.
@@ -309,3 +339,48 @@ class UnexpectedDeviceProperty(QubesException, ValueError):
 
 class StoragePoolException(QubesException):
     """A general storage exception"""
+
+
+class QubesVolumeCopyTokenNotFoundError(ProtocolError, KeyError):
+    """Domain volume copy did not specify a configured token."""
+
+    def __init__(self, msg=None):
+        super().__init__(msg or "Token to clone volume of qube was not found")
+
+
+class QubesVolumeCopyInUseError(ProtocolError, Exception):
+    """Domain volume is already being cloned."""
+
+    def __init__(self, vm, volume, msg=None):
+        super().__init__(
+            vm,
+            msg
+            or "Domain volume is being cloned already: {!r}:{!r}".format(
+                vm.name, volume
+            ),
+        )
+
+
+class QubesVolumeRevisionNotFoundError(KeyError):
+    """Specified revision not found in qube volume."""
+
+
+class QubesPoolNotFoundError(KeyError):
+    """Pool does not exist."""
+
+
+class QubesVolumeNotFoundError(KeyError):
+    """Pool does not exist."""
+
+
+class QubesInvalidLabelError(QubesValueError):
+    """Domain label is invalid."""
+
+
+class QubesLabelInUseError(ProtocolError, Exception):
+    """Cannot remove label as it is still in use."""
+
+    def __init__(self, label, msg=None):
+        super().__init__(
+            msg or "Label is still in use: {!r}".format(label),
+        )
diff --git a/qubes/tests/api_admin.py b/qubes/tests/api_admin.py
index 418f765c5..3d9a739ad 100644
--- a/qubes/tests/api_admin.py
+++ b/qubes/tests/api_admin.py
@@ -652,7 +652,7 @@ def test_080_vm_volume_info_invalid_volume(self):
             "keys.return_value": ["root", "private", "volatile", "kernel"]
         }
         self.vm.volumes.configure_mock(**volumes_conf)
-        with self.assertRaises(qubes.exc.PermissionDenied):
+        with self.assertRaises(qubes.exc.ProtocolError):
             self.call_mgmt_func(
                 b"admin.vm.volume.Info", b"test-vm1", b"no-such-volume"
             )
@@ -688,7 +688,7 @@ def test_090_vm_volume_listsnapshots_invalid_volume(self):
             "keys.return_value": ["root", "private", "volatile", "kernel"]
         }
         self.vm.volumes.configure_mock(**volumes_conf)
-        with self.assertRaises(qubes.exc.PermissionDenied):
+        with self.assertRaises(qubes.exc.ProtocolError):
             self.call_mgmt_func(
                 b"admin.vm.volume.ListSnapshots",
                 b"test-vm1",
@@ -713,7 +713,7 @@ def test_100_vm_volume_snapshot_invalid_volume(self):
             },
         }
         self.vm.volumes.configure_mock(**volumes_conf)
-        with self.assertRaises(qubes.exc.PermissionDenied):
+        with self.assertRaises(qubes.exc.ProtocolError):
             self.call_mgmt_func(
                 b"admin.vm.volume.Snapshots", b"test-vm1", b"no-such-volume"
             )
@@ -728,7 +728,7 @@ def test_100_vm_volume_snapshot_invalid_revision(self):
             "keys.return_value": ["root", "private", "volatile", "kernel"]
         }
         self.vm.volumes.configure_mock(**volumes_conf)
-        with self.assertRaises(qubes.exc.PermissionDenied):
+        with self.assertRaises(qubes.exc.ProtocolError):
             self.call_mgmt_func(
                 b"admin.vm.volume.Snapshots",
                 b"test-vm1",
@@ -783,7 +783,7 @@ def test_110_vm_volume_revert_invalid_rev(self):
         }
         self.vm.volumes.configure_mock(**volumes_conf)
         self.vm.storage = unittest.mock.Mock()
-        with self.assertRaises(qubes.exc.PermissionDenied):
+        with self.assertRaises(qubes.exc.QubesVolumeRevisionNotFoundError):
             self.call_mgmt_func(
                 b"admin.vm.volume.Revert",
                 b"test-vm1",
@@ -1069,7 +1069,7 @@ def test_160_pool_add_invalid_driver(self, mock_parameters, mock_drivers):
 
         add_pool_mock, self.app.add_pool = self.coroutine_mock()
 
-        with self.assertRaises(qubes.exc.QubesException):
+        with self.assertRaises(qubes.exc.ProtocolError):
             self.call_mgmt_func(
                 b"admin.pool.Add",
                 b"dom0",
@@ -1097,7 +1097,7 @@ def test_160_pool_add_invalid_param(self, mock_parameters, mock_drivers):
 
         add_pool_mock, self.app.add_pool = self.coroutine_mock()
 
-        with self.assertRaises(qubes.exc.QubesException):
+        with self.assertRaises(qubes.exc.ProtocolError):
             self.call_mgmt_func(
                 b"admin.pool.Add",
                 b"dom0",
@@ -1127,7 +1127,7 @@ def test_160_pool_add_missing_name(self, mock_parameters, mock_drivers):
 
         add_pool_mock, self.app.add_pool = self.coroutine_mock()
 
-        with self.assertRaises(qubes.exc.PermissionDenied):
+        with self.assertRaises(qubes.exc.ProtocolError):
             self.call_mgmt_func(
                 b"admin.pool.Add",
                 b"dom0",
@@ -1155,7 +1155,7 @@ def test_160_pool_add_existing_pool(self, mock_parameters, mock_drivers):
 
         add_pool_mock, self.app.add_pool = self.coroutine_mock()
 
-        with self.assertRaises(qubes.exc.PermissionDenied):
+        with self.assertRaises(qubes.exc.QubesPoolInUseError):
             self.call_mgmt_func(
                 b"admin.pool.Add",
                 b"dom0",
@@ -1185,7 +1185,7 @@ def test_160_pool_add_invalid_config_format(
 
         add_pool_mock, self.app.add_pool = self.coroutine_mock()
 
-        with self.assertRaises(qubes.exc.PermissionDenied):
+        with self.assertRaises(qubes.exc.ProtocolError):
             self.call_mgmt_func(
                 b"admin.pool.Add",
                 b"dom0",
@@ -1218,7 +1218,7 @@ def test_170_pool_remove_invalid_pool(self):
             "test-pool": unittest.mock.Mock(),
         }
         remove_pool_mock, self.app.remove_pool = self.coroutine_mock()
-        with self.assertRaises(qubes.exc.PermissionDenied):
+        with self.assertRaises(qubes.exc.ProtocolError):
             self.call_mgmt_func(b"admin.pool.Remove", b"dom0", b"no-such-pool")
         self.assertEqual(remove_pool_mock.mock_calls, [])
         self.assertFalse(self.app.save.called)
@@ -1285,7 +1285,7 @@ def test_200_label_create_invalid_color(self):
             "keys.return_value": range(1, 9),
         }
         self.app.labels.configure_mock(**labels_config)
-        with self.assertRaises(qubes.exc.PermissionDenied):
+        with self.assertRaises(qubes.exc.ProtocolError):
             self.call_mgmt_func(
                 b"admin.label.Create", b"dom0", b"cyan", b"abcd"
             )
@@ -1303,15 +1303,15 @@ def test_200_label_create_invalid_name(self):
             "keys.return_value": range(1, 9),
         }
         self.app.labels.configure_mock(**labels_config)
-        with self.assertRaises(qubes.exc.PermissionDenied):
+        with self.assertRaises(qubes.exc.ProtocolError):
             self.call_mgmt_func(
                 b"admin.label.Create", b"dom0", b"01", b"0xff0000"
             )
-        with self.assertRaises(qubes.exc.PermissionDenied):
+        with self.assertRaises(qubes.exc.ProtocolError):
             self.call_mgmt_func(
                 b"admin.label.Create", b"dom0", b"../xxx", b"0xff0000"
             )
-        with self.assertRaises(qubes.exc.PermissionDenied):
+        with self.assertRaises(qubes.exc.ProtocolError):
             self.call_mgmt_func(
                 b"admin.label.Create",
                 b"dom0",
@@ -1348,7 +1348,7 @@ def test_210_label_remove(self):
         self.assertTrue(self.app.save.called)
 
     def test_210_label_remove_invalid_label(self):
-        with self.assertRaises(qubes.exc.QubesValueError):
+        with self.assertRaises(qubes.exc.ProtocolError):
             self.call_mgmt_func(
                 b"admin.label.Remove", b"dom0", b"no-such-label"
             )
@@ -1357,7 +1357,7 @@ def test_210_label_remove_invalid_label(self):
     def test_210_label_remove_default_label(self):
         self.app.labels = unittest.mock.MagicMock(wraps=self.app.labels)
         self.app.get_label = unittest.mock.Mock(**{"return_value.index": 6})
-        with self.assertRaises(qubes.exc.PermissionDenied):
+        with self.assertRaises(qubes.exc.ProtocolError):
             self.call_mgmt_func(b"admin.label.Remove", b"dom0", b"blue")
         self.assertEqual(self.app.labels.mock_calls, [])
         self.assertFalse(self.app.save.called)
@@ -1365,7 +1365,7 @@ def test_210_label_remove_default_label(self):
     def test_210_label_remove_in_use(self):
         self.app.labels = unittest.mock.MagicMock(wraps=self.app.labels)
         self.app.get_label = unittest.mock.Mock(**{"return_value.index": 1})
-        with self.assertRaises(qubes.exc.PermissionDenied):
+        with self.assertRaises(qubes.exc.ProtocolError):
             self.call_mgmt_func(b"admin.label.Remove", b"dom0", b"red")
         self.assertEqual(self.app.labels.mock_calls, [])
         self.assertFalse(self.app.save.called)
@@ -1447,7 +1447,7 @@ async def coroutine_mock(*args, **kwargs):
             return func_mock(*args, **kwargs)
 
         self.vm.shutdown = coroutine_mock
-        with self.assertRaises(qubes.exc.PermissionDenied):
+        with self.assertRaises(qubes.exc.ProtocolError):
             self.call_mgmt_func(b"admin.vm.Shutdown", b"test-vm1", b"forcewait")
         func_mock.assert_not_called()
 
@@ -1988,7 +1988,7 @@ def test_336_vm_create_spurious_pool(self, storage_mock):
     @unittest.mock.patch("qubes.storage.Storage.create")
     def test_337_vm_create_duplicate_name(self, storage_mock):
         storage_mock.side_effect = self.dummy_coro
-        with self.assertRaises(qubes.exc.QubesException):
+        with self.assertRaises(qubes.exc.ProtocolError):
             self.call_mgmt_func(
                 b"admin.vm.Create.AppVM",
                 b"dom0",
@@ -2114,7 +2114,7 @@ def test_343_vm_create_in_pool_invalid_pool2(self, storage_mock):
     @unittest.mock.patch("qubes.storage.Storage.create")
     def test_344_vm_create_in_pool_invalid_volume(self, storage_mock):
         storage_mock.side_effect = self.dummy_coro
-        with self.assertRaises(qubes.exc.PermissionDenied):
+        with self.assertRaises(qubes.exc.ProtocolError):
             self.call_mgmt_func(
                 b"admin.vm.CreateInPool.AppVM",
                 b"dom0",
@@ -3199,7 +3199,7 @@ def test_521_vm_volume_clone_invalid_volume(self):
         self.vm.storage.get_volume = lambda x: x
         self.vm2.storage.get_volume = lambda x: x
 
-        with self.assertRaises(qubes.exc.PermissionDenied):
+        with self.assertRaises(qubes.exc.ProtocolError):
             self.call_mgmt_func(
                 b"admin.vm.volume.CloneFrom", b"test-vm1", b"private123", b""
             )
@@ -3219,7 +3219,7 @@ def test_522_vm_volume_clone_invalid_volume2(self):
         token = self.call_mgmt_func(
             b"admin.vm.volume.CloneFrom", b"test-vm1", b"private", b""
         )
-        with self.assertRaises(qubes.exc.PermissionDenied):
+        with self.assertRaises(qubes.exc.ProtocolError):
             self.call_mgmt_func(
                 b"admin.vm.volume.CloneTo",
                 b"test-vm1",
@@ -3250,7 +3250,7 @@ def get_volume(vid):
                 return unittest.mock.DEFAULT
 
         self.pool.get_volume.side_effect = get_volume
-        with self.assertRaises(qubes.exc.PermissionDenied):
+        with self.assertRaises(qubes.exc.QubesVolumeNotFoundError):
             self.call_mgmt_func(
                 b"admin.vm.volume.CloneTo",
                 b"test-vm1",
@@ -3270,7 +3270,7 @@ def test_524_vm_volume_clone_invlid_token(self):
         self.vm.storage.get_volume = lambda x: x
         self.vm2.storage.get_volume = lambda x: x
 
-        with self.assertRaises(qubes.exc.PermissionDenied):
+        with self.assertRaises(qubes.exc.ProtocolError):
             self.call_mgmt_func(
                 b"admin.vm.volume.CloneTo",
                 b"test-vm1",
@@ -3461,7 +3461,7 @@ def test_585_firewall_set_invalid(self):
         rules_txt = (
             "action=accept dstports=1-1024 comment=ążźł\n" "action=drop\n"
         )
-        with self.assertRaises(UnicodeDecodeError):
+        with self.assertRaises(qubes.exc.ProtocolError):
             self.call_mgmt_func(
                 b"admin.vm.firewall.Set", b"test-vm1", b"", rules_txt.encode()
             )
@@ -4212,7 +4212,7 @@ def test_655_vm_device_set_mode_invalid_value(self):
         with unittest.mock.patch.object(
             qubes.vm.qubesvm.QubesVM, "is_halted", lambda _: False
         ):
-            with self.assertRaises(qubes.exc.PermissionDenied):
+            with self.assertRaises(qubes.exc.ProtocolError):
                 self.call_mgmt_func(
                     b"admin.vm.device.testclass.Set.assignment",
                     b"test-vm1",
@@ -4262,7 +4262,7 @@ def test_663_vm_device_denied_add_multiple(self):
 
     def test_664_vm_device_denied_add_repeated(self):
         self.vm.devices_denied = "b******p012345p53**2*"
-        with self.assertRaises(qubes.exc.QubesValueError):
+        with self.assertRaises(qubes.exc.ProtocolError):
             self.call_mgmt_func(
                 b"admin.vm.device.denied.Add",
                 b"test-vm1",
@@ -4298,7 +4298,7 @@ def test_670_vm_device_denied_remove(self):
 
     def test_671_vm_device_denied_remove_repeated(self):
         self.vm.devices_denied = "b******p012345p53**2*"
-        with self.assertRaises(qubes.exc.QubesValueError):
+        with self.assertRaises(qubes.exc.ProtocolError):
             self.call_mgmt_func(
                 b"admin.vm.device.denied.Remove",
                 b"test-vm1",
@@ -4819,17 +4819,16 @@ def test_991_vm_unexpected_argument(self):
         vm_mock.qid = self.vm.qid
         vm_mock.__lt__ = lambda x, y: x.qid < y.qid
         self.app.domains._dict[self.vm.qid] = vm_mock
-        exceptions = (qubes.exc.PermissionDenied, qubes.exc.ProtocolError)
         for method in methods_with_no_argument:
             # should reject argument regardless of having payload or not
             with self.subTest(method.decode("ascii")):
-                with self.assertRaises(qubes.exc.PermissionDenied):
+                with self.assertRaises(qubes.exc.ProtocolError):
                     self.call_mgmt_func(method, b"test-vm1", b"some-arg", b"")
                 self.assertFalse(vm_mock.called)
                 self.assertFalse(self.app.save.called)
 
             with self.subTest(method.decode("ascii") + "+payload"):
-                with self.assertRaises(exceptions):
+                with self.assertRaises(qubes.exc.ProtocolError):
                     self.call_mgmt_func(
                         method, b"test-vm1", b"unexpected-arg", b"some-payload"
                     )
@@ -4898,17 +4897,16 @@ def test_993_dom0_unexpected_argument(self):
         vm_mock.qid = self.vm.qid
         vm_mock.__lt__ = lambda x, y: x.qid < y.qid
         self.app.domains._dict[self.vm.qid] = vm_mock
-        exceptions = (qubes.exc.PermissionDenied, qubes.exc.ProtocolError)
         for method in methods_with_no_argument:
             # should reject argument regardless of having payload or not
             with self.subTest(method.decode("ascii")):
-                with self.assertRaises(qubes.exc.PermissionDenied):
+                with self.assertRaises(qubes.exc.ProtocolError):
                     self.call_mgmt_func(method, b"dom0", b"some-arg", b"")
                 self.assertFalse(vm_mock.called)
                 self.assertFalse(self.app.save.called)
 
             with self.subTest(method.decode("ascii") + "+payload"):
-                with self.assertRaises(exceptions):
+                with self.assertRaises(qubes.exc.ProtocolError):
                     self.call_mgmt_func(
                         method, b"dom0", b"unexpected-arg", b"some-payload"
                     )
@@ -4954,29 +4952,28 @@ def test_994_dom0_only_calls(self):
         vm_mock.qid = self.vm.qid
         vm_mock.__lt__ = lambda x, y: x.qid < y.qid
         self.app.domains._dict[self.vm.qid] = vm_mock
-        exceptions = (qubes.exc.PermissionDenied, qubes.exc.ProtocolError)
         for method in methods_for_dom0_only:
             # should reject call regardless of having payload or not
             with self.subTest(method.decode("ascii")):
-                with self.assertRaises(exceptions):
+                with self.assertRaises(qubes.exc.ProtocolError):
                     self.call_mgmt_func(method, b"test-vm1", b"", b"")
                 self.assertFalse(vm_mock.called)
                 self.assertFalse(self.app.save.called)
 
             with self.subTest(method.decode("ascii") + "+arg"):
-                with self.assertRaises(exceptions):
+                with self.assertRaises(qubes.exc.ProtocolError):
                     self.call_mgmt_func(method, b"test-vm1", b"some-arg", b"")
                 self.assertFalse(vm_mock.called)
                 self.assertFalse(self.app.save.called)
 
             with self.subTest(method.decode("ascii") + "+payload"):
-                with self.assertRaises(exceptions):
+                with self.assertRaises(qubes.exc.ProtocolError):
                     self.call_mgmt_func(method, b"test-vm1", b"", b"payload")
                 self.assertFalse(vm_mock.called)
                 self.assertFalse(self.app.save.called)
 
             with self.subTest(method.decode("ascii") + "+arg+payload"):
-                with self.assertRaises(exceptions):
+                with self.assertRaises(qubes.exc.ProtocolError):
                     self.call_mgmt_func(
                         method, b"test-vm1", b"some-arg", b"some-payload"
                     )
@@ -5039,30 +5036,29 @@ def test_995_vm_only_calls(self):
         vm_mock.qid = self.vm.qid
         vm_mock.__lt__ = lambda x, y: x.qid < y.qid
         self.app.domains._dict[self.vm.qid] = vm_mock
-        exceptions = (qubes.exc.PermissionDenied, qubes.exc.ProtocolError)
         for method in methods_for_vm_only:
             # should reject payload regardless of having argument or not
             # should reject call regardless of having payload or not
             with self.subTest(method.decode("ascii")):
-                with self.assertRaises(exceptions):
+                with self.assertRaises(qubes.exc.ProtocolError):
                     self.call_mgmt_func(method, b"dom0", b"", b"")
                 self.assertFalse(vm_mock.called)
                 self.assertFalse(self.app.save.called)
 
             with self.subTest(method.decode("ascii") + "+arg"):
-                with self.assertRaises(exceptions):
+                with self.assertRaises(qubes.exc.ProtocolError):
                     self.call_mgmt_func(method, b"dom0", b"some-arg", b"")
                 self.assertFalse(vm_mock.called)
                 self.assertFalse(self.app.save.called)
 
             with self.subTest(method.decode("ascii") + "+payload"):
-                with self.assertRaises(exceptions):
+                with self.assertRaises(qubes.exc.ProtocolError):
                     self.call_mgmt_func(method, b"dom0", b"", b"payload")
                 self.assertFalse(vm_mock.called)
                 self.assertFalse(self.app.save.called)
 
             with self.subTest(method.decode("ascii") + "+arg+payload"):
-                with self.assertRaises(exceptions):
+                with self.assertRaises(qubes.exc.ProtocolError):
                     self.call_mgmt_func(
                         method, b"dom0", b"some-arg", b"some-payload"
                     )
diff --git a/qubes/tests/api_misc.py b/qubes/tests/api_misc.py
index cf885abcd..2787e0bd0 100644
--- a/qubes/tests/api_misc.py
+++ b/qubes/tests/api_misc.py
@@ -117,7 +117,7 @@ def test_002_features_request_invalid1(self):
             "/features-request/feature1": b"test$chars",
         }
         self.configure_qdb(qdb_entries)
-        with self.assertRaises(qubes.exc.PermissionDenied):
+        with self.assertRaises(qubes.exc.ProtocolError):
             self.call_mgmt_func(b"qubes.FeaturesRequest")
         self.assertEqual(self.app.mock_calls, [])
         self.assertEqual(
@@ -217,7 +217,7 @@ def test_015_notify_tools_invalid_value_qrexec(self):
             "/qubes-tools/default-user": b"user",
         }
         self.configure_qdb(qdb_entries)
-        with self.assertRaises(qubes.exc.PermissionDenied):
+        with self.assertRaises(qubes.exc.ProtocolError):
             self.call_mgmt_func(b"qubes.NotifyTools")
         self.assertEqual(self.app.mock_calls, [])
         self.assertEqual(
@@ -236,7 +236,7 @@ def test_016_notify_tools_invalid_value_gui(self):
             "/qubes-tools/default-user": b"user",
         }
         self.configure_qdb(qdb_entries)
-        with self.assertRaises(qubes.exc.PermissionDenied):
+        with self.assertRaises(qubes.exc.ProtocolError):
             self.call_mgmt_func(b"qubes.NotifyTools")
         self.assertEqual(self.app.mock_calls, [])
         self.assertEqual(
@@ -296,7 +296,7 @@ def test_021_notify_updates_standalone2(self):
 
     def test_022_notify_updates_invalid(self):
         del self.src.template
-        with self.assertRaises(qubes.exc.PermissionDenied):
+        with self.assertRaises(qubes.exc.ProtocolError):
             self.call_mgmt_func(b"qubes.NotifyUpdates", payload=b"")
         self.assertEqual(self.src.mock_calls, [])
         self.assertEqual(self.tpl.mock_calls, [])
@@ -304,7 +304,7 @@ def test_022_notify_updates_invalid(self):
 
     def test_023_notify_updates_invalid2(self):
         del self.src.template
-        with self.assertRaises(qubes.exc.PermissionDenied):
+        with self.assertRaises(qubes.exc.ProtocolError):
             self.call_mgmt_func(b"qubes.NotifyUpdates", payload=b"no updates")
         self.assertEqual(self.src.mock_calls, [])
         self.assertEqual(self.tpl.mock_calls, [])
diff --git a/qubes/tests/vm/dispvm.py b/qubes/tests/vm/dispvm.py
index 7644f5fb7..7f71b3eef 100644
--- a/qubes/tests/vm/dispvm.py
+++ b/qubes/tests/vm/dispvm.py
@@ -465,7 +465,7 @@ def test_011_create_direct_reject(self):
                     "__getitem__.side_effect": orig_getitem,
                 }
             )
-            with self.assertRaises(qubes.exc.QubesException):
+            with self.assertRaises(qubes.exc.ProtocolError):
                 self.app.add_new_vm(
                     qubes.vm.dispvm.DispVM,
                     name="test-dispvm",
diff --git a/qubes/utils.py b/qubes/utils.py
index 49e0d0e89..0de807a8f 100644
--- a/qubes/utils.py
+++ b/qubes/utils.py
@@ -510,3 +510,43 @@ def is_pci_path(device_id: str):
         r"(-[0-9a-f]{2}_[0-9a-f]{2}\.[0-9a-f])*\Z"
     )
     return bool(path_re.match(device_id))
+
+
+def validate_label(untrusted_label) -> None:
+    # don't confuse label name with label index
+    if str(untrusted_label).isdigit():
+        raise qubes.exc.QubesInvalidLabelError("Label must not be a digit")
+
+    allowed_chars = string.ascii_letters + string.digits + "-_."
+    if not all(c in allowed_chars for c in untrusted_label):
+        raise qubes.exc.QubesInvalidLabelError(
+            "Label must be in safe set: " + allowed_chars
+        )
+
+
+def validate_label_value(untrusted_label_value) -> None:
+    try:
+        untrusted_value = untrusted_label_value.decode(
+            "ascii", "strict"
+        ).strip()
+    except UnicodeDecodeError:
+        raise qubes.exc.QubesInvalidLabelError(
+            "Label value must be encoded in ASCII"
+        )
+
+    if len(untrusted_value) != 8:
+        raise qubes.exc.QubesInvalidLabelError(
+            "Label value must have length of 8"
+        )
+
+    if not untrusted_value.startswith("0x"):
+        raise qubes.exc.QubesInvalidLabelError(
+            "Label value must start with: 0x"
+        )
+
+    # besides prefix, only hex digits are allowed
+    if not all(x in string.hexdigits for x in untrusted_value[2:]):
+        raise qubes.exc.QubesInvalidLabelError(
+            "Label value must only contain hexadecimal digits after prefix: "
+            + string.hexdigits
+        )
diff --git a/qubes/vm/__init__.py b/qubes/vm/__init__.py
index d4751e8cb..93beb4b56 100644
--- a/qubes/vm/__init__.py
+++ b/qubes/vm/__init__.py
@@ -20,9 +20,7 @@
 # License along with this library; if not, see <https://www.gnu.org/licenses/>.
 #
 
-"""Qubes Virtual Machines
-
-"""
+"""Qubes Virtual Machines"""
 import asyncio
 import re
 import string
@@ -37,7 +35,7 @@
 import qubes.events
 import qubes.features
 import qubes.log
-from qubes.utils import is_pci_path, sbdf_to_path
+from qubes.utils import is_pci_path, sbdf_to_path, validate_label
 
 VM_ENTRY_POINT = "qubes.vm"
 
@@ -104,6 +102,7 @@ def setter_label(self, prop, value):
     if isinstance(value, str) and value.startswith("label-"):
         return self.app.labels[int(value.split("-", 1)[1])]
 
+    validate_label(untrusted_label=value)
     return self.app.get_label(value)