ssl: Add OpenSSL BIO that passes MSG_NOSIGNALssl_bio

Add an OpenSSL BIO that ignores SIGPIPE by passing MSG_NOSIGNAL to the
send() and recv() calls on platforms that support it.

Fixes #401

5 files changed, 209 insertions, 4 deletions

diff --git a/Makefile.am b/Makefile.am
index 307f261..df4f8ef 100644
--- a/Makefile.am
+++ b/Makefile.am
@@ -45,6 +45,8 @@ librabbitmq_librabbitmq_la_SOURCES += \
   librabbitmq/amqp_hostcheck.c \
   librabbitmq/amqp_hostcheck.h \
   librabbitmq/amqp_openssl.c \
+  librabbitmq/amqp_openssl_bio.c \
+  librabbitmq/amqp_openssl_bio.h \
   librabbitmq/amqp_openssl_hostname_validation.c \
   librabbitmq/amqp_openssl_hostname_validation.h
 if OS_APPLE
diff --git a/librabbitmq/CMakeLists.txt b/librabbitmq/CMakeLists.txt
index a00ee66..a09f164 100644
--- a/librabbitmq/CMakeLists.txt
+++ b/librabbitmq/CMakeLists.txt
@@ -86,6 +86,8 @@ if (ENABLE_SSL_SUPPORT)
       amqp_openssl_hostname_validation.h
       amqp_hostcheck.c
       amqp_hostcheck.h
+      amqp_openssl_bio.c
+      amqp_openssl_bio.h
   )
   include_directories(${OPENSSL_INCLUDE_DIR})
   set(AMQP_SSL_LIBS ${OPENSSL_LIBRARIES})
diff --git a/librabbitmq/amqp_openssl.c b/librabbitmq/amqp_openssl.c
index 68808bb..333c7ed 100644
--- a/librabbitmq/amqp_openssl.c
+++ b/librabbitmq/amqp_openssl.c
@@ -29,6 +29,7 @@
 #include "config.h"
 #endif
 
+#include "amqp_openssl_bio.h"
 #include "amqp_openssl_hostname_validation.h"
 #include "amqp_ssl_socket.h"
 #include "amqp_socket.h"
@@ -38,6 +39,7 @@
 
 #include <ctype.h>
 #include <limits.h>
+#include <openssl/bio.h>
 #include <openssl/conf.h>
 #include <openssl/err.h>
 #include <openssl/ssl.h>
@@ -172,6 +174,7 @@ amqp_ssl_socket_open(void *base, const char *host, int port, struct timeval *tim
   int status;
   amqp_time_t deadline;
   X509 *cert;
+  BIO *bio;
   if (-1 != self->sockfd) {
     return AMQP_STATUS_SOCKET_INUSE;
   }
@@ -197,13 +200,15 @@ amqp_ssl_socket_open(void *base, const char *host, int port, struct timeval *tim
     goto error_out1;
   }
 
-  status = SSL_set_fd(self->ssl, self->sockfd);
-  if (!status) {
-    self->internal_error = SSL_get_error(self->ssl, status);
-    status = AMQP_STATUS_SSL_ERROR;
+  bio = BIO_new(amqp_openssl_bio());
+  if (!bio) {
+    status = AMQP_STATUS_NO_MEMORY;
     goto error_out2;
   }
 
+  BIO_set_fd(bio, self->sockfd, BIO_NOCLOSE);
+  SSL_set_bio(self->ssl, bio, bio);
+
 start_connect:
   status = SSL_connect(self->ssl);
   if (status != 1) {
diff --git a/librabbitmq/amqp_openssl_bio.c b/librabbitmq/amqp_openssl_bio.c
new file mode 100644
index 0000000..9d5d43a
--- /dev/null
+++ b/librabbitmq/amqp_openssl_bio.c
@@ -0,0 +1,166 @@
+/*
+ * Portions created by Alan Antonuk are Copyright (c) 2017 Alan Antonuk.
+ * All Rights Reserved.
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a
+ * copy of this software and associated documentation files (the "Software"),
+ * to deal in the Software without restriction, including without limitation
+ * the rights to use, copy, modify, merge, publish, distribute, sublicense,
+ * and/or sell copies of the Software, and to permit persons to whom the
+ * Software is furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in
+ * all copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
+ * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
+ * DEALINGS IN THE SOFTWARE.
+ */
+
+
+#include "amqp_openssl_bio.h"
+#include "amqp_socket.h"
+#include "threads.h"
+
+#include <errno.h>
+#if ((defined(_WIN32)) || (defined(__MINGW32__)) || (defined(__MINGW64__)))
+# ifndef WIN32_LEAN_AND_MEAN
+#  define WIN32_LEAN_AND_MEAN
+# endif
+# include <winsock2.h>
+#else
+# include <sys/types.h>
+# include <sys/socket.h>
+#endif
+
+#ifdef MSG_NOSIGNAL
+# define AMQP_USE_AMQP_BIO
+#endif
+
+#ifdef AMQP_USE_AMQP_BIO
+
+#ifdef ENABLE_THREAD_SAFETY
+static pthread_once_t bio_init_once = PTHREAD_ONCE_INIT;
+#endif
+
+static int bio_initialized = 0;
+static BIO_METHOD amqp_bio_method;
+
+static int amqp_openssl_bio_should_retry(int res) {
+  if (res == -1) {
+    int err = amqp_os_socket_error();
+    if (
+#ifdef EWOULDBLOCK
+        err == EWOULDBLOCK ||
+#endif
+#ifdef WSAEWOULDBLOCK
+        err == WSAEWOULDBLOCK ||
+#endif
+#ifdef ENOTCONN
+        err == ENOTCONN ||
+#endif
+#ifdef EINTR
+        err == EINTR ||
+#endif
+#ifdef EAGAIN
+        err == EAGAIN ||
+#endif
+#ifdef EPROTO
+        err == EPROTO ||
+#endif
+#ifdef EINPROGRESS
+        err == EINPROGRESS ||
+#endif
+#ifdef EALREADY
+        err == EALREADY ||
+#endif
+        0) {
+      return 1;
+    }
+  }
+  return 0;
+}
+
+static int amqp_openssl_bio_write(BIO* b, const char *in, int inl) {
+  int flags = 0;
+  int fd;
+  int res;
+
+#ifdef MSG_NOSIGNAL
+  flags |= MSG_NOSIGNAL;
+#endif
+
+  BIO_get_fd(b, &fd);
+  res = send(fd, in, inl, flags);
+
+  BIO_clear_retry_flags(b);
+  if (res <= 0 && amqp_openssl_bio_should_retry(res)) {
+    BIO_set_retry_write(b);
+  }
+
+  return res;
+}
+
+static int amqp_openssl_bio_read(BIO* b, char* out, int outl) {
+  int flags = 0;
+  int fd;
+  int res;
+
+#ifdef MSG_NOSIGNAL
+  flags |= MSG_NOSIGNAL;
+#endif
+
+  BIO_get_fd(b, &fd);
+  res = recv(fd, out, outl, flags);
+
+  BIO_clear_retry_flags(b);
+  if (res <= 0 && amqp_openssl_bio_should_retry(res)) {
+    BIO_set_retry_read(b);
+  }
+
+  return res;
+}
+
+#if (OPENSSL_VERSION_NUMBER < 0x10100000L)
+static int BIO_meth_set_write(BIO_METHOD *biom,
+                              int (*wfn)(BIO *, const char *, int)) {
+  biom->bwrite = wfn;
+  return 0;
+}
+
+static int BIO_meth_set_read(BIO_METHOD *biom,
+                              int (*rfn)(BIO *, char *, int)) {
+  biom->bread = rfn;
+  return 0;
+}
+#endif
+
+static void amqp_openssl_bio_init(void) {
+  memcpy(&amqp_bio_method, BIO_s_socket(), sizeof(amqp_bio_method));
+  BIO_meth_set_write(&amqp_bio_method, amqp_openssl_bio_write);
+  BIO_meth_set_read(&amqp_bio_method, amqp_openssl_bio_read);
+
+  bio_initialized = 1;
+}
+
+#endif  /* AMQP_USE_AMQP_BIO */
+
+BIO_METHOD* amqp_openssl_bio(void) {
+#ifdef AMQP_USE_AMQP_BIO
+  if (!bio_initialized) {
+#ifdef ENABLE_THREAD_SAFETY
+    pthread_once(&bio_init_once, amqp_openssl_bio_init);
+#else
+    amqp_openssl_bio_init();
+#endif /* ifndef ENABLE_THREAD_SAFETY */
+  }
+
+  return &amqp_bio_method;
+#else
+  return BIO_s_socket();
+#endif
+}
diff --git a/librabbitmq/amqp_openssl_bio.h b/librabbitmq/amqp_openssl_bio.h
new file mode 100644
index 0000000..638717d
--- /dev/null
+++ b/librabbitmq/amqp_openssl_bio.h
@@ -0,0 +1,30 @@
+/*
+ * Portions created by Alan Antonuk are Copyright (c) 2017 Alan Antonuk.
+ * All Rights Reserved.
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a
+ * copy of this software and associated documentation files (the "Software"),
+ * to deal in the Software without restriction, including without limitation
+ * the rights to use, copy, modify, merge, publish, distribute, sublicense,
+ * and/or sell copies of the Software, and to permit persons to whom the
+ * Software is furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in
+ * all copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
+ * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
+ * DEALINGS IN THE SOFTWARE.
+ */
+#ifndef AMQP_OPENSSL_BIO
+#define AMQP_OPENSSL_BIO
+
+#include <openssl/bio.h>
+
+BIO_METHOD* amqp_openssl_bio(void);
+
+#endif /* ifndef AMQP_OPENSSL_BIO */