gfwleak/tango-tfe
Archived
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

TSG-148 修改录入 kafka 数据格式，修正代码格式

Browse Source
This commit is contained in:
luwenpeng
2019-12-06 15:51:03 +08:00
parent daa1899f91
commit 67c27a2fb5
3 changed files with 131 additions and 75 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
platform/include/internal/ssl_fetch_cert.h
View File

@@ -8,6 +8,6 @@
// 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);
void ssl_fetch_trusted_cert_from_chain(STACK_OF(X509) *cert_chain, X509_STORE *trusted_store, const char *hostname);
#endif //TFE_SSL_FETCH_CERT_H

202
platform/src/ssl_fetch_cert.cpp
View File

@@ -6,50 +6,82 @@
#include "tfe_utils.h"
#include <assert.h>
#include <cjson/cJSON.h>
#include <librdkafka/rdkafka.h>
#include <MESA/MESA_prof_load.h>
typedef struct x509_object_st {
int type;
union {
char *ptr;
X509 *x509;
char *ptr;
X509 *x509;
X509_CRL *crl;
EVP_PKEY *pkey;
} data;
} X509_OBJECT;
typedef struct ssl_kafka_logger_s {
int enable;
char brokerlist[TFE_STRING_MAX];
char topicname[TFE_STRING_MAX];
int enable;
rd_kafka_t *handle;
char tfe_ip[TFE_SYMBOL_MAX];
char topic_name[TFE_STRING_MAX];
char broker_list[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;
static unsigned int get_ip_by_eth(const char *eth) {
int sockfd = -1;
unsigned int ip;
struct ifreq ifr;
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));
sockfd = socket(AF_INET, SOCK_DGRAM, 0);
if (-1 == sockfd) {
goto error;
}
//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;
memset(&ifr, 0, sizoef(ifr));
strcpy(ifr.ifr_name, eth);
if (ioctl(sockfd, SIOCGIFADDR, &ifr) < 0) {
goto error;
}
ip = ((struct sockaddr_in *)&(ifr.ifr_addr))->sin_addr.s_addr;
close(sockfd);
return ip;
error:
if (sockfd > 0)
close(sockfd);
return INADDR_NONE;
}
static rd_kafka_t *create_kafka_handle(const char *broker_list) {
char errstr[1024];
rd_kafka_t *handle = NULL;
rd_kafka_conf_t *conf = NULL;
conf = rd_kafka_conf_new();
rd_kafka_conf_set(conf, "queue.buffering.max.messages", "1000000", errstr, sizeof(errstr));
rd_kafka_conf_set(conf, "topic.metadata.refresh.interval.ms", "600000", errstr, sizeof(errstr));
rd_kafka_conf_set(conf, "security.protocol", "MG", errstr, sizeof(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, conf, errstr, sizeof(errstr));
conf = NULL;
if (handle == NULL) {
return NULL;
}
if (rd_kafka_brokers_add(handle, brokerlist) == 0) {
if (rd_kafka_brokers_add(handle, broker_list) == 0) {
rd_kafka_destroy(handle);
return NULL;
}
return handle;
}
@@ -58,108 +90,132 @@ void ssl_mid_cert_kafka_logger_destory(void) {
if (g_kafka_logger->handle) {
free(g_kafka_logger->handle);
}
if (g_kafka_logger->topic) {
free(g_kafka_logger->topic)
}
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";
int ssl_mid_cert_kafka_logger_create(const char *profile, const char *section)
{
unsigned int ip;
char eth[64] = {0};
const char *errstr = "SSL mid cert cache occer error, ";
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);
MESA_load_profile_int_def(profile, section, "mc_cache_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);
MESA_load_profile_string_def(profile, section, "mc_cache_eth", eth, sizeof(eth), "eth0");
ip = get_ip_by_eth(eth);
if (ip == INADDR_NONE) {
TFE_LOG_ERROR(g_default_logger, "%s, Fail to get ip by %s.", errstr, eth);
goto error;
}
g_kafka_logger->handle = create_kafka_handle(g_kafka_logger->brokerlist);
inet_ntop(AF_INET, &ip, g_kafka_logger->tfe_ip, sizeof(g_kafka_logger->tfe_ip));
if (MESA_load_profile_string_def(profile, section, "mc_cache_broker_list", g_kafka_logger->broker_list, sizeof(g_kafka_logger->broker_list), NULL) < 0) {
TFE_LOG_ERROR(g_default_logger, "%s, Fail to get mc_cache_broker_list in profile %s section %s.", errstr, profile, section);
goto error;
}
g_kafka_logger->handle = create_kafka_handle(g_kafka_logger->broker_list);
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);
TFE_LOG_ERROR(g_default_logger, "%s, Fail to create kafka handle with broker list: %s.", errstr, g_kafka_logger->broker_list);
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);
MESA_load_profile_string_def(profile, section, "mc_cache_topic", g_kafka_logger->topic_name, sizeof(g_kafka_logger->topic_name), "MID-CERT-CACHE-LOG");
g_kafka_logger->topic = rd_kafka_topic_new(g_kafka_logger->handle, g_kafka_logger->topic_name, 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);
TFE_LOG_ERROR(g_default_logger, "%s, Fail to create kafka topic with broker list: %s.", errstr, g_kafka_logger->broker_list);
goto error;
}
return 0;
error:
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) {
void ssl_mid_cert_kafka_logger_send(const char *sni, const char *fingerprint, const char *cert)
{
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);
}
cJSON *obj = NULL;
cJSON *dup = NULL;
char *msg = NULL;
int ssl_mid_cert_kafka_logger_enable() {
if (g_kafka_logger && g_kafka_logger->enable) {
return 1;
} else {
return 0;
}
obj = cJSON_CreateObject();
cJSON_AddNumberToObject(obj, "sni", sni);
cJSON_AddNumberToObject(obj, "fingerprint", fingerprint);
cJSON_AddStringToObject(obj, "cert", cert);
cJSON_AddStringToObject(obj, "tfe_ip", g_kafka_logger->tfe_ip);
dup = cJSON_Duplicate(obj, 1);
msg = cJSON_PrintUnformatted(dup);
TFE_LOG_DEBUG(g_default_logger, "log to [%s] msg:%s", g_kafka_logger->topic_name, msg);
rd_kafka_produce(g_kafka_logger->topic, RD_KAFKA_PARTITION_UA, RD_KAFKA_MSG_F_COPY, msg, strlen(msg), NULL, 0, NULL);
free(msg);
JSON_Delete(dup);
cJSON_Delete(obj);
}
// test use http://www.360.cn/
void ssl_fetch_trusted_cert_from_chain(STACK_OF(X509) *cert_chain, X509_STORE *trusted_store) {
if (!ssl_mid_cert_kafka_logger_enable()) {
void ssl_fetch_trusted_cert_from_chain(STACK_OF(X509) * cert_chain, X509_STORE *trusted_store, const char *hostname) {
int ret;
int deep;
char *subj = NULL;
char *issuer = NULL;
char *fingerprint = NULL;
X509 *cert = NULL;
X509_LOOKUP *lookup = NULL;
X509_OBJECT stmp;
if (!g_kafka_logger || !g_kafka_logger->enable) {
return;
}
// range for [0, count - 1]
int count = sk_X509_num(cert_chain);
// don`t need call X509_LOOKUP_free(lookup)
X509_LOOKUP *lookup = X509_STORE_add_lookup(trusted_store, X509_LOOKUP_hash_dir());
lookup = X509_STORE_add_lookup(trusted_store, X509_LOOKUP_hash_dir());
if (lookup == NULL) {
return;
}
for (int i = 1; i < count; i++) {
// don1t need call X509_FREE(cert)
X509 *cert = sk_X509_value(cert_chain, i);
deep = sk_X509_num(cert_chain);
for (int i = 1; i < deep; i++) {
// need't call X509_FREE(cert)
cert = sk_X509_value(cert_chain, i);
assert(cert);
X509_OBJECT stmp;
stmp.type = X509_LU_NONE;
stmp.data.ptr = NULL;
int result = X509_LOOKUP_by_subject(lookup, X509_LU_X509, X509_get_issuer_name(cert), &stmp);
char *subj = ssl_x509_subject(cert);
char *issuer = ssl_x509_issuer(cert);
if (result) {
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 {
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) {
// printf("%s\n", string);
ssl_mid_cert_kafka_logger_send(string);
free(string);
ret = X509_LOOKUP_by_subject(lookup, X509_LU_X509, X509_get_issuer_name(cert), &stmp);
subj = ssl_x509_subject(cert);
issuer = ssl_x509_issuer(cert);
fingerprint = ssl_x509_fingerprint(cert, 0);
TFE_LOG_DEBUG(g_default_logger, "[dep:%d/%d] subject:%s; issuer:%s; fingerprint:%s; in_trusted_store:%d", i, deep,
subj ? subj : "NULL", issuer ? issuer : "NULL", fingerprint ? fingerprint : "NULL", ret);
if (!ret) {
char *pem = ssl_x509_to_pem(cert);
if (pem) {
ssl_mid_cert_kafka_logger_send(hostname, fingerprint, pem);
free(pem);
}
}
free(subj);
free(issuer);
if (subj)
free(subj);
if (issuer)
free(issuer);
if (fingerprint)
free(fingerprint);
}
}

2
platform/src/ssl_trusted_cert_storage.cpp
View File

@@ -404,7 +404,7 @@ int ssl_trusted_cert_storage_verify_conn(struct ssl_trusted_cert_storage* storag
// case cert verify success
if (ret == 1) {
ssl_fetch_trusted_cert_from_chain(cert_chain, storage->effective_store);
ssl_fetch_trusted_cert_from_chain(cert_chain, storage->effective_store, hostname);
}
X509_STORE_CTX_free(ctx);