tools/qvm-run: use prefix_data feature instead of copy_stdin subprocess

qubesadmin/app.py

@@ -688,6 +688,7 @@ def run_service(
         localcmd: str | None=None,
         wait: bool=True,
         autostart: bool=True,
+        prefix_data: bytes | None=None,
         **kwargs,
     ) -> Popen:
         """Run qrexec service in a given destination
@@ -930,6 +931,7 @@ def run_service(
         localcmd: str | None=None,
         wait: bool=True,
         autostart: bool=True,
+        prefix_data: bytes | None=None,
         **kwargs,
     ) -> Popen:
         """Run qrexec service in a given destination
@@ -949,6 +951,8 @@ def run_service(
             raise ValueError("Empty destination name allowed only from a VM")
         if not wait and localcmd:
             raise ValueError("wait=False incompatible with localcmd")
+        if prefix_data and b"\0" in prefix_data:
+            raise ValueError("prefix_data cannot contain NUL character")
         if autostart:
             try:
                 self.qubesd_call(dest, "admin.vm.Start")
@@ -973,6 +977,10 @@ def run_service(
                 raise NotImplementedError(
                     "wait=False not implemented in dom0->dom0 calls"
                 )
+            if prefix_data and kwargs.get("stdin", None) != subprocess.PIPE:
+                raise NotImplementedError(
+                    "prefix_data for dom0->dom0 calls requires stdin=PIPE"
+                )
             if user is None:
                 user = grp.getgrnam("qubes").gr_mem[0]
 
@@ -992,6 +1000,8 @@ def run_service(
                 **kwargs,
                 env=env,
             )
+            if prefix_data:
+                p.stdin.write(prefix_data)
             return p
         qrexec_opts = ["-d", dest]
         if filter_esc:
@@ -1008,6 +1018,8 @@ def run_service(
             user = "DEFAULT"
         if not wait:
             qrexec_opts.extend(["-e"])
+        if prefix_data:
+            qrexec_opts.extend(["-p", prefix_data])
         if "connect_timeout" in kwargs:
             qrexec_opts.extend(["-w", str(kwargs.pop("connect_timeout"))])
         kwargs.setdefault("stdin", subprocess.PIPE)
@@ -1087,6 +1099,7 @@ def run_service(
         localcmd: str | None=None,
         wait: bool=True,
         autostart: bool=True,
+        prefix_data: bytes | None=None,
         **kwargs,
     ) -> Popen:
         """Run qrexec service in a given destination
@@ -1109,6 +1122,8 @@ def run_service(
             raise ValueError("non-default user not possible for calls from VM")
         if not wait and localcmd:
             raise ValueError("wait=False incompatible with localcmd")
+        if prefix_data and b"\0" in prefix_data:
+            raise ValueError("prefix_data cannot contain NUL character")
         qrexec_opts = []
         if filter_esc:
             qrexec_opts.extend(["-t"])
@@ -1122,6 +1137,8 @@ def run_service(
             raise qubesadmin.exc.QubesVMNotRunningError(
                 "%s is not running", dest
             )
+        if prefix_data:
+            qrexec_opts.extend(["-p", prefix_data])
         if not wait:
             # qrexec-client-vm can only request service calls, which are
             # started using MSG_EXEC_CMDLINE qrexec protocol message; this

qubesadmin/base.py

@@ -101,6 +101,8 @@ def qubesd_call(self, dest: str | None, method: str,
                         "%s has empty 'default_dispvm' property, but it is "
                         "required when target is @dispvm", self.app.local_name
                     )
+                dest = dest.name
+
         # have the actual implementation at Qubes() instance
         return self.app.qubesd_call(dest, method, arg, payload,
             payload_stream)

qubesadmin/tests/__init__.py

@@ -62,23 +62,34 @@ def __iter__(self):
 
 
 class TestProcess:
-    def __init__(self, input_callback=None, stdout=None, stderr=None,
-            stdout_data=None):
+    def __init__(
+        self,
+        input_callback=None,
+        stdin=None,
+        stdout=None,
+        stderr=None,
+        stdout_data=None,
+    ):
         self.input_callback = input_callback
         self.got_any_input = False
-        self.stdin = io.BytesIO()
-        # don't let anyone close it, before we get the value
-        self.stdin_close = self.stdin.close
-        self.stdin.close = self.store_input
-        self.stdin.flush = self.store_input
-        if stdout == subprocess.PIPE or stdout == subprocess.DEVNULL \
-                or stdout is None:
+        if stdin == subprocess.PIPE:
+            self.stdin = io.BytesIO()
+            # don't let anyone close it, before we get the value
+            self.stdin_close = self.stdin.close
+            self.stdin.close = self.store_input
+            self.stdin.flush = self.store_input
+        else:
+            self.stdin = None
+        if stdout == subprocess.PIPE:
             self.stdout = io.BytesIO()
+        elif stdout == subprocess.DEVNULL:
+            self.stdout = None
         else:
             self.stdout = stdout
-        if stderr == subprocess.PIPE or stderr == subprocess.DEVNULL \
-                or stderr is None:
+        if stderr == subprocess.PIPE:
             self.stderr = io.BytesIO()
+        elif stderr == subprocess.DEVNULL:
+            self.stderr = None
         else:
             self.stderr = stderr
         if stdout_data:
@@ -102,12 +113,22 @@ def communicate(self, input=None, timeout=None):
         # pylint: disable=redefined-builtin,unused-argument
         if input is not None:
             self.stdin.write(input)
-        self.stdin.close()
-        self.stdin_close()
-        return self.stdout.read(), self.stderr.read()
+        if self.stdin:
+            self.stdin.close()
+            self.stdin_close()
+        if self.stdout:
+            stdout = self.stdout.read()
+        else:
+            stdout = None
+        if self.stderr:
+            stderr = self.stderr.read()
+        else:
+            stderr = None
+        return stdout, stderr
 
     def wait(self):
-        self.stdin_close()
+        if self.stdin:
+            self.stdin_close()
         return 0
 
     def poll(self):
@@ -188,6 +209,9 @@ def run_service(self, dest, service, **kwargs):
         # pylint: disable=arguments-differ
         assert all(c in QREXEC_ALLOWED_CHARS for c in service), \
             f"forbidden char in service '{service}"
+        kwargs.setdefault("stdin", subprocess.PIPE)
+        kwargs.setdefault("stdout", subprocess.PIPE)
+        kwargs.setdefault("stderr", subprocess.PIPE)
         self.service_calls.append((dest, service, kwargs))
         call_key = (dest, service)
         # TODO: consider it as a future extension, as a replacement for
@@ -200,8 +224,9 @@ def run_service(self, dest, service, **kwargs):
             kwargs['stdout_data'] = self.expected_service_calls[call_key]
         return TestProcess(lambda input: self.service_calls.append((dest,
             service, input)),
-            stdout=kwargs.get('stdout', None),
-            stderr=kwargs.get('stderr', None),
+            stdin=kwargs.get('stdin'),
+            stdout=kwargs.get('stdout'),
+            stderr=kwargs.get('stderr'),
             stdout_data=kwargs.get('stdout_data', None),
         )
 

qubesadmin/tests/app.py

@@ -188,7 +188,6 @@ def test_012_getitem_cached_object(self):
         self.assertAllCalled()
 
 
-
 class TC_10_QubesBase(qubesadmin.tests.QubesTestCase):
     def setUp(self):
         super().setUp()
@@ -1094,6 +1093,32 @@ def test_013_run_service_default_target(self):
         with self.assertRaises(ValueError):
             self.app.run_service('', 'service.name')
 
+    @mock.patch('os.isatty', lambda fd: fd == 2)
+    def test_014_run_service_prefix_data(self):
+        self.listen_and_send(b'0\0')
+        with mock.patch("subprocess.Popen") as mock_proc:
+            self.app.run_service(
+                "some-vm", "service.name", prefix_data=b"prefix data"
+            )
+            mock_proc.assert_called_once_with(
+                [
+                    qubesadmin.config.QREXEC_CLIENT,
+                    "-d",
+                    "some-vm",
+                    "-T",
+                    "-p",
+                    b"prefix data",
+                    "DEFAULT:QUBESRPC service.name+ dom0",
+                ],
+                stdin=subprocess.PIPE,
+                stdout=subprocess.PIPE,
+                stderr=subprocess.PIPE,
+            )
+
+        self.assertEqual(
+            self.get_request(), b"admin.vm.Start+ dom0 name some-vm\0"
+        )
+
 
 class TC_30_QubesRemote(unittest.TestCase):
     def setUp(self):
@@ -1276,3 +1301,15 @@ def test_015_run_service_no_autostart2(self):
             'some-vm', 'admin.vm.CurrentState'],
             stdin=subprocess.PIPE, stdout=subprocess.PIPE,
             stderr=subprocess.PIPE)
+
+    @mock.patch('os.isatty', lambda fd: fd == 2)
+    def test_016_run_service_prefix_data(self):
+        self.app.run_service(
+            "some-vm", "service.name", prefix_data=b"prefix data"
+        )
+        self.proc_mock.assert_called_once_with(
+            [qubesadmin.config.QREXEC_CLIENT_VM,
+              "-T", "-p", b"prefix data", "some-vm", "service.name"],
+            stdin=subprocess.PIPE, stdout=subprocess.PIPE,
+            stderr=subprocess.PIPE,
+        )

qubesadmin/tests/backup/dispvm.py

@@ -27,6 +27,7 @@
 from unittest.mock import call
 
 import subprocess
+from subprocess import PIPE
 
 import qubesadmin.tests
 from qubesadmin.tools import qvm_backup_restore
@@ -195,7 +196,7 @@ def test_040_register_backup_source(self):
         self.assertEqual(obj.storage_access_id, 'someid')
         self.assertEqual(self.app.service_calls, [
             ('backup-storage', 'qubes.RegisterBackupLocation',
-             {'stdin':subprocess.PIPE, 'stdout':subprocess.PIPE}),
+             {'stdin': PIPE, 'stdout': PIPE, 'stderr': PIPE}),
             ('backup-storage', 'qubes.RegisterBackupLocation',
              b'/backup/path\n'),
         ])