admin: validate feature name in vm_feature_remove

qubes/api/__init__.py

@@ -28,6 +28,7 @@
 import shutil
 import socket
 import struct
+import string
 import traceback
 from typing import Union, Any
 import uuid
@@ -165,8 +166,19 @@ def __init__(
         self.dest = decode_vm(dest, app.domains)
 
         #: argument
-        self.arg = arg.decode("ascii")
-
+        untrusted_arg = arg.decode("ascii")
+        # Validate arg against qrexec allowed characters
+        # Consistent with sanitize_name() in qrexec-daemon.c
+        _allowed_arg_chars = string.ascii_letters + string.digits + "-+_."
+        if untrusted_arg and not all(
+            c in _allowed_arg_chars for c in untrusted_arg
+        ):
+            raise ProtocolError(
+                "Argument contains characters outside safe set: "
+                + _allowed_arg_chars
+            )
+        self.arg = untrusted_arg
+        del untrusted_arg
         #: name of the method
         self.method = method_name.decode("ascii")
 
@@ -232,7 +244,7 @@ def fire_event_for_permission(self, **kwargs):
             pre_event=True,
             dest=self.dest,
             arg=self.arg,
-            **kwargs
+            **kwargs,
         )
 
     def fire_event_for_filter(self, iterable, **kwargs):

qubes/api/admin.py

@@ -100,6 +100,18 @@ def on_domain_delete(self, subject, event, vm):
         vm.remove_handler("*", self.vm_handler)
 
 
+def _validate_feature_name(name):
+    """Validate feature name against allowed character set.
+
+    Allowed characters: alphanumeric, dot, underscore, hyphen.
+    Raises ProtocolError if name is invalid.
+    """
+    if not re.match(r"\A[a-zA-Z0-9_.+-]+\Z", name):
+        raise qubes.exc.ProtocolError(
+            "feature name contains illegal characters"
+        )
+
+
 class QubesAdminAPI(qubes.api.AbstractQubesAPI):
     """Implementation of Qubes Management API calls
 
@@ -1162,8 +1174,7 @@ async def vm_feature_checkwithtpladminvm(self):
         "admin.vm.feature.Remove", no_payload=True, scope="local", write=True
     )
     async def vm_feature_remove(self):
-        # validation of self.arg done by qrexec-policy is enough
-
+        _validate_feature_name(self.arg)
         self.fire_event_for_permission()
         try:
             del self.dest.features[self.arg]
@@ -1175,10 +1186,7 @@ async def vm_feature_remove(self):
     async def vm_feature_set(self, untrusted_payload):
         untrusted_value = untrusted_payload.decode("ascii", errors="strict")
         del untrusted_payload
-        if re.match(r"\A[a-zA-Z0-9_.-]+\Z", self.arg) is None:
-            raise qubes.exc.QubesValueError(
-                "feature name contains illegal characters"
-            )
+        _validate_feature_name(self.arg)
         if re.match(r"\A[\x20-\x7E]*\Z", untrusted_value) is None:
             raise qubes.exc.QubesValueError(
                 f"{self.arg} value contains illegal characters"
@@ -2021,7 +2029,7 @@ def _send_stats_single(
         the next iteration)
         """
 
-        (info_time, info) = self.app.host.get_vm_stats(
+        info_time, info = self.app.host.get_vm_stats(
             info_time, info, only_vm=only_vm
         )
         for vm_id, vm_info in info.items():

qubes/tests/api_admin.py

@@ -1333,11 +1333,11 @@ def test_200_label_create_invalid_name(self):
             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",
@@ -1694,6 +1694,52 @@ def test_301_feature_remove_none(self):
             )
         self.assertFalse(self.app.save.called)
 
+    def test_302_feature_remove_invalid_name(self):
+        # service.* feature names with path traversal must raise ProtocolError
+        with self.assertRaises(qubes.exc.ProtocolError):
+            self.call_mgmt_func(
+                b"admin.vm.feature.Remove",
+                b"test-vm1",
+                b"service.../../../root/gotcha",
+                b"some-value",
+            )
+        self.assertFalse(self.app.save.called)
+
+    def test_302b_protocol_error_logging(self):
+        """
+        Verify that ProtocolError raised in __init__() is:
+        - logged server-side via log.warning()
+        - NOT sent to client (no send_exception())
+        See: QubesOS/qubes-issues#7186
+        """
+        self.app.log.warning = unittest.mock.Mock()
+        transport_mock = unittest.mock.Mock()
+
+        proto = qubes.api.QubesDaemonProtocol(
+            qubes.api.admin.QubesAdminAPI,
+            app=self.app,
+            debug=False,
+        )
+        proto.transport = transport_mock
+
+        loop = asyncio.get_event_loop()
+        loop.run_until_complete(
+            proto.respond(
+                b"dom0",
+                b"admin.vm.feature.Remove",
+                b"test-vm1",
+                b"service.../../../root/gotcha",
+                untrusted_payload=b"",
+            )
+        )
+
+        # ProtocolError is logged server-side
+        self.assertTrue(self.app.log.warning.called)
+        # Nothing sent to client
+        transport_mock.write.assert_not_called()
+        # Connection aborted
+        transport_mock.abort.assert_called()
+
     def test_303_feature_prohibited(self):
         del self.app.domains[0].fire_event
         feature = qubes.ext.admin.PROHIBITED_FEATURES[0]