Merge ssl branch.

gearmand/gearmand.cc

@@ -133,6 +133,7 @@ int main(int argc, char *argv[])
   int opt_keepalive_interval;
   int opt_keepalive_count;
 
+
   boost::program_options::options_description general("General options");
 
   general.add_options()

gearmand/include.am

@@ -57,3 +57,6 @@ gearmand-valgrind: gearmand/gearmand
 
 gearmand-debug: gearmand/gearmand
 	$(LIBTOOL) --mode=execute gdb gearmand/gearmand
+
+internals: tests/libgearman-1.0/internals_test gearmand/gearmand
+	@$(GDB_COMMAND) tests/libgearman-1.0/internals_test

libgearman-server/gearmand.cc

@@ -211,10 +211,10 @@ static gearmand_st *_global_gearmand= NULL;
 
 gearmand_st *Gearmand(void)
 {
-  if (!_global_gearmand)
+  if (_global_gearmand == NULL)
   {
-    gearmand_error("Gearmand() was called before it was allocated");
     assert_msg(false, "Gearmand() was called before it was allocated");
+    gearmand_error("Gearmand() was called before it was allocated");
   }
   assert(_global_gearmand);
   return _global_gearmand;
@@ -318,6 +318,7 @@ bool gearmand_exceptions(gearmand_st *gearmand)
 gearmand_error_t gearmand_port_add(gearmand_st *gearmand, const char *port,
                                    gearmand_connection_add_fn *function)
 {
+  assert(gearmand);
   gearmand->_port_list.resize(gearmand->_port_list.size() +1);
 
   strncpy(gearmand->_port_list.back().port, port, NI_MAXSERV);

libgearman-server/io.cc

@@ -819,7 +819,7 @@ gearmand_error_t gearmand_io_set_revents(gearman_server_con_st *con, short reven
 
 static gearmand_error_t _io_setsockopt(gearmand_io_st &connection)
 {
-  gearmand_log_debug(GEARMAN_DEFAULT_LOG_PARAM, "setsockopt() %d", connection.fd);
+  gearmand_log_debug(GEARMAN_DEFAULT_LOG_PARAM, "setsockopt() fd:%d", connection.fd);
   {
     int setting= 1;
     if (setsockopt(connection.fd, IPPROTO_TCP, TCP_NODELAY, &setting, (socklen_t)sizeof(int)) and errno != EOPNOTSUPP)

libgearman-server/plugins/protocol/gear/protocol.cc

@@ -307,15 +307,11 @@ private:
 
 static Geartext gear_context;
 
-#if defined(HAVE_CYASSL) && HAVE_CYASSL
-static struct CYASSL_CTX *ctx_ssl= NULL;
-#endif
-
 static gearmand_error_t _gear_con_add(gearman_server_con_st *connection)
 {
 #if defined(HAVE_CYASSL) && HAVE_CYASSL
-  assert(ctx_ssl);
-  if ((connection->_ssl = CyaSSL_new(ctx_ssl)) == NULL)
+  assert(Gearmand()->ctx_ssl());
+  if ((connection->_ssl= CyaSSL_new(Gearmand()->ctx_ssl())) == NULL)
   {
     return gearmand_log_error(GEARMAN_DEFAULT_LOG_PARAM, "CyaSSL_new() failed");
   }
@@ -356,36 +352,6 @@ Gear::Gear() :
     command_line_options().add_options()
       ("port,p", boost::program_options::value(&_port)->default_value(GEARMAN_DEFAULT_TCP_PORT_STRING),
        "Port the server should listen on.");
-
-#if defined(HAVE_CYASSL) && HAVE_CYASSL
-    CyaSSL_Init();
-
-    ctx_ssl= CyaSSL_CTX_new(CyaTLSv1_2_server_method());
-
-    if (access(CERT_PEM, R_OK) == -1)
-    {
-      assert("access()" == NULL);
-    }
-
-    if (CyaSSL_CTX_use_certificate_file(ctx_ssl, CERT_PEM, SSL_FILETYPE_PEM) != SSL_SUCCESS)
-    {   
-      CyaSSL_CTX_free(ctx_ssl);
-      gearmand_fatal("CyaSSL_CTX_use_certificate_file() cannot obtain certificate");
-    }
-
-    if (access(CERT_KEY_PEM, R_OK) == -1)
-    {
-      gearmand_fatal("access(CERT_KEY_PEM, R_OK) == -1");
-    }
-
-    if (CyaSSL_CTX_use_PrivateKey_file(ctx_ssl, CERT_KEY_PEM, SSL_FILETYPE_PEM) != SSL_SUCCESS)
-    {   
-      CyaSSL_CTX_free(ctx_ssl);
-      gearmand_fatal("CyaSSL_CTX_use_PrivateKey_file() cannot obtain certificate");
-    }
-
-    assert(ctx_ssl);
-#endif
   }
 
 Gear::~Gear()
@@ -424,6 +390,32 @@ gearmand_error_t Gear::start(gearmand_st *gearmand)
 
   gearmand_log_info(GEARMAN_DEFAULT_LOG_PARAM, "Initializing Gear on port %s", _port.c_str());
 
+#if defined(HAVE_CYASSL) && HAVE_CYASSL
+  {
+    if (access(CERT_PEM, R_OK) == -1)
+    {
+      assert("access()" == NULL);
+    }
+
+    if (CyaSSL_CTX_use_certificate_file(gearmand->ctx_ssl(), CERT_PEM, SSL_FILETYPE_PEM) != SSL_SUCCESS)
+    {   
+      gearmand_fatal("CyaSSL_CTX_use_certificate_file() cannot obtain certificate");
+    }
+
+    if (access(CERT_KEY_PEM, R_OK) == -1)
+    {
+      gearmand_fatal("access(CERT_KEY_PEM, R_OK) == -1");
+    }
+
+    if (CyaSSL_CTX_use_PrivateKey_file(gearmand->ctx_ssl(), CERT_KEY_PEM, SSL_FILETYPE_PEM) != SSL_SUCCESS)
+    {   
+      gearmand_fatal("CyaSSL_CTX_use_PrivateKey_file() cannot obtain certificate");
+    }
+
+    assert(gearmand->ctx_ssl());
+  }
+#endif
+
   rc= gearmand_port_add(gearmand, _port.c_str(), _gear_con_add);
 
   return rc;

libgearman-server/struct/gearmand.h

@@ -2,7 +2,7 @@
  * 
  *  Gearmand client and server library.
  *
- *  Copyright (C) 2011 Data Differential, http://datadifferential.com/
+ *  Copyright (C) 2011-2013 Data Differential, http://datadifferential.com/
  *  All rights reserved.
  *
  *  Redistribution and use in source and binary forms, with or without
@@ -136,6 +136,9 @@ struct gearmand_st
   gearman_server_st server;
   struct event wakeup_event;
   std::vector<gearmand_port_st> _port_list;
+  private:
+  struct CYASSL_CTX* _ctx_ssl;
+  public:
 
   gearmand_st(const char *host_,
               uint32_t threads_,
@@ -159,7 +162,8 @@ struct gearmand_st
     base(NULL),
     thread_list(NULL),
     thread_add_next(NULL),
-    free_dcon_list(NULL)
+    free_dcon_list(NULL),
+    _ctx_ssl(NULL)
   {
     if (host_)
     {
@@ -167,6 +171,15 @@ struct gearmand_st
     }
     wakeup_fd[0]= -1;
     wakeup_fd[1]= -1;
+#if defined(HAVE_CYASSL) && HAVE_CYASSL
+    CyaSSL_Init();
+    _ctx_ssl= CyaSSL_CTX_new(CyaSSLv23_server_method());
+#endif
+  }
+
+  struct CYASSL_CTX* ctx_ssl()
+  {
+    return _ctx_ssl;
   }
 
   ~gearmand_st()
@@ -175,11 +188,11 @@ struct gearmand_st
     {
       free(host);
     }
-  }
-  
+
 #if defined(HAVE_CYASSL) && HAVE_CYASSL
-  CYASSL_CTX *ctx_ssl;
+    CyaSSL_CTX_free(_ctx_ssl);
 #endif
+  }
 
   bool exceptions() const
   {

libgearman/connection.cc

@@ -122,6 +122,7 @@ gearman_connection_st::gearman_connection_st(gearman_universal_st &universal_arg
   events(0),
   revents(0),
   fd(-1),
+  _ssl(NULL),
   cached_errno(0),
   created_id(0),
   created_id_next(0),
@@ -235,10 +236,7 @@ gearman_connection_st *gearman_connection_copy(gearman_universal_st& universal,
 
 gearman_connection_st::~gearman_connection_st()
 {
-  if (fd != INVALID_SOCKET)
-  {
-    close_socket();
-  }
+  close_socket();
 
   reset_addrinfo();
 
@@ -331,6 +329,14 @@ void gearman_connection_st::set_host(const char *host_, const char* service_)
 */
 void gearman_connection_st::close_socket()
 {
+  if (_ssl)
+  {
+#if defined(HAVE_CYASSL) && HAVE_CYASSL
+    CyaSSL_free(_ssl);
+    _ssl= NULL;
+#endif
+  }
+
   if (fd == INVALID_SOCKET)
   {
     return;
@@ -693,6 +699,36 @@ gearman_return_t gearman_connection_st::lookup()
   return GEARMAN_SUCCESS;
 }
 
+gearman_return_t gearman_connection_st::enable_ssl()
+{
+#if defined(HAVE_CYASSL) && HAVE_CYASSL
+  _ssl= CyaSSL_new(universal.ctx_ssl());
+  if (_ssl == NULL)
+  {
+    close_socket();
+    return gearman_error(universal, GEARMAN_COULD_NOT_CONNECT, "CyaSSL_new() failed to return a valid object");
+  }
+
+  if (CyaSSL_set_fd(_ssl, fd) != SSL_SUCCESS)
+  {
+    close_socket();
+    char errorString[80];
+    return gearman_error(universal, GEARMAN_COULD_NOT_CONNECT, CyaSSL_ERR_error_string(CyaSSL_get_error(_ssl, 0), errorString));
+  }
+
+#if 0
+  if (CyaSSL_connect(_ssl) != SSL_SUCCESS)
+  {
+    close_socket();
+    char errorString[80];
+    return gearman_error(universal, GEARMAN_COULD_NOT_CONNECT, CyaSSL_ERR_error_string(CyaSSL_get_error(_ssl, 0), errorString));
+  }
+#endif
+#endif
+
+  return GEARMAN_SUCCESS;
+}
+
 gearman_return_t gearman_connection_st::flush()
 {
   while (1)
@@ -758,6 +794,7 @@ gearman_return_t gearman_connection_st::flush()
         {
           state= GEARMAN_CON_UNIVERSAL_CONNECTED;
           addrinfo_next= NULL;
+
           break;
         }
 
@@ -806,6 +843,12 @@ gearman_return_t gearman_connection_st::flush()
         else if (revents & POLLOUT)
         {
           state= GEARMAN_CON_UNIVERSAL_CONNECTED;
+          gearman_return_t ssl_ret;
+          if ((ssl_ret= enable_ssl()) != GEARMAN_SUCCESS)
+          {
+            return ssl_ret;
+          }
+
           break;
         }
 
@@ -832,7 +875,35 @@ gearman_return_t gearman_connection_st::flush()
     case GEARMAN_CON_UNIVERSAL_CONNECTED:
       while (send_buffer_size != 0)
       {
-        ssize_t write_size= ::send(fd, send_buffer_ptr, send_buffer_size, MSG_NOSIGNAL);
+        ssize_t write_size;
+#if defined(HAVE_CYASSL) && HAVE_CYASSL
+        write_size= 0;
+        if (_ssl)
+        {
+          write_size= CyaSSL_send(_ssl, send_buffer_ptr, send_buffer_size, MSG_NOSIGNAL);
+          if (write_size < 0)
+          {
+            int err;
+            switch ((err= CyaSSL_get_error(_ssl, 0)))
+            {
+              case SSL_ERROR_WANT_WRITE:
+              case SSL_ERROR_WANT_READ:
+                errno= EWOULDBLOCK;
+                break;
+
+              default:
+                {
+                  char errorString[80];
+                  CyaSSL_ERR_error_string(err, errorString);
+                  close_socket();
+                  return gearman_universal_set_error(universal, GEARMAN_LOST_CONNECTION, GEARMAN_AT, "SSL failure(%s)", errorString);
+                }
+            }
+          }
+        }
+#else
+        write_size= ::send(fd, send_buffer_ptr, send_buffer_size, MSG_NOSIGNAL);
+#endif
 
         if (write_size == 0) // Zero value on send()
         { }
@@ -1077,7 +1148,28 @@ size_t gearman_connection_st::recv_socket(void *data, size_t data_size, gearman_
 
   while (1)
   {
+#if defined(HAVE_CYASSL) && HAVE_CYASSL
+    if (_ssl)
+    {
+      read_size= CyaSSL_recv(_ssl, data, data_size, MSG_DONTWAIT);
+      if (read_size < 0)
+      {
+        int sendErr= CyaSSL_get_error(_ssl, 0);
+        if (sendErr != SSL_ERROR_WANT_READ)
+        {
+          char errorString[80];
+          int err = CyaSSL_get_error(_ssl, 0);
+          CyaSSL_ERR_error_string(err, errorString);
+          close_socket();
+          ret= gearman_universal_set_error(universal, GEARMAN_LOST_CONNECTION, GEARMAN_AT,
+                                           "SSL failure(%s)", errorString);
+        }
+        errno= EAGAIN;
+      }
+    }
+#else
     read_size= ::recv(fd, data, data_size, MSG_NOSIGNAL);
+#endif
 
     if (read_size == 0)
     {

libgearman/connection.hpp

@@ -2,7 +2,7 @@
  * 
  *  Gearmand client and server library.
  *
- *  Copyright (C) 2011 Data Differential, http://datadifferential.com/
+ *  Copyright (C) 2011-2013 Data Differential, http://datadifferential.com/
  *  Copyright (C) 2008 Brian Aker, Eric Day
  *  All rights reserved.
  *
@@ -42,9 +42,7 @@
 #include "libgearman/interface/packet.hpp"
 #include "libgearman/interface/universal.hpp"
 
-#if defined(HAVE_CYASSL) && HAVE_CYASSL
-# include <cyassl/ssl.h>
-#endif
+#include "libgearman/ssl.h"
 
 struct gearman_connection_st
 {
@@ -67,10 +65,7 @@ struct gearman_connection_st
   short events;
   short revents;
   int fd;
-#if defined(HAVE_CYASSL) && HAVE_CYASSL
-  CYASSL* ssl;
-  CYASSL_CTX* ctx_ssl;
-#endif
+  struct CYASSL* _ssl;
   int cached_errno;
   uint32_t created_id;
   uint32_t created_id_next;
@@ -109,6 +104,7 @@ public:
   gearman_return_t send_packet(const gearman_packet_st&, const bool flush_buffer);
   size_t send_and_flush(const void *data, size_t data_size, gearman_return_t *ret_ptr);
 
+  gearman_return_t enable_ssl();
   gearman_return_t flush();
   void close_socket();
 

libgearman/include.am

@@ -15,6 +15,7 @@ nobase_include_HEADERS+= libgearman/gearman.h
 noinst_HEADERS+= libgearman/client.hpp
 noinst_HEADERS+= libgearman/worker.hpp
 
+noinst_HEADERS+= libgearman/ssl.h
 noinst_HEADERS+= libgearman/magic.h
 noinst_HEADERS+= libgearman/client.h
 noinst_HEADERS+= libgearman/check.h

libgearman/interface/universal.hpp

@@ -85,6 +85,7 @@ struct gearman_universal_st
   gearman_allocator_t allocator;
   struct gearman_vector_st *_identifier;
   struct gearman_vector_st *_namespace;
+  struct CYASSL_CTX* _ctx_ssl;
   struct error_st {
     gearman_return_t rc;
     int last_errno;
@@ -169,7 +170,8 @@ struct gearman_universal_st
     log_context(NULL),
     allocator(gearman_default_allocator()),
     _identifier(NULL),
-    _namespace(NULL)
+    _namespace(NULL),
+    _ctx_ssl(NULL)
   {
     wakeup_fd[0]= INVALID_SOCKET;
     wakeup_fd[1]= INVALID_SOCKET;
@@ -185,14 +187,24 @@ struct gearman_universal_st
         options_++;
       }
     }
+
+    // Only does something if SSL has been enabled.
+    bool ret= init_ssl();
+    if (ret == false)
+    {
+      abort();
+    }
   }
 
-  ~gearman_universal_st()
+  bool init_ssl();
+
+  struct CYASSL_CTX* ctx_ssl() 
   {
-    gearman_string_free(_identifier);
-    gearman_string_free(_namespace);
+    return _ctx_ssl;
   }
 
+  ~gearman_universal_st();
+
   void identifier(const char *identifier_, const size_t identifier_size_);
 };