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

#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;
	union {
		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];

	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) {
	if (!ssl_mid_cert_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());
	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);
		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);
			}
		}
		free(subj);
		free(issuer);
	}
}