tango-tfe/common/src/tfe_packet_io.cpp

#include <assert.h>
#include <netinet/ip.h>
#include <netinet/udp.h>
#include <netinet/ether.h>

#include <marsio.h>
#include <cjson/cJSON.h>
#include <MESA/MESA_prof_load.h>

#include <tfe_utils.h>
// #include <tfe_proxy.h>
#include <proxy.h>
#include "tfe_acceptor_kni.h"
#include "tfe_ctrl_packet.h"
#include "tfe_raw_packet.h"
#include "io_uring.h"
#include "tfe_metrics.h"
#include "tfe_cmsg.h"
#include "tfe_tcp_restore.h"
#include "tfe_stream.h"
#include "raw_socket.h"
#include "packet_construct.h"
#include "tfe_tap_rss.h"
#include <intercept_policy.h>

/*
 *                 add: vxlan_hdr
 *                 del: marsio_buff_ctrlzone_reset()
 *  +----+         NF2SF              +----+
 *  |    |--------------------------->|    |
 *  |    |                            |    |
 *  |    |-------+                    |    |-------+
 *  | NF |       | NF2NF (undo)       | SF |       | SF2SF (del old vxlan_hdr; add new vxlan_hdr)
 *  |    |<------+                    |    |<------+
 *  |    |                            |    |
 *  |    |<---------------------------|    |
 *  |    |         SF2NF              |    |
 *  +---+          del: vxlan_hdr     +----+
 *                 add: session_id + route_ctx + sid
 */

/******************************************************************************
 * Struct
 ******************************************************************************/

#define RX_BURST_MAX 128

#define TRAFFIC_IS_DECRYPTED        (1 << 0)
#define SET_TRAFFIC_IS_DECRYPTED(field)    (field || TRAFFIC_IS_DECRYPTED)
#define CLEAR_TRAFFIC_IS_DECRYPTED(field)  (field && ~TRAFFIC_IS_DECRYPTED)

struct config
{
    int bypass_all_traffic;
    int rx_burst_max;
    char app_symbol[256];
    char dev_nf_interface[256];
};

struct device
{
    struct mr_vdev *mr_dev;
    struct mr_sendpath *mr_path;
};

struct packet_io
{
    int thread_num;
    struct mr_instance *instance;
    struct device dev_nf_interface;
    struct config config;
};

enum raw_pkt_action
{
    RAW_PKT_ERR_BYPASS,
    RAW_PKT_HIT_BYPASS,
    RAW_PKT_HIT_BLOCK,
    RAW_PKT_HIT_STEERING,
    RAW_PKT_HIT_MIRRORING,
};

enum inject_pkt_action
{
    INJT_PKT_ERR_DROP,
    INJT_PKT_MIRR_RX_DROP,
    INJT_PKT_HIT_BLOCK,
    INJT_PKT_HIT_FWD2SF, // forward to service function
    INJT_PKT_HIT_FWD2NF, // forward to network function
};

struct metadata
{
    uint64_t session_id;

    char *raw_data;
    int raw_len;

    int dir_is_e2i;
    int is_ctrl_pkt;

    uint16_t l7_offset;       // only control packet set l7_offset
    uint16_t user_field;    // only raw packet set traffic_is_decrypted (1 << 0)

    struct sids sids;
    struct route_ctx route_ctx;
};

/******************************************************************************
 * API Declaration
 ******************************************************************************/

struct packet_io *packet_io_create(const char *profile, int thread_num, cpu_set_t *coremask);
void packet_io_destory(struct packet_io *handle);

int packet_io_polling_nf_interface(struct packet_io *handle, int thread_seq, void *ctx);

extern int tcp_policy_enforce(struct tcp_policy_enforcer *tcp_enforcer, struct tfe_cmsg *cmsg);
extern int tfe_proxy_fds_accept(struct tfe_proxy * ctx, int fd_downstream, int fd_upstream, int fd_fake_c, int fd_fake_s, struct tfe_cmsg * cmsg);

extern void chaining_policy_enforce(struct chaining_policy_enforcer *enforcer, struct tfe_cmsg *cmsg, uint64_t rule_id);

// return  0 : success
// return -1 : error
static int packet_io_config(const char *profile, struct config *config);

// return  0 : success
// return -1 : error
static int packet_io_get_metadata(marsio_buff_t *tx_buff, struct metadata *meta);
// return  0 : success
// return -1 : error
static int packet_io_set_metadata(marsio_buff_t *tx_buff, struct metadata *meta);
static void packet_io_dump_metadata(marsio_buff_t *tx_buff, struct metadata *meta);

// return  0 : success
// return -1 : error
static int handle_control_packet(struct packet_io *handle, marsio_buff_t *rx_buff, int thread_seq, void *ctx);

static int handle_raw_packet_from_nf(struct packet_io *handle, marsio_buff_t *rx_buff, int thread_seq, void *ctx, int *action_bytes);

static int add_ether_header(void *raw_data, char *src_mac, char *dst_mac){
	struct ethhdr *ether_hdr = (struct ethhdr*)raw_data;
	memcpy(ether_hdr->h_dest, dst_mac, sizeof(ether_hdr->h_dest));
	memcpy(ether_hdr->h_source, src_mac, sizeof(ether_hdr->h_source));
	return 0;
}

// return  0 : success
// return -1 : error
static int handle_session_opening(struct metadata *meta, struct ctrl_pkt_parser *parser, int thread_seq, void *ctx);
// return  0 : success
// return -1 : error
static int handle_session_closing(struct metadata *meta, struct ctrl_pkt_parser *parser, int thread_seq, void *ctx);
// return  0 : success
// return -1 : error
static int handle_session_active(struct metadata *meta, struct ctrl_pkt_parser *parser, int thread_seq, void *ctx);
// return  0 : success
// return -1 : error
static int handle_session_resetall(struct metadata *meta, struct ctrl_pkt_parser *parser, int thread_seq, void *ctx);

static void session_value_free_cb(void *ctx);

// return 0 : not keepalive packet
// return 1 : is  keepalive packet
static int is_downstream_keepalive_packet(marsio_buff_t *rx_buff);


/******************************************************************************
 * API Definition
 ******************************************************************************/

struct packet_io *packet_io_create(const char *profile, int thread_num, cpu_set_t *coremask)
{
    int opt = 1;
    struct packet_io *handle = (struct packet_io *)calloc(1, sizeof(struct packet_io));
    assert(handle != NULL);
    handle->thread_num = thread_num;

    if (packet_io_config(profile, &(handle->config)) != 0)
    {
        goto error_out;
    }

    handle->instance = marsio_create();
    if (handle->instance == NULL)
    {
        TFE_LOG_ERROR(g_default_logger, "%s: unable to create marsio instance", LOG_TAG_PKTIO);
        goto error_out;
    }

    if (marsio_option_set(handle->instance, MARSIO_OPT_THREAD_MASK_IN_CPUSET, coremask, sizeof(cpu_set_t)) != 0)
    {
        TFE_LOG_ERROR(g_default_logger, "%s: unable to set MARSIO_OPT_EXIT_WHEN_ERR option for marsio instance", LOG_TAG_PKTIO);
        goto error_out;
    }

    if (marsio_option_set(handle->instance, MARSIO_OPT_EXIT_WHEN_ERR, &opt, sizeof(opt)) != 0)
    {
        TFE_LOG_ERROR(g_default_logger, "%s: unable to set MARSIO_OPT_EXIT_WHEN_ERR option for marsio instance", LOG_TAG_PKTIO);
        goto error_out;
    }

    if (marsio_init(handle->instance, handle->config.app_symbol) != 0)
    {
        TFE_LOG_ERROR(g_default_logger, "%s: unable to initialize marsio instance", LOG_TAG_PKTIO);
        goto error_out;
    }

    // Netwrok Function Interface
    handle->dev_nf_interface.mr_dev = marsio_open_device(handle->instance, handle->config.dev_nf_interface, handle->thread_num, handle->thread_num);
    if (handle->dev_nf_interface.mr_dev == NULL)
    {
        TFE_LOG_ERROR(g_default_logger, "%s: unable to open device %s", LOG_TAG_PKTIO, handle->config.dev_nf_interface);
        goto error_out;
    }

    handle->dev_nf_interface.mr_path = marsio_sendpath_create_by_vdev(handle->dev_nf_interface.mr_dev);
    if (handle->dev_nf_interface.mr_path == NULL)
    {
        TFE_LOG_ERROR(g_default_logger, "%s: unable to create sendpath for device %s", LOG_TAG_PKTIO, handle->config.dev_nf_interface);
        goto error_out;
    }

    return handle;

error_out:
    packet_io_destory(handle);
    return NULL;
}

void packet_io_destory(struct packet_io *handle)
{
    if (handle)
    {
        if (handle->dev_nf_interface.mr_path)
        {
            marsio_sendpath_destory(handle->dev_nf_interface.mr_path);
            handle->dev_nf_interface.mr_path = NULL;
        }

        if (handle->dev_nf_interface.mr_dev)
        {
            marsio_close_device(handle->dev_nf_interface.mr_dev);
            handle->dev_nf_interface.mr_dev = NULL;
        }

        if (handle->instance)
        {
            marsio_destory(handle->instance);
            handle->instance = NULL;
        }

        free(handle);
        handle = NULL;
    }
}

int packet_io_thread_init(struct packet_io *handle, struct acceptor_thread_ctx *thread_ctx)
{
    if (marsio_thread_init(handle->instance) != 0)
    {
        TFE_LOG_ERROR(g_default_logger, "%s: unable to init marsio thread %d", LOG_TAG_PKTIO, thread_ctx->thread_index);
        return -1;
    }

    return 0;
}

void packet_io_thread_wait(struct packet_io *handle, struct acceptor_thread_ctx *thread_ctx, int timeout_ms)
{
    struct mr_vdev *vdevs[] = {handle->dev_nf_interface.mr_dev};

    marsio_poll_wait(handle->instance, vdevs, 1, thread_ctx->thread_index, timeout_ms);
}

// return n_packet_recv
int packet_io_polling_nf_interface(struct packet_io *handle, int thread_seq, void *ctx)
{
    struct acceptor_thread_ctx *thread = (struct acceptor_thread_ctx *)ctx;
    struct global_metrics *g_metrics = thread->ref_metrics;

    marsio_buff_t *rx_buffs[RX_BURST_MAX];

    // nr_recv <= rx_burst_max <= RX_BURST_MAX
    int nr_recv = marsio_recv_burst(handle->dev_nf_interface.mr_dev, thread_seq, rx_buffs, handle->config.rx_burst_max);
    if (nr_recv <= 0)
    {
        return 0;
    }

    for (int j = 0; j < nr_recv; j++)
    {
        marsio_buff_t *rx_buff = rx_buffs[j];
        int raw_len = marsio_buff_datalen(rx_buff);

        if (is_downstream_keepalive_packet(rx_buff))
        {
            marsio_send_burst(handle->dev_nf_interface.mr_path, thread_seq, &rx_buff, 1);
            continue;
        }

        if (marsio_buff_is_ctrlbuf(rx_buff))
        {
            handle_control_packet(handle, rx_buff, thread_seq, ctx);
            throughput_metrics_inc(&g_metrics->ctrl_pkt_rx, 1, raw_len);
            // all control packet need bypass
            marsio_send_burst(handle->dev_nf_interface.mr_path, thread_seq, &rx_buff, 1);
        }
        else
        {
            int action_bytes = 0;
            handle_raw_packet_from_nf(handle, rx_buff, thread_seq, ctx, &action_bytes);
        }
    }

    return nr_recv;
}

void handle_decryption_packet_from_tap(const char *data, int len, void *args)
{
    struct acceptor_thread_ctx *thread = (struct acceptor_thread_ctx *)args;
    struct acceptor_ctx *acceptor_ctx = thread->ref_acceptor_ctx;
    struct packet_io *packet_io = thread->ref_io;

    struct addr_tuple4 inner_addr;
    struct raw_pkt_parser raw_parser;
    memset(&inner_addr, 0, sizeof(struct addr_tuple4));
    raw_packet_parser_init(&raw_parser, 0, LAYER_TYPE_ALL, 8);
    raw_packet_parser_parse(&raw_parser, (const void *)data, len);
    raw_packet_parser_get_most_inner_tuple4(&raw_parser, &inner_addr);

    struct session_node *node = session_table_search_by_addr(thread->session_table, &inner_addr);
    if (node == NULL)
    {
        char *addr_string = addr_tuple4_to_str(&inner_addr);
        TFE_LOG_ERROR(g_default_logger, "%s: unexpected inject packet, unable to find session %s from session table, drop !!!", LOG_TAG_PKTIO, addr_string);
        free(addr_string);
        return;
    }
    struct session_ctx *s_ctx = (struct session_ctx *)node->val_data;

    marsio_buff_t *tx_buffs[1];
	int alloc_ret = marsio_buff_malloc_device(packet_io->dev_nf_interface.mr_dev, tx_buffs, 1, 0, thread->thread_index);
    if (alloc_ret < 0){
		TFE_LOG_ERROR(g_default_logger, "Failed at alloc marsio buffer, ret = %d, thread_seq = %d", 
					alloc_ret, thread->thread_index);
		return;
	}

    char *dst = marsio_buff_append(tx_buffs[0], len);
    memcpy(dst, data, len);

    struct metadata meta = {0};
    meta.session_id = s_ctx->session_id;
    meta.raw_data = dst;
    meta.raw_len = len;
    meta.user_field = SET_TRAFFIC_IS_DECRYPTED(0);
    meta.is_ctrl_pkt = 0;
    meta.l7_offset = 0;
    meta.sids.num = 1;
    meta.sids.elems[0] = acceptor_ctx->sce_sids;

    if (memcmp(&inner_addr, &s_ctx->first_ctrl_pkt.tuple4, sizeof(struct addr_tuple4)) == 0)
        meta.dir_is_e2i = s_ctx->first_ctrl_pkt.dir_is_e2i;
    else
        meta.dir_is_e2i = !s_ctx->first_ctrl_pkt.dir_is_e2i;

    if (meta.dir_is_e2i)
    {
        route_ctx_copy(&meta.route_ctx, &s_ctx->raw_pkt_e2i_route_ctx);
    }
    else
    {
        route_ctx_copy(&meta.route_ctx, &s_ctx->raw_pkt_i2e_route_ctx);
    }
    packet_io_set_metadata(tx_buffs[0], &meta);
    marsio_send_burst(packet_io->dev_nf_interface.mr_path, thread->thread_index, tx_buffs, 1);
}

void handle_raw_packet_from_tap(const char *data, int len, void *args)
{
    struct acceptor_thread_ctx *thread = (struct acceptor_thread_ctx *)args;
    struct acceptor_ctx *acceptor_ctx = thread->ref_acceptor_ctx;
    struct packet_io *packet_io = thread->ref_io;

    struct addr_tuple4 inner_addr;
    struct raw_pkt_parser raw_parser;
    memset(&inner_addr, 0, sizeof(struct addr_tuple4));
    raw_packet_parser_init(&raw_parser, 0, LAYER_TYPE_ALL, 8);
    raw_packet_parser_parse(&raw_parser, (const void *)data, len);
    raw_packet_parser_get_most_inner_tuple4(&raw_parser, &inner_addr);

    struct session_node *node = session_table_search_by_addr(thread->session_table, &inner_addr);
    if (node == NULL)
    {
        char *addr_string = addr_tuple4_to_str(&inner_addr);
        TFE_LOG_ERROR(g_default_logger, "%s: unexpected inject packet, unable to find session %s from session table, drop !!!", LOG_TAG_PKTIO, addr_string);
        free(addr_string);
        return;
    }
    struct session_ctx *s_ctx = (struct session_ctx *)node->val_data;

    marsio_buff_t *tx_buffs[1];
	int alloc_ret = marsio_buff_malloc_device(packet_io->dev_nf_interface.mr_dev, tx_buffs, 1, 0, thread->thread_index);
    if (alloc_ret < 0){
		TFE_LOG_ERROR(g_default_logger, "Failed at alloc marsio buffer, ret = %d, thread_seq = %d", 
					alloc_ret, thread->thread_index);
		return;
	}

    char *dst = marsio_buff_append(tx_buffs[0], len + s_ctx->first_ctrl_pkt.header_len);
    memcpy(dst, s_ctx->first_ctrl_pkt.header_data, s_ctx->first_ctrl_pkt.header_len);
    memcpy(dst + s_ctx->first_ctrl_pkt.header_len, data, len);

    struct metadata meta = {0};
    meta.session_id = s_ctx->session_id;
    meta.raw_data = dst;
    meta.raw_len = len;
    meta.user_field = s_ctx->user_field;
    meta.is_ctrl_pkt = 0;
    meta.l7_offset = 0;

    char *src_mac = NULL;
    char *dst_mac = NULL;
    if (memcmp(&inner_addr, &s_ctx->first_ctrl_pkt.tuple4, sizeof(struct addr_tuple4)) == 0) {
        meta.dir_is_e2i = s_ctx->first_ctrl_pkt.dir_is_e2i;
        struct ethhdr *ether_hdr = (struct ethhdr *)(s_ctx->first_ctrl_pkt.header_data);
        src_mac = (char *)ether_hdr->h_source;
        dst_mac = (char *)ether_hdr->h_dest;
    }
    else {
        meta.dir_is_e2i = !s_ctx->first_ctrl_pkt.dir_is_e2i;
        struct ethhdr *ether_hdr = (struct ethhdr *)(s_ctx->first_ctrl_pkt.header_data);
        dst_mac = (char *)ether_hdr->h_source;
        src_mac = (char *)ether_hdr->h_dest;
    }

    if (meta.dir_is_e2i)
    {
        sids_copy(&meta.sids, &s_ctx->raw_pkt_e2i_sids);
        route_ctx_copy(&meta.route_ctx, &s_ctx->raw_pkt_e2i_route_ctx);
    }
    else
    {
        sids_copy(&meta.sids, &s_ctx->raw_pkt_i2e_sids);
        route_ctx_copy(&meta.route_ctx, &s_ctx->raw_pkt_i2e_route_ctx);
    }
    packet_io_set_metadata(tx_buffs[0], &meta);
    // add_ether_header(dst, src_mac, dst_mac);
    marsio_send_burst(packet_io->dev_nf_interface.mr_path, thread->thread_index, tx_buffs, 1);
}

// return  0 : success
// return -1 : error
static int packet_io_config(const char *profile, struct config *config)
{
    MESA_load_profile_int_def(profile, "PACKET_IO", "rx_burst_max", (int *)&(config->rx_burst_max), 1);
    MESA_load_profile_string_nodef(profile, "PACKET_IO", "app_symbol", config->app_symbol, sizeof(config->app_symbol));
    MESA_load_profile_string_nodef(profile, "PACKET_IO", "dev_nf_interface", config->dev_nf_interface, sizeof(config->dev_nf_interface));

    if (config->rx_burst_max > RX_BURST_MAX)
    {
        TFE_LOG_ERROR(g_default_logger, "%s: invalid rx_burst_max, exceeds limit %d", LOG_TAG_PKTIO, RX_BURST_MAX);
        return -1;
    }

    if (strlen(config->app_symbol) == 0)
    {
        TFE_LOG_ERROR(g_default_logger, "%s: invalid app_symbol in %s", LOG_TAG_PKTIO, profile);
        return -1;
    }

    if (strlen(config->dev_nf_interface) == 0)
    {
        TFE_LOG_ERROR(g_default_logger, "%s: invalid dev_nf_interface in %s", LOG_TAG_PKTIO, profile);
        return -1;
    }

    TFE_LOG_DEBUG(g_default_logger, "%s: PACKET_IO->bypass_all_traffic   : %d", LOG_TAG_PKTIO, config->bypass_all_traffic);
    TFE_LOG_DEBUG(g_default_logger, "%s: PACKET_IO->rx_burst_max         : %d", LOG_TAG_PKTIO, config->rx_burst_max);
    TFE_LOG_DEBUG(g_default_logger, "%s: PACKET_IO->app_symbol           : %s", LOG_TAG_PKTIO, config->app_symbol);
    TFE_LOG_DEBUG(g_default_logger, "%s: PACKET_IO->dev_nf_interface     : %s", LOG_TAG_PKTIO, config->dev_nf_interface);

    return 0;
}

// return  0 : success
// return -1 : error
static int packet_io_get_metadata(marsio_buff_t *rx_buff, struct metadata *meta)
{
    memset(meta, 0, sizeof(struct metadata));

    if (marsio_buff_get_metadata(rx_buff, MR_BUFF_SESSION_ID, &(meta->session_id), sizeof(meta->session_id)) <= 0)
    {
        TFE_LOG_ERROR(g_default_logger, "%s: unable to get session_id from metadata", LOG_TAG_PKTIO);
        return -1;
    }

    meta->raw_len = marsio_buff_datalen(rx_buff);
    meta->raw_data = marsio_buff_mtod(rx_buff);
    if (meta->raw_data == NULL || meta->raw_len == 0)
    {
        TFE_LOG_ERROR(g_default_logger, "%s: unable to get raw_data from metadata", LOG_TAG_PKTIO);
        return -1;
    }

    // 1: E2I
    // 0: I2E
    if (marsio_buff_get_metadata(rx_buff, MR_BUFF_DIR, &(meta->dir_is_e2i), sizeof(meta->dir_is_e2i)) <= 0)
    {
        TFE_LOG_ERROR(g_default_logger, "%s: unable to get buff_dir from metadata", LOG_TAG_PKTIO);
        return -1;
    }

    if (marsio_buff_is_ctrlbuf(rx_buff))
    {
        meta->is_ctrl_pkt = 1;
        // only control packet set MR_BUFF_PAYLOAD_OFFSET
        if (marsio_buff_get_metadata(rx_buff, MR_BUFF_PAYLOAD_OFFSET, &(meta->l7_offset), sizeof(meta->l7_offset)) <= 0)
        {
            TFE_LOG_ERROR(g_default_logger, "%s: unable to get l7_offset from metadata", LOG_TAG_PKTIO);
            return -1;
        }
    }
    else
    {
        meta->is_ctrl_pkt = 0;
        // if (marsio_buff_get_metadata(rx_buff, MR_BUFF_USER_0, &(meta->user_field), sizeof(meta->user_field)) <= 0)
        // {
        //     TFE_LOG_ERROR(g_default_logger, "%s: unable to get user_field from metadata", LOG_TAG_PKTIO);
        //     return -1;
        // }
    }

    meta->route_ctx.len = marsio_buff_get_metadata(rx_buff, MR_BUFF_ROUTE_CTX, meta->route_ctx.data, sizeof(meta->route_ctx.data));
    if (meta->route_ctx.len <= 0)
    {
        TFE_LOG_ERROR(g_default_logger, "%s: unable to get route_ctx from metadata", LOG_TAG_PKTIO);
        return -1;
    }

    meta->sids.num = marsio_buff_get_sid_list(rx_buff, meta->sids.elems, sizeof(meta->sids.elems) / sizeof(meta->sids.elems[0]));
    if (meta->sids.num < 0)
    {
        TFE_LOG_ERROR(g_default_logger, "%s: unable to get sid_list from metadata", LOG_TAG_PKTIO);
        return -1;
    }

    return 0;
}

// return  0 : success
// return -1 : error
static int packet_io_set_metadata(marsio_buff_t *tx_buff, struct metadata *meta)
{
    if (meta->session_id)
    {
        if (marsio_buff_set_metadata(tx_buff, MR_BUFF_SESSION_ID, &(meta->session_id), sizeof(meta->session_id)) != 0)
        {
            TFE_LOG_ERROR(g_default_logger, "%s: unable to set session_id for metadata", LOG_TAG_PKTIO);
            return -1;
        }
    }

    // 1: E2I
    // 0: I2E
#if 0
    // use MR_BUFF_ROUTE_CTX instead
    if (marsio_buff_set_metadata(tx_buff, MR_BUFF_DIR, &(meta->dir_is_e2i), sizeof(meta->dir_is_e2i)) != 0)
    {
        TFE_LOG_ERROR(g_default_logger, "%s: unable to set buff_dir for metadata", LOG_TAG_PKTIO);
        return -1;
    }
#endif

    if (meta->is_ctrl_pkt)
    {
        if (marsio_buff_set_metadata(tx_buff, MR_BUFF_PAYLOAD_OFFSET, &(meta->l7_offset), sizeof(meta->l7_offset)) != 0)
        {
            TFE_LOG_ERROR(g_default_logger, "%s: unable to set l7_offset for metadata", LOG_TAG_PKTIO);
            return -1;
        }
    }
    else
    {
        // if (marsio_buff_set_metadata(tx_buff, MR_BUFF_USER_0, &(meta->user_field), sizeof(meta->user_field)) != 0)
        // {
        //     TFE_LOG_ERROR(g_default_logger, "%s: unable to set user_field for metadata", LOG_TAG_PKTIO);
        //     return -1;
        // }
    }

    if (meta->route_ctx.len > 0)
    {
        if (marsio_buff_set_metadata(tx_buff, MR_BUFF_ROUTE_CTX, meta->route_ctx.data, meta->route_ctx.len) != 0)
        {
            TFE_LOG_ERROR(g_default_logger, "%s: unable to set route_ctx for metadata", LOG_TAG_PKTIO);
            return -1;
        }
    }

    if (meta->sids.num > 0)
    {
        if (marsio_buff_set_sid_list(tx_buff, meta->sids.elems, meta->sids.num) != 0)
        {
            TFE_LOG_ERROR(g_default_logger, "%s: unable to set sid_list for metadata", LOG_TAG_PKTIO);
            return -1;
        }
    }

    return 0;
}

static void packet_io_dump_metadata(marsio_buff_t *tx_buff, struct metadata *meta)
{
    TFE_LOG_DEBUG(g_default_logger, "%s: META={session_id: %lu, raw_len: %d, dir_is_e2i: %d, is_ctrl_pkt: %d, l7_offset: %d, user_field: %u, sids_num: %d}", LOG_TAG_PKTIO, meta->session_id, meta->raw_len, meta->dir_is_e2i, meta->is_ctrl_pkt, meta->l7_offset, meta->user_field, meta->sids.num);
}

static int tcp_restore_set_from_cmsg(struct tfe_cmsg *cmsg, struct tcp_restore_info *restore_info)
{
    int ret = 0;
    uint16_t length = 0;

    uint32_t seq;
    ret = tfe_cmsg_get_value(cmsg, TFE_CMSG_TCP_RESTORE_SEQ, (unsigned char *)&seq, sizeof(uint32_t), &length);
    if (ret == 0)
    {
        restore_info->client.seq = ntohl(seq);
        restore_info->server.ack = ntohl(seq);
    }

    uint32_t ack;
    ret = tfe_cmsg_get_value(cmsg, TFE_CMSG_TCP_RESTORE_ACK, (unsigned char *)&ack, sizeof(uint32_t), &length);
    if (ret == 0)
    {
        restore_info->client.ack = ntohl(ack);
        restore_info->server.seq = ntohl(ack);
    }

    uint8_t ts_client;
    ret = tfe_cmsg_get_value(cmsg, TFE_CMSG_TCP_RESTORE_TS_CLIENT, (unsigned char *)&ts_client, sizeof(uint8_t), &length);
    if (ret == 0)
    {
        restore_info->client.timestamp_perm = !!ts_client;
    }

    uint8_t ts_server;
    ret = tfe_cmsg_get_value(cmsg, TFE_CMSG_TCP_RESTORE_TS_SERVER, (unsigned char *)&ts_server, sizeof(uint8_t), &length);
    if (ret == 0)
    {
        restore_info->server.timestamp_perm = !!ts_server;
    }

    uint32_t ts_client_val;
    ret = tfe_cmsg_get_value(cmsg, TFE_CMSG_TCP_RESTORE_TS_CLIENT_VAL, (unsigned char *)&ts_client_val, sizeof(uint32_t), &length);
    if (ret == 0)
    {
        restore_info->client.ts_val = ntohl(ts_client_val);
    }

    uint32_t ts_server_val;
    ret = tfe_cmsg_get_value(cmsg, TFE_CMSG_TCP_RESTORE_TS_SERVER_VAL, (unsigned char *)&ts_server_val, sizeof(uint32_t), &length);
    if (ret == 0)
    {
        restore_info->server.ts_val = ntohl(ts_server_val);
    }

    uint8_t wsacle_client;
    ret = tfe_cmsg_get_value(cmsg, TFE_CMSG_TCP_RESTORE_WSACLE_CLIENT, (unsigned char *)&wsacle_client, sizeof(uint8_t), &length);
    if (ret == 0)
    {
        restore_info->client.wscale_perm = true;
        restore_info->client.wscale = wsacle_client;
    }

    uint8_t wsacle_server;
    ret = tfe_cmsg_get_value(cmsg, TFE_CMSG_TCP_RESTORE_WSACLE_SERVER, (unsigned char *)&wsacle_server, sizeof(uint8_t), &length);
    if (ret == 0)
    {
        restore_info->client.wscale_perm = true;
        restore_info->client.wscale = wsacle_server;
    }

    uint8_t sack_client;
    ret = tfe_cmsg_get_value(cmsg, TFE_CMSG_TCP_RESTORE_SACK_CLIENT, (unsigned char *)&sack_client, sizeof(uint8_t), &length);
    if (ret == 0)
    {
        restore_info->client.sack_perm = !!sack_client;
    }

    uint8_t sack_server;
    ret = tfe_cmsg_get_value(cmsg, TFE_CMSG_TCP_RESTORE_SACK_SERVER, (unsigned char *)&sack_server, sizeof(uint8_t), &length);
    if (ret == 0)
    {
        restore_info->server.sack_perm = !!sack_server;
    }

    uint16_t mss_client;
    ret = tfe_cmsg_get_value(cmsg, TFE_CMSG_TCP_RESTORE_MSS_CLIENT, (unsigned char *)&mss_client, sizeof(uint16_t), &length);
    if (ret == 0)
    {
        restore_info->client.mss = mss_client;
    }

    uint16_t mss_server;
    ret = tfe_cmsg_get_value(cmsg, TFE_CMSG_TCP_RESTORE_MSS_SERVER, (unsigned char *)&mss_server, sizeof(uint16_t), &length);
    if (ret == 0)
    {
        restore_info->server.mss = mss_server;
    }

    uint16_t window_client;
    ret = tfe_cmsg_get_value(cmsg, TFE_CMSG_TCP_RESTORE_WINDOW_CLIENT, (unsigned char *)&window_client, sizeof(uint16_t), &length);
    if (ret == 0)
    {
        restore_info->client.window = window_client;
    }

    uint16_t window_server;
    ret = tfe_cmsg_get_value(cmsg, TFE_CMSG_TCP_RESTORE_WINDOW_SERVER, (unsigned char *)&window_server, sizeof(uint16_t), &length);
    if (ret == 0)
    {
        restore_info->server.window = window_server;
    }

    uint8_t packet_cur_dir;
    ret = tfe_cmsg_get_value(cmsg, TFE_CMSG_TCP_RESTORE_INFO_PACKET_CUR_DIR, (unsigned char *)&packet_cur_dir, sizeof(uint8_t), &length);
    if (ret == 0)
    {
        restore_info->cur_dir = (enum tcp_restore_pkt_dir)packet_cur_dir;
    }

    return 0;
}

static int tcp_restore_set_from_pkg(struct addr_tuple4 *tuple4, struct tcp_restore_info *restore_info)
{
    if (tuple4->addr_type == ADDR_TUPLE4_TYPE_V4)
	{
		struct sockaddr_in *in_addr_client;
		struct sockaddr_in *in_addr_server;

		if (restore_info->cur_dir == PKT_DIR_NOT_SET || restore_info->cur_dir == PKT_DIR_C2S)
		{
			in_addr_client = (struct sockaddr_in *)&restore_info->client.addr;
			in_addr_server = (struct sockaddr_in *)&restore_info->server.addr;
		}
		else
		{
			in_addr_client = (struct sockaddr_in *)&restore_info->server.addr;
			in_addr_server = (struct sockaddr_in *)&restore_info->client.addr;
		}

		in_addr_client->sin_family = AF_INET;
		in_addr_client->sin_addr = tuple4->addr_v4.src_addr;
		in_addr_client->sin_port = tuple4->src_port;

		in_addr_server->sin_family = AF_INET;
		in_addr_server->sin_addr = tuple4->addr_v4.dst_addr;
		in_addr_server->sin_port = tuple4->dst_port;
	}

	if (tuple4->addr_type == ADDR_TUPLE4_TYPE_V6)
	{
		struct sockaddr_in6 *in6_addr_client;
		struct sockaddr_in6 *in6_addr_server;

		if (restore_info->cur_dir == PKT_DIR_NOT_SET || restore_info->cur_dir == PKT_DIR_C2S)
		{
			in6_addr_client = (struct sockaddr_in6 *)&restore_info->client.addr;
			in6_addr_server = (struct sockaddr_in6 *)&restore_info->server.addr;
		}
		else
		{
			in6_addr_client = (struct sockaddr_in6 *)&restore_info->server.addr;
			in6_addr_server = (struct sockaddr_in6 *)&restore_info->client.addr;
		}

		in6_addr_client->sin6_family = AF_INET6;
		in6_addr_client->sin6_addr = tuple4->addr_v6.src_addr;
		in6_addr_client->sin6_port = tuple4->src_port;

		in6_addr_server->sin6_family = AF_INET6;
		in6_addr_server->sin6_addr = tuple4->addr_v6.dst_addr;
		in6_addr_server->sin6_port = tuple4->dst_port;
	}

    return 0;
}

// return  0 : success
// return -1 : error
static int handle_control_packet(struct packet_io *handle, marsio_buff_t *rx_buff, int thread_seq, void *ctx)
{
    struct acceptor_thread_ctx *thread = (struct acceptor_thread_ctx *)ctx;
    struct acceptor_ctx *acceptor_ctx = thread->ref_acceptor_ctx;
    struct global_metrics *g_metrics = thread->ref_metrics;

    struct metadata meta;
    if (packet_io_get_metadata(rx_buff, &meta) == -1)
    {
        TFE_LOG_ERROR(g_default_logger, "%s: unexpected control packet, unable to get metadata", LOG_TAG_PKTIO);
        packet_io_dump_metadata(rx_buff, &meta);
        return -1;
    }
    packet_io_dump_metadata(rx_buff, &meta);

    struct ctrl_pkt_parser ctrl_parser;
    ctrl_packet_parser_init(&ctrl_parser);
    if (ctrl_packet_parser_parse(&ctrl_parser, meta.raw_data + meta.l7_offset, meta.raw_len - meta.l7_offset) == -1)
    {
        // TFE_LOG_ERROR(g_default_logger, "%s: unexpected control packet, unable to parse data", LOG_TAG_PKTIO);
        return -1;
    }
    ctrl_packet_parser_dump(&ctrl_parser);

    if (ctrl_parser.session_id != meta.session_id)
    {
        TFE_LOG_ERROR(g_default_logger, "%s: unexpected control packet, metadata's session %lu != control packet's session %lu", LOG_TAG_PKTIO, meta.session_id, ctrl_parser.session_id);
        return -1;
    }

    switch (ctrl_parser.state)
    {
    case SESSION_STATE_OPENING:
        __atomic_fetch_add(&g_metrics->ctrl_pkt_opening_num, 1, __ATOMIC_RELAXED);
        // when session opening, firewall not send policy id
        // return handle_session_opening(&meta, &ctrl_parser, thread_seq, ctx);
        break;
    case SESSION_STATE_CLOSING:
        __atomic_fetch_add(&g_metrics->ctrl_pkt_closing_num, 1, __ATOMIC_RELAXED);
        return handle_session_closing(&meta, &ctrl_parser, thread_seq, ctx);
    case SESSION_STATE_ACTIVE:
        __atomic_fetch_add(&g_metrics->ctrl_pkt_active_num, 1, __ATOMIC_RELAXED);
        return handle_session_active(&meta, &ctrl_parser, thread_seq, ctx);
    case SESSION_STATE_RESETALL:
        __atomic_fetch_add(&g_metrics->ctrl_pkt_resetall_num, 1, __ATOMIC_RELAXED);
        return handle_session_resetall(&meta, &ctrl_parser, thread_seq, ctx);
    default:
        __atomic_fetch_add(&g_metrics->ctrl_pkt_error_num, 1, __ATOMIC_RELAXED);
        break;
    }

    return 0;
}

static int handle_raw_packet_from_nf(struct packet_io *handle, marsio_buff_t *rx_buff, int thread_seq, void *ctx, int *action_bytes)
{
    int nsend = 0;
    struct acceptor_thread_ctx *thread = (struct acceptor_thread_ctx *)ctx;
    struct acceptor_ctx *acceptor_ctx = thread->ref_acceptor_ctx;
    struct addr_tuple4 inner_addr;
    memset(&inner_addr, 0, sizeof(struct addr_tuple4));

    int raw_len = marsio_buff_datalen(rx_buff);
    char *raw_data = marsio_buff_mtod(rx_buff);
    *action_bytes = 0;

    struct metadata meta;
    if (packet_io_get_metadata(rx_buff, &meta) == -1)
    {
        TFE_LOG_ERROR(g_default_logger, "%s: unexpected raw packet, unable to get metadata, bypass !!!", LOG_TAG_PKTIO);
        packet_io_dump_metadata(rx_buff, &meta);
        marsio_send_burst(handle->dev_nf_interface.mr_path, thread_seq, &rx_buff, 1);
        *action_bytes = raw_len;
        return RAW_PKT_ERR_BYPASS;
    }

    struct session_node *node = session_table_search_by_id(thread->session_table, meta.session_id);
    if (node == NULL)
    {
        TFE_LOG_ERROR(g_default_logger, "%s: unexpected raw packet, unable to find session %lu from session table, bypass !!!", LOG_TAG_PKTIO, meta.session_id);
        marsio_send_burst(handle->dev_nf_interface.mr_path, thread_seq, &rx_buff, 1);
        *action_bytes = raw_len;
        return RAW_PKT_ERR_BYPASS;
    }

    struct session_ctx *s_ctx = (struct session_ctx *)node->val_data;

    // update sids
    if (meta.dir_is_e2i)
    {
        sids_write_once(&(s_ctx->raw_pkt_e2i_sids), &(meta.sids));
        if (route_ctx_is_empty(&s_ctx->raw_pkt_e2i_route_ctx))
        {
            route_ctx_copy(&s_ctx->raw_pkt_e2i_route_ctx, &meta.route_ctx);
        }
    }
    else
    {
        sids_write_once(&(s_ctx->raw_pkt_i2e_sids), &(meta.sids));
        if (route_ctx_is_empty(&s_ctx->raw_pkt_i2e_route_ctx))
        {
            route_ctx_copy(&s_ctx->raw_pkt_i2e_route_ctx, &meta.route_ctx);
        }
    }

    struct raw_pkt_parser raw_parser;
    raw_packet_parser_init(&raw_parser, meta.session_id, LAYER_TYPE_ALL, 8);
    const void *payload = raw_packet_parser_parse(&raw_parser, (const void *)meta.raw_data, meta.raw_len);

    if (meta.user_field && TRAFFIC_IS_DECRYPTED)
    {
        // c2s
        if (memcmp(&inner_addr, &s_ctx->first_ctrl_pkt.tuple4, sizeof(struct addr_tuple4)) == 0) {
            add_ether_header(raw_data, acceptor_ctx->config->src_mac, acceptor_ctx->config->tap_s_mac);
            if (acceptor_ctx->config->enable_iouring) {
                io_uring_submit_write_entry(thread->tap_ctx->io_uring_s, raw_data, raw_len);
            }
            else {
                tfe_tap_write_per_thread(thread->tap_ctx->tap_s, raw_data, raw_len, g_default_logger);
            }
        }
        // s2c
        else {
            add_ether_header(raw_data, acceptor_ctx->config->src_mac, acceptor_ctx->config->tap_c_mac);
            if (acceptor_ctx->config->enable_iouring) {
                io_uring_submit_write_entry(thread->tap_ctx->io_uring_c, raw_data, raw_len);
            }
            else {
                tfe_tap_write_per_thread(thread->tap_ctx->tap_c, raw_data, raw_len, g_default_logger);
            }
        }
    }
    else 
    {
        add_ether_header(raw_data, acceptor_ctx->config->src_mac, acceptor_ctx->config->tap_mac);
        // tap0
        if (acceptor_ctx->config->enable_iouring) {
            io_uring_submit_write_entry(thread->tap_ctx->io_uring_fd, raw_data, raw_len);
        }
        else {
            tfe_tap_write_per_thread(thread->tap_ctx->tap_fd, raw_data, raw_len, g_default_logger);
        }
    }
}

struct tcp_option_mss {
	uint8_t kind;
	uint8_t length;
	uint16_t mss_value;
} __attribute__((__packed__));

struct tcp_option_window_scale {
	uint8_t kind;
	uint8_t length;
	uint8_t shift_count;
} __attribute__((__packed__));

struct tcp_option_sack {
	uint8_t kind;
	uint8_t length;
} __attribute__((__packed__));

struct tcp_option_time_stamp {
	uint8_t kind;
	uint8_t length;
	uint32_t tsval;
	uint32_t tsecr;
} __attribute__((__packed__));

static int fake_tcp_handshake(struct tfe_proxy *proxy, struct tcp_restore_info *restore_info)
{
    char buffer[1500] = {0};
    int length = 0;

	char tcp_option_buffer_c[40] = {0};
	char tcp_option_buffer_s[40] = {0};
	char tcp_option_buffer_c2[40] = {0};
	int tcp_option_length_c = 0;
	int tcp_option_length_s = 0;
	int tcp_option_length_c2 = 0;

	const struct tcp_restore_endpoint *client = &restore_info->client;
	const struct tcp_restore_endpoint *server = &restore_info->server;
    struct raw_socket *raw_socket_c = raw_socket_create(proxy->traffic_steering_options.device_client, proxy->traffic_steering_options.so_mask_client);
    struct raw_socket *raw_socket_s = raw_socket_create(proxy->traffic_steering_options.device_server, proxy->traffic_steering_options.so_mask_server);
    if (raw_socket_c == NULL || raw_socket_s == NULL)
    {
    	raw_socket_destory(raw_socket_c);
    	raw_socket_destory(raw_socket_s);

    	return -1;
    }

	uint32_t c_seq = client->seq - 1;
	uint32_t s_seq = server->seq - 1;

	/*
	 * Maximum segment size: Kind: 2, Length: 4
	 * +---------+---------+---------+
	 * | Kind=2  |Length=4 |mss.value|
	 * +---------+---------+---------+
	 *      1         1         2
	 */
	if (client->mss && server->mss)
	{
		struct tcp_option_mss *option_c = (struct tcp_option_mss *)(tcp_option_buffer_c + tcp_option_length_c);
		option_c->kind = 2;
		option_c->length = 4;
		option_c->mss_value = htons(client->mss);
		tcp_option_length_c += sizeof(struct tcp_option_mss);

		struct tcp_option_mss *option_s = (struct tcp_option_mss *)(tcp_option_buffer_s + tcp_option_length_s);
		option_s->kind = 2;
		option_s->length = 4;
		option_s->mss_value = htons(server->mss);
		tcp_option_length_s += sizeof(struct tcp_option_mss);
	}

	/*
	 * Window Scale option: Kind: 3, Length: 3
	 * +---------+---------+---------+
	 * | Kind=3  |Length=3 |shift.cnt|
	 * +---------+---------+---------+
	 *      1         1         1
	 */
	if (client->wscale_perm && server->wscale_perm)
	{
		// padding
		memset(tcp_option_buffer_c + tcp_option_length_c, 1, 1);
		tcp_option_length_c += 1;
		memset(tcp_option_buffer_s + tcp_option_length_s, 1, 1);
		tcp_option_length_s += 1;

		struct tcp_option_window_scale *option_c = (struct tcp_option_window_scale *)(tcp_option_buffer_c + tcp_option_length_c);
		option_c->kind = 3;
		option_c->length = 3;
		option_c->shift_count = client->wscale;
		tcp_option_length_c += sizeof(struct tcp_option_window_scale);

		struct tcp_option_window_scale *option_s = (struct tcp_option_window_scale *)(tcp_option_buffer_s + tcp_option_length_s);
		option_s->kind = 3;
		option_s->length = 3;
		option_s->shift_count = server->wscale;
		tcp_option_length_s += sizeof(struct tcp_option_window_scale);
	}

	/*
	 * SACK option: Kind: 4, Length: 2
	 * +---------+---------+
	 * | Kind=4  |Length=2 |
	 * +---------+---------+
	 *      1         1
	 */
	if (client->sack_perm && server->sack_perm)
	{
		// padding
		memset(tcp_option_buffer_c + tcp_option_length_c, 1, 2);
		tcp_option_length_c += 2;
		memset(tcp_option_buffer_s + tcp_option_length_s, 1, 2);
		tcp_option_length_s += 2;

		struct tcp_option_sack *option_c = (struct tcp_option_sack *)(tcp_option_buffer_c + tcp_option_length_c);
		option_c->kind = 4;
		option_c->length = 2;
		tcp_option_length_c += sizeof(struct tcp_option_sack);

		struct tcp_option_sack *option_s = (struct tcp_option_sack *)(tcp_option_buffer_s + tcp_option_length_s);
		option_s->kind = 4;
		option_s->length = 2;
		tcp_option_length_s += sizeof(struct tcp_option_sack);
	}

	/*
	 * Time Stamp option: Kind: 8, Length: 10
	 * +---------+---------+-----+-----+
	 * | Kind=8  |Length=10|tsval|tsecr|
	 * +---------+---------+-----+-----+
	 *      1         1       4     4
	 */
	if (client->timestamp_perm && server->timestamp_perm)
	{
		// padding
		memset(tcp_option_buffer_c + tcp_option_length_c, 1, 2);
		tcp_option_length_c += 2;
		memset(tcp_option_buffer_s + tcp_option_length_s, 1, 2);
		tcp_option_length_s += 2;
		memset(tcp_option_buffer_c2 + tcp_option_length_c2, 1, 2);
		tcp_option_length_c2 += 2;

		struct tcp_option_time_stamp *option_c = (struct tcp_option_time_stamp *)(tcp_option_buffer_c + tcp_option_length_c);
		option_c->kind = 8;
		option_c->length = 10;
		option_c->tsval = htonl(client->ts_val);
		option_c->tsecr = htonl(0);
		tcp_option_length_c += sizeof(struct tcp_option_time_stamp);

		struct tcp_option_time_stamp *option_s = (struct tcp_option_time_stamp *)(tcp_option_buffer_s + tcp_option_length_s);
		option_s->kind = 8;
		option_s->length = 10;
		option_s->tsval = htonl(server->ts_val);
		option_s->tsecr = htonl(client->ts_val);
		tcp_option_length_s += sizeof(struct tcp_option_time_stamp);

		struct tcp_option_time_stamp *option_c2 = (struct tcp_option_time_stamp *)(tcp_option_buffer_c2 + tcp_option_length_c2);
		option_c2->kind = 8;
		option_c2->length = 10;
		option_c2->tsval = htonl(client->ts_val);
		option_c2->tsecr = htonl(server->ts_val);
		tcp_option_length_c2 += sizeof(struct tcp_option_time_stamp);
	}

	if (client->addr.ss_family == AF_INET6)
	{
		struct sockaddr_in6 *sk_client = (struct sockaddr_in6 *)&client->addr;
		struct sockaddr_in6 *sk_server = (struct sockaddr_in6 *)&server->addr;
		uint16_t port_client = sk_client->sin6_port;
		uint16_t port_server = sk_server->sin6_port;

		// C -> S
		length = tcp_packet_v6_construct(
       		buffer, // buffer
       	 	&raw_socket_c->mac_addr, &raw_socket_s->mac_addr, 0, ETH_P_IPV6, // Ether
        	&sk_client->sin6_addr, &sk_server->sin6_addr, 55, // IPv6
			port_client, port_server, c_seq, 0, TCP_SYN_FLAG, client->window, // TCP Header
			tcp_option_buffer_c, tcp_option_length_c, // TCP Options
			NULL, 0); // Payload
		raw_socket_send(raw_socket_c, buffer, length);
		c_seq += 1;

		// S -> C
		length = tcp_packet_v6_construct(
       		buffer, // buffer
       	 	&raw_socket_s->mac_addr, &raw_socket_c->mac_addr, 0, ETH_P_IPV6, // Ether
        	&sk_server->sin6_addr, &sk_client->sin6_addr,  65, // IPv6
			port_server, port_client, s_seq, c_seq, TCP_SYN_FLAG | TCP_ACK_FLAG, server->window, // TCP Header
			tcp_option_buffer_s, tcp_option_length_s, // TCP Options
			NULL, 0); // Payload
		raw_socket_send(raw_socket_s, buffer, length);
		s_seq += 1;

		// C -> S
		length = tcp_packet_v6_construct(
       		buffer, // buffer
       	 	&raw_socket_c->mac_addr, &raw_socket_s->mac_addr, 0, ETH_P_IPV6, // Ether
        	&sk_client->sin6_addr, &sk_server->sin6_addr, 55,  // IPv6
			port_client, port_server, c_seq, s_seq, TCP_ACK_FLAG, client->window, // TCP Header
			tcp_option_buffer_c2, tcp_option_length_c2, // TCP Options
			NULL, 0); // Payload
		raw_socket_send(raw_socket_c, buffer, length);
	}
	else
	{
		struct sockaddr_in *sk_client = (struct sockaddr_in *)&client->addr;
		struct sockaddr_in *sk_server = (struct sockaddr_in *)&server->addr;
		uint16_t port_client = sk_client->sin_port;
		uint16_t port_server = sk_server->sin_port;

		// C -> S
		length = tcp_packet_v4_construct(
			buffer,                                                               // buffer
			&raw_socket_c->mac_addr, &raw_socket_s->mac_addr, 0, ETH_P_IP,        // Ether
			&sk_client->sin_addr, &sk_server->sin_addr, 0, 55, 0x11,              // IPv4
			port_client, port_server, c_seq, 0, TCP_SYN_FLAG, client->window, // TCP Header
			tcp_option_buffer_c, tcp_option_length_c, // TCP Options
			NULL, 0);
		raw_socket_send(raw_socket_c, buffer, length);
		c_seq += 1;

		// S -> C
		length = tcp_packet_v4_construct(
			buffer,                                                                              // buffer
			&raw_socket_s->mac_addr, &raw_socket_c->mac_addr, 0, ETH_P_IP,                       // Ether
			&sk_server->sin_addr,&sk_client->sin_addr, 0, 65, 0x12,                              // IPv4
			port_server, port_client, s_seq, c_seq, TCP_SYN_FLAG | TCP_ACK_FLAG, server->window, // TCP Header
			tcp_option_buffer_s, tcp_option_length_s, // TCP Options
			NULL, 0);
		raw_socket_send(raw_socket_s, buffer, length);
		s_seq += 1;

		// C -> S
		length = tcp_packet_v4_construct(
			buffer, // buffer
			&raw_socket_c->mac_addr, &raw_socket_s->mac_addr, 0, ETH_P_IP, // Ether
			&sk_client->sin_addr, &sk_server->sin_addr, 0, 55, 0x13, // IPv4
			port_client, port_server, c_seq, s_seq, TCP_ACK_FLAG, client->window, // TCP Header
			tcp_option_buffer_c2, tcp_option_length_c2, // TCP Options
			NULL, 0);
		raw_socket_send(raw_socket_c, buffer, length);
	}

    raw_socket_destory(raw_socket_c);
    raw_socket_destory(raw_socket_s);
    
	return 0;
}

static int overwrite_tcp_mss(struct tfe_cmsg *cmsg, struct tcp_restore_info *restore)
{
	int ret = 0;
	uint16_t size = 0;
	int server_side_mss_enable = 0;
	int server_side_mss_value = 0;
	int client_side_mss_enable = 0;
	int client_side_mss_value = 0;

	ret = tfe_cmsg_get_value(cmsg, TFE_CMSG_DOWNSTREAM_TCP_MSS_ENABLE, (unsigned char *)&client_side_mss_enable, sizeof(client_side_mss_enable), &size);
	if (ret < 0)
	{
		TFE_LOG_ERROR(g_default_logger, "failed at fetch client side tcp mss from cmsg: %s", strerror(-ret));
		return -1;
	}
	ret = tfe_cmsg_get_value(cmsg, TFE_CMSG_DOWNSTREAM_TCP_MSS_VALUE, (unsigned char *)&client_side_mss_value, sizeof(client_side_mss_value), &size);
	if (ret < 0)
	{
		TFE_LOG_ERROR(g_default_logger, "failed at fetch client side tcp mss value from cmsg: %s", strerror(-ret));
		return -1;
	}
	ret = tfe_cmsg_get_value(cmsg, TFE_CMSG_UPSTREAM_TCP_MSS_ENABLE, (unsigned char *)&server_side_mss_enable, sizeof(server_side_mss_enable), &size);
	if (ret < 0)
	{
		TFE_LOG_ERROR(g_default_logger, "failed at fetch server side tcp mss from cmsg: %s", strerror(-ret));
		return -1;
	}
	ret = tfe_cmsg_get_value(cmsg, TFE_CMSG_UPSTREAM_TCP_MSS_VALUE, (unsigned char *)&server_side_mss_value, sizeof(server_side_mss_value), &size);
	if (ret < 0)
	{
		TFE_LOG_ERROR(g_default_logger, "failed at fetch server side tcp mss value from cmsg: %s", strerror(-ret));
		return -1;
	}
	if (client_side_mss_enable)
	{
		restore->client.mss = client_side_mss_value;
	}
	if (server_side_mss_enable)
	{
		restore->server.mss = server_side_mss_value;
	}

	return 0;
}

// return  0 : success
// return -1 : error
static int handle_session_opening(struct metadata *meta, struct ctrl_pkt_parser *parser, int thread_seq, void *ctx)
{
    uint8_t *iptmp = NULL;
    int ret = 0;
    int fd_downstream = 0;
    int fd_upstream = 0;
    int fd_fake_c = 0;
    int fd_fake_s = 0;
    uint64_t rule_id = 0;
    uint16_t size = 0;

    uint8_t stream_protocol_in_char = 0;
    uint8_t enalbe_decrypted_traffic_steering = 0;
    struct session_ctx *s_ctx = NULL;
    struct addr_tuple4 tuple4;
    struct tcp_restore_info restore_info;
    memset(&tuple4, 0, sizeof(tuple4));
    memset(&restore_info, 0, sizeof(restore_info));

    struct sockaddr_in *in_addr_client = (struct sockaddr_in *)&restore_info.client.addr;
    struct sockaddr_in *in_addr_server = (struct sockaddr_in *)&restore_info.server.addr;

    struct acceptor_thread_ctx *thread = (struct acceptor_thread_ctx *)ctx;
    struct acceptor_ctx *acceptor_ctx = thread->ref_acceptor_ctx;

    struct raw_pkt_parser raw_parser;
    raw_packet_parser_init(&raw_parser, meta->session_id, LAYER_TYPE_ALL, 8);
    const void *payload = raw_packet_parser_parse(&raw_parser, (const void *)meta->raw_data, meta->raw_len);
    if ((char *)payload - meta->raw_data != meta->l7_offset)
    {
        uint16_t offset = (char *)payload - meta->raw_data;
        TFE_LOG_ERROR(g_default_logger, "%s: incorrect dataoffset in the control zone of session %lu, offset:%u, l7_offset:%u, payload:%p, raw_data:%p", LOG_TAG_PKTIO, meta->session_id, offset, meta->l7_offset, payload, meta->raw_data);
    }

    raw_packet_parser_get_most_inner_tuple4(&raw_parser, &tuple4);

    ret = tfe_cmsg_get_value(parser->cmsg, TFE_CMSG_POLICY_ID, (unsigned char *)&rule_id, sizeof(rule_id), &size);
	if (ret < 0)
	{
		TFE_LOG_ERROR(g_default_logger, "failed at fetch rule_id from cmsg: %s", strerror(-ret));
		goto end;
	}

    intercept_policy_enforce(thread->ref_proxy->int_ply_enforcer, parser->cmsg);
	tcp_policy_enforce(thread->ref_proxy->tcp_ply_enforcer, parser->cmsg);
    for (int i = 0; i < parser->sce_policy_id_num; i++) {
	    chaining_policy_enforce(thread->ref_proxy->chain_ply_enforcer, parser->cmsg, parser->sce_policy_ids[i]);
    }

    tcp_restore_set_from_cmsg(parser->cmsg, &restore_info);
    tcp_restore_set_from_pkg(&tuple4, &restore_info);

    if (overwrite_tcp_mss(parser->cmsg, &restore_info))
	{
		goto end;
	}

    iptmp = (uint8_t *)&in_addr_client->sin_addr.s_addr;
	// tcp repair C2S
    TFE_LOG_DEBUG(g_default_logger, "restore_info: client");

    TFE_LOG_DEBUG(g_default_logger, "\t addr:%d.%d.%d.%d", iptmp[0], iptmp[1], iptmp[2], iptmp[3]);
    TFE_LOG_DEBUG(g_default_logger, "\t port:%u", in_addr_client->sin_port);
    TFE_LOG_DEBUG(g_default_logger, "\t seq:%u", restore_info.client.seq);
    TFE_LOG_DEBUG(g_default_logger, "\t ack:%u", restore_info.client.ack);
    TFE_LOG_DEBUG(g_default_logger, "\t ts_val:%u", restore_info.client.ts_val);
    TFE_LOG_DEBUG(g_default_logger, "\t mss:%u", restore_info.client.mss);
    TFE_LOG_DEBUG(g_default_logger, "\t window:%u", restore_info.client.window);
    TFE_LOG_DEBUG(g_default_logger, "\t wscale:%u", restore_info.client.wscale);
    TFE_LOG_DEBUG(g_default_logger, "\t wscale_perm:%s", restore_info.client.wscale_perm > 0?"true":"false");
    TFE_LOG_DEBUG(g_default_logger, "\t timestamp_perm:%s", restore_info.client.timestamp_perm > 0?"true":"false");
    TFE_LOG_DEBUG(g_default_logger, "\t sack_perm:%s", restore_info.client.sack_perm > 0?"true":"false");


    iptmp = (uint8_t *)&in_addr_server->sin_addr.s_addr;
	// tcp repair C2S
    TFE_LOG_DEBUG(g_default_logger, "restore_info: server");

    TFE_LOG_DEBUG(g_default_logger, "\t addr:%d.%d.%d.%d", iptmp[0], iptmp[1], iptmp[2], iptmp[3]);
    TFE_LOG_DEBUG(g_default_logger, "\t port:%u", in_addr_server->sin_port);
    TFE_LOG_DEBUG(g_default_logger, "\t seq:%u", restore_info.server.seq);
    TFE_LOG_DEBUG(g_default_logger, "\t ack:%u", restore_info.server.ack);
    TFE_LOG_DEBUG(g_default_logger, "\t ts_val:%u", restore_info.server.ts_val);
    TFE_LOG_DEBUG(g_default_logger, "\t mss:%u", restore_info.server.mss);
    TFE_LOG_DEBUG(g_default_logger, "\t window:%u", restore_info.server.window);
    TFE_LOG_DEBUG(g_default_logger, "\t wscale:%u", restore_info.server.wscale);
    TFE_LOG_DEBUG(g_default_logger, "\t wscale_perm:%s", restore_info.server.wscale_perm > 0?"true":"false");
    TFE_LOG_DEBUG(g_default_logger, "\t timestamp_perm:%s", restore_info.server.timestamp_perm > 0?"true":"false");
    TFE_LOG_DEBUG(g_default_logger, "\t sack_perm:%s", restore_info.server.sack_perm > 0?"true":"false");


	fd_upstream = tfe_tcp_restore_fd_create(&(restore_info.client), &(restore_info.server), thread->ref_tap_config->tap_device, 0x65);
	if (fd_upstream < 0)
	{
		TFE_LOG_ERROR(g_default_logger, "Failed at tcp_restore_fd_create(UPSTREAM)");
		goto end;
	}

	// tcp repair S2C
	fd_downstream = tfe_tcp_restore_fd_create(&(restore_info.server), &(restore_info.client), thread->ref_tap_config->tap_device, 0x65);
	if (fd_downstream < 0)
	{
		TFE_LOG_ERROR(g_default_logger, "Failed at tcp_restore_fd_create(DOWNSTREAM)");
		goto end;
	}

    tfe_cmsg_get_value(parser->cmsg, TFE_CMSG_TCP_RESTORE_PROTOCOL, (unsigned char *)&stream_protocol_in_char, sizeof(stream_protocol_in_char), &size);
    tfe_cmsg_get_value(parser->cmsg, TFE_CMSG_TCP_DECRYPTED_TRAFFIC_STEERING, (unsigned char *)&enalbe_decrypted_traffic_steering, sizeof(enalbe_decrypted_traffic_steering), &size);
    if ((STREAM_PROTO_PLAIN == (enum tfe_stream_proto)stream_protocol_in_char && thread->ref_proxy->traffic_steering_options.enable_steering_http) || 
            (STREAM_PROTO_SSL == (enum tfe_stream_proto)stream_protocol_in_char && thread->ref_proxy->traffic_steering_options.enable_steering_ssl) || 
            enalbe_decrypted_traffic_steering == 1)
	{
		if (fake_tcp_handshake(thread->ref_proxy, &restore_info) == -1)
		{
			TFE_LOG_ERROR(g_default_logger, "Failed at fake_tcp_handshake()");
			goto end;
		}

		fd_fake_c = tfe_tcp_restore_fd_create(&(restore_info.client), &(restore_info.server), thread->ref_proxy->traffic_steering_options.device_client, thread->ref_proxy->traffic_steering_options.so_mask_client);
		if (fd_fake_c < 0)
		{
			TFE_LOG_ERROR(g_default_logger, "Failed at tcp_restore_fd_create(fd_fake_c)");
			goto end;
		}

		fd_fake_s = tfe_tcp_restore_fd_create(&(restore_info.server), &(restore_info.client), thread->ref_proxy->traffic_steering_options.device_server, thread->ref_proxy->traffic_steering_options.so_mask_server);
		if (fd_fake_s < 0)
		{
			TFE_LOG_ERROR(g_default_logger, "Failed at tcp_restore_fd_create(fd_fake_s)");
			goto end;
		}
	}

    if (tfe_proxy_fds_accept(thread->ref_proxy, fd_downstream, fd_upstream, fd_fake_c, fd_fake_s, parser->cmsg) < 0)
	{
		TFE_LOG_ERROR(g_default_logger, "Failed at tfe_proxy_fds_accept()");
		goto end;
	}

    s_ctx = session_ctx_new();
    s_ctx->ref_thread_ctx = thread;
    s_ctx->session_id = meta->session_id;
    s_ctx->cmsg = parser->cmsg;
    s_ctx->first_ctrl_pkt.dir_is_e2i = meta->dir_is_e2i;
    raw_packet_parser_get_most_inner_tuple4(&raw_parser, &(s_ctx->first_ctrl_pkt.tuple4));
    s_ctx->first_ctrl_pkt.addr_string = addr_tuple4_to_str(&(s_ctx->first_ctrl_pkt.tuple4));
    s_ctx->first_ctrl_pkt.header_data = (char *)calloc(1, meta->l7_offset);
    memcpy(s_ctx->first_ctrl_pkt.header_data, meta->raw_data, meta->l7_offset);
    s_ctx->first_ctrl_pkt.header_len = meta->l7_offset;
    sids_copy(&s_ctx->first_ctrl_pkt.sids, &meta->sids);
    route_ctx_copy(&s_ctx->first_ctrl_pkt.route_ctx, &meta->route_ctx);

    TFE_LOG_INFO(g_default_logger, "%s: session %lu %s active first", LOG_TAG_PKTIO, s_ctx->session_id, s_ctx->first_ctrl_pkt.addr_string);
    s_ctx->policy_ids = parser->tfe_policy_ids[0];

    session_table_insert(thread->session_table, s_ctx->session_id, &(s_ctx->first_ctrl_pkt.tuple4), s_ctx, session_value_free_cb);

    return 0;
end:
    return -1;
}

/*
{
  "tsync": "1.0",
  "session_id": "123456789",
  "state": "active",
  "method": "log_update",
  "params": {
    "sf_profile_ids": [
      2,
      3,
      4,
      5,
      6,
      7
    ]
  }
}
*/
static void send_event_log(struct session_ctx *s_ctx, int thread_seq, void *ctx)
{
    struct acceptor_thread_ctx *thread = (struct acceptor_thread_ctx *)ctx;
    struct acceptor_ctx *acceptor_ctx = thread->ref_acceptor_ctx;
    struct packet_io *packet_io = thread->ref_io;

    char buffer[32] = {0};
    sprintf(buffer, "%lu", s_ctx->session_id);

    cJSON *root = cJSON_CreateObject();
    cJSON_AddStringToObject(root, "tsync", "1.0");
    cJSON_AddStringToObject(root, "session_id", buffer);
    cJSON_AddStringToObject(root, "state", "closing");
    cJSON_AddStringToObject(root, "method", "log_update");
    cJSON *sf_profile_ids = cJSON_CreateArray();

    cJSON *params = cJSON_CreateObject();
    cJSON_AddItemToObject(params, "sf_profile_ids", sf_profile_ids);
    cJSON_AddItemToObject(root, "params", params);
    char *json_str = cJSON_PrintUnformatted(root);

    TFE_LOG_INFO(g_default_logger, "%s: session %lu %s event log: %s", LOG_TAG_METRICS, s_ctx->session_id, s_ctx->first_ctrl_pkt.addr_string, json_str);

    marsio_buff_t *tx_buffs[1];
    marsio_buff_malloc_device(packet_io->dev_nf_interface.mr_dev, tx_buffs, 1, 0, thread_seq);
    char *dst = marsio_buff_append(tx_buffs[0], s_ctx->first_ctrl_pkt.header_len + strlen(json_str));
    memcpy(dst, s_ctx->first_ctrl_pkt.header_data, s_ctx->first_ctrl_pkt.header_len);
    memcpy(dst + s_ctx->first_ctrl_pkt.header_len, json_str, strlen(json_str));

    struct metadata meta = {0};
    meta.session_id = s_ctx->session_id;
    meta.is_ctrl_pkt = 1;
    meta.l7_offset = s_ctx->first_ctrl_pkt.header_len;
    meta.sids.num = 1;
    meta.sids.elems[0] = acceptor_ctx->firewall_sids;
    route_ctx_copy(&meta.route_ctx, &s_ctx->first_ctrl_pkt.route_ctx);
    packet_io_set_metadata(tx_buffs[0], &meta);
    marsio_send_burst(packet_io->dev_nf_interface.mr_path, thread_seq, tx_buffs, 1);

    free(json_str);
    cJSON_Delete(root);
}

// return  0 : success
// return -1 : error
static int handle_session_closing(struct metadata *meta, struct ctrl_pkt_parser *parser, int thread_seq, void *ctx)
{
    struct acceptor_thread_ctx *thread = (struct acceptor_thread_ctx *)ctx;

    struct session_node *node = session_table_search_by_id(thread->session_table, meta->session_id);
    if (node)
    {
        struct session_ctx *s_ctx = (struct session_ctx *)node->val_data;
        TFE_LOG_INFO(g_default_logger, "%s: session %lu %s closing", LOG_TAG_PKTIO, s_ctx->session_id, s_ctx->first_ctrl_pkt.addr_string);

        send_event_log(s_ctx, thread_seq, ctx);

        session_table_delete_by_id(thread->session_table, meta->session_id);
        return 0;
    }

    return -1;
}

// return  0 : success
// return -1 : error
static int handle_session_active(struct metadata *meta, struct ctrl_pkt_parser *parser, int thread_seq, void *ctx)
{
    struct acceptor_thread_ctx *thread = (struct acceptor_thread_ctx *)ctx;

    struct session_node *node = session_table_search_by_id(thread->session_table, meta->session_id);
    if (node)
    {
        struct raw_pkt_parser raw_parser;
        raw_packet_parser_init(&raw_parser, meta->session_id, LAYER_TYPE_ALL, 8);
        const void *payload = raw_packet_parser_parse(&raw_parser, (const void *)meta->raw_data, meta->raw_len);
        if ((char *)payload - (char *)&meta->raw_data != meta->l7_offset)
        {
            TFE_LOG_ERROR(g_default_logger, "%s: incorrect dataoffset in the control zone of session %lu", LOG_TAG_PKTIO, meta->session_id);
        }

        struct session_ctx *s_ctx = (struct session_ctx *)node->val_data;
        TFE_LOG_INFO(g_default_logger, "%s: session %lu %s active again", LOG_TAG_PKTIO, s_ctx->session_id, s_ctx->first_ctrl_pkt.addr_string);
        s_ctx->policy_ids = parser->tfe_policy_ids[0];
    }
    else
    {
        return handle_session_opening(meta, parser, thread_seq, ctx);
    }

    return 0;
}

// return  0 : success
// return -1 : error
static int handle_session_resetall(struct metadata *meta, struct ctrl_pkt_parser *parser, int thread_seq, void *ctx)
{
    struct acceptor_thread_ctx *thread = (struct acceptor_thread_ctx *)ctx;
    struct acceptor_ctx *acceptor_ctx = thread->ref_acceptor_ctx;

    TFE_LOG_ERROR(g_default_logger, "%s: session %lu resetall: notification clears all session tables !!!", LOG_TAG_PKTIO, meta->session_id);

    for (int i = 0; i < acceptor_ctx->nr_worker_threads; i++)
    {
        struct acceptor_thread_ctx *thread_ctx = &acceptor_ctx->work_threads[i];
        __atomic_fetch_add(&thread_ctx->session_table_need_reset, 1, __ATOMIC_RELAXED);
    }

    return 0;
}

static void session_value_free_cb(void *ctx)
{
    struct session_ctx *s_ctx = (struct session_ctx *)ctx;
    session_ctx_free(s_ctx);
}

// return 0 : not keepalive packet
// return 1 : is  keepalive packet
static int is_downstream_keepalive_packet(marsio_buff_t *rx_buff)
{
    int raw_len = marsio_buff_datalen(rx_buff);
    char *raw_data = marsio_buff_mtod(rx_buff);
    if (raw_data == NULL || raw_len < (int)(sizeof(struct ethhdr)))
    {
        return 0;
    }

    struct ethhdr *eth_hdr = (struct ethhdr *)raw_data;
    if (eth_hdr->h_proto == 0xAAAA)
    {
        return 1;
    }
    else
    {
        return 0;
    }
}