Ensure that EOF is propagated to stdout

It is critical that EOF from the service gets propagated to
the stdout of qrexec-client-vm or qrexec-client.  Otherwise, the caller
will not recognize that EOF has happened and may wait forever for data
that will never arrive.  In QubesOS/qubes-issues#9169, this caused curl
to hang when reading a plaintext HTTP/1.0 response from tinyproxy, which
relies on EOF to delimit the response.

The bug will trigger iff all of the following conditions are met:

1. The remote side must close the writing side of the connection, either
   by closing or shutting down a socket.
2. If the remote service is executable, it must _not_ exit.
3. qrexec-client-vm or qrexec-client must _not_ get EOF on stdin.
4. Either the remote side does not close its stdin, or the local side
   does not continue to write data.
5. The same file description is used for both stdin and stdout of the
   local qrexec-client or qrexec-client-vm process.
6. qrexec-client or qrexec-client-vm connect the stdin and stdout of the
   remote process to their own stdin and stdout, not the stdin and
   stdout of a locally spawned command.

To fix this bug, add flags to qrexec-client and qrexec-client-vm that
cause the socket passed as file descriptor 0 to be used for both stdin
and stdout.

Fixes: QubesOS/qubes-issues#9169

agent/qrexec-client-vm.c

@@ -18,6 +18,7 @@
  * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA.
  *
  */
+#include <assert.h>
 #include <sys/socket.h>
 #include <sys/wait.h>
 #include <sys/un.h>
@@ -94,6 +95,7 @@ static void convert_target_name_keyword(char *target)
 enum {
     opt_no_filter_stdout = 't'+128,
     opt_no_filter_stderr = 'T'+128,
+    opt_use_stdin_socket = 'u'+128,
 };
 
 static struct option longopts[] = {
@@ -104,6 +106,7 @@ static struct option longopts[] = {
     { "no-filter-escape-chars-stderr", no_argument, 0, opt_no_filter_stderr},
     { "agent-socket", required_argument, 0, 'a'},
     { "prefix-data", required_argument, 0, 'p' },
+    { "use-stdin-socket", no_argument, 0, opt_use_stdin_socket },
     { "help", no_argument, 0, 'h' },
     { NULL, 0, 0, 0},
 };
@@ -141,6 +144,7 @@ int main(int argc, char **argv)
     int inpipe[2], outpipe[2];
     int buffer_size = 0;
     int opt;
+    int stdout_fd = 1;
     const char *agent_trigger_path = QREXEC_AGENT_TRIGGER_PATH, *prefix_data = NULL;
 
     setup_logging("qrexec-client-vm");
@@ -198,6 +202,23 @@ int main(int argc, char **argv)
                     exit(1);
                 }
                 break;
+            case opt_use_stdin_socket:
+                {
+                    int type;
+                    if (stdout_fd == 0)
+                        errx(2, "--use-stdin-socket passed twice");
+                    socklen_t len = sizeof(type);
+                    if (getsockopt(0, SOL_SOCKET, SO_TYPE, &type, &len)) {
+                        if (errno == ENOTSOCK)
+                            errx(2, "--use-stdin-socket passed, but stdin not a socket");
+                        err(2, "getsockopt(0, SOL_SOCKET, SO_TYPE)");
+                    }
+                    assert(len == sizeof(type));
+                    if (type != SOCK_STREAM)
+                        errx(2, "stdin was a socket of type %d, not SOCK_STREAM (%d)", type, SOCK_STREAM);
+                    stdout_fd = 0;
+                }
+                break;
             case '?':
                 usage(argv[0], 2);
         }
@@ -209,6 +230,10 @@ int main(int argc, char **argv)
     if (argc - optind > 2) {
         start_local_process = 1;
     }
+    if (start_local_process && stdout_fd != 1) {
+        fprintf(stderr, "cannot spawn a local process with --use-stdin-socket\n");
+        usage(argv[0], 2);
+    }
 
     if (!start_local_process) {
         if (replace_chars_stdout == -1 && isatty(1))
@@ -303,7 +328,7 @@ int main(int argc, char **argv)
     } else {
         ret = handle_data_client(MSG_SERVICE_CONNECT,
                 exec_params.connect_domain, exec_params.connect_port,
-                1, 0, -1, buffer_size, 0, prefix_data);
+                stdout_fd, 0, -1, buffer_size, 0, prefix_data);
     }
 
     close(trigger_fd);

daemon/qrexec-client.c

@@ -72,11 +72,16 @@ static _Noreturn void do_exec(const char *prog, const char *username __attribute
     PERROR("exec bash");
     exit(1);
 }
+enum {
+    opt_socket_dir = 'd'+128,
+    opt_use_stdin_socket = 'u'+128,
+};
 
 static struct option longopts[] = {
     { "help", no_argument, 0, 'h' },
-    { "socket-dir", required_argument, 0, 'd'+128 },
+    { "socket-dir", required_argument, 0, opt_socket_dir },
     { "no-exit-code", no_argument, 0, 'E' },
+    { "use-stdin-socket", no_argument, 0, opt_use_stdin_socket },
     { NULL, 0, 0, 0 },
 };
 
@@ -96,7 +101,8 @@ _Noreturn static void usage(const char *const name)
             "  -c - connect to existing process (response to trigger service call)\n"
             "  -w timeout - override default connection timeout of 5s (set 0 for no timeout)\n"
             "  -k - kill the domain right before exiting\n"
-            "  --socket-dir=PATH -  directory for qrexec socket, default: %s\n",
+            "  --socket-dir=PATH -  directory for qrexec socket, default: %s\n"
+            "  --use-stdin-socket - use fd 0 (which must be socket) for both stdin and stdout\n",
             name ? name : "qrexec-client", QREXEC_DAEMON_SOCKET_DIR);
     exit(1);
 }
@@ -217,9 +223,26 @@ int main(int argc, char **argv)
             case 'W':
                 wait_connection_end = 1;
                 break;
-            case 'd' + 128:
+            case opt_socket_dir:
                 socket_dir = strdup(optarg);
                 break;
+            case opt_use_stdin_socket:
+                {
+                    int type;
+                    if (local_stdin_fd != 1)
+                        errx(2, "--use-stdin-socket passed twice");
+                    socklen_t len = sizeof(type);
+                    if (getsockopt(0, SOL_SOCKET, SO_TYPE, &type, &len)) {
+                        if (errno == ENOTSOCK)
+                            errx(2, "--use-stdin-socket passed, but stdin not a socket");
+                        err(2, "getsockopt(0, SOL_SOCKET, SO_TYPE)");
+                    }
+                    assert(len == sizeof(type) && "wrong socket option length?");
+                    if (type != SOCK_STREAM)
+                        errx(2, "stdin was a socket of type %d, not SOCK_STREAM (%d)", type, SOCK_STREAM);
+                    local_stdin_fd = 0;
+                }
+                break;
             case 'k':
                 kill = true;
                 break;
@@ -241,6 +264,11 @@ int main(int argc, char **argv)
         usage(argv[0]);
     }
 
+    if ((local_cmdline != NULL) && (local_stdin_fd != 1)) {
+        fprintf(stderr, "ERROR: at most one of -l and --use-stdin-socket can be specified\n");
+        usage(argv[0]);
+    }
+
     if (strcmp(domname, "dom0") == 0 || strcmp(domname, "@adminvm") == 0) {
         if (request_id == NULL) {
             fprintf(stderr, "ERROR: when target domain is 'dom0', -c must be specified\n");

daemon/qrexec-daemon-common.c

@@ -155,7 +155,8 @@ bool send_service_connect(int s, const char *conn_ident,
 
 #define QREXEC_DATA_MIN_VERSION QREXEC_PROTOCOL_V2
 
-static int local_stdin_fd = 1, local_stdout_fd = 0;
+static int local_stdout_fd = 0;
+int local_stdin_fd = 1;
 static pid_t local_pid = 0;
 
 static volatile sig_atomic_t sigchld = 0;

daemon/qrexec-daemon-common.h

@@ -38,3 +38,5 @@ __attribute__((warn_unused_result))
 int handle_agent_handshake(libvchan_t *vchan, bool remote_send_first);
 __attribute__((warn_unused_result))
 int prepare_local_fds(struct qrexec_parsed_command *command, struct buffer *stdin_buffer);
+/* FD for stdout of remote process */
+extern int local_stdin_fd;

qrexec/tests/socket/agent.py

@@ -1173,11 +1173,24 @@ class TestClientVm(unittest.TestCase):
     target_domain = 43
     target_port = 1024
 
+    def assertStdoutMessages(self, target, expected_stdout: bytes, ty=qrexec.MSG_DATA_STDOUT):
+        bytes_recvd, expected = 0, len(expected_stdout)
+        while bytes_recvd < expected:
+            msg_type, msg_body = target.recv_message()
+            self.assertEqual(msg_type, ty)
+            l = len(msg_body)
+            self.assertEqual(msg_body, expected_stdout[bytes_recvd:l])
+            bytes_recvd += l
+        self.assertEqual(bytes_recvd, expected)
+
     def setUp(self):
         self.tempdir = tempfile.mkdtemp()
         self.addCleanup(shutil.rmtree, self.tempdir)
 
-    def start_client(self, args):
+    def start_client(self, args, stdio=subprocess.PIPE):
+        if stdio is not subprocess.PIPE:
+            self.assertIsInstance(stdio, socket.socket)
+            args.insert(0, "--use-stdin-socket")
         env = os.environ.copy()
         env["LD_LIBRARY_PATH"] = os.path.join(ROOT_PATH, "libqrexec")
         env["VCHAN_DOMAIN"] = str(self.domain)
@@ -1193,8 +1206,8 @@ def start_client(self, args):
         self.client = subprocess.Popen(
             cmd,
             env=env,
-            stdin=subprocess.PIPE,
-            stdout=subprocess.PIPE,
+            stdin=stdio,
+            stdout=stdio,
             stderr=subprocess.PIPE,
             pass_fds=(stderr_dup,),
         )
@@ -1219,7 +1232,7 @@ def connect_target_client(self):
         self.addCleanup(target_client.close)
         return target_client
 
-    def run_service(self, *, local_program=None, options=None):
+    def run_service(self, *, local_program=None, options=None, stdio=subprocess.PIPE):
         server = self.connect_server()
 
         args = options or []
@@ -1228,7 +1241,7 @@ def run_service(self, *, local_program=None, options=None):
         if local_program:
             args += local_program
 
-        self.start_client(args)
+        self.start_client(args, stdio=stdio)
         server.accept()
 
         message_type, data = server.recv_message()
@@ -1257,6 +1270,30 @@ def test_run_client(self):
         self.client.wait()
         self.assertEqual(self.client.returncode, 42)
 
+    def test_run_client_eof(self):
+        remote, local = socket.socketpair()
+        target_client = self.run_service(stdio=remote)
+        remote.close()
+        initial_data = b"stdout data\n"
+        target_client.send_message(qrexec.MSG_DATA_STDOUT, initial_data)
+        # FIXME: data can be received in multiple messages
+        self.assertEqual(local.recv(len(initial_data)), initial_data)
+        initial_data = b"stdin data\n"
+        local.sendall(initial_data)
+        self.assertStdoutMessages(target_client, initial_data, qrexec.MSG_DATA_STDIN)
+        target_client.send_message(qrexec.MSG_DATA_STDOUT, b"")
+        # Check that EOF got propagated
+        self.assertEqual(local.recv(1), b"")
+        # Close local
+        local.close()
+        # Check that EOF got propagated on this side too
+        self.assertEqual(target_client.recv_message(), (qrexec.MSG_DATA_STDIN, b""))
+        target_client.send_message(
+            qrexec.MSG_DATA_EXIT_CODE, struct.pack("<L", 42)
+        )
+        self.client.wait()
+        self.assertEqual(self.client.returncode, 42)
+
     def test_run_client_replace_chars(self):
         target_client = self.run_service(options=["-t"])
         target_client.send_message(
@@ -1301,10 +1338,7 @@ def test_run_client_with_local_proc(self):
         target_client = self.run_service(local_program=["/bin/cat"])
         target_client.send_message(qrexec.MSG_DATA_STDOUT, b"stdout data\n")
         target_client.send_message(qrexec.MSG_DATA_STDOUT, b"")
-        self.assertEqual(
-            target_client.recv_message(),
-            (qrexec.MSG_DATA_STDIN, b"stdout data\n"),
-        )
+        self.assertStdoutMessages(target_client, b"stdout data\n", qrexec.MSG_DATA_STDIN)
         self.assertEqual(
             target_client.recv_message(), (qrexec.MSG_DATA_STDIN, b"")
         )
@@ -1331,16 +1365,12 @@ def test_stdio_socket(self):
 """,
             ]
         )
-        self.assertEqual(
-            target.recv_message(), (qrexec.MSG_DATA_STDIN, b"hello world\n")
-        )
+        self.assertStdoutMessages(target, b"hello world\n", qrexec.MSG_DATA_STDIN)
 
         target.send_message(qrexec.MSG_DATA_STDOUT, b"stdin\n")
         target.send_message(qrexec.MSG_DATA_STDOUT, b"")
 
-        self.assertEqual(
-            target.recv_message(), (qrexec.MSG_DATA_STDIN, b"received: stdin\n")
-        )
+        self.assertStdoutMessages(target, b"received: stdin\n", qrexec.MSG_DATA_STDIN)
         self.assertEqual(target.recv_message(), (qrexec.MSG_DATA_STDIN, b""))
 
         target.send_message(qrexec.MSG_DATA_EXIT_CODE, struct.pack("<L", 42))
@@ -1412,10 +1442,7 @@ def test_run_client_vchan_disconnect(self):
         target_client = self.run_service()
         self.client.stdin.write(b"stdin data\n")
         self.client.stdin.flush()
-        self.assertEqual(
-            target_client.recv_message(),
-            (qrexec.MSG_DATA_STDIN, b"stdin data\n"),
-        )
+        self.assertStdoutMessages(target_client, b"stdin data\n", qrexec.MSG_DATA_STDIN)
         target_client.send_message(qrexec.MSG_DATA_STDOUT, b"stdout data\n")
         target_client.send_message(qrexec.MSG_DATA_STDOUT, b"")
         self.assertEqual(self.client.stdout.read(), b"stdout data\n")
@@ -1427,10 +1454,7 @@ def test_run_client_vchan_disconnect(self):
     def test_run_client_with_local_proc_disconnect(self):
         target_client = self.run_service(local_program=["/bin/cat"])
         target_client.send_message(qrexec.MSG_DATA_STDOUT, b"stdout data\n")
-        self.assertEqual(
-            target_client.recv_message(),
-            (qrexec.MSG_DATA_STDIN, b"stdout data\n"),
-        )
+        self.assertStdoutMessages(target_client, b"stdout data\n", qrexec.MSG_DATA_STDIN)
         target_client.close()
         self.client.wait()
         self.assertEqual(self.client.returncode, 255)