WL#15524 Patch #16 Use TLS for Event Listener

Adapt the MGM client and server so that the event listener thread
in ndb_mgm can use TLS.

Change-Id: I727e17e063b3d457d39b54d94a1aac9d3c2cb9b2

Author: jdduncan

storage/ndb/src/mgmclient/CommandInterpreter.cpp

@@ -972,6 +972,7 @@ printLogEvent(struct ndb_logevent* event)
 struct event_thread_param {
   NdbMgmHandle *m;
   NdbMutex **p;
+  int tls_req;
 };
 
 static int do_event_thread = 0;
@@ -984,6 +985,7 @@ event_thread_run(void* p)
   struct event_thread_param param= *(struct event_thread_param*)p;
   NdbMgmHandle handle= *(param.m);
   NdbMutex* printmutex= *(param.p);
+  int tls_req = param.tls_req;
 
   int filter[] = { 15, NDB_MGM_EVENT_CATEGORY_BACKUP,
 		   1, NDB_MGM_EVENT_CATEGORY_STARTUP,
@@ -993,6 +995,16 @@ event_thread_run(void* p)
   NdbLogEventHandle log_handle= NULL;
   struct ndb_logevent log_event;
 
+  if(tls_req != CLIENT_TLS_DEFERRED)
+  {
+    int r = ndb_mgm_start_tls(handle);
+    if(r != 0 && tls_req == CLIENT_TLS_STRICT)
+    {
+      do_event_thread = -1;
+      DBUG_RETURN(NULL);
+    }
+  }
+
   log_handle= ndb_mgm_create_logevent_handle(handle, filter);
   if (log_handle) 
   {
@@ -1037,6 +1049,7 @@ CommandInterpreter::connect(bool interactive)
     ndbout_c("Can't create handle to management server.");
     exit(-1);
   }
+  ndb_mgm_set_ssl_ctx(m_mgmsrv, m_tlsKeyManager.ctx());
 
   if((m_tls_start_type == CLIENT_TLS_STRICT) &&
      (m_tlsKeyManager.ctx() == nullptr))
@@ -1057,6 +1070,7 @@ CommandInterpreter::connect(bool interactive)
       ndb_mgm_destroy_handle(&m_mgmsrv);
       exit(-1);
     }
+    ndb_mgm_set_ssl_ctx(m_mgmsrv2, m_tlsKeyManager.ctx());
   } else {
     m_mgmsrv2 = nullptr;
   }
@@ -1126,6 +1140,7 @@ CommandInterpreter::connect(bool interactive)
       struct event_thread_param p;
       p.m= &m_mgmsrv2;
       p.p= &m_print_mutex;
+      p.tls_req= m_tls_start_type;
       m_event_thread = NdbThread_Create(event_thread_run,
                                         (void**)&p,
                                         0, // default stack size

storage/ndb/src/mgmsrv/MgmtSrvr.cpp

@@ -285,7 +285,6 @@ MgmtSrvr::MgmtSrvr(const MgmtOpts& opts) :
   /* Setup clusterlog as client[0] in m_event_listner */
   {
     Ndb_mgmd_event_service::Event_listener se;
-    ndb_socket_initialize(&(se.m_socket));
     for(size_t t = 0; t<LogLevel::LOGLEVEL_CATEGORIES; t++){
       se.m_logLevel.setLogLevel((LogLevel::EventCategory)t, 7);
     }

storage/ndb/src/mgmsrv/MgmtSrvr.hpp

@@ -49,7 +49,7 @@ class Ndb_mgmd_event_service : public EventLoggerBase
 public:
   struct Event_listener : public EventLoggerBase {
     Event_listener() {}
-    ndb_socket_t m_socket;
+    NdbSocket * m_socket_ptr {nullptr};
     Uint32 m_parsable;
   };
 
@@ -66,7 +66,7 @@ class Ndb_mgmd_event_service : public EventLoggerBase
     stop_sessions();
   }
 
-  void add_listener(const Event_listener&);
+  void add_listener(Event_listener&, NdbSocket &);
   void check_listeners();
   void update_max_log_level(const LogLevel&);
   void update_log_level(const LogLevel&);

storage/ndb/src/mgmsrv/Services.cpp

@@ -1698,16 +1698,19 @@ Ndb_mgmd_event_service::log(int eventType, const Uint32* theData,
   logevent2str(str, eventType, theData, len, nodeId, 0,
                pretty_text, sizeof(pretty_text));
 
-  Vector<ndb_socket_t> copy;
+  Vector<NdbSocket *> copy;
   m_clients.lock();
   for(i = m_clients.size() - 1; i >= 0; i--)
   {
     if(threshold <= m_clients[i].m_logLevel.getLogLevel(cat))
     {
-      if(!ndb_socket_valid(m_clients[i].m_socket))
+      if(m_clients[i].m_socket_ptr == nullptr)
         continue;
 
-      SocketOutputStream out(m_clients[i].m_socket);
+      if(!m_clients[i].m_socket_ptr->is_valid())
+        continue;
+
+      SecureSocketOutputStream out(* m_clients[i].m_socket_ptr);
 
       int r;
       if (m_clients[i].m_parsable)
@@ -1728,7 +1731,7 @@ Ndb_mgmd_event_service::log(int eventType, const Uint32* theData,
 
       if (r<0)
       {
-        copy.push_back(m_clients[i].m_socket);
+        copy.push_back(m_clients[i].m_socket_ptr);
         m_clients.erase(i, false);
       }
     }
@@ -1737,8 +1740,10 @@ Ndb_mgmd_event_service::log(int eventType, const Uint32* theData,
 
   if ((n= (int)copy.size()))
   {
-    for(i= 0; i < n; i++)
-      ndb_socket_close(copy[i]);
+    for(i= 0; i < n; i++) {
+      copy[i]->close();
+      delete copy[i];
+    }
 
     LogLevel tmp; tmp.clear();
     m_clients.lock();
@@ -1780,18 +1785,18 @@ Ndb_mgmd_event_service::check_listeners()
   m_clients.lock();
   for(i= m_clients.size() - 1; i >= 0; i--)
   {
-    if(!ndb_socket_valid(m_clients[i].m_socket))
+    if(m_clients[i].m_socket_ptr == nullptr)
       continue;
 
-    SocketOutputStream out(m_clients[i].m_socket);
+    if(!(m_clients[i].m_socket_ptr->is_valid()))
+      continue;
 
-    DBUG_PRINT("info",("%d %s",
-                       i,
-                       ndb_socket_to_string(m_clients[i].m_socket).c_str()));
+    SecureSocketOutputStream out(* m_clients[i].m_socket_ptr);
 
     if(out.println("<PING>") < 0)
     {
-      ndb_socket_close(m_clients[i].m_socket);
+      m_clients[i].m_socket_ptr->close();
+      delete m_clients[i].m_socket_ptr;
       m_clients.erase(i, false);
       n=1;
     }
@@ -1808,14 +1813,15 @@ Ndb_mgmd_event_service::check_listeners()
 }
 
 void
-Ndb_mgmd_event_service::add_listener(const Event_listener& client)
+Ndb_mgmd_event_service::add_listener(Event_listener& client, NdbSocket& socket)
 {
   DBUG_ENTER("Ndb_mgmd_event_service::add_listener");
-  DBUG_PRINT("enter",("client.m_socket: %s",
-                      ndb_socket_to_string(client.m_socket).c_str()));
 
   check_listeners();
 
+  client.m_socket_ptr = new NdbSocket();
+  NdbSocket::transfer(* client.m_socket_ptr, socket);
+
   m_clients.push_back(client);
   update_max_log_level(client.m_logLevel);
 
@@ -1826,9 +1832,9 @@ void
 Ndb_mgmd_event_service::stop_sessions(){
   m_clients.lock();
   for(int i = m_clients.size() - 1; i >= 0; i--){
-    if(ndb_socket_valid(m_clients[i].m_socket))
+    if(m_clients[i].m_socket_ptr && m_clients[i].m_socket_ptr->is_valid())
     {
-      ndb_socket_close(m_clients[i].m_socket);
+      m_clients[i].m_socket_ptr->close();
       m_clients.erase(i, false);
     }
   }
@@ -1918,7 +1924,6 @@ MgmApiSession::listen_event(Parser<MgmApiSession>::Context & ctx,
 
   Ndb_mgmd_event_service::Event_listener le;
   le.m_parsable = parsable;
-  le.m_socket = m_secure_socket.ndb_socket();
 
   Vector<BaseString> list;
   param.trim();
@@ -1982,9 +1987,9 @@ MgmApiSession::listen_event(Parser<MgmApiSession>::Context & ctx,
 
   if(result==0)
   {
-    m_mgmsrv.m_event_listner.add_listener(le);
+    m_mgmsrv.m_event_listner.add_listener(le, m_secure_socket);
     m_stop = true;
-    m_secure_socket.invalidate();
+    assert(! m_secure_socket.is_valid()); // it has been transfered to listener
   }
 }