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 +#include +#include 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 #include +#include #include #include #include @@ -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);