Merge pull request #294 from bill1600/cnheal

Add logic to save parodus connection status in health file

CHANGELOG.md

@@ -6,6 +6,10 @@ and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0.
 
 ## [Unreleased]
 - Security: Added support to use auth token during initial connect to cloud
+- Fix re-registration fails that lose a socket
+- Fix mutex error in service alive
+- Security: Mutual Authentication (mTLS or two way TLS)
+- Rename command line options for MTLS cert and Key
 
 ## [1.0.2] - 2019-02-08
 - Refactored connection.c and updated corresponding unit tests

README.md

@@ -75,6 +75,12 @@ make test
 
 - /jwt-public-key-file -JWT token validation key
 
+# if ENABLE_MUTUAL_AUTH is enabled
+
+- /mtls-client-cert-path - Provide the client cert for establishing a mutual auth secure websocket connection
+
+- /mtls-client-key-path - Provide the client cert key for establishing a mutual auth secure websocket connection
+
 ```
 
 # Sample parodus start commands:

src/client_list.c

@@ -76,7 +76,11 @@ int addToList( wrp_msg_t **msg)
             if(rc < 0)
             {
                 ParodusError ("Unable to connect socket (errno=%d, %s)\n",errno, strerror(errno));
-                nn_close (sock);
+		if (nn_close (sock) < 0)
+                {
+                   ParodusError ("nn_close socket=%d (err=%d, %s)\n", 
+			sock, errno, strerror(errno));
+		}
                 
             }
             else
@@ -87,6 +91,7 @@ int addToList( wrp_msg_t **msg)
 		{
     			memset( new_node, 0, sizeof( reg_list_item_t ) );
     			new_node->sock = sock;
+                        new_node->endpoint = rc;
     			ParodusPrint("new_node->sock is %d\n", new_node->sock);
     			
     			
@@ -219,6 +224,16 @@ int deleteFromList(char* service_name)
 			}
 			
 			ParodusPrint("Deleting the node\n");
+                        if(nn_shutdown(curr_node->sock, curr_node->endpoint) < 0)
+                        {
+                           ParodusError ("nn_shutdown socket=%d endpt=%d, err=%d\n", 
+				curr_node->sock, curr_node->endpoint, errno);
+                        }
+			if (nn_close (curr_node->sock) < 0)
+                        {
+                           ParodusError ("nn_close socket=%d err=%d\n", 
+				curr_node->sock, errno);
+                        }
 			free( curr_node );
 			curr_node = NULL;
 			ParodusInfo("Deleted successfully and returning..\n");

src/client_list.h

@@ -30,6 +30,7 @@
 typedef struct reg_list_item
 {
 	int sock;
+	int endpoint;
 	char service_name[32];
 	char url[100];
 	struct reg_list_item *next;

src/config.c

@@ -365,6 +365,9 @@ int parseCommandLine(int argc,char **argv,ParodusCfg * cfg)
 	{"client-cert-path",        required_argument, 0, 'P'},
 	{"token-server-url",        required_argument, 0, 'U'},
 	{"crud-config-file",        required_argument, 0, 'C'},
+	{"connection-health-file",  required_argument, 0, 'S'},
+	{"mtls-client-key-path",    required_argument, 0, 'K'},
+	{"mtls-client-cert-path",    required_argument, 0,'M'},
         {0, 0, 0, 0}
     };
     int c;
@@ -380,6 +383,7 @@ int parseCommandLine(int argc,char **argv,ParodusCfg * cfg)
 	cfg->jwt_algo = 0;
 	parStrncpy (cfg->jwt_key, "", sizeof(cfg->jwt_key));
 	cfg->crud_config_file = NULL;
+	cfg->connection_health_file = NULL;
 	cfg->client_cert_path = NULL;
 	cfg->token_server_url = NULL;
 	cfg->cloud_status = NULL;
@@ -390,7 +394,7 @@ int parseCommandLine(int argc,char **argv,ParodusCfg * cfg)
 
       /* getopt_long stores the option index here. */
       int option_index = 0;
-      c = getopt_long (argc, argv, "m:s:f:d:r:n:b:u:t:o:i:l:p:e:D:j:a:k:c:T:w:J:46:C",
+      c = getopt_long (argc, argv, "m:s:f:d:r:n:b:u:t:o:i:l:p:e:D:j:a:k:c:T:w:J:46:C:S:K:M",
 				long_options, &option_index);
 
       /* Detect the end of the options. */
@@ -528,10 +532,15 @@ int parseCommandLine(int argc,char **argv,ParodusCfg * cfg)
           ParodusInfo("boot_retry_wait is %d\n",cfg->boot_retry_wait);
           break;
 
-		case 'C':
-		  cfg->crud_config_file = strdup(optarg);
-		  ParodusInfo("crud_config_file is %s\n", cfg->crud_config_file);
-		  break;
+	case 'S':
+	  cfg->connection_health_file = strdup(optarg);
+	  ParodusInfo("connection_health_file is %s\n", cfg->connection_health_file);
+	  break;
+
+	case 'C':
+	  cfg->crud_config_file = strdup(optarg);
+	  ParodusInfo("crud_config_file is %s\n", cfg->crud_config_file);
+	  break;
 
 		case 'P':
 		cfg->client_cert_path = strdup(optarg);
@@ -543,6 +552,16 @@ int parseCommandLine(int argc,char **argv,ParodusCfg * cfg)
 		ParodusInfo("token_server_url is %s\n", cfg->token_server_url);
 		break;
 
+        case 'K':
+          cfg->mtls_client_key_path = strdup(optarg);
+          ParodusInfo("mtls_client_key_path is %s\n", cfg->mtls_client_key_path);
+          break;
+
+	case 'M':
+          cfg->mtls_client_cert_path = strdup(optarg);
+          ParodusInfo("mtls_client_cert_path is %s\n", cfg->mtls_client_cert_path);
+          break;
+
         case '?':
           /* getopt_long already printed an error message. */
           break;
@@ -616,6 +635,7 @@ void setDefaultValuesToCfg(ParodusCfg *cfg)
     parStrncpy(cfg->webpa_uuid, "1234567-345456546",sizeof(cfg->webpa_uuid));
     ParodusPrint("cfg->webpa_uuid is :%s\n", cfg->webpa_uuid);
     cfg->crud_config_file = NULL;
+    cfg->connection_health_file = NULL;
     cfg->client_cert_path = NULL;
     cfg->token_server_url = NULL;
 	
@@ -767,6 +787,15 @@ void loadParodusCfg(ParodusCfg * config,ParodusCfg *cfg)
     parStrncpy(cfg->webpa_uuid, "1234567-345456546",sizeof(cfg->webpa_uuid));
     ParodusPrint("cfg->webpa_uuid is :%s\n", cfg->webpa_uuid);
     
+    if(config->connection_health_file != NULL)
+    {
+        cfg->connection_health_file = strdup(config->connection_health_file);
+    }
+    else
+    {
+        ParodusPrint("connection_health_file is NULL. set to empty\n");
+    }
+
     if(config->crud_config_file != NULL)
     {
         cfg->crud_config_file = strdup(config->crud_config_file);

src/config.h

@@ -101,6 +101,9 @@ typedef struct
     char token_read_script[64];
     char *client_cert_path;
     char *token_server_url;
+    char *connection_health_file;
+    char *mtls_client_key_path;
+    char *mtls_client_cert_path;
 	char *crud_config_file;
 	char *cloud_status;
 	char *cloud_disconnect;

src/conn_interface.c

@@ -74,6 +74,7 @@ void createSocketConnection(void (* initKeypress)())
     //ParodusCfg *tmpCfg = (ParodusCfg*)config_in;
     noPollCtx *ctx;
     bool seshat_registered = false;
+    int create_conn_rtn = 0;
     unsigned int webpa_ping_timeout_ms = 1000 * get_parodus_cfg()->webpa_ping_timeout;
     unsigned int heartBeatTimer = 0;
     struct timespec start_svc_alive_timer;
@@ -94,7 +95,10 @@ void createSocketConnection(void (* initKeypress)())
     nopoll_log_set_handler (ctx, __report_log, NULL);
     #endif
 
-    if(!createNopollConnection(ctx))
+    start_conn_in_progress ();
+    create_conn_rtn = createNopollConnection(ctx);
+    stop_conn_in_progress ();
+    if(!create_conn_rtn)
     {
 		ParodusError("Unrecovered error, terminating the process\n");
 		abort();
@@ -177,7 +181,9 @@ void createSocketConnection(void (* initKeypress)())
 		free(get_parodus_cfg()->cloud_disconnect);
 		reset_cloud_disconnect_reason(get_parodus_cfg());
             }
+            start_conn_in_progress ();
             createNopollConnection(ctx);
+            stop_conn_in_progress ();
         }
        } while(!get_close_retry() && !g_shutdown);
 

src/connection.c

@@ -628,13 +628,20 @@ static char* build_extra_headers( const char *auth, const char *device_id,
 static noPollConnOpts * createConnOpts (char * extra_headers, bool secure)
 {
     noPollConnOpts * opts;
+    char * mtls_client_cert_path = NULL;
+    char * mtls_client_key_path = NULL;
     
     opts = nopoll_conn_opts_new ();
     if(secure) 
 	{
 	    if(strlen(get_parodus_cfg()->cert_path) > 0)
             {
-                nopoll_conn_opts_set_ssl_certs(opts, NULL, NULL, NULL, get_parodus_cfg()->cert_path);
+                if( ( get_parodus_cfg()->mtls_client_cert_path !=NULL && strlen(get_parodus_cfg()->mtls_client_cert_path) > 0) && (get_parodus_cfg()->mtls_client_key_path !=NULL && strlen(get_parodus_cfg()->mtls_client_key_path) > 0) )
+                {
+                       mtls_client_cert_path = get_parodus_cfg()->mtls_client_cert_path;
+                       mtls_client_key_path = get_parodus_cfg()->mtls_client_key_path;
+                }
+                nopoll_conn_opts_set_ssl_certs(opts, mtls_client_cert_path, mtls_client_key_path, NULL, get_parodus_cfg()->cert_path);
             }
 	    nopoll_conn_opts_ssl_peer_verify (opts, nopoll_true);
 	    nopoll_conn_opts_set_ssl_protocol (opts, NOPOLL_METHOD_TLSV1_2);
@@ -655,3 +662,38 @@ void close_and_unref_connection(noPollConn *conn)
     }
 }
 
+void write_conn_in_prog_file (const char *msg)
+{
+  int fd;
+  FILE *fp;
+  unsigned long timestamp;
+  ParodusCfg *cfg = get_parodus_cfg();
+
+  if (NULL == cfg->connection_health_file)
+    return;
+  fd = open (cfg->connection_health_file, O_CREAT | O_WRONLY | O_SYNC, 0666);
+  if (fd < 0) {
+    ParodusError ("Error(1) %d opening file %s\n", errno, cfg->connection_health_file);
+    return;
+  }
+  ftruncate (fd, 0);
+  fp = fdopen (fd, "w");
+  if (fp == NULL) {
+    ParodusError ("Error(2) %d opening file %s\n", errno, cfg->connection_health_file);
+    return;
+  }
+  timestamp = (unsigned long) time(NULL);
+  fprintf (fp, "{%s=%lu}\n", msg, timestamp);
+  fclose (fp);
+}
+
+void start_conn_in_progress (void)
+{
+  write_conn_in_prog_file ("START");
+}   
+
+void stop_conn_in_progress (void)
+{
+  write_conn_in_prog_file ("STOP");
+}   
+

src/connection.h

@@ -57,6 +57,13 @@ void set_global_reconnect_status(bool status);
 
 int get_cloud_disconnect_time();
 void set_cloud_disconnect_time(int disconnTime);
+
+/**
+ * @brief Interface to self heal connection in progress getting stuck
+ */
+void start_conn_in_progress (void);
+void stop_conn_in_progress (void);
+
 #ifdef __cplusplus
 }
 #endif

src/upstream.c

@@ -190,6 +190,16 @@ void *handle_upstream()
                     ParodusError("failure in allocation for message\n");
                 }
             }
+	    if(nn_shutdown(sock, bind) < 0)
+	    {
+	        ParodusError ("nn_shutdown bind socket=%d endpt=%d, err=%d\n", 
+		    sock, bind, errno);
+	    }
+	    if (nn_close (sock) < 0)
+	    {
+	        ParodusError ("nn_close bind socket=%d err=%d\n", 
+		    sock, errno);
+	    }
         }
     }
     else
@@ -234,7 +244,7 @@ void *processUpstreamMessage()
             if(rv > 0)
             {
                 msgType = msg->msg_type;				   
-                if(msgType == 9)
+                if(msgType == WRP_MSG_TYPE__SVC_REGISTRATION)
                 {
                     ParodusInfo("\n Nanomsg client Registration for Upstream\n");
                     //Extract serviceName and url & store it in a linked list for reg_clients
@@ -249,12 +259,16 @@ void *processUpstreamMessage()
                             {
                                 ParodusInfo("match found, client is already registered\n");
                                 parStrncpy(temp->url,msg->u.reg.url, sizeof(temp->url));
-                                if(nn_shutdown(temp->sock, 0) < 0)
+                                if(nn_shutdown(temp->sock, temp->endpoint) < 0)
                                 {
-                                    ParodusError ("nn_shutdown socket=%d err=%d\n", 
+                                    ParodusError ("nn_shutdown socket=%d endpt=%d, err=%d\n", 
+					temp->sock, temp->endpoint, errno);
+                                }
+				if (nn_close (temp->sock) < 0)
+                                {
+                                    ParodusError ("nn_close socket=%d err=%d\n", 
 					temp->sock, errno);
                                 }
-				nn_close (temp->sock);
 
                                 temp->sock = nn_socket(AF_SP,NN_PUSH );
                                 if(temp->sock >= 0)
@@ -272,6 +286,7 @@ void *processUpstreamMessage()
                                     }
                                     else
                                     {
+                                        temp->endpoint = rc;
                                         ParodusInfo("Client registered before. Sending ack on socket %d\n", temp->sock); 
                                         status =sendAuthStatus(temp);
 
@@ -402,9 +417,8 @@ void *processUpstreamMessage()
 								{
 									ParodusError("Failed to get device_id\n");
 								}
-						}
-						else
-						{
+						} else if (WRP_MSG_TYPE__SVC_ALIVE != msgType) {
+						  /* Don't reply to service alive message */
 							sendUpstreamMsgToServer(&message->msg, message->len);
 						}
 					}

tests/test_conn_interface.c

@@ -74,6 +74,15 @@ noPollMutexUnlock 	mutex_unlock
     UNUSED(mutex_create); UNUSED(mutex_destroy); UNUSED(mutex_lock); UNUSED(mutex_unlock);
     function_called();
 }
+
+void start_conn_in_progress (void)
+{
+}   
+
+void stop_conn_in_progress (void)
+{
+}   
+
 void packMetaData()
 {
     function_called();

tests/test_upstream.c

@@ -273,6 +273,10 @@ void test_handleUpstreamNull()
     expect_function_call(nn_bind);
     will_return(nn_recv, 12);
     expect_function_call(nn_recv);
+    will_return(nn_shutdown, 0);
+    expect_function_call(nn_shutdown);
+    will_return(nn_close, 0);
+    expect_function_call(nn_close);
     handle_upstream();
 }
 
@@ -294,6 +298,10 @@ void test_handle_upstream()
     expect_function_call(nn_bind);
     will_return(nn_recv, 12);
     expect_function_call(nn_recv);
+    will_return(nn_shutdown, 0);
+    expect_function_call(nn_shutdown);
+    will_return(nn_close, 0);
+    expect_function_call(nn_close);
     handle_upstream();
     free(UpStreamMsgQ->next);
     free(UpStreamMsgQ);