Implement connections to TCP-based services

Both IPv4 and IPv6 are supported.  The port or both host and port can be
taken from the service argument instead of the symbolic link name.  Of
course, there are full unit tests.

Fixes: QubesOS/qubes-issues#9037

Author: DemiMarie

.gitlab-ci.yml

@@ -31,7 +31,7 @@ checks:tests:
   variables:
     PYTEST_ADDOPTS: "--color=yes"
   script:
-    - xvfb-run ./run-tests
+    - DOCKER_BUG_BREAKS_IPV6=1 xvfb-run ./run-tests
   after_script:
     - (cd libqrexec; gcov *.c)
     - (cd daemon; gcov *.c)

libqrexec/exec.c

@@ -27,11 +27,12 @@
 #include <stddef.h>
 #include <limits.h>
 
-#include <sys/socket.h>
 #include <sys/types.h>
+#include <sys/socket.h>
 #include <sys/stat.h>
 #include <sys/un.h>
 #include <sys/wait.h>
+#include <netdb.h>
 #include <unistd.h>
 #include <fcntl.h>
 #include "qrexec.h"
@@ -411,7 +412,7 @@ struct qrexec_parsed_command *parse_qubes_rpc_command(
         /* Parse service name ("qubes.Service") */
 
         const char *const plus = memchr(start, '+', descriptor_len);
-        size_t const name_len = plus != NULL ? (size_t)(plus - start) : descriptor_len;
+        size_t name_len = plus != NULL ? (size_t)(plus - start) : descriptor_len;
         if (name_len > NAME_MAX) {
             LOG(ERROR, "Service name too long to execute (length %zu)", name_len);
             goto err;
@@ -423,6 +424,7 @@ struct qrexec_parsed_command *parse_qubes_rpc_command(
         cmd->service_name = memdupnul(start, name_len);
         if (!cmd->service_name)
             goto err;
+        cmd->arg = plus != NULL ? plus + 1 : NULL;
 
         /* If there is no service argument, add a trailing "+" to the descriptor */
         cmd->service_descriptor = memdupnul(start, descriptor_len + (plus == NULL));
@@ -480,7 +482,7 @@ int execute_qubes_rpc_command(const char *cmdline, int *pid, int *stdin_fd,
 }
 
 int execute_parsed_qubes_rpc_command(
-        const struct qrexec_parsed_command *cmd, int *pid, int *stdin_fd,
+        struct qrexec_parsed_command *cmd, int *pid, int *stdin_fd,
         int *stdout_fd, int *stderr_fd, struct buffer *stdin_buffer) {
     if (cmd->service_descriptor) {
         // Proper Qubes RPC call
@@ -501,9 +503,81 @@ int execute_parsed_qubes_rpc_command(
                            pid, stdin_fd, stdout_fd, stderr_fd);
     }
 }
+static bool validate_port(const char *port) {
+#define MAXPORT "65535"
+#define MAXPORTLEN (sizeof MAXPORT - 1)
+    if (*port < '1' || *port > '9')
+        return false;
+    const char *p = port + 1;
+    for (; *p != 0; ++p) {
+        if (*p < '0' || *p > '9')
+            return false;
+    }
+    if (p - port > (ptrdiff_t)MAXPORTLEN)
+        return false;
+    if (p - port < (ptrdiff_t)MAXPORTLEN)
+        return true;
+    return memcmp(port, MAXPORT, MAXPORTLEN) <= 0;
+#undef MAXPORT
+#undef MAXPORTLEN
+}
+
+static int qubes_tcp_connect(const char *host, const char *port)
+{
+    // Work around a glibc bug: overly-large port numbers not rejected
+    if (!validate_port(port)) {
+        LOG(ERROR, "Invalid port number %s", port);
+        return -1;
+    }
+    /* If there is ':' or '%' in the host, then this must be an IPv6 address, not IPv4. */
+    bool const must_be_ipv6_addr = strchr(host, ':') != NULL || strchr(host, '%') != NULL;
+    LOG(DEBUG, "Connecting to %s%s%s:%s",
+               must_be_ipv6_addr ? "[" : "",
+               host,
+               must_be_ipv6_addr ? "]" : "",
+               port);
+    struct addrinfo hints = {
+        .ai_flags = AI_NUMERICSERV | AI_NUMERICHOST,
+        .ai_family = must_be_ipv6_addr ? AF_INET6 : AF_UNSPEC,
+        .ai_socktype = SOCK_STREAM,
+        .ai_protocol = IPPROTO_TCP,
+    }, *addrs;
+    int rc = getaddrinfo(host, port, &hints, &addrs);
+    if (rc != 0) {
+        /* data comes from symlink or from qrexec service argument, which has already
+         * been sanitized */
+        LOG(ERROR, "getaddrinfo(%s, %s) failed: %s", host, port, gai_strerror(rc));
+        return -1;
+    }
+    rc = -1;
+    assert(addrs != NULL && "getaddrinfo() returned zero addresses");
+    assert(addrs->ai_next == NULL &&
+           "getaddrinfo() returned multiple addresses despite AI_NUMERICHOST | AI_NUMERICSERV");
+    int sockfd = socket(addrs->ai_family,
+                        addrs->ai_socktype | SOCK_CLOEXEC,
+                        addrs->ai_protocol);
+    if (sockfd < 0)
+        goto freeaddrs;
+    {
+        int one = 1;
+        if (setsockopt(sockfd, SOL_SOCKET, SO_REUSEADDR, &one, sizeof one) != 0)
+            abort();
+    }
+    int res = connect(sockfd, addrs->ai_addr, addrs->ai_addrlen);
+    if (res != 0) {
+        PERROR("connect");
+        close(sockfd);
+    } else {
+        rc = sockfd;
+        LOG(DEBUG, "Connection succeeded");
+    }
+freeaddrs:
+    freeaddrinfo(addrs);
+    return rc;
+}
 
 bool find_qrexec_service(
-        const struct qrexec_parsed_command *cmd,
+        struct qrexec_parsed_command *cmd,
         int *socket_fd, struct buffer *stdin_buffer) {
     assert(cmd->service_descriptor);
 
@@ -529,10 +603,11 @@ bool find_qrexec_service(
         return false;
     }
 
+    const char *desc = cmd->command + RPC_REQUEST_COMMAND_LEN + 1;
     if (S_ISSOCK(statbuf.st_mode)) {
         /* Socket-based service. */
         int s;
-        if ((s = socket(AF_UNIX, SOCK_STREAM, 0)) == -1) {
+        if ((s = socket(AF_UNIX, SOCK_STREAM | SOCK_CLOEXEC | SOCK_NONBLOCK, 0)) == -1) {
             PERROR("socket");
             return false;
         }
@@ -544,12 +619,59 @@ bool find_qrexec_service(
 
         if (cmd->send_service_descriptor) {
             /* send part after "QUBESRPC ", including trailing NUL */
-            const char *desc = cmd->command + RPC_REQUEST_COMMAND_LEN + 1;
             buffer_append(stdin_buffer, desc, strlen(desc) + 1);
         }
 
         *socket_fd = s;
         return true;
+    } else if (S_ISLNK(statbuf.st_mode)) {
+        /* TCP-based service */
+        assert(path_buffer.buflen >= (int)sizeof "/dev/tcp");
+        assert(memcmp(path_buffer.data, "/dev/tcp/", sizeof "/dev/tcp") == 0);
+        char *address = path_buffer.data + sizeof "/dev/tcp";
+        char *slash = strchr(address, '/');
+        char *host, *port;
+        if (slash == NULL) {
+            if (cmd->arg == NULL || *cmd->arg == ' ') {
+                LOG(ERROR, "No or empty argument provided, cannot connect to %s",
+                    path_buffer.data);
+                return -1;
+            }
+            char *ptr = cmd->service_descriptor + (cmd->arg - desc);
+            if (*address == '\0') {
+                /* Get both host and port from service arguments */
+                host = ptr;
+                port = strrchr(ptr, '+');
+                if (port == NULL) {
+                    LOG(ERROR, "No host provided, cannot connect at %s", path_buffer.data);
+                    return -1;
+                }
+                *port = '\0';
+                for (char *p = host; p < port; ++p) {
+                    if (*p == '_') {
+                        LOG(ERROR, "Underscore not allowed in hostname %s", host);
+                        return -1;
+                    }
+                    if (*p == '+')
+                        *p = ':';
+                }
+                port++;
+            } else {
+                /* Get just port from service arguments */
+                host = address;
+                port = ptr;
+            }
+        } else {
+            *slash = '\0';
+            host = address;
+            port = slash + 1;
+        }
+        int res = qubes_tcp_connect(host, port);
+        if (res == -1)
+            return false;
+        *socket_fd = res;
+        cmd->send_service_descriptor = false;
+        return true;
     }
 
     if (euidaccess(path_buffer.data, X_OK) == 0) {

libqrexec/libqrexec-utils.h

@@ -49,7 +49,9 @@ struct buffer {
 #define WRITE_STDIN_BUFFERED  1 /* something still in the buffer */
 #define WRITE_STDIN_ERROR     2 /* write error, errno set */
 
-/* Parsed Qubes RPC or legacy command. */
+/* Parsed Qubes RPC or legacy command.
+ * The size of this structure is not part of the public API or ABI.
+ * Only use instances allocated by libqrexec-utils. */
 struct qrexec_parsed_command {
     const char *cmdline;
 
@@ -83,6 +85,13 @@ struct qrexec_parsed_command {
 
     /* For socket-based services: Should the service descriptor be sent? */
     bool send_service_descriptor;
+
+    /* Remaining fields are private to libqrexec-utils.  Do not access them
+     * directly - they may change in any update. */
+
+    /* Pointer to the argument, or NULL if there is no argument.
+     * Same buffer as "command" and "cmdline". */
+    const char *arg;
 };
 
 /* Parse a command, return NULL on failure. Uses cmd->cmdline
@@ -142,7 +151,7 @@ int write_stdin(int fd, const char *data, int len, struct buffer *buffer);
  * nonzero on failure.
  */
 int execute_parsed_qubes_rpc_command(
-        const struct qrexec_parsed_command *cmd, int *pid, int *stdin_fd,
+        struct qrexec_parsed_command *cmd, int *pid, int *stdin_fd,
         int *stdout_fd, int *stderr_fd, struct buffer *stdin_buffer);
 
 /**

libqrexec/process_io.c

@@ -76,7 +76,7 @@ static void close_stdout(int fd, bool restore_block) {
     } else if (shutdown(fd, SHUT_RD) == -1) {
         if (errno == ENOTSOCK)
             close(fd);
-        else
+        else if (errno != ENOTCONN) /* can happen with TCP, harmless */
             PERROR("shutdown close_stdout");
     }
 }

qrexec/tests/socket/agent.py

@@ -22,6 +22,7 @@
 import os.path
 import os
 import tempfile
+import socket
 import shutil
 import struct
 import getpass
@@ -686,6 +687,95 @@ def test_connect_socket_no_metadata(self):
         )
         self.check_dom0(dom0)
 
+    def test_connect_socket_tcp(self):
+        socket_path = os.path.join(
+            self.tempdir, "rpc", "qubes.SocketService+arg"
+        )
+        port = 65534
+        host = "127.0.0.1"
+        os.symlink(f"/dev/tcp/{host}/{port}", socket_path)
+        self._test_tcp(socket.AF_INET, "qubes.SocketService+arg", host, port)
+
+    def _test_tcp_raw(self, family: int, service: str, host: str, port: int, accept=True):
+        server = socket.socket(family, socket.SOCK_STREAM, socket.IPPROTO_TCP)
+        server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+        server.bind((host, port))
+        server.listen(1)
+        server = qrexec.QrexecServer(server)
+        self.addCleanup(server.close)
+
+        target, dom0 = self.execute_qubesrpc(service, "domX")
+        if accept:
+            server.accept()
+        message = b"stdin data"
+        target.send_message(qrexec.MSG_DATA_STDIN, message)
+        target.send_message(qrexec.MSG_DATA_STDIN, b"")
+        if accept:
+            self.assertEqual(server.recvall(len(message)), message)
+            server.sendall(b"stdout data")
+        server.close()
+        messages = target.recv_all_messages()
+        self.check_dom0(dom0)
+        return util.sort_messages(messages)
+
+    def _test_tcp(self, family: int, service: str, host: str, port: int) -> None:
+        # No stderr
+        self.assertListEqual(
+            self._test_tcp_raw(family, service, host, port),
+            [
+                (qrexec.MSG_DATA_STDOUT, b"stdout data"),
+                (qrexec.MSG_DATA_STDOUT, b""),
+                (qrexec.MSG_DATA_EXIT_CODE, b"\0\0\0\0"),
+            ],
+        )
+
+    def test_connect_socket_tcp_port_from_arg(self):
+        socket_path = os.path.join(
+            self.tempdir, "rpc", "qubes.SocketService"
+        )
+        port = 65533
+        host = "127.0.0.1"
+        os.symlink(f"/dev/tcp/{host}", socket_path)
+        self._test_tcp(socket.AF_INET, f"qubes.SocketService+{port}", host, port)
+
+    def test_connect_socket_tcp_host_and_port_from_arg(self):
+        socket_path = os.path.join(
+            self.tempdir, "rpc", "qubes.SocketService"
+        )
+        port = 65535
+        host = "127.0.0.1"
+        os.symlink(f"/dev/tcp/", socket_path)
+        self._test_tcp(socket.AF_INET, f"qubes.SocketService+{host}+{port}", host, port)
+
+    @unittest.skipIf(os.environ.get("DOCKER_BUG_BREAKS_IPV6"),
+                     "Disabling IPv6 test due to Docker bug")
+    def test_connect_socket_tcp_ipv6(self):
+        socket_path = os.path.join(
+            self.tempdir, "rpc", "qubes.SocketService"
+        )
+        port = 65532
+        host = "::1"
+        os.symlink(f"/dev/tcp/", socket_path)
+        self._test_tcp(socket.AF_INET6, f"qubes.SocketService+{host.replace(':', '+')}+{port}", host, port)
+
+    def test_connect_socket_tcp_unexpected_host(self):
+        socket_path = os.path.join(
+            self.tempdir, "rpc", "qubes.SocketService"
+        )
+        port = 65535
+        host = "127.0.0.1"
+        os.symlink(f"/dev/tcp/{host}", socket_path)
+        messages = self._test_tcp_raw(socket.AF_INET, f"qubes.SocketService+{host}+{port}",
+                                      host, port, accept=False)
+        self.assertListEqual(
+            messages,
+            [
+                (qrexec.MSG_DATA_STDOUT, b""),
+                (qrexec.MSG_DATA_STDERR, b""),
+                (qrexec.MSG_DATA_EXIT_CODE, b"\177\0\0\0"),
+            ],
+        )
+
     def test_connect_socket(self):
         socket_path = os.path.join(
             self.tempdir, "rpc", "qubes.SocketService+arg"

qrexec/tests/socket/qrexec.py

@@ -147,6 +147,7 @@ def socket_server(socket_path, socket_path_alt=None):
         except FileNotFoundError:
             pass
     server = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
+    server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
     server.bind(socket_path)
     if socket_path_alt is not None:
         os.symlink(socket_path, socket_path_alt)