tango-tfe/plugin/business/pangu-http/pangu_http.cpp

#include "pangu_logger.h"

#include <tfe_stream.h>
#include <tfe_utils.h>
#include <tfe_http.h>
#include <tfe_plugin.h>

#include <MESA/Maat_rule.h>
#include <MESA/MESA_handle_logger.h>
#include <MESA/MESA_prof_load.h>
#include <MESA/stream.h>

#include <event2/event.h>
#include <event2/buffer.h>

#include <ctemplate/template.h>
#include <assert.h>
#include <time.h>
#include <stdio.h>
#include <string.h>
#include <sys/types.h>
#include <regex.h>

#define	MAX_SCAN_RESULT				16
#define	MAX_EDIT_ZONE_NUM			64
#define	MAX_EDIT_MATCHES			16

enum pangu_action//Bigger action number is prior.
{
	PG_ACTION_NONE = 0x00,
	PG_ACTION_MONIT = 0x01,
	PG_ACTION_FORWARD = 0x02,    /* N/A */
	PG_ACTION_REJECT = 0x10,
	PG_ACTION_DROP = 0x20,    /* N/A */
	PG_ACTION_REDIRECT = 0x30,
	PG_ACTION_RATELIMIT = 0x40,    /* N/A */
	PG_ACTION_REPLACE = 0x50,
	PG_ACTION_LOOP = 0x60,    /* N/A */
	PG_ACTION_WHITELIST = 0x80
};

enum scan_table
{
	PXY_CTRL_IP,
	PXY_CTRL_HTTP_URL,
	PXY_CTRL_HTTP_REQ_HDR,
	PXY_CTRL_HTTP_REQ_BODY,
	PXY_CTRL_HTTP_RES_HDR,
	PXY_CTRL_HTTP_RES_BODY,
	__SCAN_TABLE_MAX
};

struct pangu_rt
{
	Maat_feather_t maat;
	struct pangu_logger * send_logger;
	void * local_logger;
	int log_level;
	int thread_num;
	int scan_table_id[__SCAN_TABLE_MAX];
	ctemplate::Template * tpl_403, * tpl_404, * tpl_451;
	char * reject_page;
	int page_size;
};
struct pangu_rt * g_pangu_rt;

#define MAAT_INPUT_JSON			0
#define MAAT_INPUT_REDIS    	1
#define MAAT_INPUT_FILE        	2

static Maat_feather_t create_maat_feather(const char * profile, const char * section, int max_thread, void * logger)
{
	Maat_feather_t target;
	int input_mode = 0, maat_stat_on = 0, maat_perf_on = 0;
	int ret = 0, scan_detail = 0, effect_interval = 60;
	char table_info[TFE_STRING_MAX] = {0}, inc_cfg_dir[TFE_STRING_MAX] = {0}, ful_cfg_dir[TFE_STRING_MAX] = {0};
	char redis_server[TFE_STRING_MAX] = {0};
	int redis_port = 0;
	int redis_db_idx = 0;
	char json_cfg_file[TFE_STRING_MAX] = {0}, maat_stat_file[TFE_STRING_MAX] = {0};
	const char * instance_name = "pangu";
	MESA_load_profile_int_def(profile, section, "MAAT_INPUT_MODE", &(input_mode), 0);
	MESA_load_profile_int_def(profile, section, "STAT_SWITCH", &(maat_stat_on), 1);
	MESA_load_profile_int_def(profile, section, "PERF_SWITCH", &(maat_perf_on), 1);

	MESA_load_profile_string_def(profile, section, "TABLE_INFO", table_info, sizeof(table_info), "");

	MESA_load_profile_string_def(profile, section, "JSON_CFG_FILE", json_cfg_file, sizeof(json_cfg_file), "");

	MESA_load_profile_string_def(profile, section, "MAAT_REDIS_SERVER", redis_server, sizeof(redis_server), "");
	MESA_load_profile_int_def(profile, section, "MAAT_REDIS_PORT", &(redis_port), 6379);
	MESA_load_profile_int_def(profile, section, "MAAT_REDIS_DB_INDEX", &(redis_db_idx), 0);

	MESA_load_profile_string_def(profile, section, "INC_CFG_DIR", inc_cfg_dir, sizeof(inc_cfg_dir), "");
	MESA_load_profile_string_def(profile, section, "FULL_CFG_DIR", ful_cfg_dir, sizeof(ful_cfg_dir), "");

	MESA_load_profile_string_def(profile, section, "STAT_FILE", maat_stat_file, sizeof(maat_stat_file), "");
	MESA_load_profile_int_def(profile, section, "EFFECT_INTERVAL_S", &(effect_interval), 60);

	effect_interval *= 1000;//convert s to ms
	assert(strlen(inc_cfg_dir) != 0 && strlen(ful_cfg_dir) != 0);

	target = Maat_feather(max_thread, table_info, logger);
	Maat_set_feather_opt(target, MAAT_OPT_INSTANCE_NAME, instance_name, strlen(instance_name) + 1);
	switch (input_mode)
	{
		case MAAT_INPUT_JSON:
			Maat_set_feather_opt(target, MAAT_OPT_JSON_FILE_PATH, json_cfg_file, strlen(json_cfg_file) + 1);
			break;
		case MAAT_INPUT_REDIS: Maat_set_feather_opt(target, MAAT_OPT_REDIS_IP, redis_server, strlen(redis_server) + 1);
			Maat_set_feather_opt(target, MAAT_OPT_REDIS_PORT, &redis_port, sizeof(redis_port));
			Maat_set_feather_opt(target, MAAT_OPT_REDIS_INDEX, &redis_db_idx, sizeof(redis_db_idx));
			break;
		case MAAT_INPUT_FILE: Maat_set_feather_opt(target, MAAT_OPT_FULL_CFG_DIR, ful_cfg_dir, strlen(ful_cfg_dir) + 1);
			Maat_set_feather_opt(target, MAAT_OPT_INC_CFG_DIR, inc_cfg_dir, strlen(inc_cfg_dir) + 1);
			break;
		default: TFE_LOG_ERROR(logger, "Invalid MAAT Input Mode: %d.", input_mode);
			goto error_out;
			break;
	}

	if (maat_stat_on)
	{
		Maat_set_feather_opt(target, MAAT_OPT_STAT_FILE_PATH, maat_stat_file, strlen(maat_stat_file) + 1);
		Maat_set_feather_opt(target, MAAT_OPT_STAT_ON, NULL, 0);
		if (maat_perf_on)
		{
			Maat_set_feather_opt(target, MAAT_OPT_PERF_ON, NULL, 0);
		}
	}

	Maat_set_feather_opt(target, MAAT_OPT_EFFECT_INVERVAL_MS, &effect_interval, sizeof(effect_interval));
	Maat_set_feather_opt(target, MAAT_OPT_SCAN_DETAIL, &scan_detail, sizeof(scan_detail));

	ret = Maat_initiate_feather(target);
	if (ret < 0)
	{
		TFE_LOG_ERROR(logger, "%s MAAT init failed.", __FUNCTION__);
		goto error_out;
	}

	return target;
error_out:
	Maat_burn_feather(target);
	return NULL;
}

int pangu_http_init(struct tfe_proxy * proxy)
{
	const char * profile = "./pangu_conf/pangu_pxy.conf";
	const char * logfile = "./log/pangu_pxy.log";
	g_pangu_rt = ALLOC(struct pangu_rt, 1);
	g_pangu_rt->thread_num = 16;
	MESA_load_profile_int_def(profile, "DEBUG", "LOG_LEVEL", &(g_pangu_rt->log_level), 0);
	g_pangu_rt->local_logger = MESA_create_runtime_log_handle(logfile, g_pangu_rt->log_level);
	g_pangu_rt->send_logger = pangu_log_handle_create(profile, "LOG", g_pangu_rt->local_logger);
	if (!g_pangu_rt->send_logger)
	{
		goto error_out;
	}
	g_pangu_rt->maat = create_maat_feather(profile, "MAAT", g_pangu_rt->thread_num, g_pangu_rt->local_logger);
	if (!g_pangu_rt->maat)
	{
		goto error_out;
	}
	const char * table_name[__SCAN_TABLE_MAX];
	table_name[PXY_CTRL_IP] = "PXY_CTRL_IP";
	table_name[PXY_CTRL_HTTP_URL] = "PXY_CTRL_HTTP_URL";
	table_name[PXY_CTRL_HTTP_REQ_HDR] = "PXY_CTRL_HTTP_REQ_HDR";
	table_name[PXY_CTRL_HTTP_REQ_BODY] = "PXY_CTRL_HTTP_REQ_BODY";
	table_name[PXY_CTRL_HTTP_RES_HDR] = "PXY_CTRL_HTTP_RES_HDR";
	table_name[PXY_CTRL_HTTP_RES_BODY] = "PXY_CTRL_HTTP_RES_BODY";
	for (int i = 0; i < __SCAN_TABLE_MAX; i++)
	{

		g_pangu_rt->scan_table_id[i] = Maat_table_register(g_pangu_rt->maat, table_name[i]);
		if (g_pangu_rt->scan_table_id[i] < 0)
		{
			TFE_LOG_ERROR(NULL, "Pangu HTTP Maat table %s register failed.", table_name[i]);
			goto error_out;
		}
	}

	char page_path[256];
	memset(page_path, 0, sizeof(page_path));
	MESA_load_profile_string_def(profile, "TEMPLATE", "PAGE_403", page_path, sizeof(page_path),
		"./pangu_conf/template/HTTP403.html");
	g_pangu_rt->tpl_403 = ctemplate::Template::GetTemplate(page_path, ctemplate::DO_NOT_STRIP);
	memset(page_path, 0, sizeof(page_path));
	MESA_load_profile_string_def(profile, "TEMPLATE", "PAGE_404", page_path, sizeof(page_path),
		"./pangu_conf/template/HTTP404.html");
	g_pangu_rt->tpl_404 = ctemplate::Template::GetTemplate(page_path, ctemplate::DO_NOT_STRIP);
	memset(page_path, 0, sizeof(page_path));
	MESA_load_profile_string_def(profile, "TEMPLATE", "PAGE_451", page_path, sizeof(page_path),
		"./pangu_conf/template/HTTP451.html");
	g_pangu_rt->tpl_451 = ctemplate::Template::GetTemplate(page_path, ctemplate::DO_NOT_STRIP);

	TFE_LOG_INFO(NULL, "Pangu HTTP init success.");
	return 0;

error_out:
	TFE_LOG_ERROR(NULL, "Pangu HTTP init failed.");
	return -1;
}
static void _wrap_std_field_write(struct tfe_http_half * half, enum tfe_http_std_field field_id, const char * value)
{
	struct http_field_name tmp_name;
	tmp_name.field_id = field_id;
	tmp_name.field_name = NULL;
	tfe_http_field_write(half, &tmp_name, value);
	return;
}
#if 0
static void _wrap_non_std_field_write(struct tfe_http_half * half, const char* field_name, const char * value)
{
    struct http_field_name tmp_name;
    tmp_name.field_id=TFE_HTTP_UNKNOWN_FIELD;
    //todo remove force convert after tfe_http.h improved.
    tmp_name.field_name=(char*)field_name;
    tfe_http_field_write(half, &tmp_name, value);
    return;
}
#endif
enum replace_zone
{
	kZoneRequestUri = 0,
	kZoneRequestHeaders,
	kZoneRequestBody,
	kZoneResponseHeader,
	kZoneResponseBody,
	kZoneMax
};
struct replace_rule
{
	enum replace_zone zone;
	char * find;
	char * replace_with;
};
struct replace_ctx
{
	struct replace_rule * rule;
	size_t n_rule;
	struct tfe_http_half * replacing;
	struct evbuffer * http_body;
	size_t body_size;
};

struct pangu_http_ctx
{
	enum pangu_action action;
	char * action_para;
	scan_status_t mid;
	stream_para_t sp;

	struct Maat_rule_t * enforce_rules;
	size_t n_enforce;
	char * enforce_para;

	struct replace_ctx * rep_ctx;
	int thread_id;
};

static struct pangu_http_ctx * pangu_http_ctx_new(unsigned int thread_id)
{
	struct pangu_http_ctx * ctx = ALLOC(struct pangu_http_ctx, 1);
	ctx->mid = NULL;
	ctx->thread_id = (int) thread_id;
	return ctx;
}

static void pangu_http_ctx_free(struct pangu_http_ctx * ctx)
{
	if (ctx->rep_ctx != NULL)
	{
		for (size_t i = 0; i < ctx->rep_ctx->n_rule; i++)
		{
			FREE(&(ctx->rep_ctx->rule[i].find));
			FREE(&(ctx->rep_ctx->rule[i].replace_with));
		}
		evbuffer_free(ctx->rep_ctx->http_body);
		ctx->rep_ctx->http_body = NULL;
		//todo destroy http_half;
		assert(ctx->rep_ctx->replacing == NULL);
		FREE(&ctx->rep_ctx);
	}
	FREE(&ctx->enforce_rules);
	FREE(&ctx->enforce_para);
	Maat_clean_status(&(ctx->mid));
	assert(ctx->sp == NULL);
	ctx->mid = NULL;
	FREE(&ctx);
}

inline void addr_tfe2sapp(const struct tfe_stream_addr * tfe_addr, struct ipaddr * sapp_addr)
{
	sapp_addr->addrtype = tfe_addr->addrtype;
	sapp_addr->paddr = (char *) tfe_addr->paddr;
	return;
}

//enforce_rules[0] contains execute action.
static enum pangu_action decide_ctrl_action(const struct Maat_rule_t * hit_rules, size_t n_hit,
	struct Maat_rule_t ** enforce_rules, size_t * n_enforce)
{
	size_t n_monit = 0, exist_enforce_num = 0, i = 0;
	const struct Maat_rule_t * prior_rule = hit_rules;
	struct Maat_rule_t monit_rule[n_hit];
	enum pangu_action prior_action = PG_ACTION_NONE;

	for (i = 0; i < n_hit; i++)
	{
		if ((enum pangu_action) hit_rules[i].action == PG_ACTION_MONIT)
		{
			memcpy(monit_rule + n_monit, hit_rules + i, sizeof(struct Maat_rule_t));
			n_monit++;
		}
		if ((enum pangu_action) hit_rules[i].action > prior_action)
		{
			prior_rule = hit_rules + i;
			prior_action = (enum pangu_action) hit_rules[i].action;
		}
		else if ((enum pangu_action) hit_rules[i].action == prior_action)
		{
			if (hit_rules[i].config_id < prior_rule->config_id)
			{
				prior_rule = hit_rules + i;
			}
		}
		else
		{
			continue;
		}
	}

	if (prior_action == PG_ACTION_WHITELIST)
	{
		return PG_ACTION_WHITELIST;
	}

	exist_enforce_num = *n_enforce;
	if (prior_action == PG_ACTION_MONIT)
	{
		*n_enforce += n_monit;
	}
	else
	{
		*n_enforce += n_monit + 1;
	}

	*enforce_rules = (struct Maat_rule_t *) realloc(*enforce_rules, sizeof(struct Maat_rule_t) * (*n_enforce));

	if (prior_action == PG_ACTION_MONIT)
	{
		memcpy(*enforce_rules + exist_enforce_num, monit_rule, n_monit * sizeof(struct Maat_rule_t));
	}
	else
	{
		memcpy(*enforce_rules + exist_enforce_num, prior_rule, sizeof(struct Maat_rule_t));
		memcpy(*enforce_rules + exist_enforce_num + 1, monit_rule, n_monit * sizeof(struct Maat_rule_t));
	}

	return prior_action;
}
//https://github.com/AndiDittrich/HttpErrorPages
static void html_generate(int cfg_id, int status_code, char ** page_buff, size_t * page_size)
{
	ctemplate::TemplateDictionary dict("pg_page_dict");
	dict.SetIntValue("cfg_id", cfg_id);
	std::string output;
	ctemplate::Template * tpl = NULL;

	switch (status_code)
	{
		case 403: tpl = g_pangu_rt->tpl_403;
			break;
		case 404: tpl = g_pangu_rt->tpl_404;
			break;
		case 451: tpl = g_pangu_rt->tpl_451;
			break;
		default: return;
	}

	tpl->Expand(&output, &dict);
	//todo: do I need to delete dict?
	*page_size = output.length();
	*page_buff = ALLOC(char, *page_size);
	memcpy(*page_buff, output.c_str(), *page_size);
}

static void html_free(char ** page_buff)
{
	FREE(page_buff);
	return;
}
static int is_http_request(enum tfe_http_event events)
{
	if ((events & EV_HTTP_REQ_HDR) | (events & EV_HTTP_REQ_BODY_BEGIN) | (events & EV_HTTP_REQ_BODY_END)
		| (events & EV_HTTP_REQ_BODY_CONT))
	{
		return 1;
	}
	else
	{
		return 0;
	}
}

enum replace_zone zone_name_to_id(const char * name)
{
	const char * std_name[] = {"http_req_uri",
		"http_req_header",
		"http_req_body",
		"http_resp_header",
		"http_resp_body",
		"http_resp_body"};
	size_t i = 0;
	for (i = 0; i < sizeof(std_name) / sizeof(const char *); i++)
	{
		if (0 == strcasecmp(name, std_name[i]))
		{
			break;
		}
	}
	return (enum replace_zone) i;
}
static char * strchr_esc(char * s, const char delim)
{
	char * token;
	if (s == NULL)
		return NULL;
	for (token = s; *token != '\0'; token++)
	{
		if (*token == '\\')
		{
			token++;
			continue;
		}
		if (*token == delim)
			break;
	}
	if (*token == '\0')
	{
		return NULL;
	}
	else
	{
		return token;
	}
}
static char * strtok_r_esc(char * s, const char delim, char ** save_ptr)
{
	char * token;

	if (s == NULL) s = *save_ptr;

	/* Scan leading delimiters.  */
	token = strchr_esc(s, delim);
	if (token == NULL)
	{
		*save_ptr = token;
		return s;
	}
	/* Find the end of the token.  */
	*token = '\0';
	token++;
	*save_ptr = token;

	return s;
}

size_t format_replace_rule(const char * exec_para, struct replace_rule * replace, size_t n_replace)
{

	char * tmp = ALLOC(char, strlen(exec_para) + 1);
	char * token = NULL, * sub_token = NULL, * saveptr = NULL, * saveptr2 = NULL;
	size_t idx = 0;
	const char * str_zone = "replace=";
	const char * str_subs = "substitute=";
	memcpy(tmp, exec_para, strlen(exec_para));
	for (token = tmp;; token = NULL)
	{
		sub_token = strtok_r(token, ";", &saveptr);
		if (sub_token == NULL)
			break;
		if (0 == strncasecmp(sub_token, str_zone, strlen(str_zone)))
		{
			replace[idx].zone = zone_name_to_id(sub_token + strlen(str_zone));
			if (replace[idx].zone == kZoneMax)
			{
				break;
			}
		}
		if (0 == strncasecmp(sub_token, str_subs, strlen(str_subs)))
		{
			sub_token += strlen(str_subs);
			replace[idx].find = tfe_strdup(strtok_r_esc(sub_token, '/', &saveptr2));
			replace[idx].replace_with = tfe_strdup(strtok_r_esc(NULL, '/', &saveptr2));
			idx++;
			if (idx == n_replace)
			{
				break;
			}
		}
	}
	free(tmp);
	tmp = NULL;
	return idx;
}
size_t select_replace_rule(enum replace_zone zone, const struct replace_rule * replace, size_t n_replace,
	const struct replace_rule ** selected, size_t n_selected)
{
	size_t i = 0, j = 0;
	for (i = 0; i < n_replace && j < n_selected; i++)
	{
		if (replace[i].zone == zone)
		{
			selected[j] = replace + i;
			j++;
		}
	}
	return j;
}
static struct evbuffer * replace_string(const char * in, const struct replace_rule * zone)
{
	//Reference to https://www.lemoda.net/c/unix-regex/
	// Regular Expression test:  https://regex101.com/
	regex_t reg;
	int status = 0, is_replaced = 0;
	struct evbuffer * out = NULL;
	size_t in_sz = strlen(in);

	size_t replace_len = strlen(zone->replace_with);

	status = regcomp(&reg, zone->find, REG_EXTENDED | REG_NEWLINE);
	if (status != 0)
	{
		char error_message[TFE_STRING_MAX];
		regerror(status, &reg, error_message, sizeof(error_message));
		TFE_LOG_ERROR(g_pangu_rt->local_logger, "Regex error compiling '%s': %s\n",
			zone->find, error_message);
		regfree(&reg);
		return NULL;
	}

	/* "p" is a pointer into the string which points to the end of the  previous match. */
	const char * p = in;
	/* "pre_sub_expr_end" is a pointer into the string which points to the end of the  previous sub expression match. */
	const char * pre_sub_expr_end = NULL;

	/* "N_matches" is the maximum number of matches allowed. */
	const int n_matches = MAX_EDIT_MATCHES;
	/* "M" contains the matches found. */
	regmatch_t m[n_matches];
	int i = 0;

	while (1)
	{
		int nomatch = regexec(&reg, p, n_matches, m, 0);
		if (nomatch)
		{
			break;
		}
		if (is_replaced == 0)
		{
			out = evbuffer_new();
			is_replaced = 1;
		}
		assert(m[0].rm_so != -1);
		pre_sub_expr_end = p;
		if (m[1].rm_so == -1)//no sub expr, replace the entire expr.
		{
			evbuffer_add(out, pre_sub_expr_end, m[0].rm_so - (pre_sub_expr_end - p));
			evbuffer_add(out, zone->replace_with, replace_len);
			pre_sub_expr_end = p + m[0].rm_eo;
		}
		else    //have sub expr, replace the sub expr.
		{
			for (i = 1, pre_sub_expr_end = p; i < n_matches; i++)
			{
				if (m[i].rm_so == -1)
				{
					break;
				}
				evbuffer_add(out, pre_sub_expr_end, m[i].rm_so - (pre_sub_expr_end - p));
				evbuffer_add(out, zone->replace_with, replace_len);
				pre_sub_expr_end = p + m[i].rm_eo;
			}
		}
		p += m[0].rm_eo;
	}
	if (is_replaced)
	{
		evbuffer_add(out, pre_sub_expr_end, in_sz - (pre_sub_expr_end - p));
	}

	regfree(&reg);
	return out;

}
struct evbuffer * execute_replace_rule(const char * in, size_t in_sz,
	enum replace_zone zone, const struct replace_rule * rules, size_t n_rule)
{
	const struct replace_rule * todo[MAX_EDIT_ZONE_NUM];
	size_t n_todo = 0, i = 0;
	struct evbuffer * out = NULL;
	const char * interator = NULL;
	struct evbuffer * new_out = NULL, * pre_out = NULL;
	if (in == 0)
	{
		return NULL;
	}
	//Do not process buffer that contains '\0'.
	if (0 != memchr(in, '\0', in_sz))
	{
		return NULL;
	}
	n_todo = select_replace_rule(zone, rules, n_rule, todo, MAX_EDIT_ZONE_NUM);
	interator = in;
	for (i = 0; i < n_todo; i++)
	{

		new_out = replace_string(interator, todo[i]);
		if (new_out != NULL)
		{
			pre_out = out;
			out = new_out;
			interator = (char *) evbuffer_pullup(out, -1);
			evbuffer_free(pre_out);
			pre_out = NULL;
		}
	}
	return out;
}
void http_replace(const struct tfe_stream * stream, const struct tfe_http_session * session,
	enum tfe_http_event events, const unsigned char * body_frag, size_t frag_size, struct pangu_http_ctx * ctx)
{
	void * interator = NULL;
	struct http_field_name tmp_name;
	const char * buff_in = NULL;
	struct evbuffer * rewrite_url = NULL, * rewrite_buff = NULL;
	struct replace_ctx * rep_ctx = NULL;
	struct tfe_http_session * to_write_sess = NULL;

	to_write_sess = tfe_http_session_allow_write(session);
	if (to_write_sess == NULL)    //fail to wirte, abandon.
	{
		TFE_STREAM_LOG_INFO(stream, "tfe_http_session_allow_write() %s failed.", session->req->req_spec.uri);
		tfe_http_session_detach(session);
		return;
	}
	if (ctx->rep_ctx == NULL)
	{
		ctx->rep_ctx = rep_ctx = ALLOC(struct replace_ctx, 1);
		rep_ctx->rule = ALLOC(struct replace_rule, MAX_EDIT_ZONE_NUM);
		rep_ctx->n_rule = format_replace_rule(ctx->enforce_para, rep_ctx->rule, MAX_EDIT_ZONE_NUM);
	}
	if (events & EV_HTTP_REQ_HDR)
	{
		rewrite_url = execute_replace_rule(session->req->req_spec.uri, strlen(session->req->req_spec.uri),
			kZoneRequestUri, rep_ctx->rule, rep_ctx->n_rule);
	}
	if ((events & EV_HTTP_REQ_HDR) | (events & EV_HTTP_RESP_HDR))
	{

		if (events & EV_HTTP_REQ_HDR)
		{
			rep_ctx->replacing = tfe_http_session_request_create(to_write_sess, session->req->req_spec.method,
				rewrite_url != NULL ? (char *) evbuffer_pullup(rewrite_url, -1) : session->req->req_spec.uri);
			evbuffer_free(rewrite_url);
			rewrite_url = NULL;
		}
		else
		{
			rep_ctx->replacing = tfe_http_session_response_create(to_write_sess, session->resp->resp_spec.resp_code);
		}
		while (1)
		{
			buff_in = tfe_http_field_iterate(session->req, &interator, &tmp_name);
			if (tmp_name.field_id == TFE_HTTP_CONT_LENGTH)
			{
				continue;
			}
			if (buff_in != NULL)
			{
				rewrite_buff = execute_replace_rule(buff_in, strlen(buff_in),
					events & EV_HTTP_REQ_HDR ? kZoneRequestHeaders : kZoneResponseHeader, rep_ctx->rule,
					rep_ctx->n_rule);
				tfe_http_field_write(rep_ctx->replacing, &tmp_name,
					rewrite_buff != NULL ? (char *) evbuffer_pullup(rewrite_buff, -1) : buff_in);
				evbuffer_free(rewrite_buff);
				rewrite_buff = NULL;

			}
			else
			{
				break;
			}
		}
	}
	if ((events & EV_HTTP_REQ_BODY_BEGIN) | (events & EV_HTTP_RESP_BODY_BEGIN))
	{
		assert(rep_ctx->http_body == NULL);
		assert(rep_ctx->body_size = 0);
		rep_ctx->http_body = evbuffer_new();
	}
	if (body_frag != NULL)
	{
		evbuffer_add(rep_ctx->http_body, body_frag, frag_size);
		rep_ctx->body_size++;
	}
	if ((events & EV_HTTP_REQ_BODY_END) | (events & EV_HTTP_RESP_BODY_END))
	{

		assert(rep_ctx->body_size == evbuffer_get_length(rep_ctx->http_body));
		buff_in = (char *) evbuffer_pullup(rep_ctx->http_body, -1);
		rewrite_buff = execute_replace_rule(buff_in, rep_ctx->body_size,
			events & EV_HTTP_REQ_HDR ? kZoneRequestHeaders : kZoneResponseHeader, rep_ctx->rule, rep_ctx->n_rule);
		char cont_len_str[TFE_SYMBOL_MAX];
		snprintf(cont_len_str, sizeof(cont_len_str), "%lu", evbuffer_get_length(rewrite_buff));
		_wrap_std_field_write(rep_ctx->replacing, TFE_HTTP_CONT_LENGTH, cont_len_str);
		tfe_http_half_append_body(rep_ctx->replacing,
			(char *) evbuffer_pullup(rewrite_buff, -1), evbuffer_get_length(rewrite_buff), 0);
		evbuffer_free(rewrite_buff);
		rewrite_buff = NULL;
		if (is_http_request(events))
		{
			tfe_http_session_request_set(to_write_sess, rep_ctx->replacing);
		}
		else
		{
			tfe_http_session_response_set(to_write_sess, rep_ctx->replacing);
		}
		rep_ctx->replacing = NULL;//http half's ownership has been transfered to session.

		evbuffer_free(rep_ctx->http_body);
		rep_ctx->http_body = NULL;
		rep_ctx->body_size = 0;
	}
	return;

}
static void http_reject(const struct tfe_http_session * session, enum tfe_http_event events,
	struct pangu_http_ctx * ctx)
{
	int resp_code = 0, ret = 0;
	struct tfe_http_half * response = NULL;
	char * page_buff = NULL;
	size_t page_size = 0;

	char cont_len_str[TFE_STRING_MAX];
	struct tfe_http_session * to_write_sess = NULL;

	ret = sscanf(ctx->enforce_para, "code=%d;", &resp_code);
	if (ret != 1)
	{
		TFE_LOG_ERROR(g_pangu_rt->local_logger, "Invalid reject rule %d paramter %s",
			ctx->enforce_rules[0].config_id, ctx->enforce_para); goto error_out;
	}

	to_write_sess = tfe_http_session_allow_write(session);
	response = tfe_http_session_response_create(to_write_sess, resp_code);

	html_generate(ctx->enforce_rules[0].config_id, resp_code, &page_buff, &page_size);
	_wrap_std_field_write(response, TFE_HTTP_CONT_TYPE, "text/html; charset=utf-8");
	snprintf(cont_len_str, sizeof(cont_len_str), "%lu", page_size);
	_wrap_std_field_write(response, TFE_HTTP_CONT_LENGTH, cont_len_str);

	tfe_http_half_append_body(response, page_buff, page_size, 0);
	tfe_http_session_response_set(to_write_sess, response);
	tfe_http_session_detach(session);

error_out:
	html_free(&page_buff);
}

static void http_redirect(const struct tfe_http_session * session, enum tfe_http_event events,
	struct pangu_http_ctx * ctx)
{
	int resp_code = 0, ret = 0;
	char * url = NULL;

	struct tfe_http_half * response = NULL;
	struct tfe_http_session * to_write = NULL;

	url = ALLOC(char, ctx->enforce_rules[0].serv_def_len);
	ret = sscanf(ctx->enforce_para, "code=%d;url=%[^;]", &resp_code, url);

	if (ret != 2)
	{
		TFE_LOG_ERROR(g_pangu_rt->local_logger, "Invalid redirect rule %d paramter %s",
			ctx->enforce_rules[0].config_id, ctx->enforce_para);
		goto error_out;
	}

	to_write = tfe_http_session_allow_write(session);
	if (to_write == NULL)
	{
		assert(0);
	}

	response = tfe_http_session_response_create(to_write, resp_code);
	_wrap_std_field_write(response, TFE_HTTP_LOCATION, url);

	tfe_http_session_response_set(to_write, response);
	tfe_http_session_detach(session);

error_out:
	free(url);
	return;
}

enum pangu_action http_scan(const struct tfe_http_session * session, enum tfe_http_event events,
	const unsigned char * body_frag, size_t frag_size, struct pangu_http_ctx * ctx)
{
	void * interator = NULL;
	const char * field_val = NULL;
	struct http_field_name field_name;
	struct Maat_rule_t result[MAX_SCAN_RESULT];
	char buff[TFE_STRING_MAX], * p = NULL;
	int scan_ret = 0, table_id = 0, read_rule_ret = 0;
	size_t hit_cnt = 0, i = 0;

	if (events & EV_HTTP_REQ_HDR)
	{
		const char * str_url = session->req->req_spec.url;
		int str_url_length = (int) (strlen(session->req->req_spec.url));

		scan_ret = Maat_full_scan_string(g_pangu_rt->maat, g_pangu_rt->scan_table_id[PXY_CTRL_HTTP_URL],
			CHARSET_UTF8, str_url, str_url_length, result, NULL, MAX_SCAN_RESULT, &(ctx->mid), ctx->thread_id);

		if (scan_ret > 0)
		{
			hit_cnt += scan_ret;
		}
	}

	if ((events & EV_HTTP_REQ_HDR) || (events & EV_HTTP_RESP_HDR))
	{
		table_id = events & EV_HTTP_REQ_HDR ? g_pangu_rt->scan_table_id[PXY_CTRL_HTTP_REQ_HDR] : g_pangu_rt
			->scan_table_id[PXY_CTRL_HTTP_RES_HDR];
		while (hit_cnt < MAX_SCAN_RESULT)
		{
			field_val = tfe_http_field_iterate(session->req, &interator, &field_name);
			if (field_val == NULL)
			{
				break;
			}

			const char * str_field_name = http_field_to_string(&field_name);
			scan_ret = Maat_set_scan_status(g_pangu_rt->maat, &(ctx->mid), MAAT_SET_SCAN_DISTRICT,
				str_field_name, strlen(str_field_name));

			assert(scan_ret == 0);
			scan_ret = Maat_full_scan_string(g_pangu_rt->maat, table_id,
				CHARSET_UTF8, field_val, strlen(field_val),
				result + hit_cnt, NULL, MAX_SCAN_RESULT - hit_cnt, &(ctx->mid), ctx->thread_id);
			if (scan_ret > 0)
			{
				hit_cnt += scan_ret;
			}
		}
	}
	if ((events & EV_HTTP_REQ_BODY_BEGIN) | (events & EV_HTTP_RESP_BODY_BEGIN))
	{
		assert(ctx->sp == NULL);
		table_id = events & EV_HTTP_REQ_BODY_BEGIN ? g_pangu_rt->scan_table_id[PXY_CTRL_HTTP_REQ_BODY] : g_pangu_rt
			->scan_table_id[PXY_CTRL_HTTP_RES_BODY];
		ctx->sp = Maat_stream_scan_string_start(g_pangu_rt->maat, table_id, ctx->thread_id);
	}
	if (body_frag != NULL)
	{
		scan_ret = Maat_stream_scan_string(&(ctx->sp), CHARSET_UTF8, (const char *) body_frag, (int) frag_size,
			result + hit_cnt, NULL, MAX_SCAN_RESULT - hit_cnt, &(ctx->mid));
		if (scan_ret > 0)
		{
			hit_cnt += scan_ret;
		}
	}
	if ((events & EV_HTTP_REQ_BODY_END) | (events & EV_HTTP_RESP_BODY_END))
	{
		Maat_stream_scan_string_end(&(ctx->sp));
		ctx->sp = NULL;
	}

	if (hit_cnt > 0)
	{
		ctx->action = decide_ctrl_action(result, hit_cnt, &ctx->enforce_rules, &ctx->n_enforce);
		size_t __serv_def_len = (size_t)ctx->enforce_rules[0].serv_def_len;
		ctx->enforce_para = ALLOC(char, __serv_def_len);

		if (__serv_def_len > MAX_SERVICE_DEFINE_LEN)
		{
			read_rule_ret = Maat_read_rule(g_pangu_rt->maat, &ctx->enforce_rules[0], MAAT_RULE_SERV_DEFINE,
				ctx->enforce_para, ctx->enforce_rules[0].serv_def_len);
			assert(read_rule_ret == ctx->enforce_rules[0].serv_def_len);
		}
		else
		{
			strcpy(ctx->enforce_para, ctx->enforce_rules[0].service_defined);
		}

		if (hit_cnt > 1)
		{
			p = buff;
			for (i = 0; i < hit_cnt; i++)
			{
				p += snprintf(p, sizeof(buff) - (p - buff), "%d:", result[i].config_id);
			}

			*p = '\0';
			TFE_LOG_INFO(g_pangu_rt->local_logger, "Multiple rules matched: url=%s num=%lu ids=%s execute=%d.",
				session->req->req_spec.url, hit_cnt, buff, ctx->enforce_rules[0].config_id);
		}
	}

	return ctx->action;
}

void pangu_on_http_begin(const struct tfe_stream * stream,
	const struct tfe_http_session * session, unsigned int thread_id, void ** pme)
{
	struct pangu_http_ctx * ctx = *(struct pangu_http_ctx **) pme;
	struct Maat_rule_t result[MAX_SCAN_RESULT];
	struct ipaddr sapp_addr;
	int hit_cnt = 0;
	assert(ctx == NULL);
	ctx = pangu_http_ctx_new(thread_id);
	addr_tfe2sapp(stream->addr, &sapp_addr);
	hit_cnt = Maat_scan_proto_addr(g_pangu_rt->maat, g_pangu_rt->scan_table_id[PXY_CTRL_IP], &sapp_addr, 0,
		result, MAX_SCAN_RESULT, &(ctx->mid), (int) thread_id);

	if (hit_cnt > 0)
	{
		ctx->action = decide_ctrl_action(result, hit_cnt, &ctx->enforce_rules, &ctx->n_enforce);
	}
	if (ctx->action == PG_ACTION_WHITELIST)
	{
		tfe_http_session_detach(session);
	}

	*pme = ctx;
	return;
}

void pangu_on_http_end(const struct tfe_stream * stream,
	const struct tfe_http_session * session, unsigned int thread_id, void ** pme)
{
	struct pangu_http_ctx * ctx = *(struct pangu_http_ctx **) pme;
	struct pangu_log log_msg = {.stream=stream, .http=session, .result=ctx->enforce_rules, .result_num=ctx->n_enforce};
	if (ctx->action != PG_ACTION_NONE)
	{
		pangu_send_log(g_pangu_rt->send_logger, &log_msg);
	}
	pangu_http_ctx_free(ctx);
	*pme = NULL;
	return;
}

void pangu_on_http_data(const struct tfe_stream * stream, const struct tfe_http_session * session,
	enum tfe_http_event events, const unsigned char * body_frag, size_t frag_size, unsigned int thread_id, void ** pme)
{
	struct pangu_http_ctx * ctx = *(struct pangu_http_ctx **) pme;
	enum pangu_action hit_action = PG_ACTION_NONE;

Re_Enter:
	switch (ctx->action)
	{
		case PG_ACTION_NONE: hit_action = http_scan(session, events, body_frag, frag_size, ctx);
			if (hit_action != PG_ACTION_NONE)
			{
				//ctx->action changed in http_scan.
				goto Re_Enter;
			}
			break;
		case PG_ACTION_MONIT:
			//send log on close.
			break;
		case PG_ACTION_REJECT: http_reject(session, events, ctx);
			break;
		case PG_ACTION_REDIRECT: http_redirect(session, events, ctx);
			break;
		case PG_ACTION_REPLACE: http_replace(stream, session, events, body_frag, frag_size, ctx);
			break;
		case PG_ACTION_WHITELIST: tfe_http_session_detach(session);
			break;
		default: assert(0);
			break;
	}

	return;
}
struct tfe_plugin pangu_http_spec = {
	.symbol=NULL,
	.type = TFE_PLUGIN_TYPE_BUSINESS,
	.on_init = pangu_http_init,
	.on_deinit = NULL,
	.on_open = NULL,
	.on_data = NULL,
	.on_close = NULL,
	.on_session_begin=pangu_on_http_begin,
	.on_session_data=pangu_on_http_data,
	.on_session_end=pangu_on_http_end
};
TFE_PLUGIN_REGISTER(pangu_http, pangu_http_spec)