Adding basic error handling to SSL send/recv funcs

For recv/send() functions, the functions themselves return a negative
number on error, then the amqp_ssl_socket_error() can be used to return
an opaque (e.g., not very useful to API users) error code. Internally we
store this error in the last_error field of the amqp_ssl_socket_t
object.

5 files changed, 109 insertions, 18 deletions

diff --git a/librabbitmq/amqp.h b/librabbitmq/amqp.h
index 5167f2e..36c1411 100644
--- a/librabbitmq/amqp.h
+++ b/librabbitmq/amqp.h
@@ -641,7 +641,7 @@ amqp_socket_close(amqp_socket_t *self);
  *
  * \param [in,out] self A socket object.
  *
- * \return Zero upon success, non-zero otherwise.
+ * \return Zero upon success, an opaque error code otherwise
  */
 AMQP_PUBLIC_FUNCTION
 int
diff --git a/librabbitmq/amqp_cyassl.c b/librabbitmq/amqp_cyassl.c
index b04c657..2fa83e6 100644
--- a/librabbitmq/amqp_cyassl.c
+++ b/librabbitmq/amqp_cyassl.c
@@ -37,6 +37,7 @@ struct amqp_ssl_socket_t {
   int sockfd;
   char *buffer;
   size_t length;
+  int last_error;
 };
 
 static ssize_t
@@ -45,8 +46,16 @@ amqp_ssl_socket_send(void *base,
                      size_t len,
                      AMQP_UNUSED int flags)
 {
+  int status;
   struct amqp_ssl_socket_t *self = (struct amqp_ssl_socket_t *)base;
-  return CyaSSL_write(self->ssl, buf, len);
+
+  self->last_error = 0;
+  status = CyaSSL_write(self->ssl, buf, len);
+  if (status <= 0) {
+    self->last_error = ERROR_CATEGORY_SSL;
+  }
+
+  return status;
 }
 
 static ssize_t
@@ -59,6 +68,7 @@ amqp_ssl_socket_writev(void *base,
   char *bufferp;
   size_t bytes;
   int i;
+  self->last_error = 0;
   bytes = 0;
   for (i = 0; i < iovcnt; ++i) {
     bytes += iov[i].iov_len;
@@ -68,6 +78,7 @@ amqp_ssl_socket_writev(void *base,
     self->buffer = malloc(bytes);
     if (!self->buffer) {
       self->length = 0;
+      self->last_error = ERROR_NO_MEMORY;
       goto exit;
     }
     self->length = bytes;
@@ -77,7 +88,7 @@ amqp_ssl_socket_writev(void *base,
     memcpy(bufferp, iov[i].iov_base, iov[i].iov_len);
     bufferp += iov[i].iov_len;
   }
-  written = CyaSSL_write(self->ssl, self->buffer, bytes);
+  written = amqp_ssl_socket_send(self, self->buffer, bytes, 0);
 exit:
   return written;
 }
@@ -88,8 +99,16 @@ amqp_ssl_socket_recv(void *base,
                      size_t len,
                      AMQP_UNUSED int flags)
 {
+  int status;
   struct amqp_ssl_socket_t *self = (struct amqp_ssl_socket_t *)base;
-  return CyaSSL_read(self->ssl, buf, len);
+
+  self->last_error = 0;
+  status = CyaSSL_read(self->ssl, buf, len);
+  if (status <= 0) {
+    self->last_error = ERROR_CATEGORY_SSL;
+  }
+
+  return status;
 }
 
 static int
@@ -117,9 +136,10 @@ amqp_ssl_socket_close(void *base)
 }
 
 static int
-amqp_ssl_socket_error(AMQP_UNUSED void *user_data)
+amqp_ssl_socket_error(void *base)
 {
-  return ERROR_CATEGORY_SSL;
+  struct amqp_ssl_socket_t *self = (struct amqp_ssl_socket_t *)base;
+  return self->last_error;
 }
 
 char *
@@ -133,13 +153,16 @@ amqp_ssl_socket_open(void *base, const char *host, int port)
 {
   struct amqp_ssl_socket_t *self = (struct amqp_ssl_socket_t *)base;
   int status;
+  self->last_error = 0;
   self->sockfd = amqp_open_socket(host, port);
   if (0 > self->sockfd) {
+    self->last_error = - self->sockfd;
     return -1;
   }
   CyaSSL_set_fd(self->ssl, self->sockfd);
   status = CyaSSL_connect(self->ssl);
   if (SSL_SUCCESS != status) {
+    self->last_error = ERROR_CATEGORY_SSL;
     return -1;
   }
   return 0;
diff --git a/librabbitmq/amqp_gnutls.c b/librabbitmq/amqp_gnutls.c
index 8e5e1f2..fdde6d8 100644
--- a/librabbitmq/amqp_gnutls.c
+++ b/librabbitmq/amqp_gnutls.c
@@ -39,6 +39,7 @@ struct amqp_ssl_socket_t {
   char *host;
   char *buffer;
   size_t length;
+  int last_error;
 };
 
 static ssize_t
@@ -47,8 +48,15 @@ amqp_ssl_socket_send(void *base,
                      size_t len,
                      AMQP_UNUSED int flags)
 {
+  ssize_t status;
   struct amqp_ssl_socket_t *self = (struct amqp_ssl_socket_t *)base;
-  return gnutls_record_send(self->session, buf, len);
+
+  self->last_error = 0;
+  status = gnutls_record_send(self->session, buf, len);
+  if (status < 0) {
+    self->last_error = ERROR_CATEGORY_SSL;
+  }
+  return status;
 }
 
 static ssize_t
@@ -61,6 +69,7 @@ amqp_ssl_socket_writev(void *base,
   char *bufferp;
   size_t bytes;
   int i;
+  self->last_error = 0;
   bytes = 0;
   for (i = 0; i < iovcnt; ++i) {
     bytes += iov[i].iov_len;
@@ -70,6 +79,7 @@ amqp_ssl_socket_writev(void *base,
     self->buffer = malloc(bytes);
     if (!self->buffer) {
       self->length = 0;
+      self->last_error = ERROR_NO_MEMORY;
       goto exit;
     }
     self->length = 0;
@@ -79,7 +89,7 @@ amqp_ssl_socket_writev(void *base,
     memcpy(bufferp, iov[i].iov_base, iov[i].iov_len);
     bufferp += iov[i].iov_len;
   }
-  written = gnutls_record_send(self->session, self->buffer, bytes);
+  written = amqp_ssl_socket_send(self, self->buffer, bytes, 0);
 exit:
   return written;
 }
@@ -90,8 +100,16 @@ amqp_ssl_socket_recv(void *base,
                      size_t len,
                      AMQP_UNUSED int flags)
 {
+  ssize_t status;
   struct amqp_ssl_socket_t *self = (struct amqp_ssl_socket_t *)base;
-  return gnutls_record_recv(self->session, buf, len);
+
+  self->last_error = 0;
+  status = gnutls_record_recv(self->session, buf, len);
+  if (status < 0) {
+    self->last_error = ERROR_CATEGORY_SSL;
+  }
+
+  return status;
 }
 
 static int
@@ -99,8 +117,10 @@ amqp_ssl_socket_open(void *base, const char *host, int port)
 {
   struct amqp_ssl_socket_t *self = (struct amqp_ssl_socket_t *)base;
   int status;
+  self->last_error = 0;
   self->sockfd = amqp_open_socket(host, port);
   if (0 > self->sockfd) {
+    self->last_error = -self->sockfd;
     return -1;
   }
   gnutls_transport_set_ptr(self->session,
@@ -108,6 +128,11 @@ amqp_ssl_socket_open(void *base, const char *host, int port)
   do {
     status = gnutls_handshake(self->session);
   } while (status < 0 && !gnutls_error_is_fatal(status));
+
+  if (gnutls_error_is_fatal(status)) {
+    self->last_error = ERROR_CATEGORY_SSL;
+  }
+
   return status;
 }
 
@@ -130,9 +155,10 @@ amqp_ssl_socket_close(void *base)
 }
 
 static int
-amqp_ssl_socket_error(AMQP_UNUSED void *user_data)
+amqp_ssl_socket_error(void *base)
 {
-  return ERROR_CATEGORY_SSL;
+  struct amqp_ssl_socket_t *self = (struct amqp_ssl_socket_t *)base;
+  return self->last_error;
 }
 
 char *
diff --git a/librabbitmq/amqp_openssl.c b/librabbitmq/amqp_openssl.c
index 5203756..15fa6bd 100644
--- a/librabbitmq/amqp_openssl.c
+++ b/librabbitmq/amqp_openssl.c
@@ -65,6 +65,7 @@ struct amqp_ssl_socket_t {
   char *buffer;
   size_t length;
   amqp_boolean_t verify;
+  int last_error;
 };
 
 static ssize_t
@@ -76,8 +77,10 @@ amqp_ssl_socket_send(void *base,
   struct amqp_ssl_socket_t *self = (struct amqp_ssl_socket_t *)base;
   ssize_t sent;
   ERR_clear_error();
+  self->last_error = 0;
   sent = SSL_write(self->ssl, buf, len);
   if (0 > sent) {
+    self->last_error = ERROR_CATEGORY_SSL;
     switch (SSL_get_error(self->ssl, sent)) {
     case SSL_ERROR_NONE:
     case SSL_ERROR_ZERO_RETURN:
@@ -100,6 +103,7 @@ amqp_ssl_socket_writev(void *base,
   char *bufferp;
   size_t bytes;
   int i;
+  self->last_error = 0;
   bytes = 0;
   for (i = 0; i < iovcnt; ++i) {
     bytes += iov[i].iov_len;
@@ -109,6 +113,7 @@ amqp_ssl_socket_writev(void *base,
     self->buffer = malloc(bytes);
     if (!self->buffer) {
       self->length = 0;
+      self->last_error = ERROR_NO_MEMORY;
       goto exit;
     }
     self->length = bytes;
@@ -132,8 +137,10 @@ amqp_ssl_socket_recv(void *base,
   struct amqp_ssl_socket_t *self = (struct amqp_ssl_socket_t *)base;
   ssize_t received;
   ERR_clear_error();
+  self->last_error = 0;
   received = SSL_read(self->ssl, buf, len);
   if (0 > received) {
+    self->last_error = ERROR_CATEGORY_SSL;
     switch(SSL_get_error(self->ssl, received)) {
     case SSL_ERROR_WANT_READ:
     case SSL_ERROR_WANT_WRITE:
@@ -215,30 +222,37 @@ amqp_ssl_socket_open(void *base, const char *host, int port)
   struct amqp_ssl_socket_t *self = (struct amqp_ssl_socket_t *)base;
   long result;
   int status;
+  self->last_error = 0;
   self->ssl = SSL_new(self->ctx);
   if (!self->ssl) {
+    self->last_error = ERROR_CATEGORY_SSL;
     return -1;
   }
   SSL_set_mode(self->ssl, SSL_MODE_AUTO_RETRY);
   self->sockfd = amqp_open_socket(host, port);
   if (0 > self->sockfd) {
+    self->last_error = -self->sockfd;
     return -1;
   }
   status = SSL_set_fd(self->ssl, self->sockfd);
   if (!status) {
+    self->last_error = ERROR_CATEGORY_SSL;
     return -1;
   }
   status = SSL_connect(self->ssl);
   if (!status) {
+    self->last_error = ERROR_CATEGORY_SSL;
     return -1;
   }
   result = SSL_get_verify_result(self->ssl);
   if (X509_V_OK != result) {
+    self->last_error = ERROR_CATEGORY_SSL;
     return -1;
   }
   if (self->verify) {
     int status = amqp_ssl_socket_verify(self, host);
     if (status) {
+      self->last_error = ERROR_CATEGORY_SSL;
       return -1;
     }
   }
@@ -261,9 +275,10 @@ amqp_ssl_socket_close(void *base)
 }
 
 static int
-amqp_ssl_socket_error(AMQP_UNUSED void *base)
+amqp_ssl_socket_error(void *base)
 {
-  return ERROR_CATEGORY_SSL;
+  struct amqp_ssl_socket_t *self = (struct amqp_ssl_socket_t *)base;
+  return self->last_error;
 }
 
 char *
diff --git a/librabbitmq/amqp_polarssl.c b/librabbitmq/amqp_polarssl.c
index b355ddb..da3d6a2 100644
--- a/librabbitmq/amqp_polarssl.c
+++ b/librabbitmq/amqp_polarssl.c
@@ -45,6 +45,7 @@ struct amqp_ssl_socket_t {
   ssl_session *session;
   char *buffer;
   size_t length;
+  int last_error;
 };
 
 static ssize_t
@@ -53,8 +54,16 @@ amqp_ssl_socket_send(void *base,
                      size_t len,
                      AMQP_UNUSED int flags)
 {
+  ssize_t status;
   struct amqp_ssl_socket_t *self = (struct amqp_ssl_socket_t *)base;
-  return ssl_write(self->ssl, buf, len);
+
+  self->last_error = 0;
+  status = ssl_write(self->ssl, buf, len);
+  if (status < 0) {
+    self->last_error = ERROR_CATEGORY_SSL;
+  }
+
+  return status;
 }
 
 static ssize_t
@@ -67,6 +76,7 @@ amqp_ssl_socket_writev(void *base,
   char *bufferp;
   size_t bytes;
   int i;
+  self->last_error = 0;
   bytes = 0;
   for (i = 0; i < iovcnt; ++i) {
     bytes += iov[i].iov_len;
@@ -76,6 +86,7 @@ amqp_ssl_socket_writev(void *base,
     self->buffer = malloc(bytes);
     if (!self->buffer) {
       self->length = 0;
+      self->last_error = ERROR_NO_MEMORY;
       goto exit;
     }
     self->length = bytes;
@@ -85,8 +96,8 @@ amqp_ssl_socket_writev(void *base,
     memcpy(bufferp, iov[i].iov_base, iov[i].iov_len);
     bufferp += iov[i].iov_len;
   }
-  written = ssl_write(self->ssl, (const unsigned char *)self->buffer,
-                      bytes);
+  written = amqp_ssl_socket_send(self, (const unsigned char *)self->buffer,
+                      bytes, 0);
 exit:
   return written;
 }
@@ -97,16 +108,31 @@ amqp_ssl_socket_recv(void *base,
                      size_t len,
                      AMQP_UNUSED int flags)
 {
+  ssize_t status;
   struct amqp_ssl_socket_t *self = (struct amqp_ssl_socket_t *)base;
-  return ssl_read(self->ssl, buf, len);
+
+  self->last_error = 0;
+  status = ssl_read(self->ssl, buf, len);
+  if (status < 0) {
+    self->last_error = ERROR_CATEGORY_SSL;
+  }
+
+  return status;
 }
 
 static int
 amqp_ssl_socket_open(void *base, const char *host, int port)
 {
+  int status;
   struct amqp_ssl_socket_t *self = (struct amqp_ssl_socket_t *)base;
-  int status = net_connect(&self->sockfd, host, port);
+  self->last_error = 0;
+
+  status = net_connect(&self->sockfd, host, port);
   if (status) {
+    /* This isn't quite right. We should probably translate between
+     * POLARSSL_ERR_* to our internal error codes
+     */
+    self->last_error = ERROR_CATEGORY_SSL;
     return -1;
   }
   if (self->cacert) {
@@ -123,6 +149,7 @@ amqp_ssl_socket_open(void *base, const char *host, int port)
     case POLARSSL_ERR_NET_WANT_WRITE:
       continue;
     default:
+      self->last_error = ERROR_CATEGORY_SSL;
       break;
     }
   }