[Qemu-devel] [RFC][PATCH v1 10/12] guest agent: qemu-ga daemon

[Michael Roth]

This is the actual guest daemon, it listens for requests over a
virtio-serial/isa-serial/unix socket channel and routes them through
to dispatch routines, and writes the results back to the channel in
a manner similar to Qmp.

This is currently horribly broken, only the unix-listen channel method
is working at the moment (likely due to mis-use of gio channel
interfaces), and the code is in overall rough shape.

Signed-off-by: Michael Roth <address@hidden>
---
 qemu-ga.c |  522 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
 1 files changed, 522 insertions(+), 0 deletions(-)
 create mode 100644 qemu-ga.c

diff --git a/qemu-ga.c b/qemu-ga.c
new file mode 100644
index 0000000..435a1fc
--- /dev/null
+++ b/qemu-ga.c
@@ -0,0 +1,522 @@
+/*
+ * QEMU Guest Agent
+ *
+ * Copyright IBM Corp. 2011
+ *
+ * Authors:
+ *  Adam Litke        <address@hidden>
+ *  Michael Roth      <address@hidden>
+ *
+ * This work is licensed under the terms of the GNU GPL, version 2 or later.
+ * See the COPYING file in the top-level directory.
+ */
+#include <stdlib.h>
+#include <stdio.h>
+#include <stdbool.h>
+#include <glib.h>
+#include <gio/gio.h>
+#include <getopt.h>
+#include <termios.h>
+#include "qemu_socket.h"
+#include "json-streamer.h"
+#include "json-parser.h"
+#include "guest-agent.h"
+
+#define QGA_VERSION "1.0"
+#define QGA_GUEST_PATH_VIRTIO_DEFAULT "/dev/virtio-ports/va"
+#define QGA_PIDFILE_DEFAULT "/var/run/qemu-va.pid"
+#define QGA_BAUDRATE_DEFAULT B38400 /* for isa-serial channels */
+
+bool verbose_enabled = false;
+
+typedef struct GAState {
+    bool active;
+    int session_id;
+    const char *proxy_path;
+    JSONMessageParser parser;
+    GMainLoop *main_loop;
+    guint conn_id;
+    GSocket *conn_sock;
+    GIOChannel *conn_channel;
+    guint listen_id;
+    GSocket *listen_sock;
+    GIOChannel *listen_channel;
+    const char *path;
+    const char *method;
+} GAState;
+
+static void usage(const char *cmd)
+{
+    printf(
+"Usage: %s -c <channel_opts>\n"
+"QEMU virtagent guest agent %s\n"
+"\n"
+"  -c, --channel     channel method: one of unix-connect, virtio-serial, or\n"
+"                    isa-serial\n"
+"  -p, --path        channel path\n"
+"  -v, --verbose     display extra debugging information\n"
+"  -d, --daemonize   become a daemon\n"
+"  -h, --help        display this help and exit\n"
+"\n"
+"Report bugs to <address@hidden>\n"
+    , cmd, QGA_VERSION);
+}
+
+static void conn_channel_close(GAState *s);
+
+static void become_daemon(void)
+{
+    pid_t pid, sid;
+    int pidfd;
+    char *pidstr;
+
+    pid = fork();
+    if (pid < 0)
+        exit(EXIT_FAILURE);
+    if (pid > 0) {
+        exit(EXIT_SUCCESS);
+    }
+
+    pidfd = open(QGA_PIDFILE_DEFAULT, O_CREAT|O_RDWR, S_IRUSR|S_IWUSR);
+    if (!pidfd || lockf(pidfd, F_TLOCK, 0))
+        g_error("Cannot lock pid file");
+
+    if (ftruncate(pidfd, 0) || lseek(pidfd, 0, SEEK_SET))
+       g_error("Cannot truncate pid file");
+    if (asprintf(&pidstr, "%d", getpid()) == -1)
+        g_error("Cannot allocate memory");
+    if (write(pidfd, pidstr, strlen(pidstr)) != strlen(pidstr))
+        g_error("Failed to write pid file");
+    free(pidstr);
+
+    umask(0);
+    sid = setsid();
+    if (sid < 0)
+        goto fail;
+    if ((chdir("/")) < 0)
+        goto fail;
+
+    close(STDIN_FILENO);
+    close(STDOUT_FILENO);
+    close(STDERR_FILENO);
+    return;
+
+fail:
+    unlink(QGA_PIDFILE_DEFAULT);
+    g_error("failed to daemonize");
+}
+
+static int conn_channel_send_payload(GIOChannel *channel, QObject *payload)
+{
+    gsize count, written=0;
+    const char *buf;
+    QString *payload_qstr;
+    GIOStatus status;
+    GError *err = NULL;
+
+    if (!payload || !channel) {
+        return -EINVAL;
+    }
+
+    payload_qstr = qobject_to_json(payload);
+    if (!payload_qstr) {
+        return -EINVAL;
+    }
+
+    buf = qstring_get_str(payload_qstr);
+    count = strlen(buf);
+
+    while (count) {
+        g_warning("sending, count: %d", (int)count);
+        status = g_io_channel_write_chars(channel, buf, count, &written, &err);
+        if (err != NULL) {
+            g_warning("error sending payload: %s", err->message);
+            return err->code;
+        }
+        if (status == G_IO_STATUS_NORMAL) {
+            count -= written;
+        } else if (status == G_IO_STATUS_ERROR || status == G_IO_STATUS_EOF) {
+            return -EPIPE;
+        }
+    }
+    g_io_channel_flush(channel, &err);
+    if (err != NULL) {
+        g_warning("error flushing payload: %s", err->message);
+        return err->code;
+    }
+    return 0;
+}
+
+/* keep reading channel, but ignore all data until we see an appropriate
+ * ack from the host
+ */
+static void start_negotiation(GAState *s)
+{
+    QDict *payload = qdict_new();
+    int ret;
+
+    g_debug("[re]negotiating connection");
+
+    g_assert(s && s->conn_channel);
+    s->active = false;
+    s->session_id = g_random_int_range(1, G_MAXINT32);
+
+    qdict_put_obj(payload, "_xport_event",
+                  QOBJECT(qstring_from_str("guest_init")));
+    qdict_put_obj(payload, "_xport_arg_sid",
+                  QOBJECT(qint_from_int(s->session_id)));
+
+    ret = conn_channel_send_payload(s->conn_channel, QOBJECT(payload));
+    if (ret) {
+        g_warning("failed to send guest init, resetting connection");
+        conn_channel_close(s);
+    }
+}
+
+static void complete_negotiation(GAState *s)
+{
+    g_debug("connection negotiation complete");
+    s->active = true;
+}
+
+static void process_event(JSONMessageParser *parser, QList *tokens)
+{
+    GAState *s = container_of(parser, GAState, parser);
+    QObject *obj, *rsp;
+    QDict *qdict;
+    Error *err = NULL;
+    const char *cmd;
+    int session_id, ret;
+
+    g_assert(s && parser);
+
+    g_debug("process_event: called\n");
+    obj = json_parser_parse_err(tokens, NULL, &err);
+    if (!obj || qobject_type(obj) != QTYPE_QDICT) {
+        g_warning("failed to parse event");
+        return;
+    } else {
+        g_debug("parse successful");
+        qdict = qobject_to_qdict(obj);
+    }
+
+    /* check for transport-only commands/events */
+    if (qdict_haskey(qdict, "_xport_event")) {
+        cmd = qdict_get_try_str(qdict, "_xport_event");
+        if (cmd && strcmp(cmd, "host_ack") == 0) {
+            session_id = qdict_get_try_int(qdict, "_xport_arg_sid", 0);
+            if (session_id == s->session_id) {
+                /* host acknowledged us, begin normal operation */
+                complete_negotiation(s);
+            }
+            return;
+        }
+    }
+
+    /* ignore any non-xport-related events/objects */
+    if (!s->active) {
+        return;
+    }
+
+    if (qdict_haskey(qdict, "execute")) {
+        g_debug("received command");
+        rsp = qga_dispatch(QOBJECT(qdict), &err);
+        g_debug("done handling command");
+        if (err == NULL && rsp) {
+            g_debug("response:\n%s", qstring_get_str(qobject_to_json(rsp)));
+            ret = conn_channel_send_payload(s->conn_channel, rsp);
+            if (ret) {
+                g_warning("error sending payload: %s", strerror(ret));
+                /* reset/renegotiate connection, since we may be in
+                 * unrecoverable state
+                 */
+                start_negotiation(s);
+            }
+            qobject_decref(rsp);
+        } else {
+            g_warning("error getting response");
+        }
+    } else {
+        g_warning("unrecognized payload format, ignoring");
+    }
+}
+
+static gboolean conn_channel_read(GIOChannel *channel, GIOCondition condition,
+                                  gpointer data)
+{
+    GAState *s = data;
+    gchar buf[1024];
+    gsize count;
+    GError *err = NULL;
+    memset(buf, 0, 1024);
+    GIOStatus status = g_io_channel_read_chars(channel, buf, 1024,
+                                               &count, &err);
+    if (err != NULL) {
+        g_warning("error reading channel: %s", err->message);
+        conn_channel_close(s);
+        return false;
+    }
+    switch (status) {
+    case G_IO_STATUS_ERROR:
+        g_warning("problem");
+        return false;
+    case G_IO_STATUS_NORMAL:
+        json_message_parser_feed(&s->parser, (char *)buf, (int)count);
+        g_warning("count: %d, data: %s", (int)count, buf);
+    case G_IO_STATUS_AGAIN:
+        return true;
+    case G_IO_STATUS_EOF:
+        g_debug("received EOF");
+        conn_channel_close(s);
+        return false;
+    default:
+        g_warning("unknown channel read status, closing");
+        conn_channel_close(s);
+        return false;
+    }
+    return true;
+}
+
+static int conn_channel_add(GAState *s, int fd)
+{
+    GIOChannel *conn_channel;
+    guint conn_id;
+    GError *err = NULL;
+
+    g_assert(s && fd > 0 && !s->conn_channel);
+    conn_channel = g_io_channel_unix_new(fd);
+    g_assert(conn_channel);
+    g_io_channel_set_encoding(conn_channel, NULL, &err);
+    if (err != NULL) {
+        g_warning("error setting channel encoding to binary");
+        return -1;
+    }
+    conn_id = g_io_add_watch(conn_channel, G_IO_IN, conn_channel_read, s);
+    if (err != NULL) {
+        g_warning("error adding io watch: %s", err->message);
+        return -1;
+    }
+    s->conn_channel = conn_channel;
+    s->conn_id = conn_id;
+    start_negotiation(s);
+    return 0;
+}
+
+static gboolean listen_channel_accept(GIOChannel *channel,
+                                      GIOCondition condition, gpointer data)
+{
+    GAState *s = data;
+    GError *err = NULL;
+    g_assert(channel != NULL);
+    int ret;
+    bool accepted = false;
+
+    s->conn_sock = g_socket_accept(s->listen_sock, NULL, &err);
+    if (err != NULL) {
+        g_warning("error converting fd to gsocket: %s", err->message);
+        goto out;
+    }
+    ret = conn_channel_add(s, g_socket_get_fd(s->conn_sock));
+    if (ret) {
+        g_warning("error setting up connection");
+        goto out;
+    }
+    accepted = true;
+
+out:
+    /* only accept 1 connection at a time */
+    return !accepted;
+}
+
+/* start polling for readable events on listen fd, listen_fd=0
+ * indicates we should use the existing s->listen_channel
+ */
+static int listen_channel_add(GAState *s, int listen_fd)
+{
+    GError *err = NULL;
+    guint listen_id;
+
+    if (listen_fd) {
+        s->listen_channel = g_io_channel_unix_new(listen_fd);
+        if (s->listen_sock) {
+            g_object_unref(s->listen_sock);
+        }
+        s->listen_sock = g_socket_new_from_fd(listen_fd, &err);
+        if (err != NULL) {
+            g_warning("error converting fd to gsocket: %s", err->message);
+            return -1;
+        }
+    }
+    listen_id = g_io_add_watch(s->listen_channel, G_IO_IN,
+                               listen_channel_accept, s);
+    if (err != NULL) {
+        g_warning("error adding io watch: %s", err->message);
+        return -1;
+    }
+    return 0;
+}
+
+/* cleanup state for closed connection/session, start accepting new
+ * connections if we're in listening mode
+ */
+static void conn_channel_close(GAState *s)
+{
+    if (strcmp(s->method, "unix-listen") == 0) {
+        g_object_unref(s->conn_sock);
+        s->conn_sock = NULL;
+        listen_channel_add(s, 0);
+    } else if (strcmp(s->method, "virtio-serial") == 0) {
+        /* we spin on EOF for virtio-serial, so back off a bit. also,
+         * dont close the connection in this case, it'll resume normal
+         * operation when another process connects to host chardev
+         */
+        usleep(100*1000);
+        return;
+    }
+    g_io_channel_shutdown(s->conn_channel, true, NULL);
+    g_io_channel_unref(s->conn_channel);
+    s->conn_channel = NULL;
+    s->conn_id = 0;
+}
+
+static void init_guest_agent(GAState *s)
+{
+    struct termios tio;
+    int ret, fd;
+
+    if (s->method == NULL) {
+        /* try virtio-serial as our default */
+        s->method = "virtio-serial";
+    }
+
+    if (s->path == NULL) {
+        if (strcmp(s->method, "virtio-serial") != 0) {
+            g_error("must specify a path for this channel");
+        }
+        /* try the default path for the virtio-serial port */
+        s->path = QGA_GUEST_PATH_VIRTIO_DEFAULT;
+    }
+
+    if (strcmp(s->method, "virtio-serial") == 0) {
+        fd = qemu_open(s->path, O_RDWR);
+        if (fd == -1) {
+            g_error("error opening channel: %s", strerror(errno));
+        }
+        ret = fcntl(fd, F_GETFL);
+        if (ret < 0) {
+            g_error("error getting channel flags: %s", strerror(errno));
+        }
+        ret = fcntl(fd, F_SETFL, ret | O_NONBLOCK | O_ASYNC);
+        if (ret < 0) {
+            g_error("error setting channel flags: %s", strerror(errno));
+        }
+        ret = conn_channel_add(s, fd);
+        if (ret) {
+            g_error("error adding channel to main loop");
+        }
+    } else if (strcmp(s->method, "isa-serial") == 0) {
+        fd = qemu_open(s->path, O_RDWR | O_NOCTTY);
+        if (fd == -1) {
+            g_error("error opening channel: %s", strerror(errno));
+        }
+        tcgetattr(fd, &tio);
+        /* set up serial port for non-canonical, dumb byte streaming */
+        tio.c_iflag &= ~(IGNBRK | BRKINT | IGNPAR | PARMRK | INPCK | ISTRIP |
+                         INLCR | IGNCR | ICRNL | IXON | IXOFF | IXANY |
+                         IMAXBEL);
+        tio.c_oflag = 0;
+        tio.c_lflag = 0;
+        tio.c_cflag |= QGA_BAUDRATE_DEFAULT;
+        /* 1 available byte min or reads will block (we'll set non-blocking
+         * elsewhere, else we have to deal with read()=0 instead)
+         */
+        tio.c_cc[VMIN] = 1;
+        tio.c_cc[VTIME] = 0;
+        /* flush everything waiting for read/xmit, it's garbage at this point 
*/
+        tcflush(fd, TCIFLUSH);
+        tcsetattr(fd, TCSANOW, &tio);
+        ret = conn_channel_add(s, fd);
+        if (ret) {
+            g_error("error adding channel to main loop");
+        }
+    } else if (strcmp(s->method, "unix-listen") == 0) {
+        fd = unix_listen(s->path, NULL, strlen(s->path));
+        if (fd <= 0) {
+            g_error("error opening path: %s", strerror(errno));
+        }
+        ret = listen_channel_add(s, fd);
+        if (ret) {
+            g_error("error binding/listening to specified socket");
+        }
+    } else {
+        g_error("unsupported channel method/type: %s", s->method);
+    }
+
+    json_message_parser_init(&s->parser, process_event);
+    s->main_loop = g_main_loop_new(NULL, false);
+}
+
+int main(int argc, char **argv)
+{
+    const char *sopt = "hVvdc:p:", *method = NULL, *path = NULL;
+    struct option lopt[] = {
+        { "help", 0, NULL, 'h' },
+        { "version", 0, NULL, 'V' },
+        { "verbose", 0, NULL, 'v' },
+        { "channel", 0, NULL, 'c' },
+        { "path", 0, NULL, 'p' },
+        { "daemonize", 0, NULL, 'd' },
+        { NULL, 0, NULL, 0 }
+    };
+    int opt_ind = 0, ch, daemonize=0;
+    GAState *s;
+
+    g_type_init();
+
+    while ((ch = getopt_long(argc, argv, sopt, lopt, &opt_ind)) != -1) {
+        switch (ch) {
+        case 'c':
+            method = optarg;
+            break;
+        case 'p':
+            path = optarg;
+            break;
+        case 'v':
+            /* TODO: set appropriate log level */
+            verbose_enabled = true;
+            break;
+        case 'V':
+            printf("QEMU Guest Agent %s\n", QGA_VERSION);
+            return 0;
+        case 'd':
+            daemonize = 1;
+            break;
+        case 'h':
+            usage(argv[0]);
+            return 0;
+        case '?':
+            g_error("Unknown option, try '%s --help' for more information.",
+                    argv[0]);
+        }
+    }
+
+    if (daemonize) {
+        g_debug("starting daemon");
+        become_daemon();
+    }
+
+    qga_init_marshal();
+    
+    s = g_malloc(sizeof(GAState));
+    s->active = false;
+    s->session_id = 0;
+    s->conn_id = 0;
+    s->conn_channel = NULL;
+    s->path = path;
+    s->method = method;
+
+    init_guest_agent(s);
+    g_main_loop_run(s->main_loop);
+
+    return 0;
+}
-- 
1.7.0.4