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

#include <assert.h>
#include <cjson/cJSON.h>

#include <ssl_utils.h>
#include <tfe_kafka_logger.h>
#include <MESA/MESA_prof_load.h>

typedef struct x509_object_st
{
	/* one of the above types */
	X509_LOOKUP_TYPE type;
	union
	{
		char     *ptr;
		X509     *x509;
		X509_CRL *crl;
		EVP_PKEY *pkey;
	} data;
} X509_OBJECT;

enum CERT_TYPE {
	ENTITY_CERT = 0,
	INTERMEDIATE_CERT = 1,
	ROOT_CERT = 2,
	MAX_TYPE = 3,
};

static char cert_type_desc[MAX_TYPE][64] = {
	{"Entity certificate"},
	{"Intermediate certificate"},
	{"Root certificate"},
};

static tfe_kafka_logger_t *g_kafka_logger = NULL;

void ssl_mid_cert_kafka_logger_destory(void)
{
	tfe_kafka_logger_destroy(g_kafka_logger);
}

int ssl_mid_cert_kafka_logger_create(const char *profile, const char *section)
{
	int enable = 0;
	char nic_name[TFE_SYMBOL_MAX] = {0};
	char broker_list[TFE_SYMBOL_MAX] = {0};
	char topic_name[TFE_SYMBOL_MAX] = {0};
	char sasl_username[TFE_STRING_MAX] = {0};
	char sasl_passwd[TFE_STRING_MAX] = {0};

	MESA_load_profile_int_def(profile, section, "mc_cache_enable", &enable, 0);
	MESA_load_profile_string_def(profile, section, "mc_cache_eth", nic_name, sizeof(nic_name), "eth0");
	MESA_load_profile_string_def(profile, section, "mc_cache_topic", topic_name, sizeof(topic_name), "PXY-EXCH-INTERMEDIA-CERT");
	MESA_load_profile_string_def(profile, section, "SASL_USERNAME", sasl_username, sizeof(sasl_username), "");
	MESA_load_profile_string_def(profile, section, "SASL_PASSWD", sasl_passwd, sizeof(sasl_passwd), "");

	if (!enable)
		goto skip;
	if (MESA_load_profile_string_def(profile, section, "mc_cache_broker_list", broker_list, sizeof(broker_list), NULL) < 0)
	{
		TFE_LOG_ERROR(g_default_logger, "Fail to get mc_cache_broker_list in profile %s section %s.", profile, section);
		return -1;
	}
skip:
	g_kafka_logger = tfe_kafka_logger_create(enable, nic_name, broker_list, topic_name, sasl_username, sasl_passwd, g_default_logger);
	if (g_kafka_logger)
		return 0;
	else
		return -1;
}

static void ssl_mid_cert_kafka_logger_send(const char *sni, const char *fingerprint, const char *cert)
{
	if (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->local_ip_str);
	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);
	tfe_kafka_logger_send(g_kafka_logger, msg, strlen(msg));

	free(msg);
	cJSON_Delete(dup);
	cJSON_Delete(obj);
}

static int is_x509v3_ca_cert(X509 *x)
{
	/* what is Basic Constraint Extension: 
	 * http://www.pkiglobe.org/
	 * https://tools.ietf.org/html/rfc5280#section-4.2.1.9
	 * 
	 * how to visit Basic Constraint Extension: 
	 * https://stackoverflow.com/questions/40609792/how-can-i-verify-certificate-has-ca-true-basic-constraint
	 */
	int is_ca = 0;
	BASIC_CONSTRAINTS *bs = NULL;

	if (X509_get_version(x) != 2)
	{
		return is_ca;
	}

	bs = (BASIC_CONSTRAINTS *)X509_get_ext_d2i(x, NID_basic_constraints, NULL, NULL);
	if (bs)
	{
		if (bs->ca)
		{
			is_ca = 1;
		}
		BASIC_CONSTRAINTS_free(bs);
	}

	return is_ca;
}

void ssl_fetch_trusted_cert_from_chain(STACK_OF(X509) * cert_chain, X509_STORE *trusted_store, const char *hostname)
{
	int          in_store;
	int          type;
	int          deep;
	char        *pem = NULL;
	char        *subj = NULL;
	char        *issuer = NULL;
	char        *fingerprint = NULL;
	X509        *cert = NULL;
	X509_OBJECT *obj = NULL;
	if (!g_kafka_logger || !g_kafka_logger->enable)
	{
		return;
	}

	deep = sk_X509_num(cert_chain);
	for (int i = 0; i < deep; i++)
	{
		cert = sk_X509_value(cert_chain, i);
		assert(cert);

		in_store = 0;
		subj = ssl_x509_subject(cert);
		issuer = ssl_x509_issuer(cert);
		fingerprint = ssl_x509_fingerprint(cert, 0);
		pem = ssl_x509_to_pem(cert);

		if (!is_x509v3_ca_cert(cert))
		{
			type = ENTITY_CERT;
			goto end;
		}
		else
		{
			if (subj && issuer && strcmp(subj, issuer) == 0)
			{
				type = ROOT_CERT;
				goto end;
			}
			else
			{
				type = INTERMEDIATE_CERT;
			}
		}

		obj = X509_OBJECT_new();
		assert(obj);
		obj->type = X509_LU_X509;
		obj->data.x509 = (X509 *)cert;
		X509_OBJECT_up_ref_count(obj);

		// not in trusted store
		if (X509_OBJECT_retrieve_match(X509_STORE_get0_objects(trusted_store), obj) == NULL)
		{
			in_store = 0;
		}
		// in trusted store
		else
		{
			in_store = 1;
		}
		X509_OBJECT_free(obj);

		if (!in_store && fingerprint && pem)
		{
			ssl_mid_cert_kafka_logger_send(hostname, fingerprint, pem);
		}

	end:
		TFE_LOG_DEBUG(g_default_logger, "[dep:%d/%d] is %s, in_trusted_store:%d, sin:%s; subject:(%s); issuer:(%s); fingerprint:%s; cert:%s",
					  i, deep, cert_type_desc[type], in_store, (hostname ? hostname : "NULL"), (subj ? subj : "NULL"), (issuer ? issuer : "NULL"), (fingerprint ? fingerprint : "NULL"),
					  ((pem && g_kafka_logger->enable == 0x10) ? pem : " ..."));
		if (pem)
		{
			free(pem);
			pem = NULL;
		}
		if (subj)
		{
			free(subj);
			subj = NULL;
		}
		if (issuer)
		{
			free(issuer);
			issuer = NULL;
		}
		if (fingerprint)
		{
			free(fingerprint);
			fingerprint = NULL;
		}
	}
}