stellar-stellar/decoders/session_flags/session_flags.cpp

#include <stdio.h>
#include <unistd.h>
#include <assert.h>
#include <stdlib.h>
#include <string.h>
#include <time.h>
#include <math.h>
#include <cJSON.h>
#include <pthread.h>

#include "onlinemean.h"
#include "session_flags_internal.h"
#include "fet.h"
#include "stellar/stellar.h"
#include "mesa_sts.h"
#include "stellar/session.h"
#include "stellar/log.h"

#define SESSION_BUKLY_THRESHORD 0.95
#define SESSION_DOWNLOAD_THRESHORD 0.95
#define SESSION_CBR_STREAMING_THRESHORD 0.15
#define SESSION_STREAMING_THRESHORD 0.86
#define SESSION_PSEUDO_UNIDIRECTIONA_THRESHORD 0.95
#define SESSION_INTERACTIVE_THRESHORD 0.7

#define SESSION_BIDIRECTIONAL_THRESHORD_MIN 0.7
#define SESSION_BIDIRECTIONAL_THRESHORD_MAX 1.43
#define SESSION_EWMA_FACTOR 0.9


#define SESSION_RANDOM_LOOKING_PAYLOAD_LEN_MIN 100

#define SESSION_ITER_INTERVAL_MS 1000
#define SESSION_EWMA_ITER_CNT_MIN 5

#define LARGE_PKT_SIZE_MIN 600
#define LARGE_PKT_SIZE_MAX 1400

#define SESSION_FLAGS_IDENTIFY_STR_LEN 256

extern struct session_flags_init_conf g_sf_conf;

extern uint32_t sts_random_switch;

thread_local OnlineMean_t g_large_pkt_omean = {
	.mean = 0,
	.varsum = 0,
	.count = 0
};//record the mean and std of packet size, to calculate large packet size
thread_local uint64_t g_large_pkt_size_update_ms = 0;
thread_local uint32_t g_large_pkt_size = 0;
#define LARGE_PKT_SIZE_UPDATE_COUNT_MIN 10000
#define LARGE_PKT_SIZE_UPDATE_INTERVAL_MS 10000

static void session_flags_calculate_local(struct session_flags_stat *stat, enum session_direction session_dir)
{
	if (session_dir == SESSION_DIRECTION_INBOUND)
	{
		stat->result.flags |= SESSION_FLAGS_LOCAL_SERVER;
		stat->result.identify[session_flags_local_server_mask] = 1;
	}
	else 
	{
		stat->result.flags |= SESSION_FLAGS_LOCAL_CLIENT;
		stat->result.identify[session_flags_local_client_mask] = 1;
	}

	return;
}

void session_flags_stat_init(struct session_flags_stat *stat, enum session_direction session_dir)
{
	session_flags_calculate_local(stat, session_dir);

	OnlineMean_Init(&stat->iter.c2s.omean);
	OnlineMean_Init(&stat->iter.s2c.omean);

	stat->main_dir = MAIN_DIR_UNKONWN;

	return;
};

static void session_flags_EWMA_iter(float *ori_value, float iter_value)
{
	float temp = *ori_value;

	if (temp == 0)
	{
		temp = iter_value;
	}
	else
	{
		temp = SESSION_EWMA_FACTOR * temp + (1 - SESSION_EWMA_FACTOR) * iter_value;
	}

	*ori_value = temp;
}

static void session_flags_calculate_pseudo_unidirection(struct session_flags_stat *stat, uint64_t all_pkts)
{
	if (stat->result.flags & SESSION_FLAGS_PSEUDO_UNIDIRECTIONAL)
		return ;

	float c2s_ratio, s2c_ratio;
	float pseudo_unidirectional;
	uint64_t delta_payload_pkts = stat->c2s.delta_payload_pkts + stat->s2c.delta_payload_pkts;

	if (delta_payload_pkts == 0)
	{
		return;
	}

	c2s_ratio = stat->c2s.delta_payload_pkts * 1.f / delta_payload_pkts;
	s2c_ratio = stat->s2c.delta_payload_pkts * 1.f / delta_payload_pkts;

	session_flags_EWMA_iter(&stat->iter.c2s.pseudo_unidirectional, c2s_ratio);
	session_flags_EWMA_iter(&stat->iter.s2c.pseudo_unidirectional, s2c_ratio);

	if (stat->iter.iter_cnt < SESSION_EWMA_ITER_CNT_MIN || stat->main_dir == MAIN_DIR_UNKONWN)
	{
		return;
	}

	if (stat->main_dir == FLOW_TYPE_C2S)
	{
		pseudo_unidirectional = stat->iter.c2s.pseudo_unidirectional;
	}
	else
	{
		pseudo_unidirectional = stat->iter.s2c.pseudo_unidirectional;
	}

	if (pseudo_unidirectional > SESSION_PSEUDO_UNIDIRECTIONA_THRESHORD)
	{
		stat->result.flags |= SESSION_FLAGS_PSEUDO_UNIDIRECTIONAL;
		stat->result.identify[session_flags_pseudo_unidirectional_mask] = all_pkts;
	}
}

static void session_flags_calculate_unidirection(struct session_flags_stat *stat, uint64_t all_pkts)
{
	if (stat->result.flags & SESSION_FLAGS_UNIDIRECTIONAL || stat->s2c.pkts == 0 || stat->c2s.pkts == 0)
		return ;

	if (stat->c2s.payload_pkts == 0 || stat->s2c.payload_pkts == 0)
	{
		stat->result.flags |= SESSION_FLAGS_UNIDIRECTIONAL;
		stat->result.identify[session_flags_unidirectional_mask] = all_pkts;
	}
}

static void session_flags_calculate_streaming(struct session_flags_stat *stat, uint64_t all_pkts)
{
	if (stat->result.flags & SESSION_FLAGS_STREAMING)
		return;
	
	float c2s_ratio, s2c_ratio;
	float streaming;
	uint64_t delta_bytes = stat->c2s.delta_bytes + stat->s2c.delta_bytes;

	if (delta_bytes == 0)
	{
		return;
	}

	c2s_ratio = stat->c2s.delta_bytes * 1.f / delta_bytes;
	s2c_ratio = stat->s2c.delta_bytes * 1.f / delta_bytes;

	session_flags_EWMA_iter(&stat->iter.c2s.streaming, c2s_ratio);
	session_flags_EWMA_iter(&stat->iter.s2c.streaming, s2c_ratio);

	if (stat->iter.iter_cnt < SESSION_EWMA_ITER_CNT_MIN || stat->main_dir == MAIN_DIR_UNKONWN)
	{
		return;
	}

	if (stat->main_dir == FLOW_TYPE_C2S)
	{
		streaming = stat->iter.c2s.streaming;
	}
	else
	{
		streaming = stat->iter.s2c.streaming;
	}

	if (streaming > SESSION_STREAMING_THRESHORD)
	{
		stat->result.flags |= SESSION_FLAGS_STREAMING;
		stat->result.identify[session_flags_streaming_mask] = all_pkts;
	}
}

static void session_flags_calculate_download(struct session_flags_stat *stat, uint64_t all_pkts)
{
	if (stat->result.flags & SESSION_FLAGS_DOWNLOAD)
		return ;
	
	float c2s_ratio, s2c_ratio;
	float download;
	uint64_t delta_payload_pkts = stat->c2s.delta_payload_pkts + stat->s2c.delta_payload_pkts;

	if (delta_payload_pkts == 0)
	{
		return;
	}

	c2s_ratio = stat->c2s.delta_large_pkts * 1.f / delta_payload_pkts;
	s2c_ratio = stat->s2c.delta_large_pkts * 1.f / delta_payload_pkts;
	
	session_flags_EWMA_iter(&stat->iter.c2s.download, c2s_ratio);
	session_flags_EWMA_iter(&stat->iter.s2c.download, s2c_ratio);

	if (stat->iter.iter_cnt < SESSION_EWMA_ITER_CNT_MIN || stat->main_dir == MAIN_DIR_UNKONWN)
	{
		return;
	}

	if (stat->main_dir == FLOW_TYPE_C2S)
	{
		download = stat->iter.c2s.download;
	}
	else
	{
		download = stat->iter.s2c.download;
	}

	if (download > SESSION_DOWNLOAD_THRESHORD)
	{
		stat->result.flags |= SESSION_FLAGS_DOWNLOAD;
		stat->result.identify[session_flags_download_mask] = all_pkts;
	}
}

static void session_flags_calculate_bulky(struct session_flags_stat *stat, uint64_t all_pkts)
{
	if (stat->result.flags & SESSION_FLAGS_BULKY)
		return ;
	
	float bulky;

	if (stat->c2s.delta_pkts != 0)
	{
		float c2s_ratio;
		c2s_ratio = stat->c2s.delta_large_pkts * 1.f / stat->c2s.delta_pkts;
		session_flags_EWMA_iter(&stat->iter.c2s.bulky, c2s_ratio);
	}

	if (stat->s2c.delta_pkts != 0)
	{
		float s2c_ratio;
		s2c_ratio = stat->s2c.delta_large_pkts * 1.f / stat->s2c.delta_pkts;
		session_flags_EWMA_iter(&stat->iter.s2c.bulky, s2c_ratio);
	}
	
	if (stat->iter.iter_cnt < SESSION_EWMA_ITER_CNT_MIN || stat->main_dir == MAIN_DIR_UNKONWN)
	{
		return;
	}

	if (stat->main_dir == FLOW_TYPE_C2S)
	{
		bulky = stat->iter.c2s.bulky;
	}
	else
	{
		bulky = stat->iter.s2c.bulky;
	}

	if (bulky > SESSION_BUKLY_THRESHORD)
	{
		stat->result.flags |= SESSION_FLAGS_BULKY;
		stat->result.identify[session_flags_bulky_mask] = all_pkts;
	}
}

static void flow_stat_update(struct session *session, struct flow_stat *flow, uint64_t bytes)
{
	flow->bytes += bytes;
	flow->pkts++;

	flow->delta_pkts++;
	flow->delta_bytes += bytes;
	
	const struct packet *pkt = session_get0_current_packet(session);
	size_t payload_len = packet_get_payload_len(pkt);

	if (payload_len > 0)
	{
		flow->payload_pkts++;
		flow->delta_payload_pkts++;
	}

	if (bytes > g_large_pkt_size)
	{
		flow->large_pkts++;
		flow->delta_large_pkts++;
	}

	return;
}

static void session_flags_calculate_main_dir(struct session_flags_stat *stat)
{
	if (stat->main_dir != MAIN_DIR_UNKONWN)
	{
		return;
	}

	if (stat->c2s.bytes > stat->s2c.bytes)
	{
		stat->main_dir = FLOW_TYPE_C2S;
	}
	else
	{
		stat->main_dir = FLOW_TYPE_S2C;
	}
}

float session_flags_calculate_CV(OnlineMean_t * omean)
{
	float CV = -1.f;
	float mean = OnlineMean_GetMean(omean);
	if (mean!= 0)
	{
		CV = OnlineMean_GetStd(omean) / mean;
	}
	// printf("CV:%lf\n", CV);
	return CV;
}

static void session_flags_calculate_CBR(struct session_flags_stat *stat, uint64_t all_pkts)
{
	if (stat->result.flags & SESSION_FLAGS_CBR)
		return;

	float CBR;

	OnlineMean_Update(&stat->iter.c2s.omean, stat->c2s.rate);
	OnlineMean_Update(&stat->iter.s2c.omean, stat->s2c.rate);
	
	stat->iter.c2s.CBR = session_flags_calculate_CV(&stat->iter.c2s.omean);
	stat->iter.s2c.CBR = session_flags_calculate_CV(&stat->iter.s2c.omean);

	if (stat->main_dir == MAIN_DIR_UNKONWN)
	{
		return;
	}

	if (stat->main_dir == FLOW_TYPE_C2S)
	{
		CBR = stat->iter.c2s.CBR;
	}
	else
	{
		CBR = stat->iter.s2c.CBR;
	}

	if (CBR < SESSION_CBR_STREAMING_THRESHORD && CBR > 0.0)
	{
		stat->result.flags |= SESSION_FLAGS_CBR;
		stat->result.identify[session_flags_cbr_mask] = all_pkts;
	}
}

static void session_flags_calculate_interactive(struct session_flags_stat *stat, uint64_t all_pkts)
{
	if (stat->result.flags & SESSION_FLAGS_INTERACTIVE || stat->c2s.pkts == 0 || stat->s2c.pkts == 0)
		return;

	if (stat->stream_live_time_ms > g_sf_conf.interactive_starttime_ms && stat->interactive_pulse_num > g_sf_conf.interactive_pulse_num)
	{
		stat->result.flags |= SESSION_FLAGS_INTERACTIVE;
		stat->result.identify[session_flags_interactive_mask] = all_pkts;
	}
}

static void session_flags_calculate_interactive_pulse(struct session_flags_stat *stat, uint64_t cur_ms, uint64_t all_pkts)
{
	uint64_t delta_ms = 0;
	if (cur_ms > stat->last_pkt_ts_ms)
	{
		delta_ms = cur_ms - stat->last_pkt_ts_ms;
	}

	if (delta_ms > g_sf_conf.interactive_latency_ms)
	{
		stat->interactive_pulse_num++;
		session_flags_calculate_interactive(stat, all_pkts);
	}
}

static void session_flags_calculate_bound(struct session_flags_stat *stat, uint64_t all_pkts)
{
	if (stat == NULL || stat->c2s.pkts == 0 || stat->s2c.pkts == 0)
	{
		return;
	}
	
	if (!(stat->result.flags & SESSION_FLAGS_INBOUND) && !(stat->result.flags & SESSION_FLAGS_OUTBOUND))
	{
		if (stat->result.flags & SESSION_FLAGS_LOCAL_CLIENT)
		{
			if (stat->main_dir == FLOW_TYPE_C2S)
			{
				stat->result.flags |= SESSION_FLAGS_OUTBOUND;
				stat->result.identify[session_flags_outbound_mask] = all_pkts;
			}
			else
			{
				stat->result.flags |= SESSION_FLAGS_INBOUND;
				stat->result.identify[session_flags_inbound_mask] = all_pkts;
			}
		}
		else
		{
			if (stat->main_dir == FLOW_TYPE_C2S)
			{
				stat->result.flags |= SESSION_FLAGS_INBOUND;
				stat->result.identify[session_flags_inbound_mask] = all_pkts;
			}
			else
			{
				stat->result.flags |= SESSION_FLAGS_OUTBOUND;
				stat->result.identify[session_flags_outbound_mask] = all_pkts;
			}
		}
	}
}

static void session_flags_calculate_dir(struct session_flags_stat *stat, uint64_t all_pkts)
{
	if (stat->c2s.pkts != 0 && !(stat->result.flags & SESSION_FLAGS_C2S))
	{
		stat->result.flags |= SESSION_FLAGS_C2S;
		stat->result.identify[session_flags_c2s_mask] = all_pkts;
	}
	if (stat->s2c.pkts != 0 && !(stat->result.flags & SESSION_FLAGS_S2C))
	{
		stat->result.flags |= SESSION_FLAGS_S2C;
		stat->result.identify[session_flags_s2c_mask] = all_pkts;
	}
}

static void session_flags_calculate_bidirectional(struct session_flags_stat *stat, uint64_t all_pkts)
{
	if (stat->result.flags & SESSION_FLAGS_BIDIRECTIONAL)
	{
		return;
	}

	float rate_ratio = stat->c2s.rate / stat->s2c.rate;
	if (stat->iter.bidirectional == 0)
	{
		stat->iter.bidirectional = rate_ratio;
	}
	else
	{//EWMA
		stat->iter.bidirectional = SESSION_EWMA_FACTOR * stat->iter.bidirectional + (1 - SESSION_EWMA_FACTOR) * rate_ratio;
	}

	if (stat->iter.iter_cnt >= SESSION_EWMA_ITER_CNT_MIN)
	{
		if (stat->iter.bidirectional > SESSION_BIDIRECTIONAL_THRESHORD_MIN && stat->iter.bidirectional < SESSION_BIDIRECTIONAL_THRESHORD_MAX)
		{
			stat->result.flags |= SESSION_FLAGS_BIDIRECTIONAL;
			stat->result.identify[session_flags_bidirectional_mask] = all_pkts;
		}
	}
	
	return;
}

static void session_flags_calculate_randomness_by_fet(struct session_flags_stat *stat, size_t payload_len, const char *payload, uint64_t all_pkts)
{
	if (stat->random_looking_stat.has_judged_fet)
	{
		return;
	}

	stat->random_looking_stat.has_judged_fet = 1;

	if (g_sf_conf.fet_enabled)
	{
		struct fet_detail detail;
		int is_fet = is_data_fet((unsigned char *)payload, payload_len, &detail);

		stat->result.is_tls = detail.is_tls;

		if (is_fet)
		{//if payload is fet data, then it is definitely random looking
			stat->result.flags |= SESSION_FLAGS_RANDOM_LOOKING;
			stat->result.identify[session_flags_random_looking_mask] = all_pkts;
		}
	}

	return;
}

static void session_flags_calculate_randomness_by_sts(struct session_flags_plugin_info *sf_plugin_info, struct session *session, struct session_flags_stat *stat, size_t payload_len, const char *payload, uint64_t all_pkts)
{
	if (stat->random_looking_stat.has_judged_sts || sts_random_switch == 0)
	{
		return;
	}

	if (payload_len < SESSION_RANDOM_LOOKING_PAYLOAD_LEN_MIN)
	{
		return;
	}

	stat->random_looking_stat.has_judged_sts = 1;

	struct sts_result result;
	uint32_t random_flags_count = 0;
	memset(&result, 0, sizeof(result));

	mesa_statistical_test_suite((void*)payload, payload_len, &result, sts_random_switch);

	if (result.frequency)
	{
		stat->result.random_looking_flags |= SESSION_FLAGS_FREQUENCY;
		random_flags_count++;
	}
	if (result.block_frequency)
	{
		stat->result.random_looking_flags |= SESSION_FLAGS_BLOCK_FREQUENCY;
		random_flags_count++;
	}
	if (result.cumulative_sums)
	{
		stat->result.random_looking_flags |= SESSION_FLAGS_CUMULATIVE_SUMS;
		random_flags_count++;
	}
	if (result.runs)
	{
		stat->result.random_looking_flags |= SESSION_FLAGS_RUNS;
		random_flags_count++;
	}
	if (result.longest_run)
	{
		stat->result.random_looking_flags |= SESSION_FLAGS_LONGEST_RUN;
		random_flags_count++;
	}
	if (result.rank)
	{
		stat->result.random_looking_flags |= SESSION_FLAGS_RANK;
		random_flags_count++;
	}
	if (result.non_overlapping_template_matching)
	{
		stat->result.random_looking_flags |= SESSION_FLAGS_NON_OVERLAPPING_TEMPLATE_MATCHING;
		random_flags_count++;
	}
	if (result.overlapping_template_matching)
	{
		stat->result.random_looking_flags |= SESSION_FLAGS_OVERLAPPING_TEMPLATE_MATCHING;
		random_flags_count++;
	}
	if (result.universal)
	{
		stat->result.random_looking_flags |= SESSION_FLAGS_UNIVERSAL;
		random_flags_count++;
	}
	if (result.random_excursions)
	{
		stat->result.random_looking_flags |= SESSION_FLAGS_RANDOM_EXCURSIONS;
		random_flags_count++;
	}
	if (result.random_excursions_variant)
	{
		stat->result.random_looking_flags |= SESSION_FLAGS_RANDOM_EXCURSIONS_VARIANT;
		random_flags_count++;
	}
	if (result.poker_detect)
	{
		stat->result.random_looking_flags |= SESSION_FLAGS_POKER_DETECT;
		random_flags_count++;
	}
	if (result.runs_distribution)
	{
		stat->result.random_looking_flags |= SESSION_FLAGS_RUNS_DISTRIBUTION;
		random_flags_count++;
	}
	if (result.self_correlation)
	{
		stat->result.random_looking_flags |= SESSION_FLAGS_SELF_CORRELATION;
		random_flags_count++;
	}
	if (result.binary_derivative)
	{
		stat->result.random_looking_flags |= SESSION_FLAGS_BINARY_DERIVATIVE;
		random_flags_count++;
	}

	if (random_flags_count > (g_sf_conf.random_judge_flags_cnt / 2))
	{
		stat->result.flags |= SESSION_FLAGS_RANDOM_LOOKING;
		stat->result.identify[session_flags_random_looking_mask] = all_pkts;
	}

	STELLAR_LOG_DEBUG(sf_plugin_info->log_handle, SESSION_FLAGS_LOG_MODULE, "[%s] calculate random looking flags, flags:0x%x, pkts_num:%d", session_get0_readable_addr(session), stat->result.random_looking_flags, all_pkts);
}

static void session_flags_calculate_random_looking(struct session_flags_plugin_info *sf_plugin_info, struct session_flags_stat *stat, struct session *session, int topic_id, uint64_t all_pkts)
{
	if (stat->result.flags & SESSION_FLAGS_RANDOM_LOOKING)
	{
		return;
	}

	int udp_topic_id = sf_plugin_info->udp_topic_id;

	const struct packet *pkt = session_get0_current_packet(session);
	size_t payload_len = packet_get_payload_len(pkt);
	const char *payload = packet_get_payload(pkt);

	if ((topic_id == udp_topic_id) && g_sf_conf.random_looking_udp_ignore_pkts < 0)// disable random looking udp when  random_looking_udp_ignore_pkts<0
	{
		return;
	}

	if (payload_len == 0 || payload == NULL)
	{
		return;
	}

	stat->random_looking_stat.payload_pkt_num++;
	if ((topic_id == udp_topic_id) && stat->random_looking_stat.payload_pkt_num <= g_sf_conf.random_looking_udp_ignore_pkts)
	{
		return;
	}

	session_flags_calculate_randomness_by_fet(stat, payload_len, payload, all_pkts);
	if (stat->result.flags & SESSION_FLAGS_RANDOM_LOOKING)
	{
		return;
	}

	session_flags_calculate_randomness_by_sts(sf_plugin_info, session, stat, payload_len, payload, all_pkts);

	return;
}

static void session_flags_reset_delta(struct session_flags_stat *stat)
{
	stat->c2s.delta_bytes = 0;
	stat->c2s.delta_pkts = 0;
	stat->c2s.delta_large_pkts = 0;
	stat->c2s.delta_payload_pkts = 0;

	stat->s2c.delta_bytes = 0;
	stat->s2c.delta_pkts = 0;
	stat->s2c.delta_large_pkts = 0;
	stat->s2c.delta_payload_pkts = 0;

	return;
}

static void session_flags_calculate_rate(struct session_flags_stat *stat, uint64_t delta_ms)
{
	stat->c2s.rate = stat->c2s.delta_bytes * 1.f / delta_ms;
	stat->s2c.rate = stat->s2c.delta_bytes * 1.f / delta_ms;

	return;
}

static void session_flags_update_large_pkt_size(struct session_flags_plugin_info *sf_plugin_info, uint32_t pkt_len, uint64_t ms)
{
	if (pkt_len > 1500)
	{
		return;
	}

	OnlineMean_Update(&g_large_pkt_omean, pkt_len);

	if (g_large_pkt_size_update_ms == 0)
	{
		g_large_pkt_size_update_ms = ms;
		g_large_pkt_size = g_sf_conf.large_ptks_init_size;
		return;
	}

	if (ms - g_large_pkt_size_update_ms >= LARGE_PKT_SIZE_UPDATE_INTERVAL_MS && g_large_pkt_omean.count >= LARGE_PKT_SIZE_UPDATE_COUNT_MIN)
	{
		g_large_pkt_size_update_ms = ms;
		g_large_pkt_size = (uint32_t)(0.84f * OnlineMean_GetStd(&g_large_pkt_omean) + OnlineMean_GetMean(&g_large_pkt_omean));//计算公式 0.84 = (g_large_pkt_size - mean) / std, 0.84为正态分布表中80%的概率对应的z值

		if (g_large_pkt_size < LARGE_PKT_SIZE_MIN)
		{
			g_large_pkt_size = LARGE_PKT_SIZE_MIN;
		}
		else if (g_large_pkt_size > LARGE_PKT_SIZE_MAX)
		{
			g_large_pkt_size = LARGE_PKT_SIZE_MAX;
		}

		STELLAR_LOG_DEBUG(sf_plugin_info->log_handle, SESSION_FLAGS_LOG_MODULE, "session_flags thread %ld update large_pkt_size %d", pthread_self(), g_large_pkt_size);
	}

	return;
}

struct session_flags_result *session_flags(struct session_flags_plugin_info *sf_plugin_info, struct session_flags_ctx *ctx, struct session *session, int topic_id, uint32_t bytes, enum flow_type flow_type, uint64_t ms)
{
	struct session_flags_stat *stat = &ctx->stat;

	if (stat == NULL || bytes == 0)
	{
		return NULL;
	}

	session_flags_update_large_pkt_size(sf_plugin_info, bytes, ms);

	uint64_t delta_ms = 0;

	if(stat->session_start_time_ms == 0)
	{
		stat->session_start_time_ms = ms;
	}
	stat->stream_live_time_ms = ms - stat->session_start_time_ms;

	if (stat->last_iter_ts_ms == 0)
	{
		stat->last_iter_ts_ms = ms;
	}

	if (flow_type == FLOW_TYPE_C2S)
	{
		flow_stat_update(session, &stat->c2s, bytes);
	}
	else
	{
		flow_stat_update(session, &stat->s2c, bytes);
	}

	uint64_t all_pkts = stat->c2s.pkts + stat->s2c.pkts;
	session_flags_calculate_dir(stat, all_pkts);
	session_flags_calculate_random_looking(sf_plugin_info, stat, session, topic_id, all_pkts);

	if (stat->stream_live_time_ms >= START_JUDGE_TIME_MS)
	{
		if (all_pkts > g_sf_conf.main_dir_front_n_pkts)
		{
			session_flags_calculate_main_dir(stat);
		}
	}

	if (stat->c2s.pkts == 0 || stat->s2c.pkts == 0)
	{
		goto END;
	}

	if (stat->main_dir != MAIN_DIR_UNKONWN)
	{
		session_flags_calculate_bound(stat, all_pkts);
		session_flags_calculate_unidirection(stat, all_pkts);
	}

	delta_ms = ms - stat->last_iter_ts_ms;
	if (delta_ms >= SESSION_ITER_INTERVAL_MS)
	{
		session_flags_calculate_rate(stat, delta_ms);
		
		session_flags_calculate_bidirectional(stat, all_pkts);
		session_flags_calculate_CBR(stat, all_pkts);
		session_flags_calculate_download(stat, all_pkts);
		session_flags_calculate_bulky(stat, all_pkts);
		session_flags_calculate_pseudo_unidirection(stat, all_pkts);
		session_flags_calculate_streaming(stat, all_pkts);
		stat->iter.iter_cnt++;
		stat->last_iter_ts_ms = ms;

		session_flags_reset_delta(stat);
	}

	session_flags_calculate_interactive_pulse(stat, ms, all_pkts);

END:
	stat->last_pkt_ts_ms = ms;
	return &stat->result;
}

struct session_flags_result *session_flags_get_flags(struct session_flags_stat *stat)
{
	if (stat == NULL)
		return NULL;

	return &stat->result;
}

struct session_flags_message *session_flags_generate_firewall_message(uint64_t flags, const uint32_t identify[session_flags_all_mask])
{
	int flag_num = 0;
	uint32_t temp_identify[session_flags_all_mask] = {0};
	struct session_flags_message *flags_msg = (struct session_flags_message *)calloc(1, sizeof(struct session_flags_message));

	flags_msg->magic= MESSAGE_MAGIC;
	flags_msg->flags = flags;

	for (int i = 0; i < session_flags_all_mask; i++)
	{
		if (flags & (SESSION_FLAGS_START << i))
		{
			temp_identify[flag_num] = identify[i];
			flag_num++;
		}
	}

	flags_msg->packet_sequence_array = (uint32_t *)calloc(flag_num, sizeof(uint32_t));
	for (int i = 0; i < flag_num; i++)
	{
		flags_msg->packet_sequence_array[i] = temp_identify[i];
	}
	flags_msg->array_num = flag_num;

	return flags_msg;
}