From edc80dec7603b4f77a69aaa642c7159850cc827d Mon Sep 17 00:00:00 2001
From: Demi Marie Obenour <demi@invisiblethingslab.com>
Date: Sat, 13 Apr 2024 15:48:02 -0400
Subject: [PATCH 1/2] Check for dom0 messages in more agent tests

Use a utility function for doing so, instead of open-coding the checks.
Some code goes away.
---
 qrexec/tests/socket/agent.py | 71 +++++++++++++++---------------------
 1 file changed, 29 insertions(+), 42 deletions(-)

diff --git a/qrexec/tests/socket/agent.py b/qrexec/tests/socket/agent.py
index 8aaa5fbc..94dfb133 100644
--- a/qrexec/tests/socket/agent.py
+++ b/qrexec/tests/socket/agent.py
@@ -47,6 +47,15 @@ class TestAgentBase(unittest.TestCase):
     target_domain = 43
     target_port = 1024
 
+    def check_dom0(self, dom0):
+        self.assertEqual(
+            dom0.recv_message(),
+            (
+                qrexec.MSG_CONNECTION_TERMINATED,
+                struct.pack("<LL", self.target_domain, self.target_port),
+            ),
+        )
+
     def setUp(self):
         self.tempdir = tempfile.mkdtemp()
         os.mkdir(os.path.join(self.tempdir, "local-rpc"))
@@ -157,14 +166,7 @@ def test_just_exec(self):
             lambda: os.path.exists(os.path.join(self.tempdir, "new_file")),
             "file created",
         )
-
-        self.assertEqual(
-            dom0.recv_message(),
-            (
-                qrexec.MSG_CONNECTION_TERMINATED,
-                struct.pack("<LL", self.target_domain, self.target_port),
-            ),
-        )
+        self.check_dom0(dom0)
 
     def test_exec_cmdline(self):
         self.start_agent()
@@ -196,14 +198,7 @@ def test_exec_cmdline(self):
                 (qrexec.MSG_DATA_EXIT_CODE, b"\0\0\0\0"),
             ],
         )
-
-        self.assertEqual(
-            dom0.recv_message(),
-            (
-                qrexec.MSG_CONNECTION_TERMINATED,
-                struct.pack("<LL", self.target_domain, self.target_port),
-            ),
-        )
+        self.check_dom0(dom0)
 
     def test_trigger_service(self):
         self.start_agent()
@@ -229,13 +224,7 @@ def test_trigger_service(self):
         )
 
         client.close()
-        self.assertEqual(
-            dom0.recv_message(),
-            (
-                qrexec.MSG_CONNECTION_TERMINATED,
-                struct.pack("<LL", self.target_domain, self.target_port),
-            ),
-        )
+        self.check_dom0(dom0)
 
     def test_trigger_service_refused(self):
         self.start_agent()
@@ -310,15 +299,6 @@ def execute_qubesrpc(self, service: str, src_domain_name: str):
         target.handshake()
         return target, dom0
 
-    def check_dom0(self, dom0):
-        self.assertEqual(
-            dom0.recv_message(),
-            (
-                qrexec.MSG_CONNECTION_TERMINATED,
-                struct.pack("<LL", self.target_domain, self.target_port),
-            ),
-        )
-
     def make_executable_service(self, *args):
         util.make_executable_service(self.tempdir, *args)
 
@@ -332,7 +312,7 @@ def test_exec_service(self):
 echo "arg: $1, remote domain: $QREXEC_REMOTE_DOMAIN"
 """,
         )
-        target, _ = self.execute_qubesrpc("qubes.Service+arg", "domX")
+        target, dom0 = self.execute_qubesrpc("qubes.Service+arg", "domX")
         target.send_message(qrexec.MSG_DATA_STDIN, b"")
         messages = target.recv_all_messages()
         self.assertListEqual(
@@ -344,6 +324,7 @@ def test_exec_service(self):
                 (qrexec.MSG_DATA_EXIT_CODE, b"\0\0\0\0"),
             ],
         )
+        self.check_dom0(dom0)
 
     def test_exec_service_keyword(self):
         util.make_executable_service(
@@ -793,10 +774,10 @@ def execute(self, cmd: str):
 
         target = self.connect_target()
         target.handshake()
-        return target
+        return target, dom0
 
     def test_stdin_stderr(self):
-        target = self.execute('echo "stdout"; echo "stderr" >&2')
+        target, dom0 = self.execute('echo "stdout"; echo "stderr" >&2')
         target.send_message(qrexec.MSG_DATA_STDIN, b"")
 
         messages = target.recv_all_messages()
@@ -812,7 +793,7 @@ def test_stdin_stderr(self):
         )
 
     def test_pass_stdin(self):
-        target = self.execute("cat")
+        target, dom0 = self.execute("cat")
 
         target.send_message(qrexec.MSG_DATA_STDIN, b"data 1")
         self.assertEqual(
@@ -834,15 +815,16 @@ def test_pass_stdin(self):
                 (qrexec.MSG_DATA_EXIT_CODE, b"\0\0\0\0"),
             ],
         )
+        self.check_dom0(dom0)
 
     def test_close_stdin_early(self):
         # Make sure that we cover the error on writing stdin into living
         # process.
-        target = self.execute(
+        target, dom0 = self.execute(
             """
 read
 exec <&-
-echo closed stdin
+echo "closed stdin"
 sleep 1
 """
         )
@@ -862,6 +844,7 @@ def test_close_stdin_early(self):
                 (qrexec.MSG_DATA_EXIT_CODE, b"\0\0\0\0"),
             ],
         )
+        self.check_dom0(dom0)
 
     def test_buffer_stdin(self):
         # Test to trigger WRITE_STDIN_BUFFERED.
@@ -877,7 +860,7 @@ def test_buffer_stdin(self):
 
         fifo = os.path.join(self.tempdir, "fifo")
         os.mkfifo(fifo)
-        target = self.execute("read <{}; cat".format(fifo))
+        target, dom0 = self.execute("read <{}; cat".format(fifo))
 
         for i in range(0, data_size, msg_size):
             msg = data[i : i + msg_size]
@@ -911,9 +894,10 @@ def test_buffer_stdin(self):
                 (qrexec.MSG_DATA_EXIT_CODE, b"\0\0\0\0"),
             ],
         )
+        self.check_dom0(dom0)
 
     def test_close_stdout_stderr_early(self):
-        target = self.execute(
+        target, dom0 = self.execute(
             """\
 read
 echo closing stdout
@@ -946,9 +930,10 @@ def test_close_stdout_stderr_early(self):
             target.recv_message(),
             (qrexec.MSG_DATA_EXIT_CODE, struct.pack("<L", 42)),
         )
+        self.check_dom0(dom0)
 
     def test_stdio_socket(self):
-        target = self.execute(
+        target, dom0 = self.execute(
             """\
 kill -USR1 $QREXEC_AGENT_PID
 echo hello world >&0
@@ -973,11 +958,12 @@ def test_stdio_socket(self):
                 (qrexec.MSG_DATA_EXIT_CODE, b"\0\0\0\0"),
             ],
         )
+        self.check_dom0(dom0)
 
     def test_exit_before_closing_streams(self):
         fifo = os.path.join(self.tempdir, "fifo")
         os.mkfifo(fifo)
-        target = self.execute(
+        target, dom0 = self.execute(
             """\
 # duplicate original stdin to fd 3, because bash will
 # close original stdin in child process
@@ -1021,6 +1007,7 @@ def test_exit_before_closing_streams(self):
                 (qrexec.MSG_DATA_EXIT_CODE, struct.pack("<L", 42)),
             ],
         )
+        self.check_dom0(dom0)
 
 
 @unittest.skipIf(os.environ.get("SKIP_SOCKET_TESTS"), "socket tests not set up")

From dfd804f1227be23404a80de3ac8fdde9a363d6dc Mon Sep 17 00:00:00 2001
From: Demi Marie Obenour <demi@invisiblethingslab.com>
Date: Sat, 13 Apr 2024 15:24:16 -0400
Subject: [PATCH 2/2] Fix flaky qrexec agent tests

Various tests assumed that qrexec would deliver stdout in a single
message.  Qrexec does not make this guarantee: calls to write(2),
send(2), sendmsg(2), etc are not guaranteed to correspond 1-to-1
to MSG_DATA_STDOUT messages on the vchan.  This caused
https://gitlab.com/QubesOS/qubes-core-qrexec/-/jobs/6616564043 to
wrongly fail, even thouggh the code is correct.

Fix this problem by concatenating the payloads of all stdout messages
into a single bytes object before comparing with the expected stdout
value.  Also add a utility function for this, saving a lot of code in
tests.
---
 qrexec/tests/socket/agent.py | 158 +++++++----------------------------
 1 file changed, 29 insertions(+), 129 deletions(-)

diff --git a/qrexec/tests/socket/agent.py b/qrexec/tests/socket/agent.py
index 94dfb133..9c56dfe6 100644
--- a/qrexec/tests/socket/agent.py
+++ b/qrexec/tests/socket/agent.py
@@ -56,6 +56,21 @@ def check_dom0(self, dom0):
             ),
         )
 
+    def assertExpectedStdout(self, target, expected_stdout: bytes, *, exit_code=0):
+        messages = util.sort_messages(target.recv_all_messages())
+        self.assertListEqual(messages[-3:],
+            [
+                (qrexec.MSG_DATA_STDOUT, b""),
+                (qrexec.MSG_DATA_STDERR, b""),
+                (qrexec.MSG_DATA_EXIT_CODE, struct.pack("<L", exit_code))
+            ])
+        stdout_entries = []
+        for msg_type, msg_body in messages[:-3]:
+            # messages before last are not empty, hence truthy
+            self.assertTrue(msg_body)
+            self.assertEqual(msg_type, qrexec.MSG_DATA_STDOUT)
+            stdout_entries.append(msg_body)
+
     def setUp(self):
         self.tempdir = tempfile.mkdtemp()
         os.mkdir(os.path.join(self.tempdir, "local-rpc"))
@@ -188,16 +203,7 @@ def test_exec_cmdline(self):
 
         target.send_message(qrexec.MSG_DATA_STDIN, b"")
 
-        messages = target.recv_all_messages()
-        self.assertListEqual(
-            util.sort_messages(messages),
-            [
-                (qrexec.MSG_DATA_STDOUT, b"Hello world\n"),
-                (qrexec.MSG_DATA_STDOUT, b""),
-                (qrexec.MSG_DATA_STDERR, b""),
-                (qrexec.MSG_DATA_EXIT_CODE, b"\0\0\0\0"),
-            ],
-        )
+        self.assertExpectedStdout(target, b"Hello world\n")
         self.check_dom0(dom0)
 
     def test_trigger_service(self):
@@ -314,16 +320,7 @@ def test_exec_service(self):
         )
         target, dom0 = self.execute_qubesrpc("qubes.Service+arg", "domX")
         target.send_message(qrexec.MSG_DATA_STDIN, b"")
-        messages = target.recv_all_messages()
-        self.assertListEqual(
-            util.sort_messages(messages),
-            [
-                (qrexec.MSG_DATA_STDOUT, b"arg: arg, remote domain: domX\n"),
-                (qrexec.MSG_DATA_STDOUT, b""),
-                (qrexec.MSG_DATA_STDERR, b""),
-                (qrexec.MSG_DATA_EXIT_CODE, b"\0\0\0\0"),
-            ],
-        )
+        self.assertExpectedStdout(target, b"arg: arg, remote domain: domX\n")
         self.check_dom0(dom0)
 
     def test_exec_service_keyword(self):
@@ -342,20 +339,11 @@ def test_exec_service_keyword(self):
         )
         target, dom0 = self.execute_qubesrpc("qubes.Service", "domX")
         target.send_message(qrexec.MSG_DATA_STDIN, b"")
-        messages = target.recv_all_messages()
-        self.assertListEqual(
-            util.sort_messages(messages),
-            [
-                (qrexec.MSG_DATA_STDOUT, b"""arg: , remote domain: domX
+        self.assertExpectedStdout(target, b"""arg: , remote domain: domX
 target name: NONAME
 target keyword: NOKEYWORD
 target type: ''
-"""),
-                (qrexec.MSG_DATA_STDOUT, b""),
-                (qrexec.MSG_DATA_STDERR, b""),
-                (qrexec.MSG_DATA_EXIT_CODE, b"\0\0\0\0"),
-            ],
-        )
+""")
         self.check_dom0(dom0)
 
     def test_exec_service_with_config(self):
@@ -376,16 +364,7 @@ def test_exec_service_with_config(self):
 """)
         target, dom0 = self.execute_qubesrpc("qubes.Service+arg", "domX")
         target.send_message(qrexec.MSG_DATA_STDIN, b"")
-        messages = target.recv_all_messages()
-        self.assertListEqual(
-            util.sort_messages(messages),
-            [
-                (qrexec.MSG_DATA_STDOUT, b"arg: arg, remote domain: domX\n"),
-                (qrexec.MSG_DATA_STDOUT, b""),
-                (qrexec.MSG_DATA_STDERR, b""),
-                (qrexec.MSG_DATA_EXIT_CODE, b"\0\0\0\0"),
-            ],
-        )
+        self.assertExpectedStdout(target, b"arg: arg, remote domain: domX\n")
         self.check_dom0(dom0)
 
     def test_wait_for_session(self):
@@ -449,20 +428,9 @@ def _test_wait_for_session(self, config_name, service_name="qubes.Service", argu
         # Do not send EOF. Shell read doesn't need it, and this checks that
         # qrexec does not wait for EOF on stdin before sending the exit code
         # from the remote process.
-        messages = target.recv_all_messages()
-        self.assertListEqual(
-            util.sort_messages(messages),
-            [
-                (
-                    qrexec.MSG_DATA_STDOUT,
-                    b"arg: " + argument.encode("ascii", "strict")
-                    + b", remote domain: domX, input: stdin data\n",
-                ),
-                (qrexec.MSG_DATA_STDOUT, b""),
-                (qrexec.MSG_DATA_STDERR, b""),
-                (qrexec.MSG_DATA_EXIT_CODE, b"\0\0\0\0"),
-            ],
-        )
+        expected_stdout = (b"arg: " + argument.encode("ascii", "strict")
+                           + b", remote domain: domX, input: stdin data\n")
+        self.assertExpectedStdout(target, expected_stdout)
         self.check_dom0(dom0)
 
     def test_exec_service_fail(self):
@@ -599,16 +567,7 @@ def test_exec_null_argument_finds_service_for_empty_argument(self):
         )
         target, dom0 = self.execute_qubesrpc("qubes.Service", "domX")
         target.send_message(qrexec.MSG_DATA_STDIN, b"")
-        messages = target.recv_all_messages()
-        self.assertListEqual(
-            util.sort_messages(messages),
-            [
-                (qrexec.MSG_DATA_STDOUT, b"specific service: qubes.Service\n"),
-                (qrexec.MSG_DATA_STDOUT, b""),
-                (qrexec.MSG_DATA_STDERR, b""),
-                (qrexec.MSG_DATA_EXIT_CODE, b"\0\0\0\0"),
-            ],
-        )
+        self.assertExpectedStdout(target, b"specific service: qubes.Service\n")
         self.check_dom0(dom0)
 
     def test_socket_null_argument_finds_service_for_empty_argument(self):
@@ -806,16 +765,7 @@ def test_pass_stdin(self):
         )
 
         target.send_message(qrexec.MSG_DATA_STDIN, b"")
-        messages = target.recv_all_messages()
-        self.assertListEqual(
-            util.sort_messages(messages),
-            [
-                (qrexec.MSG_DATA_STDOUT, b""),
-                (qrexec.MSG_DATA_STDERR, b""),
-                (qrexec.MSG_DATA_EXIT_CODE, b"\0\0\0\0"),
-            ],
-        )
-        self.check_dom0(dom0)
+        self.assertExpectedStdout(target, b"")
 
     def test_close_stdin_early(self):
         # Make sure that we cover the error on writing stdin into living
@@ -835,15 +785,7 @@ def test_close_stdin_early(self):
         target.send_message(qrexec.MSG_DATA_STDIN, b"data 2\n")
         target.send_message(qrexec.MSG_DATA_STDIN, b"")
 
-        messages = target.recv_all_messages()
-        self.assertListEqual(
-            util.sort_messages(messages),
-            [
-                (qrexec.MSG_DATA_STDOUT, b""),
-                (qrexec.MSG_DATA_STDERR, b""),
-                (qrexec.MSG_DATA_EXIT_CODE, b"\0\0\0\0"),
-            ],
-        )
+        self.assertExpectedStdout(target, b"")
         self.check_dom0(dom0)
 
     def test_buffer_stdin(self):
@@ -871,29 +813,7 @@ def test_buffer_stdin(self):
         with open(fifo, "a") as f:
             f.write("end\n")
             f.flush()
-
-        messages = []
-        received_data = b""
-        while len(received_data) < data_size:
-            message_type, message = target.recv_message()
-            if message_type != qrexec.MSG_DATA_STDOUT:
-                messages.append((message_type, message))
-            else:
-                self.assertEqual(message_type, qrexec.MSG_DATA_STDOUT)
-                received_data += message
-
-        self.assertEqual(len(received_data), data_size)
-        self.assertEqual(received_data, data)
-
-        messages += target.recv_all_messages()
-        self.assertListEqual(
-            util.sort_messages(messages),
-            [
-                (qrexec.MSG_DATA_STDOUT, b""),
-                (qrexec.MSG_DATA_STDERR, b""),
-                (qrexec.MSG_DATA_EXIT_CODE, b"\0\0\0\0"),
-            ],
-        )
+        self.assertExpectedStdout(target, data)
         self.check_dom0(dom0)
 
     def test_close_stdout_stderr_early(self):
@@ -948,16 +868,7 @@ def test_stdio_socket(self):
         target.send_message(qrexec.MSG_DATA_STDIN, b"stdin\n")
         target.send_message(qrexec.MSG_DATA_STDIN, b"")
 
-        messages = target.recv_all_messages()
-        self.assertListEqual(
-            util.sort_messages(messages),
-            [
-                (qrexec.MSG_DATA_STDOUT, b"received: stdin\n"),
-                (qrexec.MSG_DATA_STDOUT, b""),
-                (qrexec.MSG_DATA_STDERR, b""),
-                (qrexec.MSG_DATA_EXIT_CODE, b"\0\0\0\0"),
-            ],
-        )
+        self.assertExpectedStdout(target, b"received: stdin\n")
         self.check_dom0(dom0)
 
     def test_exit_before_closing_streams(self):
@@ -995,18 +906,7 @@ def test_exit_before_closing_streams(self):
         with open(fifo, "a") as f:
             f.write("end\n")
             f.flush()
-        self.assertEqual(
-            target.recv_message(), (qrexec.MSG_DATA_STDOUT, b"child exiting\n")
-        )
-        messages = target.recv_all_messages()
-        self.assertListEqual(
-            util.sort_messages(messages),
-            [
-                (qrexec.MSG_DATA_STDOUT, b""),
-                (qrexec.MSG_DATA_STDERR, b""),
-                (qrexec.MSG_DATA_EXIT_CODE, struct.pack("<L", 42)),
-            ],
-        )
+        self.assertExpectedStdout(target, b"child exiting\n", exit_code=42)
         self.check_dom0(dom0)