//
// Created by lwp on 2019/10/16.
//

#include "ssl_utils.h"
#include "tfe_utils.h"

#include <assert.h>
#include <unistd.h>
#include <net/if.h>
#include <sys/ioctl.h>
#include <arpa/inet.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;
		X509_CRL *crl;
		EVP_PKEY *pkey;
	} data;
} X509_OBJECT;

typedef struct ssl_kafka_logger_s {
	int               enable;

	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 unsigned int get_ip_by_eth(const char *eth) {
	int          sockfd = -1;
	unsigned int ip;
	struct ifreq ifr;

	sockfd = socket(AF_INET, SOCK_DGRAM, 0);
	if (-1 == sockfd) {
		goto error;
	}

	memset(&ifr, 0, sizeof(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, broker_list) == 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);
		}
		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)
{
	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, "mc_cache_enable", &(g_kafka_logger->enable), 0); 
	if (!g_kafka_logger->enable) {
		return 0;
	}

	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;
	}
	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, Fail to create kafka handle with broker list: %s.", errstr, g_kafka_logger->broker_list);
		goto error;
	}

<<<<<<< HEAD
	MESA_load_profile_string_def(profile, section, "mc_cache_topic", g_kafka_logger->topic_name, sizeof(g_kafka_logger->topic_name), "PXY-EXCH-INTERMEDIA-CERT");
=======
	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");
>>>>>>> a3856d20e7f702b74de5173ecf23538eadd0184a
	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, Fail to create kafka topic with broker list: %s.", errstr, g_kafka_logger->broker_list);
		goto error;
	}

	return 0;

error:
	ssl_mid_cert_kafka_logger_destory();
	return -1;
}

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;
	}
	cJSON *obj = NULL;
	cJSON *dup = NULL;
	char  *msg = NULL;

	obj = cJSON_CreateObject();
	cJSON_AddStringToObject(obj, "sni", sni);
	cJSON_AddStringToObject(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);
	cJSON_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, 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;
	}

	// don`t need call X509_LOOKUP_free(lookup)
	lookup = X509_STORE_add_lookup(trusted_store, X509_LOOKUP_hash_dir());
	if (lookup == NULL) {
		return;
	}

	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);

		stmp.type = X509_LU_NONE;
		stmp.data.ptr = NULL;
		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);
			}
		}
		if (subj)
			free(subj);
		if (issuer)
			free(issuer);
		if (fingerprint)
			free(fingerprint);
	}
}