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 <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 <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];
	char* reject_page;
	int page_size;
};
struct pangu_rt *g_pangu_rt;
static Maat_feather_t create_maat_feather(const char* profile, const char* section,int max_thread, void* logger)
{
	Maat_feather_t target;
	int maat_json_switch=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 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_JSON_SWITCH", &(maat_json_switch),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,"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,"JSON_CFG_FILE",json_cfg_file, sizeof(json_cfg_file),"");
	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);
	if(maat_json_switch==1)
	{
		Maat_set_feather_opt(target, MAAT_OPT_JSON_FILE_PATH, json_cfg_file, strlen(json_cfg_file)+1);
	}
	else
	{
		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);
	}
	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__);
		return NULL;
	}
	return target;
}
void pangu_http_init(struct tfe_proxy * proxy)
{
	const char* profile="./pangu/pangu_pxy.conf";
	const char* logfile="./log/pangu_pxy.log";
	g_pangu_rt=ALLOC(struct pangu_rt,1);
	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_send_log_init(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(proxy->logger, "Pangu HTTP Maat table %s register failed.", table_name[i]);
			goto error_out;
		}
	}
	TFE_LOG_INFO(proxy->logger, "Pangu HTTP init success.");
	return;
	
error_out:
	TFE_LOG_ERROR(proxy->logger, "Pangu HTTP init failed.");
	return;
}
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;
}
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;
	tmp_name.field_name=field_name;
	tfe_http_field_write(half, tmp_name, value);
	return;
}
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* exec_rule;
	struct Maat_rule_t* monit_rule;
	int monit_num;

	char* exec_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));
	}
	Maat_clean_status(&(ctx->mid));
	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=tfe_addr->paddr;
	return;
}

/*
20180909记录
1）const char * tfe_http_field_iterate(const struct tfe_http_half * half, void * interator, struct http_field_name * name); void**?
2)http入口函数返回值增加 enum tfe_bussiness_action
3)struct http_field_name中的field_name内存是如何分配到？
4）http对上层要暴露evbuffer吗？
5）	uint64_t cont_len;
	uint64_t cont_range_from;
	uint64_t cont_range_to;
需要保留吗？accept_encoding?
6）entry函数中缺少thread id;
7) plugin init and deinit cb;
8) add http session open and close cb;
9)enum tfe_bussiness_action pangu_http_entry(const struct tfe_stream * stream, const struct tfe_http_session * session,
   						 uint64_t event, const char* body_frag, size_t frag_size, unsigned int thread_id void ** pme)
*/

static enum pangu_action decide_ctrl_action(const Maat_rule_t* hit_result,int cnt,const Maat_rule_t**enforce_rule)
{
	int i=0;
	int idx=0;
	const Maat_rule_t* tmp_rule=hit_result;
	enum pangu_action tmp_action=PG_ACTION_NONE;

	for(i=0;i<cnt;i++)
	{
		if(hit_result[i].action>tmp_action)
		{
			tmp_rule=hit_result+i;
			tmp_action=hit_result[i].action;
		}
		else if(hit_result[i].action==tmp_action)
		{
			if(hit_result[i].config_id<tmp_rule->config_id)
			{
				tmp_rule=hit_result+i;
			}

		}
		else
		{
			continue;
		}
	}
	*enforce_rule=tmp_rule;
	return tmp_action;
}
static void html_generate(const char* enforce_para, char** page_buff,size_t *page_size)
{
	*page_buff=g_pangu_rt->reject_page;
	*page_size=g_pangu_rt->page_size;
}
static void html_free(char** page_buff)
{
	return;
}
static int is_http_request(uint64 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"};
	int 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, 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);
		goto error_out;
	}
	
	/* "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));
	}

error_out:
	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];
	enum replace_zone zone;
	int ret=0, is_replaced=0;
	size_t n_todo=0, i=0, tmp_size=0;
	struct evbuffer* out=NULL;
	const char* interator=NULL;
	size_t interator_sz=0;
	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=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,
   						 uint64_t events, const char* body_frag, size_t frag_size,struct pangu_http_ctx* ctx)
{
	struct replace_rule* ctx->rep_ctx=ALLOC(struct replace_rule, MAX_EDIT_ZONE_NUM);
	struct tfe_http_half * req=NULL;
	void* interator=NULL;
	struct http_field_name tmp_name;
	const struct replace_rule** todo[MAX_EDIT_ZONE_NUM];
	size_t n_todo=0,i=0, out_size=0;
	char* buff_in=NULL, buff_out=NULL;
	struct evbuffer* rewrite_url=NULL, *rewrite_buff=NULL;
	struct replace_ctx* rep_ctx=NULL;
	int is_replaced=0;
	if(ctx->rep_ctx==NULL)
	{
		ctx->rep_ctx=rep_ctx=ALLOC(struct replace_ctx, 1);
		
		rep_ctx->n_rule=format_replace_rule(ctx->exec_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(session, session->req->req_spec.method, 
					rewrite_url!=NULL ? evbuffer_pullup(rewrite_url,-1) : session->req->req_spec.uri);
			evbuffer_free(rewrite_url);
			rewrite_url=NULL;
			tfe_http_allow_write(session->req);
		}
		else
		{
			rep_ctx->replacing=tfe_http_session_response_create(session, session->resp->resp_spec.resp_code);
			tfe_http_allow_write(session->resp);
		}
		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 ? evbuffer_pullup(rewrite_buff) : 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=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, "%llu", 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,
					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(session, rep_ctx->replacing);
		}
		else
		{
			tfe_http_session_response_set(session, 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;
	}
	
	
	
}
static void http_reject(const struct tfe_http_session * session, uint64_t events, struct pangu_http_ctx* ctx)
{
	
	enum tfe_bussiness_action plugin_ret=BIZ_ACTION_FORWARD;
	int resp_code=0,ret=0;
	struct tfe_http_half* response=NULL;
	struct http_field_name tmp_name;
	char* page_buff=NULL;
	size_t page_size=0;
	char cont_len_str[TFE_STRING_MAX];
	struct tfe_http_session* to_write=NULL;

	ret=sscanf(ctx->exec_para,"code=%d%[^;];",&resp_code);
	if(ret!=1)
	{
		TFE_LOG_ERROR(g_pangu_rt->local_logger, "Invalid reject rule %d paramter %s", 
								ctx->exec_rule->config_id, ctx->exec_para);
		goto error_out;
	}
	to_write=tfe_http_session_allow_write(session);
	response=tfe_http_session_response_create(to_write, resp_code);
	
	html_generate(ctx->exec_para, &page_buff, &page_size);
	_wrap_std_field_write(response, TFE_HTTP_CONT_TYPE, "text/html; charset=utf-8");
	snprinf(cont_len_str,sizeof(cont_len_str), "%llu", 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(session, response);
	response=NULL;
error_out:
	html_free(page_buff);
	return plugin_ret;
}
static void http_redirect(const struct tfe_http_session * session, uint64_t 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->exec_rule->serv_def_len);
	ret=sscanf(ctx->exec_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->exec_rule->config_id, ctx->exec_para);
		goto error_out;
	}
	
	to_write=tfe_http_session_allow_write(session);
	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);
	response=NULL;
	
error_out:
	free(url);
	return;
}

enum pangu_action http_scan(const struct tfe_http_session * session, uint64_t events, 
						const 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], *choosen=NULL;
	char buff[TFE_STRING_MAX], *p=NULL;
	int scan_ret=0, hit_cnt=0, table_id=0, read_rule_ret=0;
	unsigned int i=0;
	if(events&EV_HTTP_REQ_HDR)
	{
		scan_ret=Maat_full_scan_string(g_pangu_rt->maat, g_pangu_rt->scan_table_id[PXY_CTRL_HTTP_URL],
			CHARSET_UTF8, session->req->req_spec.url, strlen(session->req->req_spec.url),
			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];
		for(i=0; i<session->req->field_cnt && hit_cnt<MAX_SCAN_RESULT; i++)
		{
			field_val=tfe_http_field_iterate(session->req, &interator, &field_name);
			scan_ret=Maat_set_scan_status(g_pangu_rt->maat, ctx->mid, MAAT_SET_SCAN_DISTRICT, 
						field_name.field_name,strlen(field_name.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, 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, &choosen);
		ctx->exec_rule=ALLOC(struct Maat_rule_t, 1);
		memcpy(ctx->exec_rule, choosen, sizeof(struct Maat_rule_t));
		if(ctx->exec_rule.serv_def_len>MAX_SERVICE_DEFINE_LEN)
		{
			ctx->exec_para=ALLOC(char, ctx->exec_rule.serv_def_len);
			read_rule_ret=Maat_read_rule(g_pangu_rt->maat, ctx->exec_rule,
				MAAT_RULE_SERV_DEFINE, ctx->exec_para, ctx->exec_rule.serv_def_len);
			assert(read_rule_ret== ctx->exec_rule.serv_def_len);
		}
		if(hit_cnt>1)
		{
			p=buff;
			for(i=0;i<(unsigned int)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=%d ids=%s enforce=%d .", 
						session->req->req_spec.url, buff, hit_cnt, ctx->exec_rule->config_id);
		}
	}
	return ctx->action;
}

 enum tfe_bussiness_action 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 Maat_rule_t* choosen=NULL;
	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, &choosen);
	}
	if(ctx->action==PG_ACTION_WHITELIST)
	{
		
	}
}


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->exec_rule, .result_num=1};
	if(ctx->action!=PG_ACTION_NONE)
	{
		pangu_send_log(g_pangu_rt->send_logger, &log_msg, NULL, 0);
	}
	pangu_http_ctx_free(ctx);
	*pme=NULL;
	return;
}

void pangu_on_http_data(const struct tfe_stream * stream, const struct tfe_http_session * session,
   						 uint64_t events, const char* body_frag, size_t frag_size, unsigned int thread_id, void ** pme)
{
	struct pangu_http_ctx* ctx=*(struct pangu_http_ctx**)pme;
	int hit_cnt=0;
	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);
		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;
}