qrexec: new protocol - direct data vchan connections

qrexec/Makefile

@@ -3,8 +3,8 @@ CFLAGS+=-I. -g -Wall -Wextra -Werror -pie -fPIC `pkg-config --cflags vchan-$(BAC
 LIBS=`pkg-config --libs vchan-$(BACKEND_VMM)` -lqrexec-utils
 
 all: qrexec-agent qrexec-client-vm
-qrexec-agent: qrexec-agent.o
-	$(CC) -pie -g -o qrexec-agent qrexec-agent.o $(LIBS)
+qrexec-agent: qrexec-agent.o qrexec-agent-data.o
+	$(CC) -pie -g -o qrexec-agent qrexec-agent.o qrexec-agent-data.o $(LIBS)
 qrexec-client-vm: qrexec-client-vm.o
 	$(CC) -pie -g -o qrexec-client-vm qrexec-client-vm.o
 clean:

qrexec/qrexec-agent-data.c

@@ -0,0 +1,418 @@
+/*
+ * The Qubes OS Project, http://www.qubes-os.org
+ *
+ * Copyright (C) 2013  Marek Marczykowski-Górecki  <marmarek@invisiblethingslab.com>
+ *
+ * This program is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU General Public License
+ * as published by the Free Software Foundation; either version 2
+ * of the License, or (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program; if not, write to the Free Software
+ * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA.
+ *
+ */
+
+#include <stdio.h>
+#include <stdlib.h>
+#include <signal.h>
+#include <unistd.h>
+#include <errno.h>
+#include <string.h>
+#include <sys/types.h>
+#include <sys/stat.h>
+#include <sys/wait.h>
+#include <sys/select.h>
+#include <fcntl.h>
+#include <libvchan.h>
+#include "qrexec.h"
+#include "libqrexec-utils.h"
+#include "qrexec-agent.h"
+
+#define VCHAN_BUFFER_SIZE 65536
+
+static volatile int child_exited;
+int stdout_msg_type = MSG_DATA_STDOUT;
+pid_t child_process_pid;
+
+static void sigchld_handler(int __attribute__((__unused__))x)
+{
+	child_exited = 1;
+	signal(SIGCHLD, sigchld_handler);
+}
+
+
+void no_colon_in_cmd()
+{
+	fprintf(stderr,
+		"cmdline is supposed to be in user:command form\n");
+	exit(1);
+}
+
+void do_exec(char *cmd)
+{
+	char buf[strlen(QUBES_RPC_MULTIPLEXER_PATH) + strlen(cmd) - strlen(RPC_REQUEST_COMMAND) + 1];
+	char *realcmd = index(cmd, ':');
+	if (!realcmd)
+		no_colon_in_cmd();
+	/* mark end of username and move to command */
+	*realcmd = 0;
+	realcmd++;
+	/* ignore "nogui:" prefix in linux agent */
+	if (strncmp(realcmd, "nogui:", 6) == 0)
+		realcmd+=6;
+	/* replace magic RPC cmd with RPC multiplexer path */
+	if (strncmp(realcmd, RPC_REQUEST_COMMAND " ", strlen(RPC_REQUEST_COMMAND)+1)==0) {
+		strcpy(buf, QUBES_RPC_MULTIPLEXER_PATH);
+		strcpy(buf + strlen(QUBES_RPC_MULTIPLEXER_PATH), realcmd + strlen(RPC_REQUEST_COMMAND));
+		realcmd = buf;
+	}
+	signal(SIGCHLD, SIG_DFL);
+	signal(SIGPIPE, SIG_DFL);
+
+	execl("/bin/su", "su", "-", cmd, "-c", realcmd, NULL);
+	perror("execl");
+	exit(1);
+}
+
+int handle_just_exec(char *cmdline)
+{
+	int fdn, pid;
+
+    switch (pid = fork()) {
+        case -1:
+            perror("fork");
+            return -1;
+        case 0:
+            fdn = open("/dev/null", O_RDWR);
+            fix_fds(fdn, fdn, fdn);
+            do_exec(cmdline);
+            perror("execl");
+            exit(1);
+        default:;
+    }
+	fprintf(stderr, "executed (nowait) %s pid %d\n", cmdline, pid);
+    return 0;
+}
+
+void send_exit_code(libvchan_t *data_vchan, int status)
+{
+	struct msg_header hdr;
+	hdr.type = MSG_DATA_EXIT_CODE;
+	hdr.len = sizeof(status);
+	if (libvchan_send(data_vchan, &hdr, sizeof(hdr)) < 0)
+		handle_vchan_error("write hdr");
+	if (libvchan_send(data_vchan, &status, sizeof(status)) < 0)
+		handle_vchan_error("write status");
+	fprintf(stderr, "send exit code %d\n", status);
+}
+
+/* handle data from specified FD and send over vchan link
+ * Return:
+ *  -1 - vchan error occurred
+ *  0 - EOF received, do not attempt to access this FD again
+ *  1 - some data processed, call it again when buffer space and more data
+ *      available
+ */
+int handle_input(libvchan_t *vchan, int fd, int msg_type)
+{
+	char buf[MAX_DATA_CHUNK];
+    int len;
+    struct msg_header hdr;
+
+    hdr.type = msg_type;
+    while (libvchan_buffer_space(vchan) > (int)sizeof(struct msg_header)) {
+        len = libvchan_buffer_space(vchan)-sizeof(struct msg_header);
+        if (len > (int)sizeof(buf))
+            len = sizeof(buf);
+        len = read(fd, buf, len);
+        if (len < 0) {
+            if (errno == EAGAIN || errno == EWOULDBLOCK)
+                return 1;
+            else
+                return -1;
+        }
+        hdr.len = len;
+        if (libvchan_send(vchan, &hdr, sizeof(hdr)) < 0)
+            return -1;
+
+        if (len && !write_vchan_all(vchan, buf, len))
+            return -1;
+
+        if (len == 0) {
+            close(fd);
+            return 0;
+        }
+    }
+    return 1;
+}
+
+/* handle data from vchan and send it to specified FD
+ * Return:
+ *  -2 - remote process terminated, do not send more data to it
+ *  -1 - vchan error occurred
+ *  0 - EOF received, do not attempt to access this FD again
+ *  1 - some data processed, call it again when buffer space and more data
+ *      available
+ */
+int handle_remote_data(libvchan_t *data_vchan, int stdin_fd)
+{
+	struct msg_header hdr;
+    char buf[MAX_DATA_CHUNK];
+    int status;
+
+    /* TODO: set stdin_fd to non-blocking mode and handle its buffering */
+    while (libvchan_data_ready(data_vchan) > 0) {
+        if (libvchan_recv(data_vchan, &hdr, sizeof(hdr)) < 0)
+            return -1;
+        if (hdr.len > MAX_DATA_CHUNK) {
+            fprintf(stderr, "Too big data chunk received: %d > %d\n",
+                    hdr.len, MAX_DATA_CHUNK);
+            return -1;
+        }
+        if (!read_vchan_all(data_vchan, buf, hdr.len))
+            return -1;
+
+        switch (hdr.type) {
+            /* handle both directions because this can be either server or client
+             * of VM-VM connection */
+            case MSG_DATA_STDIN:
+            case MSG_DATA_STDOUT:
+                if (stdin_fd < 0)
+                    /* discard the data */
+                    continue;
+                if (hdr.len == 0) {
+                    close(stdin_fd);
+                    stdin_fd = -1;
+                    return 0;
+                } else {
+                    /* FIXME: use buffered write here to prevent deadlock */
+                    if (!write_all(stdin_fd, buf, hdr.len)) {
+                        if (errno == EPIPE) {
+                            close(stdin_fd);
+                            stdin_fd = -1;
+                        } else {
+                            perror("write");
+                        }
+                        return 0;
+                    }
+                }
+                break;
+            case MSG_DATA_STDERR:
+                /* stderr of remote service, log locally */
+                if (!write_all(2, buf, hdr.len)) {
+                    perror("write");
+                    /* only log the error */
+                }
+                break;
+            case MSG_DATA_EXIT_CODE:
+                /* remote process exited, so there is no sense to send any data
+                 * to it */
+                status = *(unsigned int *)buf;
+                fprintf(stderr, "Remote service process exited with code %d\n", status);
+                return -2;
+        }
+    }
+    return 1;
+}
+
+void process_child_io(libvchan_t *data_vchan,
+        int stdin_fd, int stdout_fd, int stderr_fd)
+{
+    fd_set rdset, wrset;
+    int vchan_fd;
+    sigset_t selectmask;
+    int child_process_status = -1;
+    int ret, max_fd;
+    struct timespec zero_timeout = { 0, 0 };
+
+    sigemptyset(&selectmask);
+    sigaddset(&selectmask, SIGCHLD);
+    sigprocmask(SIG_BLOCK, &selectmask, NULL);
+    sigemptyset(&selectmask);
+
+    set_nonblock(stdout_fd);
+    set_nonblock(stderr_fd);
+
+    while (1) {
+        if (child_exited) {
+            pid_t pid;
+            int status;
+            while ((pid = waitpid(-1, &status, WNOHANG)) > 0) {
+                if (pid == child_process_pid) {
+                    child_process_status = WEXITSTATUS(status);
+                    if (stdin_fd >= 0) {
+                        close(stdin_fd);
+                        stdin_fd = -1;
+                    }
+                }
+            }
+            child_exited = 0;
+        }
+
+        /* if all done, exit the loop */
+        if ((!child_process_pid || child_process_status > -1) &&
+                stdin_fd == -1 && stdout_fd == -1 && stderr_fd == -1) {
+            if (child_process_status > -1) {
+                send_exit_code(data_vchan, child_process_status);
+            }
+            break;
+        }
+        /* otherwise handle the events */
+
+        FD_ZERO(&rdset);
+        FD_ZERO(&wrset);
+        max_fd = -1;
+        vchan_fd = libvchan_fd_for_select(data_vchan);
+        if (libvchan_buffer_space(data_vchan) > (int)sizeof(struct msg_header)) {
+            if (stdout_fd >= 0) {
+                FD_SET(stdout_fd, &rdset);
+                if (stdout_fd > max_fd)
+                    max_fd = stdout_fd;
+            }
+            if (stderr_fd >= 0) {
+                FD_SET(stderr_fd, &rdset);
+                if (stderr_fd > max_fd)
+                    max_fd = stderr_fd;
+            }
+        }
+        FD_SET(vchan_fd, &rdset);
+        if (vchan_fd > max_fd)
+            max_fd = vchan_fd;
+
+        if (libvchan_data_ready(data_vchan) > 0) {
+            /* check for other FDs, but exit immediately */
+            ret = pselect(max_fd + 1, &rdset, &wrset, NULL, &zero_timeout, &selectmask);
+        } else
+            ret = pselect(max_fd + 1, &rdset, &wrset, NULL, NULL, &selectmask);
+        if (ret < 0) {
+            if (errno == EINTR)
+                continue;
+            else {
+                perror("pselect");
+                /* TODO */
+                break;
+            }
+        }
+
+        /* clear event pending flag */
+        if (FD_ISSET(vchan_fd, &rdset)) {
+            if (libvchan_wait(data_vchan) < 0)
+                handle_vchan_error("wait");
+        }
+
+        if (stdout_fd >= 0 && FD_ISSET(stdout_fd, &rdset)) {
+            switch (handle_input(data_vchan, stdout_fd, stdout_msg_type)) {
+                case -1:
+                    handle_vchan_error("send");
+                    break;
+                case 0:
+                    stdout_fd = -1;
+                    break;
+            }
+        }
+        if (stderr_fd >= 0 && FD_ISSET(stderr_fd, &rdset)) {
+            switch (handle_input(data_vchan, stderr_fd, MSG_DATA_STDERR)) {
+                case -1:
+                    handle_vchan_error("send");
+                    break;
+                case 0:
+                    stderr_fd = -1;
+                    break;
+            }
+        }
+        /* handle_remote_data will check if any data is available */
+        switch (handle_remote_data(data_vchan, stdin_fd)) {
+            case -1:
+                handle_vchan_error("read");
+                break;
+            case 0:
+                stdin_fd = -1;
+                break;
+            case -2:
+                /* remote process exited, no sense in sending more data to it */
+                close(stdout_fd);
+                stdout_fd = -1;
+                close(stderr_fd);
+                stderr_fd = -1;
+                break;
+        }
+    }
+}
+
+pid_t handle_new_process(int type, int connect_domain, int connect_port,
+        char *cmdline, int cmdline_len)
+{
+    struct service_params *svc_params = (struct service_params*)cmdline;
+    libvchan_t *data_vchan;
+    pid_t pid;
+    int stdin_fd, stdout_fd, stderr_fd;
+
+    if (type == MSG_SERVICE_CONNECT) {
+        if (cmdline_len != sizeof(*svc_params)) {
+            fprintf(stderr, "Invalid MSG_SERVICE_CONNECT packet (cmdline len %d)\n", cmdline_len);
+            return -1;
+        }
+        sscanf(cmdline, "%d %d %d", &stdin_fd, &stdout_fd, &stderr_fd);
+    }
+
+    switch (pid=fork()){
+        case -1:
+            perror("fork");
+            return -1;
+        case 0:
+            break;
+        default:
+            if (type == MSG_SERVICE_CONNECT) {
+                /* no longer needed in parent process */
+                close(stdin_fd);
+                close(stdout_fd);
+                close(stderr_fd);
+            }
+            return pid;
+    }
+
+    /* child process */
+    if (type == MSG_SERVICE_CONNECT) {
+        data_vchan = libvchan_server_init(connect_domain, connect_port,
+                VCHAN_BUFFER_SIZE, VCHAN_BUFFER_SIZE);
+        if (data_vchan)
+            libvchan_wait(data_vchan);
+    } else {
+        data_vchan = libvchan_client_init(connect_domain, connect_port);
+    }
+    if (!data_vchan) {
+        fprintf(stderr, "Data vchan connection failed\n");
+        exit(1);
+    }
+    handle_handshake(data_vchan);
+
+    signal(SIGCHLD, sigchld_handler);
+
+    switch (type) {
+        case MSG_JUST_EXEC:
+            send_exit_code(data_vchan, handle_just_exec(cmdline));
+            libvchan_close(data_vchan);
+            break;
+        case MSG_EXEC_CMDLINE:
+            do_fork_exec(cmdline, &pid, &stdin_fd, &stdout_fd, &stderr_fd);
+            fprintf(stderr, "executed %s pid %d\n", cmdline, pid);
+            child_process_pid = pid;
+            process_child_io(data_vchan, stdin_fd, stdout_fd, stderr_fd);
+            break;
+        case MSG_SERVICE_CONNECT:
+            child_process_pid = 0;
+            stdout_msg_type = MSG_DATA_STDIN;
+            process_child_io(data_vchan, stdin_fd, stdout_fd, stderr_fd);
+            break;
+    }
+    exit(0);
+    /* suppress warning */
+    return 0;
+}

qrexec/qrexec-agent.c

@@ -31,75 +31,103 @@
 #include <pwd.h>
 #include <grp.h>
 #include <sys/stat.h>
+#include <assert.h>
 #include "qrexec.h"
 #include <libvchan.h>
 #include "libqrexec-utils.h"
+#include "qrexec-agent.h"
 
-enum fdtype {
-    FDTYPE_INVALID,
-    FDTYPE_STDOUT,
-    FDTYPE_STDERR
-};
-
-struct _process_fd {
-    int client_id;
-    int type;
-    int is_blocked;
-};
-struct _client_info {
-    int stdin_fd;
-    int stdout_fd;
-    int stderr_fd;
-
-    int exit_status;
-    int is_exited;
+struct _connection_info {
     int pid;
-    int is_blocked;
-    int is_close_after_flush_needed;
-    struct buffer buffer;
+    int connect_domain;
+    int connect_port;
 };
 
 int max_process_fd = -1;
 
-/* indexed by file descriptor */
-struct _process_fd process_fd[MAX_FDS];
+/*  */
+struct _connection_info connection_info[MAX_FDS];
 
-/* indexed by client id, which is descriptor number of a client in daemon */
-struct _client_info client_info[MAX_FDS];
-
-libvchan_t *vchan;
+libvchan_t *ctrl_vchan;
 
 int trigger_fd;
 int passfd_socket;
 
 int meminfo_write_started = 0;
 
-void do_exec(const char *cmd);
-void handle_vchan_error(const char *op) {
+void handle_vchan_error(const char *op)
+{
     fprintf(stderr, "Error while vchan %s, exiting\n", op);
     exit(1);
 }
 
+int handle_handshake(libvchan_t *ctrl)
+{
+    struct msg_header hdr;
+    struct peer_info info;
+
+    /* send own HELLO */
+    hdr.type = MSG_HELLO;
+    hdr.len = sizeof(info);
+    info.version = QREXEC_PROTOCOL_VERSION;
+
+    if (libvchan_send(ctrl, &hdr, sizeof(hdr)) != sizeof(hdr)) {
+        fprintf(stderr, "Failed to send HELLO hdr to agent\n");
+        return -1;
+    }
+
+    if (libvchan_send(ctrl, &info, sizeof(info)) != sizeof(info)) {
+        fprintf(stderr, "Failed to send HELLO hdr to agent\n");
+        return -1;
+    }
+
+    /* receive MSG_HELLO from remote */
+    if (libvchan_recv(ctrl, &hdr, sizeof(hdr)) != sizeof(hdr)) {
+        fprintf(stderr, "Failed to read agent HELLO hdr\n");
+        return -1;
+    }
+
+    if (hdr.type != MSG_HELLO || hdr.len != sizeof(info)) {
+        fprintf(stderr, "Invalid HELLO packet received: type %d, len %d\n", hdr.type, hdr.len);
+        return -1;
+    }
+
+    if (libvchan_recv(ctrl, &info, sizeof(info)) != sizeof(info)) {
+        fprintf(stderr, "Failed to read agent HELLO body\n");
+        return -1;
+    }
+
+    if (info.version != QREXEC_PROTOCOL_VERSION) {
+        fprintf(stderr, "Incompatible agent protocol version (remote %d, local %d)\n", info.version, QREXEC_PROTOCOL_VERSION);
+        return -1;
+    }
+
+
+    return 0;
+}
+
 void init()
 {
     /* FIXME: This 0 is remote domain ID */
-    vchan = libvchan_server_init(0, REXEC_PORT, 4096, 4096);
-    if (!vchan)
+    ctrl_vchan = libvchan_server_init(0, VCHAN_BASE_PORT, 4096, 4096);
+    if (!ctrl_vchan)
         handle_vchan_error("server_init");
+    if (handle_handshake(ctrl_vchan) < 0)
+        exit(1);
     umask(0);
     mkfifo(QREXEC_AGENT_TRIGGER_PATH, 0666);
     passfd_socket = get_server_socket(QREXEC_AGENT_FDPASS_PATH);
     umask(077);
     trigger_fd =
         open(QREXEC_AGENT_TRIGGER_PATH, O_RDONLY | O_NONBLOCK);
-    register_exec_func(do_exec);
 
     /* wait for qrexec daemon */
-    while (!libvchan_is_open(vchan))
-        libvchan_wait(vchan);
+    while (!libvchan_is_open(ctrl_vchan))
+        libvchan_wait(ctrl_vchan);
 }
 
-void wake_meminfo_writer() {
+void wake_meminfo_writer()
+{
     FILE *f;
     int pid;
 
@@ -130,266 +158,82 @@ void wake_meminfo_writer() {
     meminfo_write_started = 1;
 }
 
-void no_colon_in_cmd()
-{
-    fprintf(stderr,
-            "cmdline is supposed to be in user:command form\n");
-    exit(1);
-}
-
-void do_exec(const char *cmd)
+void register_vchan_connection(pid_t pid, int domain, int port)
 {
-    char buf[strlen(QUBES_RPC_MULTIPLEXER_PATH) + strlen(cmd) - strlen(QUBES_RPC_MAGIC_CMD) + 1];
-    char *realcmd = index(cmd, ':'), *user;
-    if (!realcmd)
-        no_colon_in_cmd();
-    /* mark end of username and move to command */
-    user=strndup(cmd,realcmd-cmd);
-    realcmd++;
-    /* ignore "nogui:" prefix in linux agent */
-    if (strncmp(realcmd, "nogui:", 6) == 0)
-        realcmd+=6;
-    /* replace magic RPC cmd with RPC multiplexer path */
-    if (strncmp(realcmd, QUBES_RPC_MAGIC_CMD " ", strlen(QUBES_RPC_MAGIC_CMD)+1)==0) {
-        strcpy(buf, QUBES_RPC_MULTIPLEXER_PATH);
-        strcpy(buf + strlen(QUBES_RPC_MULTIPLEXER_PATH), realcmd + strlen(QUBES_RPC_MAGIC_CMD));
-        realcmd = buf;
-    }
-    signal(SIGCHLD, SIG_DFL);
-    signal(SIGPIPE, SIG_DFL);
-
-    execl("/bin/su", "su", "-", user, "-c", realcmd, NULL);
-    perror("execl");
-    exit(1);
-}
+    int i;
 
-void handle_just_exec(int len)
-{
-    char buf[len];
-    int fdn, pid;
-
-    if (libvchan_recv(vchan, buf, len) < 0)
-        handle_vchan_error("read");
-    switch (pid = fork()) {
-        case -1:
-            perror("fork");
-            exit(1);
-        case 0:
-            fdn = open("/dev/null", O_RDWR);
-            fix_fds(fdn, fdn, fdn);
-            do_exec(buf);
-            perror("execl");
-            exit(1);
-        default:;
+    for (i = 0; i < MAX_FDS; i++) {
+        if (connection_info[i].pid == 0) {
+            connection_info[i].pid = pid;
+            connection_info[i].connect_domain = domain;
+            connection_info[i].connect_port = port;
+            return;
+        }
     }
-    fprintf(stderr, "executed (nowait) %s pid %d\n", buf, pid);
-}
 
-void create_info_about_client(int client_id, int pid, int stdin_fd,
-        int stdout_fd, int stderr_fd)
-{
-    process_fd[stdout_fd].client_id = client_id;
-    process_fd[stdout_fd].type = FDTYPE_STDOUT;
-    process_fd[stdout_fd].is_blocked = 0;
-    process_fd[stderr_fd].client_id = client_id;
-    process_fd[stderr_fd].type = FDTYPE_STDERR;
-    process_fd[stderr_fd].is_blocked = 0;
-
-    if (stderr_fd > max_process_fd)
-        max_process_fd = stderr_fd;
-    if (stdout_fd > max_process_fd)
-        max_process_fd = stdout_fd;
-
-    set_nonblock(stdin_fd);
-
-    client_info[client_id].stdin_fd = stdin_fd;
-    client_info[client_id].stdout_fd = stdout_fd;
-    client_info[client_id].stderr_fd = stderr_fd;
-    client_info[client_id].exit_status = 0;
-    client_info[client_id].is_exited = 0;
-    client_info[client_id].pid = pid;
-    client_info[client_id].is_blocked = 0;
-    client_info[client_id].is_close_after_flush_needed = 0;
-    buffer_init(&client_info[client_id].buffer);
+    fprintf(stderr, "No free slot for child %d (connection to %d:%d)\n", pid, domain, port);
 }
 
-void handle_exec(int client_id, int len)
+void handle_server_exec_request(struct msg_header *hdr)
 {
-    char buf[len];
-    int pid, stdin_fd, stdout_fd, stderr_fd;
+    struct exec_params params;
+    char buf[hdr->len-sizeof(params)];
+    pid_t child_agent;
 
-    if (libvchan_recv(vchan, buf, len) < 0)
-        handle_vchan_error("read");
+    assert(hdr->len >= sizeof(params));
 
-    do_fork_exec(buf, &pid, &stdin_fd, &stdout_fd, &stderr_fd);
+    if (libvchan_recv(ctrl_vchan, &params, sizeof(params)) < 0)
+        handle_vchan_error("read exec params");
+    if (libvchan_recv(ctrl_vchan, buf, hdr->len-sizeof(params)) < 0)
+        handle_vchan_error("read exec cmd");
 
-    create_info_about_client(client_id, pid, stdin_fd, stdout_fd,
-            stderr_fd);
-
-    fprintf(stderr, "executed %s pid %d\n", buf, pid);
+    child_agent = handle_new_process(hdr->type,
+            params.connect_domain, params.connect_port,
+            buf, hdr->len-sizeof(params));
 
+    register_vchan_connection(child_agent,
+            params.connect_domain, params.connect_port);
 }
 
-void handle_connect_existing(int client_id, int len)
+void handle_service_refused(struct msg_header *hdr)
 {
+    struct service_params params;
     int stdin_fd, stdout_fd, stderr_fd;
-    char buf[len];
-    if (libvchan_recv(vchan, buf, len) < 0)
-        handle_vchan_error("read");
-    sscanf(buf, "%d %d %d", &stdin_fd, &stdout_fd, &stderr_fd);
-    create_info_about_client(client_id, -1, stdin_fd, stdout_fd,
-            stderr_fd);
-    client_info[client_id].is_exited = 1;	//do not wait for SIGCHLD
-}
-
-void update_max_process_fd()
-{
-    int i;
-    for (i = max_process_fd;
-            i >= 0 && process_fd[i].type == FDTYPE_INVALID; i--);
-    max_process_fd = i;
-}
-
-void send_exit_code(int client_id, int status)
-{
-    struct server_header s_hdr;
-    s_hdr.type = MSG_AGENT_TO_SERVER_EXIT_CODE;
-    s_hdr.client_id = client_id;
-    s_hdr.len = sizeof status;
-    if (libvchan_send(vchan, &s_hdr, sizeof(s_hdr)) < 0)
-        handle_vchan_error("write hdr");
-    if (libvchan_send(vchan, &status, sizeof(status)) < 0)
-        handle_vchan_error("write status");
-    fprintf(stderr, "send exit code %d for client_id %d pid %d\n",
-            status, client_id, client_info[client_id].pid);
-}
-
-
-// erase process data structures, possibly forced by remote
-void remove_process(int client_id, int status)
-{
-    int i;
-    if (!client_info[client_id].pid)
-        return;
-    if (client_info[client_id].stdin_fd >= 0)
-        fork_and_flush_stdin(client_info[client_id].stdin_fd,
-                &client_info[client_id].buffer);
-#if 0
-    //      let's let it die by itself, possibly after it has received buffered stdin
-    kill(client_info[client_id].pid, SIGKILL);
-#endif
-    if (status != -1)
-        send_exit_code(client_id, status);
-
-
-    close(client_info[client_id].stdin_fd);
-    client_info[client_id].pid = 0;
-    client_info[client_id].stdin_fd = -1;
-    client_info[client_id].is_blocked = 0;
-    buffer_free(&client_info[client_id].buffer);
-
-    for (i = 0; i <= max_process_fd; i++)
-        if (process_fd[i].type != FDTYPE_INVALID
-                && process_fd[i].client_id == client_id) {
-            process_fd[i].type = FDTYPE_INVALID;
-            process_fd[i].client_id = -1;
-            process_fd[i].is_blocked = 0;
-            close(i);
-        }
-    update_max_process_fd();
-}
-
-// remove process not immediately after it has exited, but after its stdout and stderr has been drained
-// previous method implemented in flush_out_err was broken - it cannot work when peer signalled it is blocked
-void possibly_remove_process(int client_id)
-{
-    if (client_info[client_id].stdout_fd == -1 &&
-            client_info[client_id].stderr_fd == -1 &&
-            client_info[client_id].is_exited)
-        remove_process(client_id,
-                client_info[client_id].exit_status);
-}
-
-
-void handle_input(int client_id, int len)
-{
-    char buf[len];
-
-    if (libvchan_recv(vchan, buf, len) < 0)
-        handle_vchan_error("read");
-    if (!client_info[client_id].pid || client_info[client_id].stdin_fd == -1)
-        return;
 
-    if (len == 0) {
-        if (client_info[client_id].is_blocked)
-            client_info[client_id].is_close_after_flush_needed
-                = 1;
-        else {
-            close(client_info[client_id].stdin_fd);
-            client_info[client_id].stdin_fd = -1;
-        }
-        return;
+    if (hdr->len != sizeof(params)) {
+        fprintf(stderr, "Invalid msg 0x%x length (%d)\n", MSG_SERVICE_REFUSED, hdr->len);
+        exit(1);
     }
 
-    switch (write_stdin
-            (vchan, client_info[client_id].stdin_fd, client_id, buf, len,
-             &client_info[client_id].buffer)) {
-        case WRITE_STDIN_OK:
-            break;
-        case WRITE_STDIN_BUFFERED:
-            client_info[client_id].is_blocked = 1;
-            break;
-        case WRITE_STDIN_ERROR:
-            // do not remove process, as it still can write data to stdout
-            close(client_info[client_id].stdin_fd);
-            client_info[client_id].stdin_fd = -1;
-            client_info[client_id].is_blocked = 0;
-            break;
-        default:
-            fprintf(stderr, "unknown write_stdin?\n");
-            exit(1);
-    }
+    if (libvchan_recv(ctrl_vchan, &params, sizeof(params)) < 0)
+        handle_vchan_error("read exec params");
 
+    sscanf(params.ident, "%d %d %d", &stdin_fd, &stdout_fd, &stderr_fd);
+    /* TODO: send some signal? some response? */
+    close(stdin_fd);
+    close(stdout_fd);
+    close(stderr_fd);
 }
 
-void set_blocked_outerr(int client_id, int val)
+void handle_server_cmd()
 {
-    process_fd[client_info[client_id].stdout_fd].is_blocked = val;
-    process_fd[client_info[client_id].stderr_fd].is_blocked = val;
-}
+    struct msg_header s_hdr;
 
-void handle_server_data()
-{
-    struct server_header s_hdr;
-    if (libvchan_recv(vchan, &s_hdr, sizeof(s_hdr)) < 0)
+    if (libvchan_recv(ctrl_vchan, &s_hdr, sizeof(s_hdr)) < 0)
         handle_vchan_error("read s_hdr");
 
     //      fprintf(stderr, "got %x %x %x\n", s_hdr.type, s_hdr.client_id,
     //              s_hdr.len);
 
     switch (s_hdr.type) {
-        case MSG_XON:
-            set_blocked_outerr(s_hdr.client_id, 0);
-            break;
-        case MSG_XOFF:
-            set_blocked_outerr(s_hdr.client_id, 1);
-            break;
-        case MSG_SERVER_TO_AGENT_CONNECT_EXISTING:
-            handle_connect_existing(s_hdr.client_id, s_hdr.len);
-            break;
-        case MSG_SERVER_TO_AGENT_EXEC_CMDLINE:
-            wake_meminfo_writer();
-            handle_exec(s_hdr.client_id, s_hdr.len);
-            break;
-        case MSG_SERVER_TO_AGENT_JUST_EXEC:
+        case MSG_EXEC_CMDLINE:
+        case MSG_JUST_EXEC:
+        case MSG_SERVICE_CONNECT:
             wake_meminfo_writer();
-            handle_just_exec(s_hdr.len);
-            break;
-        case MSG_SERVER_TO_AGENT_INPUT:
-            handle_input(s_hdr.client_id, s_hdr.len);
+            handle_server_exec_request(&s_hdr);
             break;
-        case MSG_SERVER_TO_AGENT_CLIENT_END:
-            remove_process(s_hdr.client_id, -1);
+        case MSG_SERVICE_REFUSED:
+            handle_service_refused(&s_hdr);
             break;
         default:
             fprintf(stderr, "msg type from daemon is %d ?\n",
@@ -398,54 +242,6 @@ void handle_server_data()
     }
 }
 
-void handle_process_data(int fd)
-{
-    struct server_header s_hdr;
-    char buf[MAX_DATA_CHUNK];
-    int ret;
-    unsigned int len;
-
-    len = libvchan_buffer_space(vchan);
-    if (len <= sizeof s_hdr)
-        return;
-
-    ret = read(fd, buf, len - sizeof s_hdr);
-    s_hdr.client_id = process_fd[fd].client_id;
-
-    if (process_fd[fd].type == FDTYPE_STDOUT)
-        s_hdr.type = MSG_AGENT_TO_SERVER_STDOUT;
-    else if (process_fd[fd].type == FDTYPE_STDERR)
-        s_hdr.type = MSG_AGENT_TO_SERVER_STDERR;
-    else {
-        fprintf(stderr, "fd=%d, client_id=%d, type=%d ?\n", fd,
-                process_fd[fd].client_id, process_fd[fd].type);
-        exit(1);
-    }
-    s_hdr.len = ret;
-    if (ret >= 0) {
-        if (libvchan_send(vchan, &s_hdr, sizeof(s_hdr)) < 0)
-            handle_vchan_error("write hdr");
-        if (libvchan_send(vchan, buf, ret) < 0)
-            handle_vchan_error("write buf");
-    }
-    if (ret == 0) {
-        int client_id = process_fd[fd].client_id;
-        if (process_fd[fd].type == FDTYPE_STDOUT)
-            client_info[client_id].stdout_fd = -1;
-        else
-            client_info[client_id].stderr_fd = -1;
-
-        process_fd[fd].type = FDTYPE_INVALID;
-        process_fd[fd].client_id = -1;
-        process_fd[fd].is_blocked = 0;
-        close(fd);
-        update_max_process_fd();
-        possibly_remove_process(client_id);
-    }
-    if (ret < 0)
-        remove_process(process_fd[fd].client_id, 127);
-}
-
 volatile int child_exited;
 
 void sigchld_handler(int x __attribute__((__unused__)))
@@ -454,37 +250,36 @@ void sigchld_handler(int x __attribute__((__unused__)))
     signal(SIGCHLD, sigchld_handler);
 }
 
-int find_info(int pid)
+int find_connection(int pid)
 {
     int i;
     for (i = 0; i < MAX_FDS; i++)
-        if (client_info[i].pid == pid)
+        if (connection_info[i].pid == pid)
             return i;
     return -1;
 }
 
 
-void handle_process_data_all(fd_set * select_fds)
-{
-    int i;
-    for (i = 0; i <= max_process_fd; i++)
-        if (process_fd[i].type != FDTYPE_INVALID
-                && FD_ISSET(i, select_fds))
-            handle_process_data(i);
-}
-
 void reap_children()
 {
     int status;
     int pid;
-    int client_id;
+    int id;
+    struct msg_header hdr;
+    struct exec_params params;
     while ((pid = waitpid(-1, &status, WNOHANG)) > 0) {
-        client_id = find_info(pid);
-        if (client_id < 0)
+        id = find_connection(pid);
+        if (id < 0)
             continue;
-        client_info[client_id].is_exited = 1;
-        client_info[client_id].exit_status = status;
-        possibly_remove_process(client_id);
+        hdr.type = MSG_CONNECTION_TERMINATED;
+        hdr.len = sizeof(struct exec_params);
+        params.connect_domain = connection_info[id].connect_domain;
+        params.connect_port = connection_info[id].connect_port;
+        if (libvchan_send(ctrl_vchan, &hdr, sizeof(hdr)) < 0)
+            handle_vchan_error("send");
+        if (libvchan_send(ctrl_vchan, &params, sizeof(params)) < 0)
+            handle_vchan_error("send");
+        connection_info[id].pid = 0;
     }
     child_exited = 0;
 }
@@ -492,62 +287,18 @@ void reap_children()
 int fill_fds_for_select(fd_set * rdset, fd_set * wrset)
 {
     int max = -1;
-    int fd, i;
     FD_ZERO(rdset);
     FD_ZERO(wrset);
 
-    for (i = 0; i <= max_process_fd; i++)
-        if (process_fd[i].type != FDTYPE_INVALID
-                && !process_fd[i].is_blocked) {
-            FD_SET(i, rdset);
-            max = i;
-        }
-
     FD_SET(trigger_fd, rdset);
     if (trigger_fd > max)
         max = trigger_fd;
     FD_SET(passfd_socket, rdset);
     if (passfd_socket > max)
         max = passfd_socket;
-
-    for (i = 0; i < MAX_FDS; i++)
-        if (client_info[i].pid && client_info[i].is_blocked) {
-            fd = client_info[i].stdin_fd;
-            FD_SET(fd, wrset);
-            if (fd > max)
-                max = fd;
-        }
     return max;
 }
 
-void flush_client_data_agent(int client_id)
-{
-    struct _client_info *info = &client_info[client_id];
-    switch (flush_client_data
-            (vchan, info->stdin_fd, client_id, &info->buffer)) {
-        case WRITE_STDIN_OK:
-            info->is_blocked = 0;
-            if (info->is_close_after_flush_needed) {
-                close(info->stdin_fd);
-                info->stdin_fd = -1;
-                info->is_close_after_flush_needed = 0;
-            }
-            break;
-        case WRITE_STDIN_ERROR:
-            // do not remove process, as it still can write data to stdout
-            info->is_blocked = 0;
-            close(info->stdin_fd);
-            info->stdin_fd = -1;
-            info->is_close_after_flush_needed = 0;
-            break;
-        case WRITE_STDIN_BUFFERED:
-            break;
-        default:
-            fprintf(stderr, "unknown flush_client_data?\n");
-            exit(1);
-    }
-}
-
 void handle_new_passfd()
 {
     int fd = do_accept(passfd_socket);
@@ -561,21 +312,19 @@ void handle_new_passfd()
     }
 }
 
-
 void handle_trigger_io()
 {
-    struct server_header s_hdr;
-    struct trigger_connect_params params;
+    struct msg_header hdr;
+    struct trigger_service_params params;
     int ret;
 
-    s_hdr.client_id = 0;
-    s_hdr.len = 0;
+    hdr.len = sizeof(params);
     ret = read(trigger_fd, &params, sizeof(params));
     if (ret == sizeof(params)) {
-        s_hdr.type = MSG_AGENT_TO_SERVER_TRIGGER_CONNECT_EXISTING;
-        if (libvchan_send(vchan, &s_hdr, sizeof(s_hdr)) < 0)
+        hdr.type = MSG_TRIGGER_SERVICE;
+        if (libvchan_send(ctrl_vchan, &hdr, sizeof(hdr)) < 0)
             handle_vchan_error("write hdr");
-        if (libvchan_send(vchan, &params, sizeof(params)) < 0)
+        if (libvchan_send(ctrl_vchan, &params, sizeof(params)) < 0)
             handle_vchan_error("write params");
     }
     // trigger_fd is nonblock - so no need to reopen
@@ -591,7 +340,6 @@ int main()
 {
     fd_set rdset, wrset;
     int max;
-    int i;
     sigset_t chld_set;
 
     init();
@@ -606,27 +354,20 @@ int main()
         if (child_exited)
             reap_children();
         max = fill_fds_for_select(&rdset, &wrset);
-        if (libvchan_buffer_space(vchan) <=
-                sizeof(struct server_header))
+        if (libvchan_buffer_space(ctrl_vchan) <=
+                (int)sizeof(struct msg_header))
             FD_ZERO(&rdset);
 
-        wait_for_vchan_or_argfd(vchan, max, &rdset, &wrset);
+        wait_for_vchan_or_argfd(ctrl_vchan, max, &rdset, &wrset);
         sigprocmask(SIG_UNBLOCK, &chld_set, NULL);
 
         if (FD_ISSET(passfd_socket, &rdset))
             handle_new_passfd();
 
-        while (libvchan_data_ready(vchan))
-            handle_server_data();
+        while (libvchan_data_ready(ctrl_vchan))
+            handle_server_cmd();
 
         if (FD_ISSET(trigger_fd, &rdset))
             handle_trigger_io();
-
-        handle_process_data_all(&rdset);
-        for (i = 0; i <= MAX_FDS; i++)
-            if (client_info[i].pid
-                    && client_info[i].is_blocked
-                    && FD_ISSET(client_info[i].stdin_fd, &wrset))
-                flush_client_data_agent(i);
     }
 }

qrexec/qrexec-agent.h

@@ -0,0 +1,27 @@
+/*
+ * The Qubes OS Project, http://www.qubes-os.org
+ *
+ * Copyright (C) 2013  Marek Marczykowski-Górecki  <marmarek@invisiblethingslab.com>
+ *
+ * This program is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU General Public License
+ * as published by the Free Software Foundation; either version 2
+ * of the License, or (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program; if not, write to the Free Software
+ * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA.
+ *
+ */
+
+int handle_handshake(libvchan_t *ctrl);
+void handle_vchan_error(const char *op);
+
+pid_t handle_new_process(int type,
+        int connect_domain, int connect_port,
+        char *cmdline, int cmdline_len);

qrexec/qrexec-client-vm.c

@@ -60,7 +60,7 @@ char *get_program_name(char *prog)
 int main(int argc, char **argv)
 {
     int trigger_fd;
-    struct trigger_connect_params params;
+    struct trigger_service_params params;
     int local_fd[3], remote_fd[3];
     int i;
     char *abs_exec_path;
@@ -93,15 +93,16 @@ int main(int argc, char **argv)
             putenv(env);
             dup2(local_fd[i], i);
             close(local_fd[i]);
-        }
+        } else
+            close(local_fd[i]);
     }
 
     memset(&params, 0, sizeof(params));
-    strncpy(params.exec_index, argv[2], sizeof(params.exec_index));
-    strncpy(params.target_vmname, argv[1],
-            sizeof(params.target_vmname));
-    snprintf(params.process_fds.ident,
-            sizeof(params.process_fds.ident), "%d %d %d",
+    strncpy(params.service_name, argv[2], sizeof(params.service_name));
+    strncpy(params.target_domain, argv[1],
+            sizeof(params.target_domain));
+    snprintf(params.request_id.ident,
+            sizeof(params.request_id.ident), "%d %d %d",
             remote_fd[0], remote_fd[1], remote_fd[2]);
 
     if (write(trigger_fd, &params, sizeof(params)) < 0) {