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

#include "pangu_logger.h"
#include "pattern_replace.h"
#include "pangu_web_cache.h"

#include <tfe_proxy.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 <cjson/cJSON.h>

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

#include "ratelimiter.h"

#define	MAX_SCAN_RESULT				16
#define	MAX_EDIT_ZONE_NUM			64

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_MANIPULATE = 0x30,
	PG_ACTION_RATELIMIT = 0x40,    /* N/A */
	PG_ACTION_LOOP = 0x60,    /* N/A */
	PG_ACTION_WHITELIST = 0x80,
	__PG_ACTION_MAX
};

enum manipulate_action
{
	MA_ACTION_REDIRECT = 0,
	MA_ACTION_BLOCK,
	MA_ACTION_REPLACE,
	MA_ACTION_HIJACK,
	MA_ACTION_INSERT,
	__MA_ACTION_MAX
};

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

enum pangu_http_stat
{
	STAT_SESSION,
	STAT_LOG_NUM,
	STAT_ACTION_MONIT,
	STAT_ACTION_REJECT,
	STAT_ACTION_REDIRECT,
	STAT_ACTION_PRE_REPLACE,
	STAT_ACTION_REPLACE,
	STAT_ACTION_HIJACK,
	STAT_ACTION_INSERT,
	STAT_ACTION_WHITELSIT,
	STAT_SUSPENDING,
	__PG_STAT_MAX
};

enum manipulate_profile_table
{
	POLICY_PROFLIE_TABLE_REJECT,
	POLICY_PROFILE_TABLE_INSERT,
	POLICY_PROFILE_TABLE_HIJACK,
	POLICY_PROFILE_TABLE_MAX
};

struct manipulate_profile
{
	int profile_id;
	int ref_cnt;
	size_t msg_len;
	char *profile_name;
	char *profile_msg;
	char *profile_type;
	char *profile_position;
	ctemplate::Template * tpl;
	pthread_mutex_t lock;
};

struct policy_action_param
{
	int ref_cnt;
	enum manipulate_action action;

	char *message;
	char *position;

	float enforcement_ratio;
	int profile_id;
	int status_code;

	size_t n_rule;
	struct replace_rule *rule;
	pthread_mutex_t lock;
};

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

	long long suspend_max;
	int cache_enabled;
	struct cache_handle* cache;

	screen_stat_handle_t fs_handle;
	long long stat_val[__PG_STAT_MAX];
	int fs_id[__PG_STAT_MAX];
	struct event_base* gc_evbase;
	struct event* gcev;

	Ratelimiter_handle_t ratelimiter;
	int enable_rate;

	int ctrl_compile_idx;
	int ca_store_reseting;

};
struct pangu_rt * g_pangu_rt;

Maat_feather_t g_business_maat;


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

static Maat_feather_t create_maat_feather(const char * instance_name, 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};
	char redis_port_range[TFE_STRING_MAX] = {0};
	char accept_tags[TFE_STRING_MAX] = {0};
	int redis_port_begin=0, redis_port_end=0;
	int redis_port_select=0;
	int redis_db_idx = 0;
	char json_cfg_file[TFE_STRING_MAX] = {0}, maat_stat_file[TFE_STRING_MAX] = {0};
	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, "accept_tags", accept_tags, sizeof(accept_tags), "");

	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_string_def(profile, section, "maat_redis_port_range", redis_port_range, sizeof(redis_server), "6379");
	ret=sscanf(redis_port_range,"%d-%d", &redis_port_begin, &redis_port_end);
	if(ret==1)
	{
		redis_port_select=redis_port_begin;
	}
	else if(ret==2)
	{
		srand(time(NULL));
		redis_port_select=redis_port_begin+rand()%(redis_port_end-redis_port_begin);
	}
	else
	{
		TFE_LOG_ERROR(logger, "Invalid redis port range %s, MAAT init failed.", redis_port_range);
	}
	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 || strlen(redis_server)!=0 || strlen(json_cfg_file)!=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_select, sizeof(redis_port_select));
			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;
	}


	Maat_set_feather_opt(target, MAAT_OPT_FOREIGN_CONT_DIR, "./pangu_files", strlen("./pangu_files")+1);
	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));
	if(strlen(accept_tags)>0)
	{
		Maat_set_feather_opt(target, MAAT_OPT_ACCEPT_TAGS, &accept_tags, sizeof(accept_tags));
	}

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

static void pangu_http_gc_cb(evutil_socket_t fd, short what, void * arg)
{
	int i=0;

	for(i=0;i<__PG_STAT_MAX;i++)
	{
		FS_operate(g_pangu_rt->fs_handle, g_pangu_rt->fs_id[i], 0, FS_OP_SET, ATOMIC_READ(&(g_pangu_rt->stat_val[i])));
	}
	return;
}

static void pangu_http_stat_init(struct pangu_rt * pangu_runtime)
{
	int i=0;
	struct timeval gc_delay = {0, 500*1000};	//Microseconds, we set 500 miliseconds here.
	const char* spec[__PG_STAT_MAX]={0};
	spec[STAT_SESSION]="http_sess";
	spec[STAT_LOG_NUM]="log_num";
	spec[STAT_ACTION_MONIT]="monitor";
	spec[STAT_ACTION_REJECT]="reject";
	spec[STAT_ACTION_REDIRECT]="redirect";
	spec[STAT_ACTION_PRE_REPLACE]="pre_replace";
	spec[STAT_ACTION_REPLACE]="replace";
	spec[STAT_ACTION_HIJACK]="hijack";
	spec[STAT_ACTION_INSERT]="insert";
	spec[STAT_ACTION_WHITELSIT]="whitelist";
	spec[STAT_SUSPENDING]="suspending";

	for(i=0;i<__PG_STAT_MAX;i++)
	{
		if(spec[i]!=NULL)
		{
			pangu_runtime->fs_id[i]=FS_register(pangu_runtime->fs_handle, FS_STYLE_FIELD, FS_CALC_CURRENT, spec[i]);
		}
	}
	g_pangu_rt->gcev = event_new(pangu_runtime->gc_evbase, -1, EV_PERSIST, pangu_http_gc_cb, NULL);
    evtimer_add(g_pangu_rt->gcev, &gc_delay);

	return;
}

void trusted_CA_update_start_cb(int update_type, void* u_para)
{
	if(update_type==MAAT_RULE_UPDATE_TYPE_FULL)
	{
		if(g_pangu_rt->ca_store_reseting==0)
		{
			tfe_proxy_ssl_reset_trust_ca();
			TFE_LOG_INFO(g_pangu_rt->local_logger, "Trusted CA Store Reset Start.");
		}
		g_pangu_rt->ca_store_reseting++;
	}

}
void trusted_CA_update_cert_cb(int table_id, const char* table_line, void* u_para)
{
	int ret=0, cfg_id=0, is_valid=0;
	char cert_name[128]={0}, cert_file[1024]={0};
	ret=sscanf(table_line, "%d\t%s\t%s\t%d", &cfg_id, cert_name, cert_file, &is_valid);
	if(ret!=4)
	{
		TFE_LOG_ERROR(g_pangu_rt->local_logger, "Trusted CA Store parse cert config failed: %s", table_line);
		return;
	}
	if(is_valid==1)
	{
		ret=tfe_proxy_ssl_add_trust_ca(cert_file);
		if(ret<0)
		{
			TFE_LOG_ERROR(g_pangu_rt->local_logger, "Trusted CA Store add cert failed %d:%s:%s", cfg_id, cert_name, cert_file);
		}
		else
		{
			TFE_LOG_INFO(g_pangu_rt->local_logger, "Trusted CA Store add cert success %d:%s:%s", cfg_id, cert_name, cert_file);
		}
	}
	else
	{
		ret=tfe_proxy_ssl_del_trust_ca(cert_file);
		if(ret<0)
		{
			TFE_LOG_ERROR(g_pangu_rt->local_logger, "Trusted CA Store del cert failed %d:%s:%s", cfg_id, cert_name, cert_file);
		}
		else
		{
			TFE_LOG_INFO(g_pangu_rt->local_logger, "Trusted CA Store del cert success %d:%s:%s", cfg_id, cert_name, cert_file);
		}
	}
	return;
}
void trusted_CA_update_crl_cb(int table_id,const char* table_line,void* u_para)
{
	int ret=0, crl_id=0, cert_id=0, is_valid=0;
	char crl_file[1024]={0};
	ret=sscanf(table_line, "%d\t%d\t%s\t%d", &crl_id, &cert_id, crl_file, &is_valid);
	if(ret!=4)
	{
		TFE_LOG_ERROR(g_pangu_rt->local_logger, "Trusted CA Store parse crl config failed: %s", table_line);
		return;
	}
	if(is_valid==1)
	{
		ret=tfe_proxy_ssl_add_crl(crl_file);
		if(ret<0)
		{
			TFE_LOG_ERROR(g_pangu_rt->local_logger, "Trusted CA Store add crl failed %d:%d:%s", crl_id, cert_id, crl_file);
		}
		else
		{
			TFE_LOG_INFO(g_pangu_rt->local_logger, "Trusted CA Store add crl success %d:%d:%s", crl_id, cert_id, crl_file);
		}
	}
	else
	{
		ret=tfe_proxy_ssl_del_crl(crl_file);
		if(ret<0)
		{
			TFE_LOG_ERROR(g_pangu_rt->local_logger, "Trusted CA Store del crl failed %d:%d:%s", crl_id, cert_id, crl_file);
		}
		else
		{
			TFE_LOG_INFO(g_pangu_rt->local_logger, "Trusted CA Store del crl success %d:%d:%s", crl_id, cert_id, crl_file);
		}
	}
	return;
}
void trusted_CA_update_finish_cb(void* u_para)
{
	if(g_pangu_rt->ca_store_reseting>0)
	{
		g_pangu_rt->ca_store_reseting--;
		if(g_pangu_rt->ca_store_reseting==0)
		{
			TFE_LOG_INFO(g_pangu_rt->local_logger, "Trusted CA Store Reset Finish.");
		}
	}
}

static int get_column_pos(const char* line, int column_seq, size_t *offset, size_t *len)
{
	const char* seps=" \t";
	char* saveptr=NULL, *subtoken=NULL, *str=NULL;
	char* dup_line=tfe_strdup(line);
	int i=0, ret=-1;
	for (str = dup_line; ; str = NULL)
	{
		subtoken = strtok_r(str, seps, &saveptr);
		if (subtoken == NULL)
			break;
		if(i==column_seq-1)
		{
			*offset=subtoken-dup_line;
			*len=strlen(subtoken);
			ret=0;
			break;
		}
		i++;
	}
	free(dup_line);
	return ret;
}

void subscribe_id_dup_cb(int table_id, MAAT_PLUGIN_EX_DATA* to, MAAT_PLUGIN_EX_DATA* from, long argl, void* argp)
{
	*to = tfe_strdup((char*)*from);
	return;
}
void subscribe_id_new_cb(int table_id, const char* key, const char* table_line, MAAT_PLUGIN_EX_DATA* ad, long argl, void* argp)
{
	int ret=0;
	size_t subscribe_id_offset, len;
	ret=get_column_pos(table_line, 4, &subscribe_id_offset, &len);
	if(ret<0)
	{
		TFE_LOG_ERROR(g_pangu_rt->local_logger, "Add subscribe ID faild: %s", table_line);
		return;
	}
	*ad=ALLOC(char, len+1);
	memcpy(*ad, table_line+subscribe_id_offset, len);
	TFE_LOG_INFO(g_pangu_rt->local_logger, "Add subscribe ID: %s", (char*)*ad);
	return;
}
void subscribe_id_free_cb(int table_id, MAAT_PLUGIN_EX_DATA* ad, long argl, void* argp)
{
	TFE_LOG_INFO(g_pangu_rt->local_logger, "Delete subscribe ID: %s", (char*)*ad);
	free(*ad);
	*ad=NULL;
}

static enum manipulate_action manipulate_action_str2idx(const char *action_str)
{
	const char *clue_action_map[__MA_ACTION_MAX];
	clue_action_map[MA_ACTION_REDIRECT]= "redirect";
	clue_action_map[MA_ACTION_BLOCK]= "block";
	clue_action_map[MA_ACTION_REPLACE]= "replace";
	clue_action_map[MA_ACTION_HIJACK]= "hijack";
	clue_action_map[MA_ACTION_INSERT]= "insert";

	size_t i = 0;

	for (i = 0; i < sizeof(clue_action_map) / sizeof(const char *); i++)
	{
		if (0 == strcasecmp(action_str, clue_action_map[i]))
			break;
	}
	return (enum manipulate_action)i;
}

void policy_action_param_new(int idx, const struct Maat_rule_t* rule, const char* srv_def_large,
					 MAAT_RULE_EX_DATA* ad, long argl, void *argp)
{
	struct policy_action_param* param=NULL;

	*ad=NULL;
	if((unsigned int)rule->serv_def_len<strlen("{}")+1)
	{
		return;
	}
	if((unsigned char)rule->action!=PG_ACTION_MANIPULATE&&(unsigned char)rule->action!=PG_ACTION_REJECT)
	{
		return;
	}
	int rule_id;
	cJSON *json=NULL, *rules=NULL, *item=NULL;
	json=cJSON_Parse(srv_def_large);
	if(json==NULL)
	{
		TFE_LOG_ERROR(g_pangu_rt->local_logger, "invalid policy parameter: id = %d", rule->config_id);
		return;
	}

	item=cJSON_GetObjectItem(json, "protocol");
	if(unlikely(!item || !cJSON_IsString(item)))
	{
		TFE_LOG_ERROR(g_pangu_rt->local_logger, "Invalid policy parameter: %d invalid protocol format", rule->config_id);
        goto error_out;
	}
	if(0!=strcasecmp(item->valuestring, "http"))
	{
		goto error_out;
	}

	param=ALLOC(struct policy_action_param, 1);
 	param->ref_cnt=1;
	pthread_mutex_init(&(param->lock), NULL);

	item=cJSON_GetObjectItem(json,"method");
	if(item && item->type==cJSON_String)
	{
		param->action =manipulate_action_str2idx(item->valuestring);
	}

	switch(param->action)
	{
		case MA_ACTION_REDIRECT:
			item=cJSON_GetObjectItem(json,"code");
			if(item && item->type==cJSON_Number)
			{
				param->status_code = item->valueint;
			}
			item=cJSON_GetObjectItem(json,"to");
			if(item && item->type==cJSON_String)
			{
				param->message = tfe_strdup(item->valuestring);
			}
			item=cJSON_GetObjectItem(json,"enforcement_ratio");
			if(item && item->type==cJSON_Number)
			{
				param->enforcement_ratio = item->valuedouble;
			}
			else
			{
				param->enforcement_ratio = 1;
			}
			break;
		case MA_ACTION_BLOCK:
			item=cJSON_GetObjectItem(json,"code");
			if(item && item->type==cJSON_Number)
			{
				param->status_code = item->valueint;
			}
			item=cJSON_GetObjectItem(json,"message");
			if(item && item->type==cJSON_String)
			{
				param->message = tfe_strdup(item->valuestring);
			}
			item=cJSON_GetObjectItem(json,"html_profile");
			if(item && item->type==cJSON_Number)
			{
				param->profile_id = item->valueint;
			}
			break;
		case MA_ACTION_REPLACE:
			item=cJSON_GetObjectItem(json,"enforcement_ratio");
			if(item && item->type==cJSON_Number)
			{
				param->enforcement_ratio = item->valuedouble;
			}
			else
			{
				param->enforcement_ratio = 1;
			}
			rules = cJSON_GetObjectItem(json, "rules");
			if(rules == NULL)
			{
				break;
			}
			rule_id = 0;
			param->rule = ALLOC(struct replace_rule, MAX_EDIT_ZONE_NUM);
			for (item = rules->child; item != NULL; item = item->next)
			{
				char * search  = cJSON_GetObjectItem(item , "search_in")->valuestring;
				if (search == NULL) break;

				param->rule[rule_id].zone = zone_name_to_id(search);
				if (param->rule[rule_id].zone == kZoneMax)
				{
					break;
				}
				param->rule[rule_id].find = tfe_strdup(cJSON_GetObjectItem(item , "find")->valuestring);
				param->rule[rule_id].replace_with = tfe_strdup(cJSON_GetObjectItem(item , "replace_with")->valuestring);
				rule_id++;
			}
			param->n_rule = rule_id;
			break;
		case MA_ACTION_HIJACK:
			item=cJSON_GetObjectItem(json,"hijack_profile");
			if(item && item->type==cJSON_Number){
				param->profile_id = item->valueint;
			}
			item=cJSON_GetObjectItem(json,"enforcement_ratio");
			if(item && item->type==cJSON_Number)
			{
				param->enforcement_ratio = item->valuedouble;
			}
			else
			{
				param->enforcement_ratio = 1;
			}
			break;
		case MA_ACTION_INSERT:
			item=cJSON_GetObjectItem(json,"insert_profile");
			if(item && item->type==cJSON_Number){
				param->profile_id = item->valueint;
			}
			item=cJSON_GetObjectItem(json,"position");
			if(item && item->type==cJSON_String){
				param->position = tfe_strdup(item->valuestring);
			}
			item=cJSON_GetObjectItem(json,"enforcement_ratio");
			if(item && item->type==cJSON_Number)
			{
				param->enforcement_ratio = item->valuedouble;
			}
			else
			{
				param->enforcement_ratio = 1;
			}
			break;
		default: assert(0);
			break;
	}
	*ad=param;
	TFE_LOG_INFO(g_pangu_rt->local_logger, "Add ctrl policy: %d", rule->config_id);
error_out:
	cJSON_Delete(json);
	return;
}

void policy_action_param_free_cb(int table_id, const struct Maat_rule_t* rule, const char* srv_def_large, MAAT_RULE_EX_DATA* ad, long argl, void *argp)
{
	unsigned int i=0;
	if(*ad==NULL)
	{
		return;
	}
	struct policy_action_param* param=(struct policy_action_param*)*ad;
	pthread_mutex_lock(&(param->lock));
	param->ref_cnt--;
	if(param->ref_cnt>0)
	{
		pthread_mutex_unlock(&(param->lock));
		return;
	}
	pthread_mutex_unlock(&(param->lock));
	pthread_mutex_destroy(&(param->lock));
	for(i=0; i<param->n_rule; i++)
	{
		FREE(&(param->rule[i].find));
		FREE(&(param->rule[i].replace_with));
	}

	if (param->message)
		FREE(&(param->message));
	if (param->position)
		FREE(&(param->position));
	FREE(&(param));
	return;
}

void policy_action_param_free(struct policy_action_param* param)
{
	policy_action_param_free_cb(0, NULL, NULL, (void**)&param, 0, NULL);
	return;
}

void policy_action_param_dup(int idx,	MAAT_RULE_EX_DATA *to, MAAT_RULE_EX_DATA *from, long argl, void *argp)
{
	struct policy_action_param* from_param=*((struct policy_action_param**)from);
	if(from_param==NULL)
	{
		*to=NULL;
		return;
	}
	pthread_mutex_lock(&(from_param->lock));
	from_param->ref_cnt++;
	pthread_mutex_unlock(&(from_param->lock));
	*((struct policy_action_param**)to)=from_param;
	return;
}

void ma_profile_table_new_cb(int table_id, const char* key, const char* table_line, MAAT_PLUGIN_EX_DATA* ad, long argl, void* argp)
{
	int ret=0, profile_id=0, is_valid=0;
	char profile_name[128]={0}, formate[128]={0};
	char profile_path[TFE_PATH_MAX]={0};

	ret=sscanf(table_line, "%d\t%s\t%s\t%s\t%d", &profile_id, profile_name, formate, profile_path, &is_valid);
	if(ret!=5)
	{
		TFE_LOG_ERROR(g_pangu_rt->local_logger, "Policy table parse config failed: %s", table_line);
		return;
	}

	struct manipulate_profile* ply_profile=ALLOC(struct manipulate_profile, 1);
	memset(ply_profile, 0, sizeof(struct manipulate_profile));

	ply_profile->profile_id=profile_id;
	ply_profile->ref_cnt=1;
	pthread_mutex_init(&(ply_profile->lock), NULL);

	if(strcasecmp(formate, "template") == 0)
	{
		ply_profile->tpl = ctemplate::Template::GetTemplate(profile_path, ctemplate::DO_NOT_STRIP);
	}else
	{
		ply_profile->profile_msg = tfe_read_file(profile_path, &ply_profile->msg_len);
		if (ply_profile->profile_msg == NULL)
		{
			TFE_LOG_ERROR(g_pangu_rt->local_logger, "Read file failed %d:%s:%s", profile_id, profile_name, profile_path);
		}
	}
	ply_profile->profile_name=tfe_strdup(profile_name);
	ply_profile->profile_type=tfe_strdup(formate);

	TFE_LOG_INFO(g_pangu_rt->local_logger, "Policy table add success %d", profile_id);

	*ad = ply_profile;
	return;
}

void ma_insert_profile_table_new_cb(int table_id, const char* key, const char* table_line, MAAT_PLUGIN_EX_DATA* ad, long argl, void* argp)
{
	int ret=0, profile_id=0, is_valid=0;
	char profile_name[128]={0}, formate[128]={0};
	char profile_path[TFE_PATH_MAX]={0},profile_position[TFE_PATH_MAX]={0};

	ret=sscanf(table_line, "%d\t%s\t%s\t%s\t%s\t%d", &profile_id, profile_name, formate, profile_path, profile_position, &is_valid);
	if(ret!=6)
	{
		TFE_LOG_ERROR(g_pangu_rt->local_logger, "Policy table parse config failed: %s", table_line);
		return;
	}
	struct manipulate_profile* ply_profile=ALLOC(struct manipulate_profile, 1);
	memset(ply_profile, 0, sizeof(struct manipulate_profile));

	ply_profile->profile_id=profile_id;
	ply_profile->ref_cnt=1;
	pthread_mutex_init(&(ply_profile->lock), NULL);

	if(strcasecmp(formate, "template") == 0)
	{
		ply_profile->tpl = ctemplate::Template::GetTemplate(profile_path, ctemplate::DO_NOT_STRIP);
	}else
	{
		ply_profile->profile_msg = tfe_read_file(profile_path, &ply_profile->msg_len);
		if (ply_profile->profile_msg == NULL)
		{
			TFE_LOG_ERROR(g_pangu_rt->local_logger, "Read file failed %d:%s:%s", profile_id, profile_name, profile_path);
		}
	}
	ply_profile->profile_name=tfe_strdup(profile_name);
	ply_profile->profile_type=tfe_strdup(formate);
	ply_profile->profile_position=tfe_strdup(profile_position);

	TFE_LOG_INFO(g_pangu_rt->local_logger, "Policy table add success %d", profile_id);

	*ad = ply_profile;
	return;
}

void ma_hijack_profile_table_new_cb(int table_id, const char* key, const char* table_line, MAAT_PLUGIN_EX_DATA* ad, long argl, void* argp)
{
	int ret=0, profile_id=0, is_valid=0;
	char profile_name[128]={0}, formate[128]={0};
	char profile_path[TFE_PATH_MAX]={0},hijack_name[128]={0};

	ret=sscanf(table_line, "%d\t%s\t%s\t%s\t%s\t%d", &profile_id, profile_name, hijack_name, formate, profile_path, &is_valid);
	if(ret!=6)
	{
		TFE_LOG_ERROR(g_pangu_rt->local_logger, "Policy table parse config failed: %s", table_line);
		return;
	}
	struct manipulate_profile* ply_profile=ALLOC(struct manipulate_profile, 1);
	ply_profile->ref_cnt=1;
	pthread_mutex_init(&(ply_profile->lock), NULL);

	ply_profile->profile_id=profile_id;
	ply_profile->profile_msg=tfe_strdup(profile_path);
	ply_profile->profile_name=tfe_strdup(hijack_name);
	ply_profile->profile_type=tfe_strdup(formate);

	TFE_LOG_INFO(g_pangu_rt->local_logger, "Policy table add success %d", profile_id);

	*ad = ply_profile;
	return;
}

void ma_profile_table_free_cb(int table_id, MAAT_PLUGIN_EX_DATA* ad, long argl, void *argp)
{
	if(*ad==NULL)
	{
		return;
	}
	struct manipulate_profile* ply_obj=(struct manipulate_profile*)(*ad);
	pthread_mutex_lock(&(ply_obj->lock));
	ply_obj->ref_cnt--;
	if(ply_obj->ref_cnt>0)
	{
		pthread_mutex_unlock(&(ply_obj->lock));
		return;
	}
	pthread_mutex_unlock(&(ply_obj->lock));
	pthread_mutex_destroy(&(ply_obj->lock));

	FREE(&ply_obj->profile_type);
	FREE(&ply_obj->profile_msg);
	FREE(&ply_obj->profile_name);
	if (ply_obj->profile_position)
		FREE(&ply_obj->profile_position);
	FREE(&ply_obj);
	*ad=NULL;
	return;
}

void ma_profile_table_free(struct manipulate_profile* ply_obj)
{
	ma_profile_table_free_cb(0, (void **)&ply_obj, 0, NULL);
}

void ma_profile_table_dup_cb(int table_id,	MAAT_PLUGIN_EX_DATA *to, MAAT_PLUGIN_EX_DATA *from, long argl, void *argp)
{
	struct manipulate_profile* ply_obj=(struct manipulate_profile*)(*from);
	pthread_mutex_lock(&(ply_obj->lock));
	ply_obj->ref_cnt++;
	pthread_mutex_unlock(&(ply_obj->lock));
	*to=ply_obj;
}

int maat_table_init(const char* table_name,
					Maat_start_callback_t *start, Maat_update_callback_t *update,Maat_finish_callback_t *finish,
					void *u_para)
{
	int table_id=0;

	table_id=Maat_table_register(g_pangu_rt->maat, table_name);
	if(table_id>=0)
	{
		Maat_table_callback_register(g_pangu_rt->maat, table_id,
									 start, update, finish,	u_para);
	}
	return table_id;
}

int maat_table_ex_init(int profile_idx,
					   Maat_plugin_EX_new_func_t* new_func,
					   Maat_plugin_EX_free_func_t* free_func,
					   Maat_plugin_EX_dup_func_t* dup_func)
{
	int table_id = 0;

	const char *table_name_map[] = {"TSG_PROFILE_RESPONSE_PAGES",
									"PXY_PROFILE_INSERT_SCRIPTS",
									"PXY_PROFILE_HIJACK_FILES"};

	table_id=g_pangu_rt->plolicy_table_id[profile_idx]=Maat_table_register(g_pangu_rt->maat, table_name_map[profile_idx]);
	if(table_id >= 0)
	{
		table_id=Maat_plugin_EX_register(g_pangu_rt->maat,
									     table_id,
									     new_func,
									     free_func,
									     dup_func,
									     NULL, 0, NULL);
		return 0;
	}
	TFE_LOG_INFO(NULL, "Pangu HTTP register table %s failed.", table_name_map[profile_idx]);

	return -1;
}

int pangu_policy_init(const char* profile_path, const char* static_section, const char* dynamic_section)
{
	int ret = 0;

	g_pangu_rt->maat = create_maat_feather("static", profile_path, static_section, 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] = "TSG_SECURITY_ADDR";
	table_name[PXY_CTRL_HTTP_URL] = "TSG_FIELD_HTTP_URL";
	table_name[PXY_CTRL_HTTP_FQDN] = "TSG_FIELD_HTTP_HOST";
	table_name[PXY_CTRL_HTTP_REQ_HDR] = "TSG_FIELD_HTTP_REQ_HDR";
	table_name[PXY_CTRL_HTTP_REQ_BODY] = "TSG_FIELD_HTTP_REQ_CONTENT";
	table_name[PXY_CTRL_HTTP_RES_HDR] = "TSG_FIELD_HTTP_RES_HDR";
	table_name[PXY_CTRL_HTTP_RES_BODY] = "TSG_FIELD_HTTP_RES_CONTENT";
	table_name[PXY_CTRL_SUBSCRIBE_ID] = "TSG_OBJ_SUBSCRIBER_ID";
	table_name[PXY_CTRL_APP_ID] = "TSG_OBJ_APP_ID";
	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;
		}
	}

	g_pangu_rt->ctrl_compile_idx=Maat_rule_get_ex_new_index(g_pangu_rt->maat, "PXY_CTRL_COMPILE",
													         policy_action_param_new,
													         policy_action_param_free_cb,
													         policy_action_param_dup,
													  		 0, NULL);

	ret = maat_table_init("PXY_PROFILE_TRUSTED_CA_CERT",
						   trusted_CA_update_start_cb,
						   trusted_CA_update_cert_cb,
						   trusted_CA_update_finish_cb,
						   g_pangu_rt);
	if(ret<0)
	{
		TFE_LOG_INFO(NULL, "Pangu HTTP register table PXY_OBJ_TRUSTED_CA_CERT failed.");
		goto error_out;
	}

	ret = maat_table_init("PXY_OBJ_TRUSTED_CA_CRL",
						   trusted_CA_update_start_cb,
						   trusted_CA_update_crl_cb,
						   trusted_CA_update_finish_cb,
						   g_pangu_rt);
	if(ret<0)
	{
		TFE_LOG_INFO(NULL, "Pangu HTTP register table PXY_OBJ_TRUSTED_CA_CRL failed.");
		goto error_out;
	}


	ret = maat_table_ex_init(POLICY_PROFLIE_TABLE_REJECT,
							 ma_profile_table_new_cb,
							 ma_profile_table_free_cb,
							 ma_profile_table_dup_cb);
	if(ret<0)
	{
		goto error_out;
	}

	ret = maat_table_ex_init(POLICY_PROFILE_TABLE_INSERT,
							 ma_insert_profile_table_new_cb,
							 ma_profile_table_free_cb,
							 ma_profile_table_dup_cb);
	if(ret<0)
	{
		goto error_out;
	}

	ret = maat_table_ex_init(POLICY_PROFILE_TABLE_HIJACK,
							 ma_hijack_profile_table_new_cb,
							 ma_profile_table_free_cb,
							 ma_profile_table_dup_cb);
	if(ret<0)
	{
		goto error_out;
	}

	g_pangu_rt->dyn_maat = create_maat_feather("dyn", profile_path, dynamic_section, g_pangu_rt->thread_num, g_pangu_rt->local_logger);
	if (!g_pangu_rt->maat)
	{
		goto error_out;
	}
	g_pangu_rt->subscriber_id_table_id=Maat_table_register(g_pangu_rt->dyn_maat, "TSG_DYN_SUBSCRIBER_IP");
	ret=Maat_plugin_EX_register(g_pangu_rt->dyn_maat,
							g_pangu_rt->subscriber_id_table_id,
							subscribe_id_new_cb,
							subscribe_id_free_cb,
							subscribe_id_dup_cb,
							NULL,
							0,
							NULL);
	if(ret!=0)
	{
		TFE_LOG_ERROR(NULL, "Pangu HTTP Dynamic Maat  TSG_DYN_SUBSCRIBER_IP EX data register failed.");
		goto error_out;
	}

error_out:
	return ret;
}

Ratelimiter_handle_t ratelimit_handle_create(const char* profile_path, const char* static_section)
{
	int ret=0, field_stat=0;
	int redis_db_idx = 0;
	int redis_port = 0, interval_sec = 0;
	char redis_server[TFE_STRING_MAX] = {0};
	char token_name[TFE_STRING_MAX] = {0};

	Ratelimiter_handle_t ratelimit = NULL;

	MESA_load_profile_int_def(profile_path, static_section, "enable", &(g_pangu_rt->enable_rate), 0);
	MESA_load_profile_int_def(profile_path, static_section, "redis_port", &(redis_port), 6379);
	MESA_load_profile_string_def(profile_path, static_section, "redis_server", redis_server, sizeof(redis_server), "");
	MESA_load_profile_string_def(profile_path, static_section, "token_name", token_name, sizeof(token_name), "");
	MESA_load_profile_int_def(profile_path, static_section, "redis_db_index", &(redis_db_idx), 0);

	MESA_load_profile_int_def(profile_path, static_section, "interval_sec", &(interval_sec), 1);
	if (g_pangu_rt->enable_rate != 1)
	{
		return NULL;
	}
	ratelimit=Ratelimiter_create(token_name, g_pangu_rt->local_logger);
	Ratelimiter_set_opt(ratelimit, RATELIMITER_OPT_INTERVAL_SEC, &interval_sec, sizeof(interval_sec));
	Ratelimiter_set_opt(ratelimit, RATELIMITER_OPT_REDIS_IP, redis_server, strlen(redis_server) + 1);
	Ratelimiter_set_opt(ratelimit, RATELIMITER_OPT_REDIS_PORT, &redis_port, sizeof(redis_port));
	Ratelimiter_set_opt(ratelimit, RATELIMITER_OPT_REDIS_INDEX, &redis_db_idx, sizeof(redis_db_idx));

	/*field stata for debug test**/
	MESA_load_profile_int_def(profile_path, static_section, "field_stat", &(field_stat), 0);
	if (field_stat == 1)
	{
		MESA_load_profile_int_def(profile_path, static_section, "stat_port", &(redis_port), 6379);
		MESA_load_profile_string_def(profile_path, static_section, "stat_server", redis_server, sizeof(redis_server), "");
		Ratelimiter_set_opt(ratelimit, RATELIMITER_OPT_FIELD_STAT, &field_stat, sizeof(field_stat));
		Ratelimiter_set_opt(ratelimit, RATELIMITER_OPT_STAT_IP, redis_server, strlen(redis_server));
		Ratelimiter_set_opt(ratelimit, RATELIMITER_OPT_STAT_PORT, &redis_port, sizeof(redis_port));
		Ratelimiter_set_opt(ratelimit, RATELIMITER_OPT_STAT_PATH, token_name, sizeof(token_name));
		Ratelimiter_stat_init(ratelimit);
	}
	ret = Ratelimiter_start(ratelimit);
	if (ret < 0)
	{
		TFE_LOG_ERROR(g_pangu_rt->local_logger, "%s Ratelimiter init failed.", __FUNCTION__);
		goto error_out;
	}
	return ratelimit;
error_out:
	Ratelimiter_stop(ratelimit);
	return NULL;
}

int pangu_http_init(struct tfe_proxy * proxy)
{
	const char * profile_path = "./conf/pangu/pangu_pxy.conf";
	const char * logfile = "./log/pangu_pxy.log";
	int temp=0;

	g_pangu_rt = ALLOC(struct pangu_rt, 1);

	g_pangu_rt->thread_num = tfe_proxy_get_work_thread_count();
	g_pangu_rt->gc_evbase=tfe_proxy_get_gc_evbase();

	MESA_load_profile_int_def(profile_path, "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_path, "LOG", g_pangu_rt->local_logger);
	if (!g_pangu_rt->send_logger)
	{
		goto error_out;
	}

	g_pangu_rt->fs_handle = tfe_proxy_get_fs_handle();
	g_pangu_rt->ratelimiter=ratelimit_handle_create(profile_path, "ratelimit");

	pangu_http_stat_init(g_pangu_rt);
	if(pangu_policy_init(profile_path, "MAAT", "DYNAMIC_MAAT")<0)
	{
		goto error_out;
	}

	char page_path[256];
	memset(page_path, 0, sizeof(page_path));
	MESA_load_profile_string_def(profile_path, "TEMPLATE", "PAGE_403", page_path, sizeof(page_path),
		"./resource/pangu/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_path, "TEMPLATE", "PAGE_404", page_path, sizeof(page_path),
		"./resource/pangu/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_path, "TEMPLATE", "PAGE_451", page_path, sizeof(page_path),
		"./resource/pangu/HTTP451.html");
	g_pangu_rt->tpl_451 = ctemplate::Template::GetTemplate(page_path, ctemplate::DO_NOT_STRIP);

	MESA_load_profile_int_def(profile_path, "TANGO_CACHE", "suspend_max", &(temp), 1024*1024);
	g_pangu_rt->suspend_max=temp;

	MESA_load_profile_int_def(profile_path, "TANGO_CACHE", "enable_cache", &(g_pangu_rt->cache_enabled), 1);
	if(g_pangu_rt->cache_enabled)
	{
		g_pangu_rt->cache = create_web_cache_handle(profile_path, "TANGO_CACHE", g_pangu_rt->gc_evbase,
															g_pangu_rt->maat, g_pangu_rt->local_logger);
		if(!g_pangu_rt->cache)
		{
			TFE_LOG_INFO(NULL, "Tango Cache init failed.");
			goto error_out;
		}
		TFE_LOG_INFO(NULL, "Tango Cache Enabled.");
	}
	TFE_LOG_INFO(NULL, "Pangu HTTP init success.");
	g_business_maat=g_pangu_rt->maat;
	return 0;

error_out:
	TFE_LOG_ERROR(NULL, "Pangu HTTP init failed.");
	return -1;
}

struct replace_ctx
{
	struct replace_rule * rule;
	size_t n_rule;
	struct tfe_http_half * replacing;
	struct evbuffer * http_body;
	int actually_replaced;
};

struct insert_ctx
{
	struct insert_rule *rule;
	struct tfe_http_half * replacing;
	struct evbuffer *http_body;
	int actually_replaced;
};

struct pangu_http_ctx
{
	int magic_num;
	enum pangu_action action;
	char * action_para;
	scan_status_t scan_mid;
	stream_para_t sp;
	struct cache_mid* cmid;
	struct Maat_rule_t * enforce_rules;
	size_t n_enforce;
	struct policy_action_param *param;
	struct evbuffer* log_req_body, *log_resp_body;

	size_t inject_sz;
	int manipulate_replaced;
	struct replace_ctx * rep_ctx;
	struct insert_ctx * ins_ctx;

	int (* resumed_cb)(const struct tfe_stream * stream,
	const struct tfe_http_session * session, enum tfe_http_event event, const unsigned char * data,
	size_t datalen, unsigned int thread_id, struct pangu_http_ctx* ctx);

	enum cache_pending_result pending_result;
	struct future *f_cache_pending, *f_cache_query;
	struct tfe_http_session * ref_session;
	struct tfe_http_half* cache_revalidate_req;
	struct tfe_http_half* cached_response;
	size_t cache_result_declared_sz, cache_result_actual_sz;
	struct cache_write_context* cache_write_ctx;
	int cache_wirte_result;

	int thread_id;
};


void http_repl_ctx_free(struct replace_ctx* rep_ctx)
{
	if (rep_ctx->http_body)
	{
		evbuffer_free(rep_ctx->http_body);
		rep_ctx->http_body = NULL;
	}
	FREE(&rep_ctx);
	return;
}

void http_ins_ctx_free(struct insert_ctx* ins_ctx)
{
	if (ins_ctx->rule->script)
		FREE(&ins_ctx->rule->script);
	FREE(&ins_ctx->rule->type);
	if (ins_ctx->rule->position)
		FREE(&ins_ctx->rule->position);
	FREE(&(ins_ctx->rule));
	if (ins_ctx->http_body)
	{
		evbuffer_free(ins_ctx->http_body);
		ins_ctx->http_body = NULL;
	}
	FREE(&ins_ctx);
	return;
}


#define	HTTP_CTX_MAGIC_NUM	20181021
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->magic_num=HTTP_CTX_MAGIC_NUM;
	ctx->scan_mid = NULL;
	ctx->thread_id = (int) thread_id;
	return ctx;
}

static void pangu_http_ctx_free(struct pangu_http_ctx * ctx)
{
	assert(ctx->magic_num==HTTP_CTX_MAGIC_NUM);
	if (ctx->rep_ctx)
	{
		http_repl_ctx_free(ctx->rep_ctx);
		ctx->rep_ctx = NULL;
	}
	if (ctx->ins_ctx)
	{
		http_ins_ctx_free(ctx->ins_ctx);
		ctx->ins_ctx = NULL;
	}
	ctx->manipulate_replaced=0;
	FREE(&ctx->enforce_rules);
	policy_action_param_free(ctx->param);
	ctx->param=NULL;
	Maat_clean_status(&(ctx->scan_mid));
	ctx->scan_mid = NULL;

	if(ctx->sp)
	{
		Maat_stream_scan_string_end(&(ctx->sp));
	}
	if(ctx->cache_write_ctx)
	{
		web_cache_write_end(ctx->cache_write_ctx);
		ctx->cache_write_ctx=NULL;
	}

	//On session recycle
	if(ctx->cache_revalidate_req)
	{
		tfe_http_half_free(ctx->cache_revalidate_req);
		ctx->cache_revalidate_req=NULL;
	}
	if(ctx->cached_response)
	{
		//Dirty close
		ctx->cached_response=NULL;
	}

	if(ctx->f_cache_query)
	{
		future_destroy(ctx->f_cache_query);
		ctx->f_cache_query=NULL;
	}
	if(ctx->f_cache_pending)
	{
		future_destroy(ctx->f_cache_pending);
		ctx->f_cache_pending=NULL;
	}
	if(ctx->log_req_body)
	{
		evbuffer_free(ctx->log_req_body);
		ctx->log_req_body=NULL;
	}
	if(ctx->log_resp_body)
	{
		evbuffer_free(ctx->log_resp_body);
		ctx->log_resp_body=NULL;
	}
	ctx->magic_num=0;
	FREE(&ctx);
}

static inline void addr_tfe2sapp(const struct tfe_stream_addr * tfe_addr, struct ipaddr * sapp_addr)
{
	if(tfe_addr->addrtype==TFE_ADDR_STREAM_TUPLE4_V4||tfe_addr->addrtype==TFE_ADDR_IPV4)
	{
		sapp_addr->addrtype = ADDR_TYPE_IPV4;
	}
	else
	{
		sapp_addr->addrtype=ADDR_TYPE_IPV6;
	}
	sapp_addr->paddr = (char *) tfe_addr->paddr;
	return;
}

static struct manipulate_profile* get_profile_by_id(int profile_table, int profile_id)
{
	struct manipulate_profile* result=NULL;

	char cfg_id_str[16] = {0};
	snprintf(cfg_id_str, sizeof(cfg_id_str), "%d", profile_id);
	int table_id = g_pangu_rt->plolicy_table_id[profile_table];

	result = (struct manipulate_profile*)Maat_plugin_get_EX_data(g_pangu_rt->maat, table_id, (const char*)cfg_id_str);
	return result;
}

static unsigned long long try_send_by_token(int inject_sz)
{
	if (g_pangu_rt->enable_rate != 1)
	{
		return 1;
	}
	return Ratelimiter_customer_factory(g_pangu_rt->ratelimiter, inject_sz);
}

static int pangu_action_weight[__PG_ACTION_MAX] = {0};
void __pangu_action_weight_init()  __attribute__((constructor, used));
void __pangu_action_weight_init()
{
	pangu_action_weight[PG_ACTION_NONE] = 0;
	pangu_action_weight[PG_ACTION_MONIT] = 1;
	pangu_action_weight[PG_ACTION_MANIPULATE] = 2;
	pangu_action_weight[PG_ACTION_REJECT] = 3;
	pangu_action_weight[PG_ACTION_WHITELIST] = 4;
}

static inline int action_cmp(enum pangu_action a1, enum pangu_action a2)
{
	return pangu_action_weight[a1] - pangu_action_weight[a2];
}

//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, struct policy_action_param **param)
{
	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++)
	{
		unsigned char __expand_action = (unsigned char) hit_rules[i].action;
		enum pangu_action __action = (enum pangu_action) __expand_action;

		if (__action == PG_ACTION_MONIT)
		{
			memcpy(monit_rule + n_monit, hit_rules + i, sizeof(struct Maat_rule_t));
			n_monit++;
		}
		if (action_cmp(__action, prior_action) > 0)
		{
			prior_rule = hit_rules + i;
			prior_action = __action;
		}
		else if (action_cmp(__action, prior_action) == 0)
		{
			if (hit_rules[i].config_id > prior_rule->config_id)
			{
				prior_rule = hit_rules + i;
			}
		}
		else
		{
			continue;
		}
	}

	if (prior_action == PG_ACTION_WHITELIST)
	{
		if(*n_enforce==0)
		{
			*enforce_rules=ALLOC(struct Maat_rule_t, 1);
		}
		*enforce_rules[0]=*prior_rule;
		*n_enforce=1;
		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
	{
		memmove(*enforce_rules+1, *enforce_rules, exist_enforce_num*sizeof(struct Maat_rule_t));
		memcpy(*enforce_rules, prior_rule, sizeof(struct Maat_rule_t));
		memcpy(*enforce_rules + exist_enforce_num + 1, monit_rule, n_monit * sizeof(struct Maat_rule_t));
	}

	void *ex_data=Maat_rule_get_ex_data(g_pangu_rt->maat, prior_rule, g_pangu_rt->ctrl_compile_idx);
	if(ex_data!=NULL)
	{
		*param=(struct policy_action_param*)ex_data;
	}

	return prior_action;
}
//HTML template is downloaded from https://github.com/AndiDittrich/HttpErrorPages
static void template_generate(int status_code, int cfg_id, const char* msg, char ** page_buff, size_t * page_size)
{
	ctemplate::TemplateDictionary dict("pg_page_dict");	//dict is automatically finalized after function returned.
	dict.SetIntValue("cfg_id", cfg_id);

	if (NULL == msg)
	{
		dict.SetValue("msg", "NULL");
	}
	else
	{
		dict.SetValue("msg", msg);
	}
	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);
	*page_size = output.length() + 1;
	*page_buff = tfe_strdup(output.c_str());
}

void manipulate_profile_free(struct manipulate_profile* ma_profile)
{
	FREE(&ma_profile->profile_type);
	FREE(&ma_profile->profile_msg);
	FREE(&ma_profile->profile_name);
	FREE(&ma_profile);
}

static int html_generate(int profile_id, const char* msg, char ** page_buff, size_t * page_size)
{
	int ret = 0;

	struct manipulate_profile* block_profile=get_profile_by_id(POLICY_PROFLIE_TABLE_REJECT, profile_id);
	if(block_profile==NULL)
	{
		ret=-1;
		return ret;
	}
	if(!strncmp(block_profile->profile_type, "template", strlen(block_profile->profile_type)))
	{
		ctemplate::TemplateDictionary dict("pg_page_dict");	//dict is automatically finalized after function returned.
		dict.SetIntValue("cfg_id", profile_id);
		dict.SetValue("msg", msg);
		std::string output;

		block_profile->tpl->Expand(&output, &dict);
		*page_size = output.length() + 1;
		*page_buff = tfe_strdup(output.c_str());
	}
	else
	{
		*page_size = block_profile->msg_len;
		*page_buff = tfe_strdup(block_profile->profile_msg);
	}
	ma_profile_table_free(block_profile);
	block_profile=NULL;

 	return ret;
}

static void html_free(char ** page_buff)
{
	FREE(page_buff);
	return;
}

static int http_enforcement_ratio(float enforcement_ratio)
{
    int enforcement_ratio_temp = 0;
    enforcement_ratio_temp = enforcement_ratio * 10000;

    srand(time(NULL));
    int random = rand() % (10000-1);
    if (random >=0 && random <= enforcement_ratio_temp)
    {
        return 1;
    }
    return 0;
}

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)
{
	struct tfe_http_session * to_write_sess = NULL;
	char * rewrite_buff = NULL;
	size_t rewrite_sz = 0;

	struct policy_action_param *param = ctx->param;
	int ratio = http_enforcement_ratio(param->enforcement_ratio);
	if (ratio != 1)
	{
		TFE_LOG_DEBUG(g_pangu_rt->local_logger, "enforcement ratio:%f", param->enforcement_ratio);
		ctx->action = PG_ACTION_NONE;
		return;
	}
	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);
		ctx->action = PG_ACTION_NONE;
		tfe_http_session_detach(session); return;
	}

	struct replace_ctx * rep_ctx = ctx->rep_ctx;
	if (ctx->rep_ctx == NULL)
	{
		/* we must determinate the replace action on HTTP header, otherwise,
		 * the header has been forwarded, only replace the body but not modify header will raise exception */
		if ((events & EV_HTTP_REQ_HDR) || (events & EV_HTTP_RESP_HDR))
		{
			struct policy_action_param *param = ctx->param;
			ctx->rep_ctx = rep_ctx = ALLOC(struct replace_ctx, 1);
			rep_ctx->rule = param->rule;
			rep_ctx->n_rule = param->n_rule;
		}
		else
		{
			TFE_STREAM_LOG_INFO(stream, "Can only setup replace on REQ/RESP headers, detached.");
			ctx->action = PG_ACTION_NONE;
			tfe_http_session_detach(session); return;
		}
	}

	struct tfe_http_half * in_req_half = session->req;
	struct tfe_http_half * in_resp_half = session->resp;
	struct tfe_http_req_spec * in_req_spec = &in_req_half->req_spec;
	struct tfe_http_resp_spec * in_resp_spec = &in_resp_half->resp_spec;

	if ((events & EV_HTTP_REQ_HDR) || (events & EV_HTTP_RESP_HDR))
	{
		char * rewrite_uri = NULL;
		size_t rewrite_uri_sz=0;
		if (tfe_http_in_request(events))
		{
			rewrite_uri_sz = execute_replace_rule(in_req_spec->uri, strlen(in_req_spec->uri),
				kZoneRequestUri, rep_ctx->rule, rep_ctx->n_rule, &rewrite_uri, 1);
			if(rewrite_uri_sz>0) rep_ctx->actually_replaced=1;
			rep_ctx->replacing = tfe_http_session_request_create(to_write_sess, in_req_spec->method,
				rewrite_uri_sz >0 ? rewrite_uri : in_req_spec->uri);

			tfe_http_session_request_set(to_write_sess, rep_ctx->replacing);
		}
		else
		{
			rep_ctx->replacing = tfe_http_session_response_create(to_write_sess, in_resp_spec->resp_code);
			tfe_http_session_response_set(to_write_sess, rep_ctx->replacing);
		}

		if (rewrite_uri != NULL)
		{
			FREE(&rewrite_uri);
		}

		enum replace_zone zone = tfe_http_in_request(events) ? kZoneRequestHeaders : kZoneResponseHeader;
		struct tfe_http_half * in_half = tfe_http_in_request(events) ? in_req_half : in_resp_half;

		struct http_field_name in_header_field{};
		const char * in_header_value = NULL;
		void * iterator = NULL;

		while (true)
		{
			if ((in_header_value = tfe_http_field_iterate(in_half, &iterator, &in_header_field)) == NULL)
			{
				break;
			}

			rewrite_buff = NULL;
			rewrite_sz = 0;
			rewrite_sz=execute_replace_rule(in_header_value,
				strlen(in_header_value), zone, rep_ctx->rule, rep_ctx->n_rule, &rewrite_buff, 1);

			if(rewrite_sz>0) rep_ctx->actually_replaced=1;
			tfe_http_field_write(rep_ctx->replacing, &in_header_field, rewrite_sz>0? rewrite_buff : in_header_value);
			if(rewrite_buff != NULL)
			{
				FREE(&rewrite_buff);
			}
		}
	}

	if ((events & EV_HTTP_REQ_BODY_BEGIN) || (events & EV_HTTP_RESP_BODY_BEGIN))
	{
		assert(rep_ctx->http_body == NULL);
		rep_ctx->http_body = evbuffer_new();
	}

	if ((events & EV_HTTP_REQ_BODY_CONT) || (events & EV_HTTP_RESP_BODY_CONT))
	{
		evbuffer_add(rep_ctx->http_body, body_frag, frag_size);
	}

	if ((events & EV_HTTP_REQ_BODY_END) || (events & EV_HTTP_RESP_BODY_END))
	{
		int options = 0;
		char * __http_body = (char *) evbuffer_pullup(rep_ctx->http_body, -1);
		size_t __http_body_len = evbuffer_get_length(rep_ctx->http_body);

		enum replace_zone r_zone = tfe_http_in_request(events) ? kZoneRequestBody : kZoneResponseBody;

		rewrite_buff = NULL;
		rewrite_sz = 0;

		if (tfe_http_in_response(events) && in_resp_spec->content_type != NULL
			&& strcasestr(in_resp_spec->content_type, "utf-8"))
		{
			options = 1;
		}
		rewrite_sz = execute_replace_rule(__http_body, __http_body_len, r_zone,
									rep_ctx->rule, rep_ctx->n_rule, &rewrite_buff, options);

		if (rewrite_sz >0 )
		{
			tfe_http_half_append_body(rep_ctx->replacing, rewrite_buff, rewrite_sz, 0);
			rep_ctx->actually_replaced=1;
		}
		else
		{
			tfe_http_half_append_body(rep_ctx->replacing, __http_body, __http_body_len, 0);
		}

		if (rewrite_buff != NULL)
		{
			FREE(&rewrite_buff);
		}

		if (rep_ctx->http_body != NULL)
		{
			evbuffer_free(rep_ctx->http_body);
			rep_ctx->http_body = NULL;
		}
	}

	if ((events & EV_HTTP_REQ_END) || (events & EV_HTTP_RESP_END))
	{
		tfe_http_half_append_body(rep_ctx->replacing, NULL, 0, 0);
		rep_ctx->replacing = NULL;
	}
}

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

	char cont_len_str[16];

	char *msg = NULL;
	struct tfe_http_session * to_write_sess = NULL;

	struct policy_action_param *param = ctx->param;

	resp_code = param->status_code;
	msg = param->message;
	if (resp_code <= 0){
		TFE_LOG_ERROR(g_pangu_rt->local_logger, "Invalid block rule %d", ctx->enforce_rules[0].config_id);
		ctx->action = PG_ACTION_NONE;
		return;
	}

	if(events & EV_HTTP_RESP_HDR || tfe_http_in_request(events))
	{
		to_write_sess = tfe_http_session_allow_write(session);
		response = tfe_http_session_response_create(to_write_sess, resp_code);

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

		tfe_http_half_append_body(response, page_buff, page_size, 0);
		tfe_http_half_append_body(response, NULL, 0, 0);
		tfe_http_session_response_set(to_write_sess, response);
		tfe_http_session_detach(session);
		html_free(&page_buff);
	}
	else
	{
		to_write_sess = tfe_http_session_allow_write(session);
		tfe_http_session_kill(to_write_sess);
	}
	return;
}

static void http_redirect(const struct tfe_http_session * session, enum tfe_http_event events,
						  struct pangu_http_ctx * ctx)
{
	struct tfe_http_half * response = NULL;
	struct tfe_http_session * to_write = NULL;

	struct policy_action_param *param = ctx->param;

	int resp_code   = param->status_code;
	char *rd_url    = param->message;

	int ratio = http_enforcement_ratio(param->enforcement_ratio);
	if (ratio != 1)
	{
		TFE_LOG_DEBUG(g_pangu_rt->local_logger, "enforcement ratio:%f", param->enforcement_ratio);
		ctx->action = PG_ACTION_NONE;
		return;
	}
	ctx->manipulate_replaced = 1;
	if (resp_code <= 0 || rd_url == NULL){
		TFE_LOG_ERROR(g_pangu_rt->local_logger, "Invalid redirect rule %d paramter",
			ctx->enforce_rules[0].config_id);
		goto error_out;
	}

	if ((events & EV_HTTP_RESP_BODY_BEGIN) || (events & EV_HTTP_RESP_BODY_CONT)
		 || (events & EV_HTTP_RESP_BODY_END) || (events & EV_HTTP_RESP_END))
	{
		ctx->action = PG_ACTION_NONE;
		return;
	}

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

	response = tfe_http_session_response_create(to_write, resp_code);
	tfe_http_std_field_write(response, TFE_HTTP_LOCATION, rd_url);
	tfe_http_std_field_write(response, TFE_HTTP_CONT_LENGTH, "0");
	tfe_http_half_append_body(response, NULL, 0, 0);

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

error_out:
	return;
}

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

	char cont_len_str[16];

	struct policy_action_param *param = ctx->param;

	int resp_code = param->status_code;
	int profile_id    = param->profile_id;
	char *message = param->message;

	if (resp_code <= 0 || profile_id < 0){
		TFE_LOG_ERROR(g_pangu_rt->local_logger, "Invalid block rule %d", ctx->enforce_rules[0].config_id);
		ctx->action = PG_ACTION_NONE;
		return;
	}
	struct tfe_http_session * to_write_sess = NULL;
	if (events & EV_HTTP_RESP_HDR || tfe_http_in_hdr(events))
	{
		to_write_sess = tfe_http_session_allow_write(session);
		response = tfe_http_session_response_create(to_write_sess, resp_code);

		ret = html_generate(profile_id, message, &page_buff, &page_size);
		if (ret != 0)
		{
			/*read local configuration**/
			template_generate(resp_code, ctx->enforce_rules[0].config_id, message, &page_buff, &page_size);
		}
		tfe_http_std_field_write(response, TFE_HTTP_CONT_TYPE, "text/html; charset=utf-8");
		snprintf(cont_len_str, sizeof(cont_len_str), "%lu", page_size);
		tfe_http_std_field_write(response, TFE_HTTP_CONT_LENGTH, cont_len_str);

		tfe_http_half_append_body(response, page_buff, page_size, 0);
		tfe_http_half_append_body(response, NULL, 0, 0);
		tfe_http_session_response_set(to_write_sess, response);
		tfe_http_session_detach(session);
		html_free(&page_buff);
	}
	else
	{
		to_write_sess = tfe_http_session_allow_write(session);
		tfe_http_session_kill(to_write_sess);
	}
	return;
}

static void http_hijack(const struct tfe_http_session * session, enum tfe_http_event events,
					 	struct pangu_http_ctx * ctx)
{
	struct policy_action_param *param = ctx->param;
	struct tfe_http_half * response = NULL;
	struct tfe_http_session * to_write_sess = NULL;

	if (param->profile_id <= 0)
	{
		TFE_LOG_ERROR(g_pangu_rt->local_logger, "Invalid hijack rule %d",
		ctx->enforce_rules[0].config_id);
		ctx->action = PG_ACTION_NONE;
		return;
	}

	if (http_enforcement_ratio(param->enforcement_ratio) != 1)
	{
		TFE_LOG_DEBUG(g_pangu_rt->local_logger, "enforcement ratio:%f", param->enforcement_ratio);
		ctx->action = PG_ACTION_NONE;
		return;
	}
	ctx->manipulate_replaced = 1;
	if (tfe_http_in_request(events))
	{
		return;
	}

	if(events & EV_HTTP_RESP_HDR)
	{
		struct manipulate_profile* hijack_profile=get_profile_by_id(POLICY_PROFILE_TABLE_HIJACK, param->profile_id);
		if (NULL == hijack_profile)
		{
			TFE_LOG_ERROR(g_pangu_rt->local_logger, "get table obj faild, profile_id = %d", param->profile_id);
			ctx->action = PG_ACTION_NONE;
			return;
		}

		char * hijack_buff=NULL; size_t hijack_size=0;

		hijack_buff = tfe_read_file(hijack_profile->profile_msg, &hijack_size);
		if (NULL == hijack_buff){
			TFE_LOG_ERROR(g_pangu_rt->local_logger, "read hijack file faild, path = %s", hijack_profile->profile_msg);
			ctx->action = PG_ACTION_NONE;
			return;
		}
		if (try_send_by_token(hijack_size) <= 0)
		{
			TFE_LOG_ERROR(g_pangu_rt->local_logger, "No token is available to send data, profile_id = %d", param->profile_id);
			ctx->action = PG_ACTION_NONE;
			return;
		}
		ctx->inject_sz = hijack_size;

		char cont_len_str[16];

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

		if (0!=strcasecmp(hijack_profile->profile_name, "null"))
		{
			int hijack_file_len = strlen(hijack_profile->profile_name)+strlen("filename=\"\"")+1;
			char *hijack_file_name = ALLOC(char, hijack_file_len);
			snprintf(hijack_file_name, hijack_file_len, "filename=\"%s\"", hijack_profile->profile_name);
			tfe_http_nonstd_field_write(response, "Content-Disposition", hijack_file_name);
			FREE(&hijack_file_name);
		}
		const char* cont_disposition_val=tfe_http_std_field_read(to_write_sess->resp, TFE_HTTP_CONT_DISPOSITION);
		if (cont_disposition_val != NULL)
		{
			tfe_http_std_field_write(response, TFE_HTTP_CONT_DISPOSITION, cont_disposition_val);
		}
		tfe_http_std_field_write(response, TFE_HTTP_CONT_TYPE, hijack_profile->profile_type);
		snprintf(cont_len_str, sizeof(cont_len_str), "%lu", hijack_size);
		tfe_http_std_field_write(response, TFE_HTTP_CONT_LENGTH, cont_len_str);

		tfe_http_half_append_body(response, hijack_buff, hijack_size, 0);
		tfe_http_half_append_body(response, NULL, 0, 0);
		tfe_http_session_response_set(to_write_sess, response);
		tfe_http_session_detach(session);
		ma_profile_table_free(hijack_profile);
		hijack_profile = NULL;
	}
	else
	{
		to_write_sess = tfe_http_session_allow_write(session);
		tfe_http_session_kill(to_write_sess);
	}

	return;
}

static int format_insert_rule(int profile_id, struct insert_rule *rule)
{
	int ret = 0;

	struct manipulate_profile* insert_profile=get_profile_by_id(POLICY_PROFILE_TABLE_INSERT, profile_id);
	if(insert_profile==NULL)
	{
		ret=-1;
		return ret;
	}
	rule->script = tfe_strdup(insert_profile->profile_msg);
	rule->type = tfe_strdup(insert_profile->profile_type);
	rule->position = tfe_strdup(insert_profile->profile_position);
	rule->inject_sz = insert_profile->msg_len;

	ma_profile_table_free(insert_profile);
	insert_profile = NULL;
	return ret;
}

static void http_insert(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)
{
	struct tfe_http_session * to_write_sess = NULL;
	char * rewrite_buff = NULL;
	size_t rewrite_sz = 0;

	struct policy_action_param *param = ctx->param;
	int ratio = http_enforcement_ratio(param->enforcement_ratio);
	if (ratio != 1)
	{
		TFE_LOG_DEBUG(g_pangu_rt->local_logger, "enforcement ratio:%f", param->enforcement_ratio);
		ctx->action = PG_ACTION_NONE;
		return;
	}
	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);
		ctx->action = PG_ACTION_NONE;
		tfe_http_session_detach(session); return;
	}

	struct insert_ctx *ins_ctx = ctx->ins_ctx;
	if (ctx->ins_ctx == NULL)
	{
		/* we must determinate the replace action on HTTP header, otherwise,
		 * the header has been forwarded, only replace the body but not modify header will raise exception */
		if ((events & EV_HTTP_REQ_HDR) || (events & EV_HTTP_RESP_HDR))
		{
			ctx->ins_ctx = ins_ctx = ALLOC(struct insert_ctx, 1);
			ins_ctx->rule = ALLOC(struct insert_rule, 1);
			int ret=format_insert_rule(param->profile_id, ins_ctx->rule);
			if (ret<0)
			{
				TFE_LOG_ERROR(g_pangu_rt->local_logger, "Failed to get policy table, profile_id = %d", param->profile_id);
				ctx->action = PG_ACTION_NONE;
				return;
			}
			if (try_send_by_token(ins_ctx->rule->inject_sz) <= 0)
			{
				TFE_LOG_ERROR(g_pangu_rt->local_logger, "No token is available to send data, profile_id = %d", param->profile_id);
				ctx->action = PG_ACTION_NONE;
				return;
			}
			ctx->inject_sz = ins_ctx->rule->inject_sz;
		}
		else
		{
			TFE_STREAM_LOG_INFO(stream, "Can only setup replace on REQ/RESP headers, detached.");
			ctx->action = PG_ACTION_NONE;
			tfe_http_session_detach(session); return;
		}
	}

	struct tfe_http_half * in_req_half = session->req;
	struct tfe_http_half * in_resp_half = session->resp;
	struct tfe_http_req_spec * in_req_spec = &in_req_half->req_spec;
	struct tfe_http_resp_spec * in_resp_spec = &in_resp_half->resp_spec;

	if ((events & EV_HTTP_REQ_HDR) || (events & EV_HTTP_RESP_HDR))
	{
		if (tfe_http_in_request(events))
		{
			ins_ctx->replacing = tfe_http_session_request_create(to_write_sess, in_req_spec->method, in_req_spec->uri);
			tfe_http_session_request_set(to_write_sess, ins_ctx->replacing);
		}
		else
		{
			ins_ctx->replacing = tfe_http_session_response_create(to_write_sess, in_resp_spec->resp_code);
			tfe_http_session_response_set(to_write_sess, ins_ctx->replacing);
		}
		struct tfe_http_half * in_half = tfe_http_in_request(events) ? in_req_half : in_resp_half;

		struct http_field_name in_header_field{};
		const char * in_header_value = NULL;
		void * iterator = NULL;

		while (true)
		{
			if ((in_header_value = tfe_http_field_iterate(in_half, &iterator, &in_header_field)) == NULL)
			{
				break;
			}
			tfe_http_field_write(ins_ctx->replacing, &in_header_field, in_header_value);
		}
	}

	if ((events & EV_HTTP_REQ_BODY_BEGIN) || (events & EV_HTTP_RESP_BODY_BEGIN))
	{
		assert(ins_ctx->http_body == NULL);
		ins_ctx->http_body = evbuffer_new();
	}

	if ((events & EV_HTTP_REQ_BODY_CONT) || (events & EV_HTTP_RESP_BODY_CONT))
	{
		evbuffer_add(ins_ctx->http_body, body_frag, frag_size);
	}

	if ((events & EV_HTTP_REQ_BODY_END) || (events & EV_HTTP_RESP_BODY_END))
	{
		char * http_body = (char *) evbuffer_pullup(ins_ctx->http_body, -1);
		size_t http_body_len = evbuffer_get_length(ins_ctx->http_body);
		const char* cont_type_val=tfe_http_std_field_read(ins_ctx->replacing, TFE_HTTP_CONT_TYPE);

		rewrite_buff = NULL;
		rewrite_sz = 0;

		if (cont_type_val != NULL && strstr(cont_type_val, "text/html") != NULL)
		{
			rewrite_sz = execute_insert_rule(http_body, http_body_len, ins_ctx->rule, &rewrite_buff);
		}
		if (rewrite_sz >0)
		{
			tfe_http_half_append_body(ins_ctx->replacing, rewrite_buff, rewrite_sz, 0);
			ins_ctx->actually_replaced=1;
		}
		else
		{
			tfe_http_half_append_body(ins_ctx->replacing, http_body, http_body_len, 0);
		}

		if (rewrite_buff != NULL)
		{
			FREE(&rewrite_buff);
		}

		if (ins_ctx->http_body != NULL)
		{
			evbuffer_free(ins_ctx->http_body);
			ins_ctx->http_body = NULL;
		}
	}

	if ((events & EV_HTTP_REQ_END) || (events & EV_HTTP_RESP_END))
	{
		tfe_http_half_append_body(ins_ctx->replacing, NULL, 0, 0);
		ins_ctx->replacing = NULL;
	}

	return;
}

static void http_manipulate(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)
{
	struct policy_action_param *param = ctx->param;
	if (param == NULL)
	{
		TFE_LOG_ERROR(g_pangu_rt->local_logger, "Failed to get the json format parsed. config_id = %d",
					  ctx->enforce_rules[0].config_id);
		ctx->action = PG_ACTION_NONE;
		return;
	}

	switch(param->action)
	{
		case MA_ACTION_REDIRECT:
			http_redirect(session, events, ctx);
			ATOMIC_INC(&(g_pangu_rt->stat_val[STAT_ACTION_REDIRECT]));
			break;
		case MA_ACTION_BLOCK:
			http_block(session, events, ctx);
			ATOMIC_INC(&(g_pangu_rt->stat_val[STAT_ACTION_REJECT]));
			break;
		case MA_ACTION_REPLACE:
			http_replace(stream, session, events, body_frag, frag_size, ctx);
			ATOMIC_INC(&(g_pangu_rt->stat_val[STAT_ACTION_PRE_REPLACE]));
			break;
		case MA_ACTION_HIJACK:
			http_hijack(session, events, ctx);
			ATOMIC_INC(&(g_pangu_rt->stat_val[STAT_ACTION_HIJACK]));
			break;
		case MA_ACTION_INSERT:
			http_insert(stream, session, events, body_frag, frag_size, ctx);
			ATOMIC_INC(&(g_pangu_rt->stat_val[STAT_ACTION_INSERT]));
			break;
		default: assert(0);
			break;
	}
	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 * iterator = NULL;
	const char * field_val = NULL;
	struct http_field_name field_name;
	struct tfe_http_half * http_half;
	struct Maat_rule_t result[MAX_SCAN_RESULT];
	char buff[TFE_STRING_MAX], * p = NULL;
	int scan_ret = 0, table_id = 0;
	size_t hit_cnt = 0, i = 0;

	if (events & EV_HTTP_REQ_HDR)
	{
		const char *str_host = session->req->req_spec.host;
		if (str_host != NULL)
		{
			int str_host_length = (int) (strlen(session->req->req_spec.host));
			scan_ret = Maat_full_scan_string(g_pangu_rt->maat, g_pangu_rt->scan_table_id[PXY_CTRL_HTTP_FQDN],
								CHARSET_UTF8, str_host, str_host_length, result, NULL, MAX_SCAN_RESULT, &(ctx->scan_mid), ctx->thread_id);
			if (scan_ret > 0)
			{
				hit_cnt += scan_ret;
			}
		}

		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 + hit_cnt, NULL, MAX_SCAN_RESULT - hit_cnt, &(ctx->scan_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];
		http_half = events & EV_HTTP_REQ_HDR ? session->req : session->resp;
		while (hit_cnt < MAX_SCAN_RESULT)
		{
			field_val = tfe_http_field_iterate(http_half, &iterator, &field_name);
			if (field_val == NULL)
			{
				break;
			}

			const char * str_field_name = http_field_name_to_string(&field_name);
			scan_ret = Maat_set_scan_status(g_pangu_rt->maat, &(ctx->scan_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->scan_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->scan_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, &ctx->param);
		if (ctx->action == PG_ACTION_WHITELIST)
		{
			TFE_LOG_INFO(g_pangu_rt->local_logger, "Bypass rules matched: url=%s policy id=%d.",
				session->req->req_spec.url, ctx->enforce_rules[0].config_id);
			goto __out;
		}
		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);
		}
	}


__out:
	return ctx->action;
}

void enforce_control_policy(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,
					struct pangu_http_ctx * ctx)
{
	if(ctx->action==PG_ACTION_NONE||ctx->action==PG_ACTION_MONIT)
	{
		//ctx->action changed in http_scan.
		http_scan(session, events, body_frag, frag_size, ctx);
	}
	switch (ctx->action)
	{
		case PG_ACTION_NONE:
			break;
		case PG_ACTION_MONIT:
			ATOMIC_INC(&(g_pangu_rt->stat_val[STAT_ACTION_MONIT]));
			//send log on close.
			break;
		case PG_ACTION_REJECT:
			http_block(session, events, ctx);
			ATOMIC_INC(&(g_pangu_rt->stat_val[STAT_ACTION_REJECT]));
			break;
		case PG_ACTION_MANIPULATE:
			http_manipulate(stream, session, events, body_frag, frag_size, ctx);
			break;
		case PG_ACTION_WHITELIST:
			tfe_http_session_detach(session);
			ATOMIC_INC(&(g_pangu_rt->stat_val[STAT_ACTION_WHITELSIT]));
			break;
		default: assert(0);
			break;
	}

	/* Don't store reqeust/response body when NOT hit or hit whitelist */
	if(ctx->action == PG_ACTION_NONE || ctx->action == PG_ACTION_WHITELIST)
	{
		return;
	}

	/* Otherwise, store body */
	if(events & EV_HTTP_REQ_BODY_CONT)
	{
		if(ctx->log_req_body == NULL) ctx->log_req_body = evbuffer_new();
		evbuffer_add(ctx->log_req_body, body_frag, frag_size);
	}

	if(events & EV_HTTP_RESP_BODY_CONT)
	{
		if(ctx->log_resp_body == NULL) ctx->log_resp_body = evbuffer_new();
		evbuffer_add(ctx->log_resp_body, body_frag, frag_size);
	}

	return;
}
#define RESUMED_CB_NO_MORE_CALLS	0
#define	RESUMED_CB_MORE_CALLS		1
int make_revalidate_request(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, struct pangu_http_ctx * ctx)
{
	assert(ctx->cache_revalidate_req);
	if(events & EV_HTTP_REQ_BODY_BEGIN)
	{
		tfe_http_half_write_body_begin(ctx->cache_revalidate_req, 1);
	}
	if(events & EV_HTTP_REQ_BODY_CONT)
	{
		tfe_http_half_write_body_data(ctx->cache_revalidate_req, body_frag, frag_size);
	}
	if(events & EV_HTTP_REQ_BODY_END)
	{
		tfe_http_half_write_body_end(ctx->cache_revalidate_req);
		ctx->cache_revalidate_req=NULL;
		return RESUMED_CB_NO_MORE_CALLS;
	}
	if(events & EV_HTTP_REQ_END && ctx->cache_revalidate_req)
	{
		ctx->cache_revalidate_req=NULL;
		return RESUMED_CB_NO_MORE_CALLS;
	}
	return RESUMED_CB_MORE_CALLS;
}
int dummy_resume(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, struct pangu_http_ctx * ctx)
{
	return RESUMED_CB_NO_MORE_CALLS;
}

static void cache_read_on_succ(future_result_t * result, void * user)
{
	struct pangu_http_ctx * ctx = (struct pangu_http_ctx *)user;
	const struct cached_meta* meta=NULL;
	enum cache_query_result_type type=cache_query_result_get_type(result);
	const unsigned char* data=NULL;
	size_t data_sz=0;
	char temp[TFE_STRING_MAX];

	switch(type)
	{
		case CACHE_QUERY_RESULT_META:
			meta=cache_query_result_read_meta(result);
			ctx->cache_result_declared_sz=meta->content_length;
			ctx->resumed_cb=dummy_resume;
			tfe_http_session_resume(ctx->ref_session);
			ATOMIC_DEC(&(g_pangu_rt->stat_val[STAT_SUSPENDING]));

			ctx->cached_response=tfe_http_session_response_create(ctx->ref_session, 200);
			tfe_http_std_field_write(ctx->cached_response, TFE_HTTP_CONT_TYPE, meta->content_type);
			tfe_http_std_field_write(ctx->cached_response, TFE_HTTP_LAST_MODIFIED, meta->last_modified);
			tfe_http_std_field_write(ctx->cached_response, TFE_HTTP_ETAG, meta->etag);

			tfe_http_nonstd_field_write(ctx->cached_response, "X-TG-Cache-Lookup", "HIT");
			snprintf(temp, sizeof(temp), "%lu", meta->content_length);
			tfe_http_std_field_write(ctx->cached_response, TFE_HTTP_CONT_LENGTH, temp);

			//Dirty code here.
			tfe_http_session_response_set(ctx->ref_session, ctx->cached_response);
			//From now, ownership of cached_response has been transfered to http session,
			//bussines plugin only hold this pointer as an reference for writing response body.
			tfe_http_half_write_body_begin(ctx->cached_response, 1);

			meta=NULL;
			break;
		case CACHE_QUERY_RESULT_DATA:
			data_sz=cache_query_result_get_data(result, &data);
			tfe_http_half_write_body_data(ctx->cached_response, data, data_sz);
			ctx->cache_result_actual_sz+=data_sz;
			break;
		case CACHE_QUERY_RESULT_END:
			assert(ctx->cached_response!=NULL);
			tfe_http_half_write_body_end(ctx->cached_response);
			//ownership has been transferred to http session, set to NULL.
			ctx->pending_result=PENDING_RESULT_HIT;
			ctx->cached_response=NULL;
			assert(ctx->cache_result_actual_sz==ctx->cache_result_declared_sz);
			future_destroy(ctx->f_cache_query);
			ctx->f_cache_query=NULL;
			break;
		case CACHE_QUERY_RESULT_MISS:
			ctx->pending_result=PENDING_RESULT_MISS;
			ctx->resumed_cb=dummy_resume;
			tfe_http_session_resume(ctx->ref_session);
			ATOMIC_DEC(&(g_pangu_rt->stat_val[STAT_SUSPENDING]));
			future_destroy(ctx->f_cache_query);
			ctx->f_cache_query=NULL;
			break;
		default:
			break;
	}
	return;
}
static void cache_read_on_fail(enum e_future_error err, const char * what, void * user)
{
	struct pangu_http_ctx * ctx = (struct pangu_http_ctx *)user;
	future_destroy(ctx->f_cache_query);
	ctx->f_cache_query=NULL;
	if(!ctx->cached_response)
	{
		tfe_http_session_resume(ctx->ref_session);
		ctx->resumed_cb=dummy_resume;
		ATOMIC_DEC(&(g_pangu_rt->stat_val[STAT_SUSPENDING]));
	}
	else
	{
		tfe_http_half_write_body_end(ctx->cached_response);
		ctx->cached_response=NULL;
	}
	ctx->pending_result=PENDING_RESULT_MISS;
	printf("cache query failed: %s %s\n", ctx->ref_session->req->req_spec.url, what);
}
static void cache_pend_on_succ(future_result_t * result, void * user)
{
	struct pangu_http_ctx * ctx = (struct pangu_http_ctx *)user;
	const struct cached_meta* meta=NULL;
	meta=cache_pending_result_read_meta(result, ctx->cmid);
	ctx->resumed_cb=dummy_resume;
	tfe_http_session_resume(ctx->ref_session);
	ATOMIC_DEC(&(g_pangu_rt->stat_val[STAT_SUSPENDING]));
	future_destroy(ctx->f_cache_pending);
	ctx->f_cache_pending=NULL;
	if(meta==NULL)
	{
		ctx->pending_result = PENDING_RESULT_MISS;
		return;
	}
	if( meta->etag==NULL && meta->last_modified==NULL)
	{
		ctx->pending_result = PENDING_RESULT_MISS;
		return;
	}
	ctx->pending_result=PENDING_RESULT_REVALIDATE;
	struct http_field_name in_field_name;
	const char * in_header_value = NULL;
	void * iterator = NULL;
	ctx->cache_revalidate_req=tfe_http_session_request_create(ctx->ref_session,
							ctx->ref_session->req->req_spec.method, ctx->ref_session->req->req_spec.uri);
	while (true)
	{
		if ((in_header_value = tfe_http_field_iterate(ctx->ref_session->req, &iterator, &in_field_name)) == NULL)
		{
			break;
		}
		if(in_field_name.field_id==TFE_HTTP_IF_MATCH || in_field_name.field_id==TFE_HTTP_IF_NONE_MATCH
				|| in_field_name.field_id==TFE_HTTP_IF_MODIFIED_SINCE
				|| in_field_name.field_id==TFE_HTTP_IF_UNMODIFIED_SINCE)
		{
			continue;
		}
		tfe_http_field_write(ctx->cache_revalidate_req, &in_field_name, in_header_value);
	}
	if(meta->etag) tfe_http_std_field_write(ctx->cache_revalidate_req, TFE_HTTP_IF_NONE_MATCH, meta->etag);
	if(meta->last_modified)  tfe_http_std_field_write(ctx->cache_revalidate_req, TFE_HTTP_IF_MODIFIED_SINCE, meta->last_modified);

	tfe_http_session_request_set(ctx->ref_session, ctx->cache_revalidate_req);
	ctx->resumed_cb=make_revalidate_request;
	return;
}
static void cache_pend_on_fail(enum e_future_error err, const char * what, void * user)
{
	struct pangu_http_ctx * ctx = (struct pangu_http_ctx *)user;

	ctx->pending_result=PENDING_RESULT_FOBIDDEN;
	tfe_http_session_resume(ctx->ref_session);
	ATOMIC_DEC(&(g_pangu_rt->stat_val[STAT_SUSPENDING]));
	ctx->resumed_cb=dummy_resume;
	future_destroy(ctx->f_cache_pending);
	ctx->f_cache_pending=NULL;

	return;
}

void cache_pend(const struct tfe_http_session * session,	unsigned int thread_id, struct pangu_http_ctx * ctx)
{
	if(g_pangu_rt->stat_val[STAT_SUSPENDING]>g_pangu_rt->suspend_max)
	{
		ctx->pending_result=PENDING_RESULT_FOBIDDEN;
		return;
	}
	ctx->f_cache_pending=future_create("cache_pend", cache_pend_on_succ, cache_pend_on_fail, ctx);
	ctx->ref_session=tfe_http_session_allow_write(session);
	ctx->pending_result=web_cache_async_pending(g_pangu_rt->cache, thread_id, session->req, &(ctx->cmid), ctx->f_cache_pending);
	switch(ctx->pending_result)
	{
		case PENDING_RESULT_REVALIDATE:
			tfe_http_session_suspend(ctx->ref_session);
			ATOMIC_INC(&(g_pangu_rt->stat_val[STAT_SUSPENDING]));
			break;
		case PENDING_RESULT_ALLOWED:
		case PENDING_RESULT_FOBIDDEN:
		case PENDING_RESULT_MISS:
			future_destroy(ctx->f_cache_pending);
			ctx->f_cache_pending=NULL;
			break;
		default:
			break;
	}
	return;
}
void cache_read(const struct tfe_http_session * session,	unsigned int thread_id, struct pangu_http_ctx * ctx)
{
	if(g_pangu_rt->stat_val[STAT_SUSPENDING]>g_pangu_rt->suspend_max)
	{
		return;
	}
	ctx->f_cache_query=future_create("cache_read", cache_read_on_succ, cache_read_on_fail, ctx);
	int ret=web_cache_async_read(g_pangu_rt->cache, thread_id, session->req, &(ctx->cmid), ctx->f_cache_query);
	if(ret==0)
	{
		ctx->ref_session=tfe_http_session_allow_write(session);
		tfe_http_session_suspend(ctx->ref_session);
		ATOMIC_INC(&(g_pangu_rt->stat_val[STAT_SUSPENDING]));
	}
	else
	{
		future_destroy(ctx->f_cache_query);
		ctx->f_cache_query=NULL;
	}
}

void cache_write(const struct tfe_http_session * session, enum tfe_http_event events,
								const unsigned char * body_frag, size_t frag_size,
								unsigned int thread_id, struct pangu_http_ctx * ctx)
{

	if(events & EV_HTTP_RESP_BODY_BEGIN)
	{
		ctx->cache_write_ctx=web_cache_write_start(g_pangu_rt->cache, thread_id, session, &(ctx->cmid));
	}
	if(events & EV_HTTP_RESP_BODY_CONT && ctx->cache_write_ctx!=NULL)
	{
		web_cache_write(ctx->cache_write_ctx, body_frag, frag_size);
	}
	if(events & EV_HTTP_RESP_BODY_END && ctx->cache_write_ctx!=NULL)
	{
		ctx->cache_wirte_result=web_cache_write_end(ctx->cache_write_ctx);
		ctx->cache_write_ctx=NULL;
		//printf("cache update success: %s\n", ctx->ref_session->req->req_spec.url);
	}

}

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, scan_ret=0;
	UNUSED int tmp=0;
	assert(ctx == NULL);
	ATOMIC_INC(&(g_pangu_rt->stat_val[STAT_SESSION]));
	ctx = pangu_http_ctx_new(thread_id);
	char* addr_string=tfe_stream_addr_to_str(stream->addr);
	const char* sip=NULL, *dip=NULL;
	char* source_subscribe_id=NULL, *dest_subscribe_id=NULL;

	tmp=tfe_stream_addr_str_split(addr_string, &sip, NULL, &dip, NULL);
	assert(tmp==0);
	source_subscribe_id=(char*)Maat_plugin_get_EX_data(g_pangu_rt->dyn_maat, g_pangu_rt->subscriber_id_table_id, sip);
	dest_subscribe_id=(char*)Maat_plugin_get_EX_data(g_pangu_rt->dyn_maat, g_pangu_rt->subscriber_id_table_id, dip);

	if(source_subscribe_id!=NULL)
	{
		scan_ret = Maat_full_scan_string(g_pangu_rt->maat, g_pangu_rt->scan_table_id[PXY_CTRL_SUBSCRIBE_ID],
											CHARSET_UTF8, source_subscribe_id, strlen(source_subscribe_id),
											result+hit_cnt, NULL, MAX_SCAN_RESULT-hit_cnt,
											&(ctx->scan_mid), (int) thread_id);
		if(scan_ret>0)
		{
			hit_cnt+=scan_ret;
		}
	}
	if(dest_subscribe_id!=NULL)
	{
		scan_ret = Maat_full_scan_string(g_pangu_rt->maat, g_pangu_rt->scan_table_id[PXY_CTRL_SUBSCRIBE_ID],
											CHARSET_UTF8, dest_subscribe_id, strlen(dest_subscribe_id),
											result+hit_cnt, NULL, MAX_SCAN_RESULT-hit_cnt,
											&(ctx->scan_mid), (int) thread_id);
		if(scan_ret>0)
		{
			hit_cnt+=scan_ret;
		}
	}
	const char *app_id = "http.";
	scan_ret = Maat_full_scan_string(g_pangu_rt->maat, g_pangu_rt->scan_table_id[PXY_CTRL_APP_ID],
										CHARSET_UTF8, app_id, strlen(app_id),
										result+hit_cnt, NULL, MAX_SCAN_RESULT-hit_cnt,
										&(ctx->scan_mid), (int) thread_id);
	if(scan_ret>0)
	{
		hit_cnt+=scan_ret;
	}

	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+hit_cnt, MAX_SCAN_RESULT-hit_cnt, &(ctx->scan_mid), (int) thread_id);

	if (hit_cnt > 0)
	{
		ctx->action = decide_ctrl_action(result, hit_cnt, &ctx->enforce_rules, &ctx->n_enforce, &ctx->param);
	}
	if (ctx->action == PG_ACTION_WHITELIST)
	{
		TFE_LOG_INFO(g_pangu_rt->local_logger, "Bypass rules matched on http begin: url=%s policy id=%d.",
			session->req->req_spec.url, ctx->enforce_rules[0].config_id);
		tfe_http_session_detach(session);
	}

	*pme = ctx;
	free(addr_string);
	free(source_subscribe_id);
	free(dest_subscribe_id);
	return;
}

static inline int ctx_actually_replaced(struct pangu_http_ctx * ctx)
{

	if(ctx->action == PG_ACTION_MANIPULATE &&
	   ctx->param->action == MA_ACTION_REPLACE &&
	   ctx->n_enforce==1 && ctx->rep_ctx->actually_replaced==0)
	{
		return 1;
	}
	else
	{
		return 0;
	}
}

static inline int ctx_actually_inserted(struct pangu_http_ctx * ctx)
{

	if(ctx->action == PG_ACTION_MANIPULATE &&
	   ctx->param->action == MA_ACTION_INSERT &&
	   ctx->n_enforce==1 && ctx->ins_ctx->actually_replaced==0)
	{
		return 1;
	}
	else
	{
		return 0;
	}
}

static inline int ctx_actually_manipulate(struct pangu_http_ctx * ctx)
{
	if(ctx->action == PG_ACTION_MANIPULATE &&
	   (ctx->param->action == MA_ACTION_REDIRECT ||
	    ctx->param->action == MA_ACTION_HIJACK)&&
	    ctx->n_enforce==1 && ctx->manipulate_replaced==0)
	{
		return 1;
	}
	else
	{
		return 0;
	}
}

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;
	size_t i=0, j=0;
	int ret=0;

	if(ctx->action == PG_ACTION_MANIPULATE && ctx->param->action == MA_ACTION_REPLACE && ctx->rep_ctx->actually_replaced==0)
	{
		for(i=0; i< ctx->n_enforce; i++)
		{
			if((unsigned char)ctx->enforce_rules[i].action == PG_ACTION_MANIPULATE)
			{
				if(i+1 > ctx->n_enforce)
				{
					memmove(ctx->enforce_rules+i, ctx->enforce_rules+i+1, sizeof(struct Maat_rule_t));
				}
				j++;
			}
		}
		ctx->n_enforce-=j;
		if(ctx->n_enforce==0)
		{
			ctx->action = PG_ACTION_NONE;
			FREE(&(ctx->enforce_rules));
		}
	}
	struct pangu_log log_msg =  {.stream=stream, .http=session, .result=ctx->enforce_rules, .result_num=ctx->n_enforce,
								.req_body=ctx->log_req_body, .resp_body=ctx->log_resp_body, .action=0, .inject_sz=ctx->inject_sz};

	if(ctx->action == PG_ACTION_MANIPULATE)
	{
		log_msg.action = ctx->param->action;
	}

	if ((ctx->action != PG_ACTION_NONE&&
		!(ctx_actually_replaced(ctx))) ||
		(ctx->action != PG_ACTION_NONE&&
		!(ctx_actually_inserted(ctx))) ||
		(ctx->action != PG_ACTION_NONE&&
		!(ctx_actually_manipulate(ctx))))
	{
		ret=pangu_send_log(g_pangu_rt->send_logger, &log_msg);
		ATOMIC_ADD(&(g_pangu_rt->stat_val[STAT_LOG_NUM]), ret);
	}
	if(ctx->rep_ctx && ctx->rep_ctx->actually_replaced==0)
	{
		ATOMIC_INC(&(g_pangu_rt->stat_val[STAT_ACTION_REPLACE]));
	}
	TFE_LOG_DEBUG(g_pangu_rt->local_logger, "cache %s %s upload=%d",
											session->req->req_spec.url,
											cache_pending_result_string(ctx->pending_result),
											ctx->cache_wirte_result);
	cache_mid_clear(&(ctx->cmid));
	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;
	int ret=0;
	if(ctx->resumed_cb)
	{
		ret=ctx->resumed_cb(stream, session, events, body_frag, frag_size,thread_id, ctx);
		if(ret==RESUMED_CB_NO_MORE_CALLS)
		{
			ctx->resumed_cb=NULL;
		}
		return;
	}

	enforce_control_policy(stream, session, events, body_frag, frag_size,thread_id, ctx);

	if(g_pangu_rt->cache_enabled && ctx->action == PG_ACTION_NONE)
	{
		if(events & EV_HTTP_REQ_HDR)
		{
			cache_pend(session, thread_id, ctx);
			if(ctx->pending_result==PENDING_RESULT_ALLOWED)
			{
				cache_read(session, thread_id, ctx);
			}
		}
		if(events & EV_HTTP_RESP_HDR && ctx->pending_result==PENDING_RESULT_REVALIDATE)
		{
			if(session->resp->resp_spec.resp_code==TFE_HTTP_STATUS_NOT_MODIFIED)
			{
				cache_read(session, thread_id, ctx);
			}
		}

		if(tfe_http_in_response(events) && ctx->pending_result==PENDING_RESULT_MISS)
		{
			cache_write(session, events, body_frag, frag_size, thread_id, ctx);
		}
	}
	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)