#177 在用户访问的过程中，缓存未见到过、可信的中间证书到 kafka（TODO kafka 数据录入格式待沟通，待联调测试）

3 files changed, 123 insertions, 7 deletions

diff --git a/platform/include/internal/ssl_fetch_cert.h b/platform/include/internal/ssl_fetch_cert.h
index 750582e..203c7a6 100644
--- a/platform/include/internal/ssl_fetch_cert.h
+++ b/platform/include/internal/ssl_fetch_cert.h
@@ -5,7 +5,9 @@
 #ifndef TFE_SSL_FETCH_CERT_H
 #define TFE_SSL_FETCH_CERT_H
 
-void ssl_fetch_cert_url_by_aia(X509 *cert);
+// return 0 for success, return -1 for failed
+int ssl_mid_cert_kafka_logger_create(const char *profile, const char *section);
+void ssl_mid_cert_kafka_logger_destory(void);
 void ssl_fetch_trusted_cert_from_chain(STACK_OF(X509) * cert_chain, X509_STORE *trusted_store);
 
 #endif //TFE_SSL_FETCH_CERT_H
\ No newline at end of file
diff --git a/platform/src/ssl_fetch_cert.cpp b/platform/src/ssl_fetch_cert.cpp
index 32811e5..d6728e7 100644
--- a/platform/src/ssl_fetch_cert.cpp
+++ b/platform/src/ssl_fetch_cert.cpp
@@ -3,7 +3,11 @@
 //
 
 #include "ssl_utils.h"
+#include "tfe_utils.h"
+
 #include <assert.h>
+#include <librdkafka/rdkafka.h>
+#include <MESA/MESA_prof_load.h>
 
 typedef struct x509_object_st {
 	int type;
@@ -15,12 +19,113 @@ typedef struct x509_object_st {
 	} data;
 } X509_OBJECT;
 
+typedef struct ssl_kafka_logger_s {
+	int enable;
+	char brokerlist[TFE_STRING_MAX];
+	char topicname[TFE_STRING_MAX];
+
+	rd_kafka_t *handle;
+	rd_kafka_topic_t *topic;
+} ssl_kafka_logger_t;
+
+static ssl_kafka_logger_t *g_kafka_logger = NULL;
+
+static rd_kafka_t *create_kafka_handle(const char *brokerlist) {
+	char kafka_errstr[1024];
+	rd_kafka_t *handle = NULL;
+	rd_kafka_conf_t *rdkafka_conf = NULL;
+
+	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));
+
+	//The conf object is freed by this function and must not be used or destroyed by the application sub-sequently.
+	handle = rd_kafka_new(RD_KAFKA_PRODUCER, rdkafka_conf, kafka_errstr, sizeof(kafka_errstr));
+	rdkafka_conf = NULL;
+	if (handle == NULL) {
+		return NULL;
+	}
+	if (rd_kafka_brokers_add(handle, brokerlist) == 0) {
+		rd_kafka_destroy(handle);
+		return NULL;
+	}
+	return handle;
+}
+
+void ssl_mid_cert_kafka_logger_destory(void) {
+	if (g_kafka_logger) {
+		if (g_kafka_logger->handle) {
+			free(g_kafka_logger->handle);
+		}
+		free(g_kafka_logger);
+	}
+}
+
+int ssl_mid_cert_kafka_logger_create(const char *profile, const char *section) {
+	const char *errstr = "ssl mid cert cache kafka logger create failed";
+
+	g_kafka_logger = ALLOC(ssl_kafka_logger_t, 1);
+	assert(g_kafka_logger);
+
+	MESA_load_profile_int_def(profile, section, "mid_cert_cache_kafka_enable", &(g_kafka_logger->enable), 0);
+	if (!g_kafka_logger->enable) {
+		return 0;
+	}
+
+	if (MESA_load_profile_string_def(profile, section, "mid_cert_cache_kafka_brokerlist", g_kafka_logger->brokerlist,
+									 sizeof(g_kafka_logger->brokerlist), NULL) < 0) {
+		TFE_LOG_ERROR(g_default_logger, "%s, No brokerlist in profile %s section %s.", errstr, profile, section);
+		goto error;
+	}
+	g_kafka_logger->handle = create_kafka_handle(g_kafka_logger->brokerlist);
+	if (g_kafka_logger->handle == NULL) {
+		TFE_LOG_ERROR(g_default_logger, "%s, Cannot create kafka handle with brokerlist: %s.", errstr,
+					  g_kafka_logger->brokerlist);
+		goto error;
+	}
+
+
+	MESA_load_profile_string_def(profile, section, "mid_cert_cache_kafka_topic", g_kafka_logger->topicname,
+								 sizeof(g_kafka_logger->topicname), "MID-CERT-CACHE-LOG");
+	g_kafka_logger->topic = rd_kafka_topic_new(g_kafka_logger->handle, g_kafka_logger->topicname, NULL);
+	if (g_kafka_logger->topic == NULL) {
+		TFE_LOG_ERROR(g_default_logger, "%s, Cannot create kafka handle with brokerlist: %s.", errstr,
+					  g_kafka_logger->brokerlist);
+		goto error;
+	}
+
+	return 0;
+
+	error:
+	ssl_mid_cert_kafka_logger_destory();
+	return -1;
+}
+
+void ssl_mid_cert_kafka_logger_send(char *msg) {
+	if (g_kafka_logger == NULL || g_kafka_logger->enable == 0) {
+		return;
+	}
+	rd_kafka_produce(g_kafka_logger->topic, RD_KAFKA_PARTITION_UA, RD_KAFKA_MSG_F_COPY, msg, strlen(msg), NULL, 0,
+					 NULL);
+}
+
+int ssl_mid_cert_kafka_logger_enable() {
+	if (g_kafka_logger && g_kafka_logger->enable) {
+		return 1;
+	} else {
+		return 0;
+	}
+}
 
 // test use http://www.360.cn/
 void ssl_fetch_trusted_cert_from_chain(STACK_OF(X509) *cert_chain, X509_STORE *trusted_store) {
-	// 证书链中的证书下标为 [0, count - 1]，下标为 count - 1 的证书不一定在可信证书列表中
+	if (!ssl_mid_cert_kafka_logger_enable()) {
+		return;
+	}
+
+	// range for [0, count - 1]
 	int count = sk_X509_num(cert_chain);
-	printf("------------------ max depth is : %d\n", count);
 
 	// don`t need call X509_LOOKUP_free(lookup)
 	X509_LOOKUP *lookup = X509_STORE_add_lookup(trusted_store, X509_LOOKUP_hash_dir());
@@ -40,15 +145,17 @@ void ssl_fetch_trusted_cert_from_chain(STACK_OF(X509) *cert_chain, X509_STORE *t
 		char *subj = ssl_x509_subject(cert);
 		char *issuer = ssl_x509_issuer(cert);
 		if (result) {
-			printf("[dep:%d] subject:%s; issure:%s; in_trusted_store:1\n", i, subj, issuer);
+			TFE_LOG_ERROR(g_default_logger, "[dep:%d/%d] subject:%s; issure:%s; in_trusted_store:1\n", i, count, subj,
+						  issuer);
 			// not use continue, case the intermediate certificate is exist and the root certificate is not exist.
 			/* continue; */
 		} else {
-			printf("[dep:%d] subject:%s; issure:%s; in_trusted_store:0\n", i, subj, issuer);
+			TFE_LOG_ERROR(g_default_logger, "[dep:%d/%d] subject:%s; issure:%s; in_trusted_store:0\n", i, count, subj,
+						  issuer);
 			char *string = ssl_x509_to_str(cert);
 			if (string) {
-				// TODO log kafka
-				printf("%s\n", string);
+				// printf("%s\n", string);
+				ssl_mid_cert_kafka_logger_send(string);
 				free(string);
 			}
 		}
diff --git a/platform/src/ssl_stream.cpp b/platform/src/ssl_stream.cpp
index 70429ac..ce37fcb 100644
--- a/platform/src/ssl_stream.cpp
+++ b/platform/src/ssl_stream.cpp
@@ -31,6 +31,7 @@
 #include <MESA/MESA_htable.h>
 #include <MESA/MESA_prof_load.h>
 
+#include <ssl_fetch_cert.h>
 #include <tfe_stream.h>
 #include <tfe_utils.h>
 #include <tfe_future.h>
@@ -622,6 +623,12 @@ struct ssl_mgr * ssl_manager_init(const char * ini_profile, const char * section
 	char version_str[TFE_SYMBOL_MAX] = {};
 	mgr->logger = logger;
 	mgr->ev_base_gc=ev_base_gc;
+
+	if (ssl_mid_cert_kafka_logger_create(ini_profile, section))
+	{
+		goto error_out;
+	}
+
 	MESA_load_profile_string_def(ini_profile, section, "ssl_min_version", version_str, sizeof(version_str), "ssl3");
 	mgr->ssl_min_version = sslver_str2num(version_str);