use qubes.VMRootExec/qubes.VMRootShell services when root needed

When running a command as root is requested, use
qubes.VMRootExec/qubes.VMRootShell services, instead of specifying
target user directly. The explicit target user can be specified only
from dom0, but alternative qrexec service can be used also from GUI
domain - if policy allows of course.

QubesOS/qubes-issues#4186

Author: marmarek

qubesadmin/tests/tools/qvm_run.py

@@ -1444,3 +1444,114 @@ def test_032_argparse_bug_workaround_unnamed_dispvm(self):
             ],
         )
         self.assertAllCalled()
+
+    def test_040_run_root_shell(self):
+        self.app.expected_calls[("dom0", "admin.vm.List", None, None)] = (
+            b"0\x00test-vm class=AppVM state=Running\n"
+        )
+        self.app.expected_calls[
+            ("test-vm", "admin.vm.feature.CheckWithTemplate", "os", None)
+        ] = b"2\x00QubesFeatureNotFoundError\x00\x00Feature 'os' not set\x00"
+        self.app.expected_calls[
+            ("test-vm", "admin.vm.CurrentState", None, None)
+        ] = b"0\x00power_state=Running"
+        ret = qubesadmin.tools.qvm_run.main(
+            ["--no-gui", "-u", "root", "test-vm", "shell command"], app=self.app
+        )
+        self.assertEqual(ret, 0)
+        self.assertEqual(
+            self.app.service_calls,
+            [
+                (
+                    "test-vm",
+                    "qubes.VMRootShell",
+                    {
+                        "stdout": subprocess.DEVNULL,
+                        "stderr": subprocess.DEVNULL,
+                        "user": None,
+                    },
+                ),
+                ("test-vm", "qubes.VMRootShell", b"shell command; exit\n"),
+            ],
+        )
+        self.assertAllCalled()
+
+    def test_041_run_root_exec(self):
+        self.app.expected_calls[("dom0", "admin.vm.List", None, None)] = (
+            b"0\x00test-vm class=AppVM state=Running\n"
+        )
+        self.app.expected_calls[
+            ("test-vm", "admin.vm.feature.CheckWithTemplate", "vmexec", None)
+        ] = b"0\x001"
+        self.app.expected_calls[
+            (
+                "test-vm",
+                "admin.vm.feature.CheckWithTemplate",
+                "supported-rpc.qubes.VMRootExec",
+                None,
+            )
+        ] = b"0\x001"
+        self.app.expected_calls[
+            ("test-vm", "admin.vm.CurrentState", None, None)
+        ] = b"0\x00power_state=Running"
+        ret = qubesadmin.tools.qvm_run.main(
+            ["--no-gui", "-u", "root", "test-vm", "command", "arg"],
+            app=self.app,
+        )
+        self.assertEqual(ret, 0)
+        self.assertEqual(
+            self.app.service_calls,
+            [
+                (
+                    "test-vm",
+                    "qubes.VMRootExec+command+arg",
+                    {
+                        "stdout": subprocess.DEVNULL,
+                        "stderr": subprocess.DEVNULL,
+                        "user": None,
+                    },
+                ),
+                ("test-vm", "qubes.VMRootExec+command+arg", b""),
+            ],
+        )
+        self.assertAllCalled()
+
+    def test_041_run_root_exec_not_supported(self):
+        self.app.expected_calls[("dom0", "admin.vm.List", None, None)] = (
+            b"0\x00test-vm class=AppVM state=Running\n"
+        )
+        self.app.expected_calls[
+            ("test-vm", "admin.vm.feature.CheckWithTemplate", "vmexec", None)
+        ] = b"0\x001"
+        self.app.expected_calls[
+            (
+                "test-vm",
+                "admin.vm.feature.CheckWithTemplate",
+                "supported-rpc.qubes.VMRootExec",
+                None,
+            )
+        ] = b"2\x00QubesFeatureNotFoundError\x00\x00Feature '...' not set\x00"
+        self.app.expected_calls[
+            ("test-vm", "admin.vm.CurrentState", None, None)
+        ] = b"0\x00power_state=Running"
+        ret = qubesadmin.tools.qvm_run.main(
+            ["--no-gui", "-u", "root", "test-vm", "command", "arg"],
+            app=self.app,
+        )
+        self.assertEqual(ret, 0)
+        self.assertEqual(
+            self.app.service_calls,
+            [
+                (
+                    "test-vm",
+                    "qubes.VMExec+command+arg",
+                    {
+                        "stdout": subprocess.DEVNULL,
+                        "stderr": subprocess.DEVNULL,
+                        "user": "root",
+                    },
+                ),
+                ("test-vm", "qubes.VMExec+command+arg", b""),
+            ],
+        )
+        self.assertAllCalled()

qubesadmin/tools/qvm_run.py

@@ -271,16 +271,27 @@ def run_command_single(args, vm):
             service = "qubes.VMExec"
             if args.gui and args.dispvm:
                 service = "qubes.VMExecGUI"
+            elif args.user == "root" and vm.features.check_with_template(
+                "supported-rpc.qubes.VMRootExec", False
+            ):
+                service = "qubes.VMRootExec"
+                args.user = None
             service += "+" + qubesadmin.utils.encode_for_vmexec(all_args)
         else:
             service = "qubes.VMShell"
             if args.gui and args.dispvm:
                 service += "+WaitForSession"
+            elif args.user == "root":
+                service = "qubes.VMRootShell"
+                args.user = None
             shell_cmd = " ".join(shlex.quote(arg) for arg in all_args)
     else:
         service = "qubes.VMShell"
         if args.gui and args.dispvm:
             service += "+WaitForSession"
+        elif args.user == "root":
+            service = "qubes.VMRootShell"
+            args.user = None
         shell_cmd = args.cmd
 
     proc = vm.run_service(service, user=args.user, **run_kwargs)

qubesadmin/vm/__init__.py

@@ -355,8 +355,12 @@ def run(self, command, input=None, **kwargs):
         """Run a shell command inside the domain using qubes.VMShell qrexec."""
         # pylint: disable=redefined-builtin
         try:
+            service = "qubes.VMShell"
+            if kwargs.get("user", None) == "root":
+                kwargs.pop("user")
+                service = "qubes.VMRootShell"
             return self.run_service_for_stdio(
-                "qubes.VMShell",
+                service,
                 input=self.prepare_input_for_vmshell(command, input),
                 **kwargs
             )
@@ -374,9 +378,15 @@ def run_with_args(self, *args, **kwargs):
         """  # pylint: disable=redefined-builtin
         if self.features.check_with_template("vmexec", False):
             try:
+                service = "qubes.VMExec+"
+                if kwargs.get("user", None) == "root":
+                    if self.features.check_with_template(
+                        "supported-rpc.qubes.VMRootExec", False
+                    ):
+                        kwargs.pop("user")
+                        service = "qubes.VMRootExec+"
                 return self.run_service_for_stdio(
-                    "qubes.VMExec+" + qubesadmin.utils.encode_for_vmexec(args),
-                    **kwargs
+                    service + qubesadmin.utils.encode_for_vmexec(args), **kwargs
                 )
             except subprocess.CalledProcessError as e:
                 e.cmd = str(args)