增加SSL采集字段，包含所有SSL解析层目前提供的字段

2 files changed, 396 insertions, 287 deletions

diff --git a/src/ntc_ssl_collect.c b/src/ntc_ssl_collect.c
index d1e994e..bc27908 100644
--- a/src/ntc_ssl_collect.c
+++ b/src/ntc_ssl_collect.c
@@ -1,6 +1,5 @@
 #include "ntc_ssl_collect.h"
 
-
 #include <assert.h>
 #include <stdio.h>
 #include <unistd.h>
@@ -22,7 +21,6 @@
 #include "MESA_handle_logger.h"
 #include "field_stat2.h"
 #include <rdkafka.h>
-#include <cJSON.h>
 
 #define PLUGIN_NAME "NTC_SSL_COLLECT"
 #define PROFILE_PATH "./t1conf/main.conf"
@@ -39,347 +37,455 @@ static rd_kafka_topic_t *g_ntc_kafka_topic = NULL;
 
 static unsigned int get_ip_by_eth_name(const char *ifname)
 {
-	int sockfd;
-	struct ifreq ifr;
-	unsigned int ip;
+    int sockfd;
+    struct ifreq ifr;
+    unsigned int ip;
 
-	sockfd = socket(AF_INET, SOCK_DGRAM, 0);
-	if (-1 == sockfd) {
-		goto error;
-	}
+    sockfd = socket(AF_INET, SOCK_DGRAM, 0);
+    if (-1 == sockfd)
+    {
+        goto error;
+    }
 
-	strcpy(ifr.ifr_name,ifname);
-	if (ioctl(sockfd, SIOCGIFADDR, &ifr) < 0) {
-		goto error;
-	}
+    strcpy(ifr.ifr_name, ifname);
+    if (ioctl(sockfd, SIOCGIFADDR, &ifr) < 0)
+    {
+        goto error;
+    }
 
-	ip = ((struct sockaddr_in*)&(ifr.ifr_addr))->sin_addr.s_addr;
-	close(sockfd);
-	return ip;
+    ip = ((struct sockaddr_in *)&(ifr.ifr_addr))->sin_addr.s_addr;
+    close(sockfd);
+    return ip;
 
 error:
-	close(sockfd);
-	return INADDR_NONE;
+    close(sockfd);
+    return INADDR_NONE;
 }
 
 static int ntc_addStreamInfo_to_jsonObj(cJSON *json_obj, const struct streaminfo *a_stream)
 {
-	int ret = 0;
-	const char *addr_proto = NULL;
-	const char* null_addr="0";
-	short null_port=0;
-	unsigned short tunnel_type=0;
-	char nest_addr_buf[1024];
-	int tunnel_type_size=sizeof(tunnel_type);
-	const struct layer_addr *addr=NULL;
-	char src_ip_str[128] = {0}, dst_ip_str[128] = {0};
-
-	cJSON_AddNumberToObject(json_obj, "stream_dir", a_stream->dir);
-
-	addr=&(a_stream->addr);
-	switch(addr->addrtype)
-	{
-		case ADDR_TYPE_IPV4:
-		case __ADDR_TYPE_IP_PAIR_V4:
-			inet_ntop(AF_INET, &addr->ipv4->saddr, src_ip_str, sizeof(src_ip_str));
-			inet_ntop(AF_INET, &addr->ipv4->daddr, dst_ip_str, sizeof(dst_ip_str));
-			cJSON_AddStringToObject(json_obj, "s_ip", src_ip_str);					
-			cJSON_AddStringToObject(json_obj, "d_ip", dst_ip_str);
-			cJSON_AddNumberToObject(json_obj, "s_port", ntohs(addr->ipv4->source));
-			cJSON_AddNumberToObject(json_obj, "d_port", ntohs(addr->ipv4->dest));
-			break;
-		case ADDR_TYPE_IPV6:
-		case __ADDR_TYPE_IP_PAIR_V6:
-			cJSON_AddNumberToObject(json_obj, "addr_type", addr->addrtype);
-			inet_ntop(AF_INET6, addr->ipv6->saddr, src_ip_str, sizeof(src_ip_str));
-			inet_ntop(AF_INET6, addr->ipv6->daddr, dst_ip_str, sizeof(dst_ip_str));
-			cJSON_AddStringToObject(json_obj, "s_ip", src_ip_str);					
-			cJSON_AddStringToObject(json_obj, "d_ip", dst_ip_str);
-			cJSON_AddNumberToObject(json_obj, "s_port", ntohs(addr->ipv6->source));
-			cJSON_AddNumberToObject(json_obj, "d_port", ntohs(addr->ipv6->dest));
-			break;
-		case ADDR_TYPE_VLAN:
-		case ADDR_TYPE_GRE:
-		case ADDR_TYPE_MPLS:
-		case ADDR_TYPE_PPPOE_SES:
-		case ADDR_TYPE_L2TP:
-		case ADDR_TYPE_PPP:
-			cJSON_AddNumberToObject(json_obj, "addr_type", addr->addrtype);
-			cJSON_AddStringToObject(json_obj, "s_ip", null_addr);					
-			cJSON_AddStringToObject(json_obj, "d_ip", null_addr);
-			cJSON_AddNumberToObject(json_obj, "s_port", null_port);
-			cJSON_AddNumberToObject(json_obj, "d_port", null_port);
-			break;
-		case ADDR_TYPE_PPTP:
-			cJSON_AddNumberToObject(json_obj, "addr_type", addr->addrtype);
-			cJSON_AddStringToObject(json_obj, "s_ip", null_addr);					
-			cJSON_AddStringToObject(json_obj, "d_ip", null_addr);
-			cJSON_AddNumberToObject(json_obj, "s_port", ntohs(addr->pptp->C2S_call_id));
-			cJSON_AddNumberToObject(json_obj, "d_port", ntohs(addr->pptp->S2C_call_id));
-			break;	
-		default:
-			break;
-	}
-
-	addr_proto = layer_addr_prefix_ntop(a_stream);
-	cJSON_AddStringToObject(json_obj, "trans_proto", addr_proto);
-
-	ret=MESA_get_stream_opt(a_stream, MSO_STREAM_TUNNEL_TYPE, &tunnel_type, &tunnel_type_size);
-	assert(ret==0);
-	if(tunnel_type==STREAM_TUNNLE_NON)
-	{	
-		layer_addr_ntop_r(a_stream,nest_addr_buf, sizeof(nest_addr_buf));
-	}
-	else
-	{		
-		stream_addr_list_ntop(a_stream,nest_addr_buf, sizeof(nest_addr_buf));
-	}
-
-	cJSON_AddStringToObject(json_obj, "addr_list", nest_addr_buf);
-
-	return 0;
-}
+    int ret = 0;
+    const char *addr_proto = NULL;
+    const char *null_addr = "0";
+    short null_port = 0;
+    unsigned short tunnel_type = 0;
+    char nest_addr_buf[1024];
+    int tunnel_type_size = sizeof(tunnel_type);
+    const struct layer_addr *addr = NULL;
+    char src_ip_str[128] = {0}, dst_ip_str[128] = {0};
+
+    cJSON_AddNumberToObject(json_obj, "stream_dir", a_stream->dir);
+
+    addr = &(a_stream->addr);
+    switch (addr->addrtype)
+    {
+        case ADDR_TYPE_IPV4:
+        case __ADDR_TYPE_IP_PAIR_V4:
+            inet_ntop(AF_INET, &addr->ipv4->saddr, src_ip_str, sizeof(src_ip_str));
+            inet_ntop(AF_INET, &addr->ipv4->daddr, dst_ip_str, sizeof(dst_ip_str));
+            cJSON_AddStringToObject(json_obj, "s_ip", src_ip_str);
+            cJSON_AddStringToObject(json_obj, "d_ip", dst_ip_str);
+            cJSON_AddNumberToObject(json_obj, "s_port", ntohs(addr->ipv4->source));
+            cJSON_AddNumberToObject(json_obj, "d_port", ntohs(addr->ipv4->dest));
+            break;
+        case ADDR_TYPE_IPV6:
+        case __ADDR_TYPE_IP_PAIR_V6:
+            cJSON_AddNumberToObject(json_obj, "addr_type", addr->addrtype);
+            inet_ntop(AF_INET6, addr->ipv6->saddr, src_ip_str, sizeof(src_ip_str));
+            inet_ntop(AF_INET6, addr->ipv6->daddr, dst_ip_str, sizeof(dst_ip_str));
+            cJSON_AddStringToObject(json_obj, "s_ip", src_ip_str);
+            cJSON_AddStringToObject(json_obj, "d_ip", dst_ip_str);
+            cJSON_AddNumberToObject(json_obj, "s_port", ntohs(addr->ipv6->source));
+            cJSON_AddNumberToObject(json_obj, "d_port", ntohs(addr->ipv6->dest));
+            break;
+        case ADDR_TYPE_VLAN:
+        case ADDR_TYPE_GRE:
+        case ADDR_TYPE_MPLS:
+        case ADDR_TYPE_PPPOE_SES:
+        case ADDR_TYPE_L2TP:
+        case ADDR_TYPE_PPP:
+            cJSON_AddNumberToObject(json_obj, "addr_type", addr->addrtype);
+            cJSON_AddStringToObject(json_obj, "s_ip", null_addr);
+            cJSON_AddStringToObject(json_obj, "d_ip", null_addr);
+            cJSON_AddNumberToObject(json_obj, "s_port", null_port);
+            cJSON_AddNumberToObject(json_obj, "d_port", null_port);
+            break;
+        case ADDR_TYPE_PPTP:
+            cJSON_AddNumberToObject(json_obj, "addr_type", addr->addrtype);
+            cJSON_AddStringToObject(json_obj, "s_ip", null_addr);
+            cJSON_AddStringToObject(json_obj, "d_ip", null_addr);
+            cJSON_AddNumberToObject(json_obj, "s_port", ntohs(addr->pptp->C2S_call_id));
+            cJSON_AddNumberToObject(json_obj, "d_port", ntohs(addr->pptp->S2C_call_id));
+            break;
+        default:
+            break;
+    }
 
+    addr_proto = layer_addr_prefix_ntop(a_stream);
+    cJSON_AddStringToObject(json_obj, "trans_proto", addr_proto);
+
+    ret = MESA_get_stream_opt(a_stream, MSO_STREAM_TUNNEL_TYPE, &tunnel_type, &tunnel_type_size);
+    assert(ret == 0);
+    if (tunnel_type == STREAM_TUNNLE_NON)
+    {
+        layer_addr_ntop_r(a_stream, nest_addr_buf, sizeof(nest_addr_buf));
+    }
+    else
+    {
+        stream_addr_list_ntop(a_stream, nest_addr_buf, sizeof(nest_addr_buf));
+    }
+
+    cJSON_AddStringToObject(json_obj, "addr_list", nest_addr_buf);
+
+    return 0;
+}
 
 void ntc_ssl_collect_send_kafka_log(rd_kafka_topic_t *topic, struct streaminfo *a_stream, comm_context_t *ctx, char *dpkt_buf, int dpkt_buflen, ssl_stream *a_ssl)
 {
-    if(a_stream == NULL || a_ssl == NULL)
+    if (a_stream == NULL || a_ssl == NULL)
     {
         return;
     }
-    if(a_ssl->stClientHello == NULL || a_ssl->stClientHello->server_name == NULL)
-    {
-        return ;
-    }
-      
-	cJSON *log_obj = cJSON_CreateObject();
 
-	cJSON_AddNumberToObject(log_obj, "service", g_ssl_collect_item.service);
-	cJSON_AddStringToObject(log_obj, "cap_ip", g_ssl_collect_item.local_ip_str);
-	cJSON_AddNumberToObject(log_obj, "entrance_id", g_ssl_collect_item.entry_id);
+    cJSON *log_obj = cJSON_CreateObject();
+
+    //cJSON_AddNumberToObject(log_obj, "service", g_ssl_collect_item.service);
+    cJSON_AddStringToObject(log_obj, "cap_ip", g_ssl_collect_item.local_ip_str);
+    cJSON_AddNumberToObject(log_obj, "entrance_id", g_ssl_collect_item.entry_id);
 
-	cJSON_AddNumberToObject(log_obj, "found_time", g_CurrentTime);
-	cJSON_AddNumberToObject(log_obj, "recv_time", g_CurrentTime);    
-	ntc_addStreamInfo_to_jsonObj(log_obj, a_stream);
+    cJSON_AddNumberToObject(log_obj, "found_time", g_CurrentTime);
+    cJSON_AddNumberToObject(log_obj, "recv_time", g_CurrentTime);
+    ntc_addStreamInfo_to_jsonObj(log_obj, a_stream);
 
     char tmp[128];
     sprintf(tmp, "%llu", ctx->c2s_pkts);
-	cJSON_AddStringToObject(log_obj, "c2s_pkt_num", tmp);	
+    cJSON_AddStringToObject(log_obj, "c2s_pkt_num", tmp);
     sprintf(tmp, "%llu", ctx->s2c_pkts);
-	cJSON_AddStringToObject(log_obj, "s2c_pkt_num", tmp);
-
-	sprintf(tmp, "%llu", ctx->c2s_bytes);
-	cJSON_AddStringToObject(log_obj, "c2s_byte_num", tmp);
-	
-	sprintf(tmp, "%llu", ctx->s2c_bytes);
-	cJSON_AddStringToObject(log_obj, "s2c_byte_num", tmp);
-
-	cJSON_AddStringToObject(log_obj, "SNI", (const char *)a_ssl->stClientHello->server_name);
-	
-	if(dpkt_buf != NULL && dpkt_buflen > 0)
-	{
-		cJSON_AddStringToObject(log_obj, "app_label", dpkt_buf);
-	}
-
-	cJSON_AddNumberToObject(log_obj, "create_time", a_stream->ptcpdetail->createtime);
-	cJSON_AddNumberToObject(log_obj, "lastmtime", a_stream->ptcpdetail->lastmtime);
-	
-	
-	//char *payload = cJSON_Print(log_obj);
-	char *payload = cJSON_PrintUnformatted(log_obj);
-	int paylen = strlen(payload);
-	rd_kafka_produce(g_ntc_kafka_topic, RD_KAFKA_PARTITION_UA, RD_KAFKA_MSG_F_COPY, payload, paylen, NULL, 0, NULL);
-
-	free(payload);
-	cJSON_Delete(log_obj);
-	log_obj = NULL;
-	return ;
-}
+    cJSON_AddStringToObject(log_obj, "s2c_pkt_num", tmp);
+
+    sprintf(tmp, "%llu", ctx->c2s_bytes);
+    cJSON_AddStringToObject(log_obj, "c2s_byte_num", tmp);
 
+    sprintf(tmp, "%llu", ctx->s2c_bytes);
+    cJSON_AddStringToObject(log_obj, "s2c_byte_num", tmp);
 
-int ntc_get_dpkt_label(struct streaminfo *a_stream, const char* label_name, char *label_buf, int *label_buflen)
+    if (a_ssl->stClientHello != NULL && a_ssl->stClientHello->server_name != NULL)
+    {
+        cJSON_AddStringToObject(log_obj, "SNI", (const char *)a_ssl->stClientHello->server_name);
+    }
+
+    if (dpkt_buf != NULL && dpkt_buflen > 0)
+    {
+        cJSON_AddStringToObject(log_obj, "app_label", dpkt_buf);
+    }
+
+    cJSON_AddNumberToObject(log_obj, "create_time", a_stream->ptcpdetail->createtime);
+    cJSON_AddNumberToObject(log_obj, "lastmtime", a_stream->ptcpdetail->lastmtime);
+
+    if (ctx->server_cert != NULL)
+    {
+        cJSON_AddItemToObject(log_obj, "server_cert", ctx->server_cert);
+    }
+    if (ctx->client_cert != NULL)
+    {
+        cJSON_AddItemToObject(log_obj, "client_cert", ctx->client_cert);
+    }
+    //char *payload = cJSON_Print(log_obj);
+    char *payload = cJSON_PrintUnformatted(log_obj);
+    int paylen = strlen(payload);
+    if(g_ssl_collect_item.comm_log_mode > 0)
+    {
+        rd_kafka_produce(g_ntc_kafka_topic, RD_KAFKA_PARTITION_UA, RD_KAFKA_MSG_F_COPY, payload, paylen, NULL, 0, NULL);
+    }
+    MESA_handle_runtime_log(g_ssl_collect_item.log_handle, RLOG_LV_DEBUG,   __FUNCTION__ , "%s", payload);
+    free(payload);
+    cJSON_Delete(log_obj);
+    log_obj = NULL;
+    return;
+}
+
+int ntc_get_dpkt_label(struct streaminfo *a_stream, const char *label_name, char *label_buf, int *label_buflen)
 {
-	dpkt_lable_t *dpkt_info = (dpkt_lable_t*)project_req_get_struct(a_stream,g_ssl_collect_item.dpkt_project_id);
-	if(dpkt_info == NULL)
-	{
-		*label_buflen = 0;
-		return -1;
-	}
-	snprintf(label_buf, *label_buflen, "PROTO_ID=%u;APP_ID=%u;OS_ID=%u;BS_ID=%u;WEB_ID=%u;BEHAV_ID=%u;",
-			dpkt_info->dpkt_proto_type,	        
-			dpkt_info->dpkt_app_type,	        
-			dpkt_info->dpkt_op_type,	        
-			dpkt_info->dpkt_browser_type,	        
-			dpkt_info->dpkt_web_type,	        
-			dpkt_info->dpkt_behavior_type);
-	*label_buflen = strlen(label_buf);
-	return 0;
+    dpkt_lable_t *dpkt_info = (dpkt_lable_t *)project_req_get_struct(a_stream, g_ssl_collect_item.dpkt_project_id);
+    if (dpkt_info == NULL)
+    {
+        *label_buflen = 0;
+        return -1;
+    }
+    snprintf(label_buf, *label_buflen, "PROTO_ID=%u;APP_ID=%u;OS_ID=%u;BS_ID=%u;WEB_ID=%u;BEHAV_ID=%u;",
+            dpkt_info->dpkt_proto_type,
+            dpkt_info->dpkt_app_type,
+            dpkt_info->dpkt_op_type,
+            dpkt_info->dpkt_browser_type,
+            dpkt_info->dpkt_web_type,
+            dpkt_info->dpkt_behavior_type);
+    *label_buflen = strlen(label_buf);
+    return 0;
 }
 
 int nct_get_flow_stat(comm_context_t *ctx, struct streaminfo *a_stream, int flow_id)
 {
-	struct tcp_flow_stat *tflow_project;
-	struct udp_flow_stat *uflow_project;	
-	if(flow_id < 0 )
-	{
-	    return -1;
-	}
-	if(a_stream->type == STREAM_TYPE_TCP)
-	{
-        tflow_project = (struct tcp_flow_stat *)project_req_get_struct(a_stream,flow_id);
-        if(tflow_project != NULL)
+    struct tcp_flow_stat *tflow_project;
+    struct udp_flow_stat *uflow_project;
+    if (flow_id < 0)
+    {
+        return -1;
+    }
+    if (a_stream->type == STREAM_TYPE_TCP)
+    {
+        tflow_project = (struct tcp_flow_stat *)project_req_get_struct(a_stream, flow_id);
+        if (tflow_project != NULL)
         {
-                    ctx->c2s_bytes = tflow_project->C2S_data_byte;
-                    ctx->s2c_bytes = tflow_project->S2C_data_byte;
-                    ctx->c2s_pkts = tflow_project->C2S_data_pkt;
-                    ctx->s2c_pkts = tflow_project->S2C_data_pkt;                    
+            ctx->c2s_bytes = tflow_project->C2S_data_byte;
+            ctx->s2c_bytes = tflow_project->S2C_data_byte;
+            ctx->c2s_pkts = tflow_project->C2S_data_pkt;
+            ctx->s2c_pkts = tflow_project->S2C_data_pkt;
         }
-	}
-	else if(a_stream->type == STREAM_TYPE_UDP)
-	{
+    }
+    else if (a_stream->type == STREAM_TYPE_UDP)
+    {
         uflow_project = (struct udp_flow_stat *)project_req_get_struct(a_stream, flow_id);
-        if(uflow_project != NULL)
+        if (uflow_project != NULL)
         {
-                    ctx->c2s_bytes = uflow_project->C2S_byte;
-                    ctx->s2c_bytes = uflow_project->S2C_byte;
-                    ctx->c2s_pkts = uflow_project->C2S_pkt;
-                    ctx->s2c_pkts = uflow_project->S2C_pkt;                    
-        }        
-	}
-	else
-	{
-	    return -1;
-	}
-	return 0;
+            ctx->c2s_bytes = uflow_project->C2S_byte;
+            ctx->s2c_bytes = uflow_project->S2C_byte;
+            ctx->c2s_pkts = uflow_project->C2S_pkt;
+            ctx->s2c_pkts = uflow_project->S2C_pkt;
+        }
+    }
+    else
+    {
+        return -1;
+    }
+    return 0;
+}
+
+int ntc_ssl_store_cert(comm_context_t *ctx, struct streaminfo *a_stream, ssl_stream *a_ssl)
+{
+    st_cert_t *cert = a_ssl->stSSLCert;
+    if(cert == NULL)return -1;
+
+    if (cert->cert_type != CERT_TYPE_INDIVIDUAL)
+    {
+        return 0;
+    }
+    cJSON *cert_obj = cJSON_CreateObject();
+
+    if(strlen(cert->SSLVersion) > 0)cJSON_AddStringToObject(cert_obj, "version", cert->SSLVersion);
+
+    if(strlen(cert->SSLSerialNum) > 0)
+    {
+        char serial_buf[512] = "0x";
+        for(unsigned int i = 0; i < strlen(cert->SSLSerialNum); i++)
+        {
+           sprintf(serial_buf, "%s%02X", serial_buf, cert->SSLSerialNum[i]); 
+        }
+        cJSON_AddStringToObject(cert_obj, "serial_number", serial_buf);
+        //cJSON_AddStringToObject(cert_obj, "serial_number", cert->SSLSerialNum);
+    }
+
+    if(strlen(cert->SSLFPAg) > 0)cJSON_AddStringToObject(cert_obj, "algorithm", cert->SSLFPAg);
+
+    if(strlen(cert->SSLAgID) > 0)cJSON_AddStringToObject(cert_obj, "algorithm_id", cert->SSLAgID);
+
+    if(strlen(cert->SSLIssuer) > 0)cJSON_AddStringToObject(cert_obj, "issuer", cert->SSLIssuer);
+
+    if(strlen(cert->SSLIssuerC) > 0)cJSON_AddStringToObject(cert_obj, "issue_country", cert->SSLIssuerC);
+
+    if(strlen(cert->SSLIssuerO) > 0)cJSON_AddStringToObject(cert_obj, "issue_organize", cert->SSLIssuerO);
+
+    if(strlen(cert->SSLIssuerCN) > 0)cJSON_AddStringToObject(cert_obj, "issue_cname", cert->SSLIssuerCN);
+
+    if(strlen(cert->SSLSub) > 0)cJSON_AddStringToObject(cert_obj, "sub", cert->SSLSub);
+
+    if(strlen(cert->SSLSubC) > 0)cJSON_AddStringToObject(cert_obj, "sub_country", cert->SSLSubC);
+
+    if(strlen(cert->SSLSubO) > 0)cJSON_AddStringToObject(cert_obj, "sub_organize", cert->SSLSubO);
+
+    if(strlen(cert->SSLSubCN) > 0)cJSON_AddStringToObject(cert_obj, "sub_cname", cert->SSLSubCN);
+
+    if(strlen(cert->SSLFrom) > 0)cJSON_AddStringToObject(cert_obj, "start_time", cert->SSLFrom);
+
+    if(strlen(cert->SSLTo) > 0)cJSON_AddStringToObject(cert_obj, "expire_time", cert->SSLTo);
+
+    if(cert->SSLSubAltName != NULL && cert->SSLSubAltName->count > 0)
+    {
+        //cJSON *san = cJSON_CreateStringArray((const char **)cert->SSLSubAltName->san_array, cert->SSLSubAltName->count);
+        cJSON *san = cJSON_CreateArray();
+        cJSON *member = NULL;
+        for(int i = 0; i < cert->SSLSubAltName->count; i++)
+        {
+            member = cJSON_CreateString(cert->SSLSubAltName->san_array[i].san);
+            cJSON_AddItemToArray(san, member);
+        }
+
+        cJSON_AddItemToObject(cert_obj, "san_array", san);
+    }
+    if (a_stream->curdir == DIR_C2S)
+    {
+        ctx->client_cert = cert_obj;
+    }
+    else if (a_stream->curdir == DIR_S2C)
+    {
+        ctx->server_cert = cert_obj; 
+    }
+    else
+    {
+        cJSON_Delete(cert_obj);
+        cert_obj = NULL;
+        return -1;
+    }
+    return 1;
+}
+
+void ntc_ssl_collect_close( comm_context_t * ctx, struct streaminfo *a_stream, ssl_stream *a_ssl)
+{
+    char label_buf[128] = {0};
+    int label_buflen = sizeof(label_buf);
+    nct_get_flow_stat(ctx, a_stream, g_ssl_collect_item.tcp_flow_id);
+    ntc_get_dpkt_label(a_stream, g_ssl_collect_item.dpkt_label, label_buf, &label_buflen);
+    ntc_ssl_collect_send_kafka_log(g_ntc_kafka_topic, a_stream, ctx, label_buf, label_buflen, a_ssl);
+    if(ctx != NULL)
+    {
+        free(ctx);
+        ctx=NULL;
+    }
+    return;
 }
 
 extern "C" UCHAR ntc_ssl_collect_entry(stSessionInfo *session_info, void **param, int thread_seq, struct streaminfo *a_stream, void *a_packet)
 {
-      unsigned char ret = APP_STATE_GIVEME;
-      char label_buf[128] = {0};
-      int label_buflen = sizeof(label_buf);      
-      comm_context_t ctx;
-      ssl_stream *a_ssl;
-      switch (a_stream->opstate)
-      {           
-          case OP_STATE_CLOSE:
-                    nct_get_flow_stat(&ctx, a_stream, g_ssl_collect_item.tcp_flow_id);                    
-                    ntc_get_dpkt_label(a_stream, g_ssl_collect_item.dpkt_label, label_buf, &label_buflen);       
-                    a_ssl = (ssl_stream *)session_info->app_info;
-            		ntc_ssl_collect_send_kafka_log(g_ntc_kafka_topic, a_stream, &ctx, label_buf, label_buflen, a_ssl);
-            		ret= APP_STATE_DROPME;	
-    		    break;
+    unsigned char ret = PROT_STATE_GIVEME;
+    comm_context_t *ctx = NULL;
+    if ((session_info->session_state & SESSION_STATE_PENDING) == SESSION_STATE_PENDING)
+    {
+        *param = calloc(sizeof(comm_context_t), 1);
+    }
+    ctx = (comm_context_t *)*param;
+    switch (session_info->prot_flag)
+    {
+        case SSL_CERTIFICATE_DETAIL:
+            if (session_info->app_info != NULL)
+            {
+                ntc_ssl_store_cert(ctx, a_stream, (ssl_stream *)session_info->app_info);
+            }
+            break;
+        case SSL_APPLICATION_DATA:
+            ntc_ssl_collect_close(ctx, a_stream, (ssl_stream *)session_info->app_info);
+            ret = PROT_STATE_DROPME;
         default:
-                break;
-      }
-	return ret ;
+            break;
+    }
+    if((session_info->session_state & SESSION_STATE_CLOSE) == SESSION_STATE_CLOSE && ret != PROT_STATE_DROPME)
+    {
+        ntc_ssl_collect_close(ctx, a_stream, (ssl_stream *)session_info->app_info);
+        ret = PROT_STATE_DROPME;
+    }
+    return ret;
 }
 
 int ntc_ssl_collect_load_profile()
 {
-	MESA_load_profile_string_def(PROFILE_PATH, PLUGIN_NAME, "log_path", g_ssl_collect_item.log_path, sizeof(g_ssl_collect_item.log_path), "./ntclog/ip_comm_log");
-	MESA_load_profile_uint_def(PROFILE_PATH,PLUGIN_NAME, "log_level", &g_ssl_collect_item.log_level, 30);
-
-	char nic_name[64];
-	MESA_load_profile_string_def(PROFILE_PATH,"SYSTEM", "NIC_NAME",nic_name,sizeof(nic_name),"eth0");
-	g_ssl_collect_item.local_ip_nr=get_ip_by_eth_name(nic_name);
-	if(g_ssl_collect_item.local_ip_nr==INADDR_NONE)
-	{
-		printf("get_ip_by_eth_name in %s return NULL, exit!\n", nic_name);
-            return -1;
-	}
-	inet_ntop(AF_INET,&(g_ssl_collect_item.local_ip_nr),g_ssl_collect_item.local_ip_str,sizeof(g_ssl_collect_item.local_ip_str));
-	MESA_load_profile_int_def(PROFILE_PATH,"SYSTEM", "ENTRANCE_ID",&(g_ssl_collect_item.entry_id),0);
+    MESA_load_profile_string_def(PROFILE_PATH, PLUGIN_NAME, "log_path", g_ssl_collect_item.log_path, sizeof(g_ssl_collect_item.log_path), "./ntclog/ip_comm_log");
+    MESA_load_profile_uint_def(PROFILE_PATH, PLUGIN_NAME, "log_level", &g_ssl_collect_item.log_level, 30);
 
-	MESA_load_profile_int_def(PROFILE_PATH,PLUGIN_NAME, "service", &g_ssl_collect_item.service, 0);
-	MESA_load_profile_string_def(PROFILE_PATH, PLUGIN_NAME, "dpkt_label", g_ssl_collect_item.dpkt_label, sizeof(g_ssl_collect_item.dpkt_label), "DPKT_PROJECT");
-	return 0;
+    char nic_name[64];
+    MESA_load_profile_string_def(PROFILE_PATH, "SYSTEM", "NIC_NAME", nic_name, sizeof(nic_name), "eth0");
+    g_ssl_collect_item.local_ip_nr = get_ip_by_eth_name(nic_name);
+    if (g_ssl_collect_item.local_ip_nr == INADDR_NONE)
+    {
+        printf("get_ip_by_eth_name in %s return NULL, exit!\n", nic_name);
+        return -1;
+    }
+    inet_ntop(AF_INET, &(g_ssl_collect_item.local_ip_nr), g_ssl_collect_item.local_ip_str, sizeof(g_ssl_collect_item.local_ip_str));
+    MESA_load_profile_int_def(PROFILE_PATH, "SYSTEM", "ENTRANCE_ID", &(g_ssl_collect_item.entry_id), 0);
+
+    MESA_load_profile_int_def(PROFILE_PATH, PLUGIN_NAME, "service", &g_ssl_collect_item.service, 0);
+    MESA_load_profile_string_def(PROFILE_PATH, PLUGIN_NAME, "dpkt_label", g_ssl_collect_item.dpkt_label, sizeof(g_ssl_collect_item.dpkt_label), "DPKT_PROJECT");
+    return 0;
 }
 
 int ntc_ssl_collect_kaka_init()
 {
-	char kafka_errstr[1024];	
-	MESA_load_profile_uint_def(PROFILE_PATH,PLUGIN_NAME, "kafka_mode", &g_ssl_collect_item.comm_log_mode, 0);
-	MESA_load_profile_string_def(PROFILE_PATH, PLUGIN_NAME, "kafka_topic", g_ssl_collect_item.kafka_topic, sizeof(g_ssl_collect_item.kafka_topic), NTC_SSL_COLLECT_TOPIC);
+    char kafka_errstr[1024];
+    MESA_load_profile_uint_def(PROFILE_PATH, PLUGIN_NAME, "kafka_mode", &g_ssl_collect_item.comm_log_mode, 3);
+    MESA_load_profile_string_def(PROFILE_PATH, PLUGIN_NAME, "kafka_topic", g_ssl_collect_item.kafka_topic, sizeof(g_ssl_collect_item.kafka_topic), NTC_SSL_COLLECT_TOPIC);
+
+    if(g_ssl_collect_item.comm_log_mode == 0)return 0;
+    if ((g_ssl_collect_item.comm_log_mode & INDIE_KAFKA) == INDIE_KAFKA)
+    {
+        if (0 > MESA_load_profile_string_nodef(PROFILE_PATH, PLUGIN_NAME, "kafka_brokelist", g_ssl_collect_item.kafka_brokelist, sizeof(g_ssl_collect_item.kafka_brokelist)))
+        {
+            return -1;
+        }
+        rd_kafka_conf_t *rdkafka_conf = rd_kafka_conf_new();
+        rd_kafka_conf_set(rdkafka_conf, "queue.buffering.max.messages", "1000000", kafka_errstr, sizeof(kafka_errstr));
+        rd_kafka_conf_set(rdkafka_conf, "topic.metadata.refresh.interval.ms", "600000", kafka_errstr, sizeof(kafka_errstr));
+        rd_kafka_conf_set(rdkafka_conf, "security.protocol", "MG", kafka_errstr, sizeof(kafka_errstr));
 
-	if((g_ssl_collect_item.comm_log_mode&INDIE_KAFKA)  == INDIE_KAFKA)
+        if (!(g_ntc_kafka_handle = rd_kafka_new(RD_KAFKA_PRODUCER, rdkafka_conf, kafka_errstr, sizeof(kafka_errstr))))
+        {
+            return -1;
+        }
+
+        if (rd_kafka_brokers_add(g_ntc_kafka_handle, g_ssl_collect_item.kafka_brokelist) == 0)
+        {
+            return -1;
+        }
+    }
+    else
     {
-    	if(0 > MESA_load_profile_string_nodef(PROFILE_PATH, PLUGIN_NAME, "kafka_brokelist", g_ssl_collect_item.kafka_brokelist, sizeof(g_ssl_collect_item.kafka_brokelist)))
-    	{
-    		return -1;
-    	}
-    	rd_kafka_conf_t *rdkafka_conf = rd_kafka_conf_new();
-    	rd_kafka_conf_set(rdkafka_conf, "queue.buffering.max.messages", "1000000", kafka_errstr, sizeof(kafka_errstr));
-    	rd_kafka_conf_set(rdkafka_conf, "topic.metadata.refresh.interval.ms", "600000",kafka_errstr, sizeof(kafka_errstr));
-    	rd_kafka_conf_set(rdkafka_conf, "security.protocol", "MG", kafka_errstr, sizeof(kafka_errstr));
-
-    	if (!(g_ntc_kafka_handle = rd_kafka_new(RD_KAFKA_PRODUCER, rdkafka_conf, kafka_errstr, sizeof(kafka_errstr))))
-    	{
-    		return -1;
-    	}
-
-    	if (rd_kafka_brokers_add(g_ntc_kafka_handle, g_ssl_collect_item.kafka_brokelist) == 0)
-    	{
-    		return -1;
-    	}
-	}
-	else
-	{
-		if(0 > MESA_load_profile_string_nodef(PROFILE_PATH, PLUGIN_NAME, "kafka_handle_provide_path", g_ssl_collect_item.kafka_handle_provide_path, sizeof(g_ssl_collect_item.kafka_handle_provide_path)))
-		{
-			return -1;
-		}
-		void * dl_handle = dlopen(g_ssl_collect_item.kafka_handle_provide_path, RTLD_NOW|RTLD_GLOBAL);
-		if(dl_handle == NULL)
-		{
-			return -1;
-		}
-		if(0 > MESA_load_profile_string_nodef(PROFILE_PATH, PLUGIN_NAME, "kafka_handle_name", g_ssl_collect_item.kafka_handle_name, sizeof(g_ssl_collect_item.kafka_handle_name)))
-		{
-			return -1;
-		}
-		void *dl_return = dlsym(dl_handle, g_ssl_collect_item.kafka_handle_name);
-
-		if(dl_return == NULL)
-		{
-			return -1;
-		}
-		g_ntc_kafka_handle = *(rd_kafka_t **)dl_return;
-		if(g_ntc_kafka_handle == NULL)
-		{
-			return -1;
-		}        
-
-	}	
-	rd_kafka_topic_conf_t*topic_conf = rd_kafka_topic_conf_new();
-	g_ntc_kafka_topic = rd_kafka_topic_new(g_ntc_kafka_handle, g_ssl_collect_item.kafka_topic, topic_conf);
-	return 0;
+        if (0 > MESA_load_profile_string_nodef(PROFILE_PATH, PLUGIN_NAME, "kafka_handle_provide_path", g_ssl_collect_item.kafka_handle_provide_path, sizeof(g_ssl_collect_item.kafka_handle_provide_path)))
+        {
+            return -1;
+        }
+        void *dl_handle = dlopen(g_ssl_collect_item.kafka_handle_provide_path, RTLD_NOW | RTLD_GLOBAL);
+        if (dl_handle == NULL)
+        {
+            return -1;
+        }
+        if (0 > MESA_load_profile_string_nodef(PROFILE_PATH, PLUGIN_NAME, "kafka_handle_name", g_ssl_collect_item.kafka_handle_name, sizeof(g_ssl_collect_item.kafka_handle_name)))
+        {
+            return -1;
+        }
+        void *dl_return = dlsym(dl_handle, g_ssl_collect_item.kafka_handle_name);
+
+        if (dl_return == NULL)
+        {
+            return -1;
+        }
+        g_ntc_kafka_handle = *(rd_kafka_t **)dl_return;
+        if (g_ntc_kafka_handle == NULL)
+        {
+            return -1;
+        }
+    }
+    rd_kafka_topic_conf_t *topic_conf = rd_kafka_topic_conf_new();
+    g_ntc_kafka_topic = rd_kafka_topic_new(g_ntc_kafka_handle, g_ssl_collect_item.kafka_topic, topic_conf);
+    return 0;
 }
 
 extern "C" int ntc_ssl_collect_init()
 {
-	memset(&g_ssl_collect_item, 0, sizeof(g_ssl_collect_item));
-	if(0 != ntc_ssl_collect_load_profile())
-	{
-	    return ERROR;
-	}
-	g_ssl_collect_item.log_handle = MESA_create_runtime_log_handle(g_ssl_collect_item.log_path, g_ssl_collect_item.log_level);
-	if(g_ssl_collect_item.log_handle == NULL)
-		return ERROR;
-
-
-	g_ssl_collect_item.dpkt_project_id = project_customer_register(g_ssl_collect_item.dpkt_label, "struct");
-	g_ssl_collect_item.tcp_flow_id = project_customer_register("tcp_flow_stat", "struct");		
-
-	if(0 != ntc_ssl_collect_kaka_init())
-		return ERROR;
-	return OK;
-}
+    memset(&g_ssl_collect_item, 0, sizeof(g_ssl_collect_item));
+    if (0 != ntc_ssl_collect_load_profile())
+    {
+        return ERROR;
+    }
+    g_ssl_collect_item.log_handle = MESA_create_runtime_log_handle(g_ssl_collect_item.log_path, g_ssl_collect_item.log_level);
+    if (g_ssl_collect_item.log_handle == NULL)
+        return ERROR;
 
+    g_ssl_collect_item.dpkt_project_id = project_customer_register(g_ssl_collect_item.dpkt_label, "struct");
+    g_ssl_collect_item.tcp_flow_id = project_customer_register("tcp_flow_stat", "struct");
 
+    if (0 != ntc_ssl_collect_kaka_init())
+        return ERROR;
+    return OK;
+}
 
 extern "C" void ntc_ssl_collect_destroy(void)
 {
-	MESA_destroy_runtime_log_handle(g_ssl_collect_item.log_handle);
-	return;
+    MESA_destroy_runtime_log_handle(g_ssl_collect_item.log_handle);
+    return;
 }
diff --git a/src/ntc_ssl_collect.h b/src/ntc_ssl_collect.h
index 9b3c9b7..74115ec 100644
--- a/src/ntc_ssl_collect.h
+++ b/src/ntc_ssl_collect.h
@@ -9,6 +9,7 @@ extern "C"
 
 #include "stream.h"
 //#include "soq/identify_v2.h"
+#include <cJSON.h>
 
 	typedef   struct  ntc_ssl_collect_global_item
 	{
@@ -74,6 +75,8 @@ typedef struct _comm_context_t
     unsigned long long s2c_pkts;
     unsigned long long c2s_bytes;
     unsigned long long s2c_bytes;
+	cJSON *client_cert;
+	cJSON *server_cert;
  }comm_context_t;