commit 330600b4cafd37f51d109728f9377bc921a6aa7f

Author: Andrew Beekhof <andrew@beekhof.net>

Date:   Mon Oct 14 13:50:41 2013 +1100

    Feature: remote: Properly version the remote connection protocol

    (cherry picked from commit aa024490a023bd0d0d5568f08b2c60b61adb15f5)

diff --git a/include/crm/common/ipcs.h b/include/crm/common/ipcs.h

index 3e78d51..84249a3 100644

--- a/include/crm/common/ipcs.h

+++ b/include/crm/common/ipcs.h

@@ -41,14 +41,17 @@ enum client_type {

 struct crm_remote_s {

     /* Shared */

     char *buffer;

+    size_t buffer_size;

+    size_t buffer_offset;

     int auth_timeout;

-    bool authenticated;         /* CIB-only */

+    int tcp_socket;

     mainloop_io_t *source;

-    char *token;                /* CIB Only */

-

-    int tcp_socket;

+    /* CIB-only */

+    bool authenticated;

+    char *token;

+    /* TLS only */

 #  ifdef HAVE_GNUTLS_GNUTLS_H

     gnutls_session_t *tls_session;

     bool tls_handshake_complete;

diff --git a/lib/common/remote.c b/lib/common/remote.c

index 8b00f16..297c75a 100644

--- a/lib/common/remote.c

+++ b/lib/common/remote.c

@@ -34,6 +34,8 @@

 #include <fcntl.h>

 #include <glib.h>

+#include <bzlib.h>

+

 #include <crm/common/ipcs.h>

 #include <crm/common/xml.h>

 #include <crm/common/mainloop.h>

@@ -57,6 +59,17 @@ const int anon_tls_kx_order[] = {

     0

 };

+struct crm_remote_header_v0 

+{

+    uint64_t id;

+    uint64_t flags;

+    uint32_t error;

+    uint32_t version;

+    uint32_t size_total;

+    uint32_t payload_uncompressed;

+    uint32_t payload_compressed;

+};

+

 int

 crm_initiate_client_tls_handshake(crm_remote_t * remote, int timeout_ms)

 {

@@ -176,97 +189,6 @@ crm_send_tls(gnutls_session_t * session, const char *buf, size_t len)

     return rc < 0 ? rc : total_send;

 }

-

-/*!

- * \internal

- * \brief Read bytes off non blocking tls session.

- *

- * \param session - tls session to read

- * \param max_size - max bytes allowed to read for buffer. 0 assumes no limit

- *

- * \note only use with NON-Blocking sockets. Should only be used after polling socket.

- *       This function will return once max_size is met, the socket read buffer

- *       is empty, or an error is encountered.

- *

- * \retval '\0' terminated buffer on success

- */

-static char *

-crm_recv_tls(gnutls_session_t * session, size_t max_size, size_t * recv_len, int *disconnected)

-{

-    char *buf = NULL;

-    int rc = 0;

-    size_t len = 0;

-    size_t chunk_size = max_size ? max_size : 1024;

-    size_t buf_size = 0;

-    size_t read_size = 0;

-

-    if (session == NULL) {

-        if (disconnected) {

-            *disconnected = 1;

-        }

-        goto done;

-    }

-

-    buf = calloc(1, chunk_size + 1);

-    buf_size = chunk_size;

-

-    while (TRUE) {

-        read_size = buf_size - len;

-

-        /* automatically grow the buffer when needed if max_size is not set. */

-        if (!max_size && (read_size < (chunk_size / 2))) {

-            buf_size += chunk_size;

-            crm_trace("Grow buffer by %d more bytes. buf is now %d bytes", (int)chunk_size,

-                      buf_size);

-            buf = realloc(buf, buf_size + 1);

-            CRM_ASSERT(buf != NULL);

-

-            read_size = buf_size - len;

-        }

-

-        rc = gnutls_record_recv(*session, buf + len, read_size);

-

-        if (rc > 0) {

-            crm_trace("Got %d more bytes.", rc);

-            len += rc;

-            /* always null terminate buffer, the +1 to alloc always allows for this. */

-            buf[len] = '\0';

-        }

-        if (max_size && (max_size == read_size)) {

-            crm_trace("Buffer max read size %d met", max_size);

-            goto done;

-        }

-

-        /* process any errors. */

-        if (rc == GNUTLS_E_INTERRUPTED) {

-            crm_trace("EINTR encoutered, retry tls read");

-        } else if (rc == GNUTLS_E_AGAIN) {

-            crm_trace("non-blocking, exiting read on rc = %d", rc);

-            goto done;

-        } else if (rc <= 0) {

-            if (rc == 0) {

-                crm_debug("EOF encoutered during TLS read");

-            } else {

-                crm_debug("Error receiving message: %s (%d)", gnutls_strerror(rc), rc);

-            }

-            if (disconnected) {

-                *disconnected = 1;

-            }

-            goto done;

-        }

-    }

-

-  done:

-    if (recv_len) {

-        *recv_len = len;

-    }

-    if (!len) {

-        free(buf);

-        buf = NULL;

-    }

-    return buf;

-

-}

 #endif

 static int

@@ -310,141 +232,57 @@ crm_send_plaintext(int sock, const char *buf, size_t len)

 }

-/*!

- * \internal

- * \brief Read bytes off non blocking socket.

- *

- * \param session - tls session to read

- * \param max_size - max bytes allowed to read for buffer. 0 assumes no limit

- *

- * \note only use with NON-Blocking sockets. Should only be used after polling socket.

- *       This function will return once max_size is met, the socket read buffer

- *       is empty, or an error is encountered.

- *

- * \retval '\0' terminated buffer on success

- */

-static char *

-crm_recv_plaintext(int sock, size_t max_size, size_t * recv_len, int *disconnected)

-{

-    char *buf = NULL;

-    ssize_t rc = 0;

-    ssize_t len = 0;

-    ssize_t chunk_size = max_size ? max_size : 1024;

-    size_t buf_size = 0;

-    size_t read_size = 0;

-

-    if (sock <= 0) {

-        if (disconnected) {

-            *disconnected = 1;

-        }

-        goto done;

-    }

-

-    buf = calloc(1, chunk_size + 1);

-    buf_size = chunk_size;

-

-    while (TRUE) {

-        errno = 0;

-        read_size = buf_size - len;

-

-        /* automatically grow the buffer when needed if max_size is not set. */

-        if (!max_size && (read_size < (chunk_size / 2))) {

-            buf_size += chunk_size;

-            crm_trace("Grow buffer by %d more bytes. buf is now %d bytes", (int)chunk_size,

-                      buf_size);

-            buf = realloc(buf, buf_size + 1);

-            CRM_ASSERT(buf != NULL);

-

-            read_size = buf_size - len;

-        }

-

-        rc = read(sock, buf + len, chunk_size);

-

-        if (rc > 0) {

-            crm_trace("Got %d more bytes. errno=%d", (int)rc, errno);

-            len += rc;

-            /* always null terminate buffer, the +1 to alloc always allows for this. */

-            buf[len] = '\0';

-        }

-        if (max_size && (max_size == read_size)) {

-            crm_trace("Buffer max read size %d met", max_size);

-            goto done;

-        }

-

-        if (rc > 0) {

-            continue;

-        } else if (rc == 0) {

-            if (disconnected) {

-                *disconnected = 1;

-            }

-            crm_trace("EOF encoutered during read");

-            goto done;

-        }

-

-        /* process errors */

-        if (errno == EINTR) {

-            crm_trace("EINTER encoutered, retry socket read.");

-        } else if (errno == EAGAIN) {

-            crm_trace("non-blocking, exiting read on rc = %d", rc);

-            goto done;

-        } else if (errno <= 0) {

-            if (disconnected) {

-                *disconnected = 1;

-            }

-            crm_debug("Error receiving message: %d", (int)rc);

-            goto done;

-        }

-    }

-

-  done:

-    if (recv_len) {

-        *recv_len = len;

-    }

-    if (!len) {

-        free(buf);

-        buf = NULL;

-    }

-    return buf;

-}

-

 static int

-crm_remote_send_raw(crm_remote_t * remote, const char *buf, size_t len)

+crm_remote_sendv(crm_remote_t * remote, struct iovec * iov, int iovs)

 {

+    int lpc = 0;

     int rc = -ESOCKTNOSUPPORT;

-    if (remote->tcp_socket) {

-        rc = crm_send_plaintext(remote->tcp_socket, buf, len);

+    for(; lpc < iovs; lpc++) {

+        if (remote->tcp_socket) {

+            rc = crm_send_plaintext(remote->tcp_socket, iov[lpc].iov_base, iov[lpc].iov_len);

 #ifdef HAVE_GNUTLS_GNUTLS_H

-    } else if (remote->tls_session) {

-        rc = crm_send_tls(remote->tls_session, buf, len);

+        } else if (remote->tls_session) {

+            rc = crm_send_tls(remote->tls_session, iov[lpc].iov_base, iov[lpc].iov_len);

 #endif

-    } else {

-        crm_err("Unsupported connection type");

+        } else {

+            crm_err("Unsupported connection type");

+        }

     }

     return rc;

 }

+#define PCMK_TLS_VERSION 1

+

 int

 crm_remote_send(crm_remote_t * remote, xmlNode * msg)

 {

+    static uint64_t id = 0;

     int rc = -1;

-    char *xml_text = NULL;

-    int len = 0;

+    char *xml_text = dump_xml_unformatted(msg);

-    xml_text = dump_xml_unformatted(msg);

-    if (xml_text) {

-        len = strlen(xml_text);

-    } else {

+    struct iovec iov[2];

+    struct crm_remote_header_v0 *header = calloc(1, sizeof(struct crm_remote_header_v0));

+

+    if (xml_text == NULL) {

         crm_err("Invalid XML, can not send msg");

         return -1;

     }

-    rc = crm_remote_send_raw(remote, xml_text, len);

-    if (rc >= 0) {

-        rc = crm_remote_send_raw(remote, REMOTE_MSG_TERMINATOR, strlen(REMOTE_MSG_TERMINATOR));

-    }

+    iov[0].iov_base = header;

+    iov[0].iov_len = sizeof(struct crm_remote_header_v0);

+

+    iov[1].iov_base = xml_text;

+    iov[1].iov_len = 1 + strlen(xml_text);

+    id++;

+    header->id = id;

+    header->version = PCMK_TLS_VERSION;

+    header->size_total = iov[0].iov_len + iov[1].iov_len;

+    header->payload_uncompressed = iov[1].iov_len;

+

+    rc = crm_remote_sendv(remote, iov, 2);

     if (rc < 0) {

         crm_err("Failed to send remote msg, rc = %d", rc);

     }

@@ -461,44 +299,55 @@ crm_remote_send(crm_remote_t * remote, xmlNode * msg)

 xmlNode *

 crm_remote_parse_buffer(crm_remote_t * remote)

 {

-    char *buf = NULL;

-    char *start = NULL;

-    char *end = NULL;

     xmlNode *xml = NULL;

+    size_t offset = sizeof(struct crm_remote_header_v0);

+    struct crm_remote_header_v0 *header = NULL;

     if (remote->buffer == NULL) {

         return NULL;

+

+    } else if(remote->buffer_offset < sizeof(struct crm_remote_header_v0)) {

+        return NULL;

     }

     /* take ownership of the buffer */

-    buf = remote->buffer;

-    remote->buffer = NULL;

-

-    /* MSGS are separated by a '\r\n\r\n'. Split a message off the buffer and return it. */

-    start = buf;

-    end = strstr(start, REMOTE_MSG_TERMINATOR);

+    remote->buffer_offset = 0;

+

+    header = (struct crm_remote_header_v0 *)remote->buffer;

+    /* Support compression on the receiving end now, in case we ever want to add it later */

+    if (header->payload_compressed) {

+        int rc = 0;

+        unsigned int size_u = 1 + header->payload_uncompressed;

+        char *uncompressed = calloc(1, offset + size_u);

+

+        crm_trace("Decompressing message data %d bytes into %d bytes",

+                 header->payload_compressed, size_u);

+

+        rc = BZ2_bzBuffToBuffDecompress(uncompressed + offset, &size_u,

+                                        remote->buffer + offset,

+                                        header->payload_compressed, 1, 0);

+

+        if (rc != BZ_OK) {

+            crm_err("Decompression failed: %s (%d)", bz2_strerror(rc), rc);

+            free(uncompressed);

+            return NULL;

+        }

-    while (!xml && end) {

+        CRM_ASSERT(size_u == header->payload_uncompressed);

-        /* grab the message */

-        end[0] = '\0';

-        end += strlen(REMOTE_MSG_TERMINATOR);

+        memcpy(uncompressed, remote->buffer, offset);       /* Preserve the header */

+        header = (struct crm_remote_header_v0 *)uncompressed;

-        xml = string2xml(start);

-        if (xml == NULL) {

-            crm_err("Couldn't parse: '%.120s'", start);

-        }

-        start = end;

-        end = strstr(start, REMOTE_MSG_TERMINATOR);

+        free(remote->buffer);

+        remote->buffer_size = offset + size_u;

+        remote->buffer = uncompressed;

     }

-    if (xml && start) {

-        /* we have msgs left over, save it until next time */

-        remote->buffer = strdup(start);

-        free(buf);

-    } else if (!xml) {

-        /* no msg present */

-        remote->buffer = buf;

+    CRM_ASSERT(remote->buffer[sizeof(struct crm_remote_header_v0) + header->payload_uncompressed - 1] == 0);

+

+    xml = string2xml(remote->buffer + offset);

+    if (xml == NULL) {

+        crm_err("Couldn't parse: '%.120s'", remote->buffer + offset);

     }

     return xml;

@@ -559,6 +408,105 @@ crm_remote_ready(crm_remote_t * remote, int timeout /* ms */ )

     return rc;

 }

+

+/*!

+ * \internal

+ * \brief Read bytes off non blocking remote connection.

+ *

+ * \note only use with NON-Blocking sockets. Should only be used after polling socket.

+ *       This function will return once max_size is met, the socket read buffer

+ *       is empty, or an error is encountered.

+ *

+ * \retval number of bytes received

+ */

+static size_t

+crm_remote_recv_once(crm_remote_t * remote)

+{

+    int rc = 0;

+    size_t read_len = sizeof(struct crm_remote_header_v0);

+

+    if(remote->buffer_offset >= sizeof(struct crm_remote_header_v0)) {

+        struct crm_remote_header_v0 *hdr = (struct crm_remote_header_v0 *)remote->buffer;

+

+        read_len = hdr->size_total;

+    }

+

+    /* automatically grow the buffer when needed */

+    if(remote->buffer_size < read_len) {

+        remote->buffer_size = 2 * read_len;

+        crm_trace("Expanding buffer to %u bytes", remote->buffer_size);

+

+        remote->buffer = realloc(remote->buffer, remote->buffer_size + 1);

+        CRM_ASSERT(remote->buffer != NULL);

+    }

+

+    if (remote->tcp_socket) {

+        errno = 0;

+        rc = read(remote->tcp_socket,

+                  remote->buffer + remote->buffer_offset,

+                  remote->buffer_size - remote->buffer_offset);

+        if(rc < 0) {

+            rc = -errno;

+        }

+

+#ifdef HAVE_GNUTLS_GNUTLS_H

+    } else if (remote->tls_session) {

+        rc = gnutls_record_recv(*(remote->tls_session),

+                                remote->buffer + remote->buffer_offset,

+                                remote->buffer_size - remote->buffer_offset);

+        if (rc == GNUTLS_E_INTERRUPTED) {

+            rc = -EINTR;

+        } else if (rc == GNUTLS_E_AGAIN) {

+            rc = -EAGAIN;

+        } else if (rc < 0) {

+            crm_info("TLS receive failed: %s (%d)", gnutls_strerror(rc), rc);

+            rc = -pcmk_err_generic;

+        }

+#endif

+    } else {

+        crm_err("Unsupported connection type");

+        return -ESOCKTNOSUPPORT;

+    }

+

+    /* process any errors. */

+    if (rc > 0) {

+        remote->buffer_offset += rc;

+        /* always null terminate buffer, the +1 to alloc always allows for this. */

+        remote->buffer[remote->buffer_offset] = '\0';

+        crm_trace("Received %u more bytes, %u total", rc, remote->buffer_offset);

+

+    } else if (rc == -EINTR || rc == -EAGAIN) {

+        crm_trace("non-blocking, exiting read: %s (%d)", pcmk_strerror(rc), rc);

+

+    } else if (rc == 0) {

+        crm_debug("EOF encoutered after %u bytes", remote->buffer_offset);

+        return -ENOTCONN;

+

+    } else if (rc <= 0) {

+        crm_debug("Error receiving message after %u bytes: %s (%d)",

+                  remote->buffer_offset, gnutls_strerror(rc), rc);

+        return -ENOTCONN;

+    }

+

+    if(remote->buffer_offset < sizeof(struct crm_remote_header_v0)) {

+        crm_trace("Not enough data to fill header: %u < %u bytes",

+                  remote->buffer_offset, sizeof(struct crm_remote_header_v0));

+        return -EAGAIN;

+

+    } else {

+        struct crm_remote_header_v0 *hdr = (struct crm_remote_header_v0 *)remote->buffer;

+

+        if(remote->buffer_offset < hdr->size_total) {

+            crm_trace("Read less than the advertised length: %u < %u bytes",

+                      remote->buffer_offset, hdr->size_total);

+            return -EAGAIN;

+        }

+    }

+

+    crm_trace("Read full message of %u bytes", remote->buffer_offset);

+    return remote->buffer_offset;

+}

+

 /*!

  * \internal

  * \brief Read data off the socket until at least one full message is present or timeout occures.

@@ -569,10 +517,8 @@ crm_remote_ready(crm_remote_t * remote, int timeout /* ms */ )

 gboolean

 crm_remote_recv(crm_remote_t * remote, int total_timeout /*ms */ , int *disconnected)

 {

-    int ret;

-    size_t request_len = 0;

+    int rc;

     time_t start = time(NULL);

-    char *raw_request = NULL;

     int remaining_timeout = 0;

     if (total_timeout == 0) {

@@ -588,57 +534,30 @@ crm_remote_recv(crm_remote_t * remote, int total_timeout /*ms */ , int *disconne

         /* read some more off the tls buffer if we still have time left. */

         crm_trace("waiting to receive remote msg, starting timeout %d, remaining_timeout %d",

                   total_timeout, remaining_timeout);

-        ret = crm_remote_ready(remote, remaining_timeout);

-        raw_request = NULL;

+        rc = crm_remote_ready(remote, remaining_timeout);

-        if (ret == 0) {

+        if (rc == 0) {

             crm_err("poll timed out (%d ms) while waiting to receive msg", remaining_timeout);

             return FALSE;

-        } else if (ret < 0) {

-            if (errno != EINTR) {

-                crm_debug("poll returned error while waiting for msg, rc: %d, errno: %d", ret,

-                          errno);

-                *disconnected = 1;

-                return FALSE;

-            }

-            crm_debug("poll EINTR encountered during poll, retrying");

-

-        } else if (remote->tcp_socket) {

-            raw_request = crm_recv_plaintext(remote->tcp_socket, 0, &request_len, disconnected);

+        } else if(rc < 0) {

+            crm_debug("poll() failed: %s (%d)", pcmk_strerror(rc), rc);

-#ifdef HAVE_GNUTLS_GNUTLS_H

-        } else if (remote->tls_session) {

-            raw_request = crm_recv_tls(remote->tls_session, 0, &request_len, disconnected);

-#endif

         } else {

-            crm_err("Unsupported connection type");

-        }

-

-        remaining_timeout = remaining_timeout - ((time(NULL) - start) * 1000);

-

-        if (!raw_request) {

-            crm_debug("Empty msg received after poll");

-            continue;

+            rc = crm_remote_recv_once(remote);

+            if(rc > 0) {

+                return TRUE;

+            } else if (rc < 0) {

+                crm_debug("recv() failed: %s (%d)", pcmk_strerror(rc), rc);

+            }

         }

-        if (remote->buffer) {

-            int old_len = strlen(remote->buffer);

-

-            crm_trace("Expanding recv buffer from %d to %d", old_len, old_len + request_len);

-

-            remote->buffer = realloc(remote->buffer, old_len + request_len + 1);

-            memcpy(remote->buffer + old_len, raw_request, request_len);

-            *(remote->buffer + old_len + request_len) = '\0';

-            free(raw_request);

-

-        } else {

-            remote->buffer = raw_request;

+        if(rc == -ENOTCONN) {

+            *disconnected = 1;

+            return FALSE;

         }

-        if (strstr(remote->buffer, REMOTE_MSG_TERMINATOR)) {

-            return TRUE;

-        }

+        remaining_timeout = remaining_timeout - ((time(NULL) - start) * 1000);

     }

     return FALSE;