#include #include #include #include #include #include #include "log.h" #include "sce.h" #include "g_vxlan.h" #include "ctrl_packet.h" /* * add: vxlan_hdr * del: marsio_buff_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 struct config { int bypass_all_traffic; int rx_burst_max; char app_symbol[256]; char dev_endpoint[256]; char dev_nf_interface[256]; char default_src_ip[16]; char default_dst_ip[16]; char default_src_mac[32]; char default_dst_mac[32]; }; 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 device dev_endpoint; struct config config; }; enum raw_pkt_action { RAW_PKT_ERR_BYPASS, RAW_PKT_HIT_BYPASS, RAW_PKT_HIT_BLOCK, RAW_PKT_HIT_FORWARD, }; enum inject_pkt_action { INJT_PKT_ERR_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; int l7_offset; // only control packet set l7_offset int traffic_is_decrypted; // only raw packet set traffic_is_decrypted struct sids sids; char route_ctx[64]; }; /****************************************************************************** * API Declaration ******************************************************************************/ struct packet_io *packet_io_create(const char *profile, int thread_num); void packet_io_destory(struct packet_io *handle); int packet_io_polling_nf_interface(struct packet_io *handle, int thread_seq, void *ctx); int packet_io_polling_endpoint(struct packet_io *handle, int thread_seq, void *ctx); 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); // 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); // return : RAW_PKT_ERR_BYPASS // return : RAW_PKT_HIT_BYPASS // return : RAW_PKT_HIT_BLOCK // reutrn : RAW_PKT_HIT_FORWARD static enum raw_pkt_action handle_raw_packet(struct packet_io *handle, marsio_buff_t *rx_buff, int thread_seq, void *ctx); // return : INJT_PKT_ERR_DROP // return : INJT_PKT_HIT_BLOCK // return : INJT_PKT_HIT_FWD2SF // return : INJT_PKT_HIT_FWD2NF static enum inject_pkt_action handle_inject_packet(struct packet_io *handle, marsio_buff_t *rx_buff, int thread_seq, void *ctx); // return 0 : success // return -1 : error static int forward_packet_to_sf(struct packet_io *handle, marsio_buff_t *rx_buff, struct metadata *meta, int chain_index, int thread_seq, void *ctx); // return 0 : success // return -1 : error static int forward_packet_to_nf(struct packet_io *handle, marsio_buff_t *rx_buff, struct metadata *meta, int chain_index, int thread_seq, void *ct); static void forward_all_nf_packet_to_sf(struct packet_io *handle, marsio_buff_t *rx_buff, int thread_seq, void *ctx); static void forward_all_sf_packet_to_nf(struct packet_io *handle, marsio_buff_t *rx_buff, int thread_seq, void *ctx); // 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); /****************************************************************************** * API Definition ******************************************************************************/ struct packet_io *packet_io_create(const char *profile, int thread_num) { 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) { LOG_ERROR("%s: unable to create marsio instance", LOG_TAG_PKTIO); goto error_out; } if (marsio_option_set(handle->instance, MARSIO_OPT_EXIT_WHEN_ERR, &opt, sizeof(opt)) != 0) { LOG_ERROR("%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) { LOG_ERROR("%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) { LOG_ERROR("%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) { LOG_ERROR("%s: unable to create sendpath for device %s", LOG_TAG_PKTIO, handle->config.dev_nf_interface); goto error_out; } // EndPoint Interface handle->dev_endpoint.mr_dev = marsio_open_device(handle->instance, handle->config.dev_endpoint, handle->thread_num, handle->thread_num); if (handle->dev_endpoint.mr_dev == NULL) { LOG_ERROR("%s: unable to open device %s", LOG_TAG_PKTIO, handle->config.dev_endpoint); goto error_out; } handle->dev_endpoint.mr_path = marsio_sendpath_create_by_vdev(handle->dev_endpoint.mr_dev); if (handle->dev_endpoint.mr_path == NULL) { LOG_ERROR("%s: unable to create sendpath for device %s", LOG_TAG_PKTIO, handle->config.dev_endpoint); 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->dev_endpoint.mr_path) { marsio_sendpath_destory(handle->dev_endpoint.mr_path); handle->dev_endpoint.mr_path = NULL; } if (handle->dev_endpoint.mr_dev) { marsio_close_device(handle->dev_endpoint.mr_dev); handle->dev_endpoint.mr_dev = NULL; } if (handle->instance) { marsio_destory(handle->instance); handle->instance = NULL; } free(handle); handle = NULL; } } // return n_packet_recv int packet_io_polling_nf_interface(struct packet_io *handle, int thread_seq, void *ctx) { struct thread_ctx *thread = (struct 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; } if (handle->config.bypass_all_traffic == 1) { for (int j = 0; j < nr_recv; j++) { if (marsio_buff_is_ctrlbuf(rx_buffs[j])) { int raw_len = marsio_buff_datalen(rx_buffs[j]); throughput_metrics_inc(&g_metrics->dev_nf_interface_rx, 1, raw_len); throughput_metrics_inc(&g_metrics->dev_nf_interface_tx, 1, raw_len); } } marsio_send_burst(handle->dev_nf_interface.mr_path, thread_seq, rx_buffs, nr_recv); return nr_recv; } else if (handle->config.bypass_all_traffic == 2) { for (int j = 0; j < nr_recv; j++) { marsio_buff_t *rx_buff = rx_buffs[j]; int raw_len = marsio_buff_datalen(rx_buff); throughput_metrics_inc(&g_metrics->dev_nf_interface_rx, 1, raw_len); throughput_metrics_inc(&g_metrics->dev_endpoint_tx, 1, raw_len); forward_all_nf_packet_to_sf(handle, rx_buff, thread_seq, ctx); } return nr_recv; } for (int j = 0; j < nr_recv; j++) { marsio_buff_t *rx_buff = rx_buffs[j]; if (marsio_buff_is_ctrlbuf(rx_buff)) { handle_control_packet(handle, rx_buff, thread_seq, ctx); // all control packet need bypass marsio_send_burst(handle->dev_nf_interface.mr_path, thread_seq, &rx_buff, 1); } else { int raw_len = marsio_buff_datalen(rx_buff); throughput_metrics_inc(&g_metrics->dev_nf_interface_rx, 1, raw_len); enum raw_pkt_action action = handle_raw_packet(handle, rx_buff, thread_seq, ctx); switch (action) { case RAW_PKT_ERR_BYPASS: throughput_metrics_inc(&g_metrics->dev_nf_interface_err_bypass, 1, raw_len); break; case RAW_PKT_HIT_BYPASS: throughput_metrics_inc(&g_metrics->hit_bypass_policy, 1, raw_len); break; case RAW_PKT_HIT_BLOCK: throughput_metrics_inc(&g_metrics->hit_block_policy, 1, raw_len); break; case RAW_PKT_HIT_FORWARD: break; } } } return nr_recv; } // return n_packet_recv int packet_io_polling_endpoint(struct packet_io *handle, int thread_seq, void *ctx) { struct thread_ctx *thread = (struct 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_endpoint.mr_dev, thread_seq, rx_buffs, handle->config.rx_burst_max); if (nr_recv <= 0) { return 0; } if (handle->config.bypass_all_traffic == 1) { for (int j = 0; j < nr_recv; j++) { int raw_len = marsio_buff_datalen(rx_buffs[j]); throughput_metrics_inc(&g_metrics->dev_endpoint_rx, 1, raw_len); throughput_metrics_inc(&g_metrics->dev_endpoint_tx, 1, raw_len); } marsio_send_burst(handle->dev_endpoint.mr_path, thread_seq, rx_buffs, nr_recv); return nr_recv; } else if (handle->config.bypass_all_traffic == 2) { for (int j = 0; j < nr_recv; j++) { marsio_buff_t *rx_buff = rx_buffs[j]; int raw_len = marsio_buff_datalen(rx_buff); throughput_metrics_inc(&g_metrics->dev_endpoint_rx, 1, raw_len); throughput_metrics_inc(&g_metrics->dev_nf_interface_tx, 1, raw_len); forward_all_sf_packet_to_nf(handle, rx_buff, thread_seq, ctx); } return nr_recv; } for (int j = 0; j < nr_recv; j++) { marsio_buff_t *rx_buff = rx_buffs[j]; int data_len = marsio_buff_datalen(rx_buff); throughput_metrics_inc(&g_metrics->dev_endpoint_rx, 1, data_len); enum inject_pkt_action action = handle_inject_packet(handle, rx_buff, thread_seq, ctx); switch (action) { case INJT_PKT_ERR_DROP: throughput_metrics_inc(&g_metrics->dev_endpoint_err_drop, 1, data_len); break; case INJT_PKT_HIT_BLOCK: throughput_metrics_inc(&g_metrics->hit_block_policy, 1, data_len); break; case INJT_PKT_HIT_FWD2SF: // forward to next service function throughput_metrics_inc(&g_metrics->dev_endpoint_tx, 1, data_len); break; case INJT_PKT_HIT_FWD2NF: // forward to network function throughput_metrics_inc(&g_metrics->dev_nf_interface_tx, 1, data_len); break; } } return nr_recv; } // return -1 : error // return 0 : success static int packet_io_config(const char *profile, struct config *config) { MESA_load_profile_int_def(profile, "PACKET_IO", "bypass_all_traffic", (int *)&(config->bypass_all_traffic), 0); 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_endpoint", config->dev_endpoint, sizeof(config->dev_endpoint)); MESA_load_profile_string_nodef(profile, "PACKET_IO", "dev_nf_interface", config->dev_nf_interface, sizeof(config->dev_nf_interface)); MESA_load_profile_string_nodef(profile, "PACKET_IO", "default_src_ip", config->default_src_ip, sizeof(config->default_src_ip)); MESA_load_profile_string_nodef(profile, "PACKET_IO", "default_dst_ip", config->default_dst_ip, sizeof(config->default_dst_ip)); MESA_load_profile_string_nodef(profile, "PACKET_IO", "default_src_mac", config->default_src_mac, sizeof(config->default_src_mac)); MESA_load_profile_string_nodef(profile, "PACKET_IO", "default_dst_mac", config->default_dst_mac, sizeof(config->default_dst_mac)); if (config->rx_burst_max > RX_BURST_MAX) { LOG_ERROR("%s: invalid rx_burst_max, exceeds limit %d", LOG_TAG_PKTIO, RX_BURST_MAX); return -1; } if (strlen(config->app_symbol) == 0) { LOG_ERROR("%s: invalid app_symbol in %s", LOG_TAG_PKTIO, profile); return -1; } if (strlen(config->dev_endpoint) == 0) { LOG_ERROR("%s: invalid dev_endpoint in %s", LOG_TAG_PKTIO, profile); return -1; } if (strlen(config->dev_nf_interface) == 0) { LOG_ERROR("%s: invalid dev_nf_interface in %s", LOG_TAG_PKTIO, profile); return -1; } LOG_DEBUG("%s: PACKET_IO->bypass_all_traffic : %d", LOG_TAG_PKTIO, config->bypass_all_traffic); LOG_DEBUG("%s: PACKET_IO->rx_burst_max : %d", LOG_TAG_PKTIO, config->rx_burst_max); LOG_DEBUG("%s: PACKET_IO->app_symbol : %s", LOG_TAG_PKTIO, config->app_symbol); LOG_DEBUG("%s: PACKET_IO->dev_endpoint : %s", LOG_TAG_PKTIO, config->dev_endpoint); LOG_DEBUG("%s: PACKET_IO->dev_nf_interface : %s", LOG_TAG_PKTIO, config->dev_nf_interface); LOG_DEBUG("%s: PACKET_IO->default_src_ip : %s", LOG_TAG_PKTIO, config->default_src_ip); LOG_DEBUG("%s: PACKET_IO->default_dst_ip : %s", LOG_TAG_PKTIO, config->default_dst_ip); LOG_DEBUG("%s: PACKET_IO->default_src_mac : %s", LOG_TAG_PKTIO, config->default_src_mac); LOG_DEBUG("%s: PACKET_IO->default_dst_mac : %s", LOG_TAG_PKTIO, config->default_dst_mac); 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) { LOG_ERROR("%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) { LOG_ERROR("%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) { LOG_ERROR("%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_L7_OFFSET // TODO #if 0 if (marsio_buff_get_metadata(rx_buff, MR_L7_OFFSET, &(meta->l7_offset), sizeof(meta->l7_offset)) != 0) { LOG_ERROR("%s: unable to get l7_offset from metadata", LOG_TAG_PKTIO); return -1; } #endif } else { meta->is_ctrl_pkt = 0; // only raw packet set MR_IS_DECRYPTED // TODO #if 0 if (marsio_buff_get_metadata(rx_buff, MR_IS_DECRYPTED, &(meta->traffic_is_decrypted), sizeof(meta->traffic_is_decrypted)) != 0) { LOG_ERROR("%s: unable to get traffic_is_decrypted from metadata", LOG_TAG_PKTIO); return -1; } #endif } if (marsio_buff_get_metadata(rx_buff, MR_BUFF_ROUTE_CTX, meta->route_ctx, sizeof(meta->route_ctx)) != 0) { LOG_ERROR("%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) { LOG_ERROR("%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) { LOG_ERROR("%s: unable to set session_id for metadata", LOG_TAG_PKTIO); return -1; } } // 1: E2I // 0: I2E if (marsio_buff_set_metadata(tx_buff, MR_BUFF_DIR, &(meta->dir_is_e2i), sizeof(meta->dir_is_e2i)) != 0) { LOG_ERROR("%s: unable to set buff_dir for metadata", LOG_TAG_PKTIO); return -1; } if (meta->is_ctrl_pkt) { // TODO #if 0 if (marsio_buff_set_metadata(tx_buff, MR_L7_OFFSET, &(meta->l7_offset), sizeof(meta->l7_offset)) != 0) { LOG_ERROR("%s: unable to set l7_offset for metadata", LOG_TAG_PKTIO); return -1; } #endif } else { // TODO #if 0 if (marsio_buff_set_metadata(tx_buff, MR_IS_DECRYPTED, &(meta->traffic_is_decrypted), sizeof(meta->traffic_is_decrypted)) != 0) { LOG_ERROR("%s: unable to set traffic_is_decrypted for metadata", LOG_TAG_PKTIO); return -1; } #endif } if (strlen(meta->route_ctx)) { if (marsio_buff_set_metadata(tx_buff, MR_BUFF_ROUTE_CTX, meta->route_ctx, sizeof(meta->route_ctx)) != 0) { LOG_ERROR("%s: unable to set route_ctx for metadata", LOG_TAG_PKTIO); return -1; } } if (meta->sids.num) { if (marsio_buff_set_sid_list(tx_buff, meta->sids.elems, meta->sids.num) != 0) { LOG_ERROR("%s: unable to set sid_list for metadata", LOG_TAG_PKTIO); return -1; } } 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 metadata meta; if (packet_io_get_metadata(rx_buff, &meta) == -1) { LOG_ERROR("%s: unexpected control packet, unable to get metadata", LOG_TAG_PKTIO); return -1; } 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) { LOG_ERROR("%s: unexpected control packet, unable to parse data", LOG_TAG_PKTIO); return -1; } if (ctrl_parser.session_id != meta.session_id) { LOG_ERROR("%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: // when session opening, firewall not send policy id // return handle_session_opening(&meta, &ctrl_parser, thread_seq, ctx); break; case SESSION_STATE_CLONING: return handle_session_closing(&meta, &ctrl_parser, thread_seq, ctx); case SESSION_STATE_ACTIVE: return handle_session_active(&meta, &ctrl_parser, thread_seq, ctx); case SESSION_STATE_RESETALL: return handle_session_resetall(&meta, &ctrl_parser, thread_seq, ctx); } return 0; } // return : RAW_PKT_ERR_BYPASS // return : RAW_PKT_HIT_BYPASS // return : RAW_PKT_HIT_BLOCK // reutrn : RAW_PKT_HIT_FORWARD static enum raw_pkt_action handle_raw_packet(struct packet_io *handle, marsio_buff_t *rx_buff, int thread_seq, void *ctx) { struct thread_ctx *thread = (struct thread_ctx *)ctx; struct metadata meta; if (packet_io_get_metadata(rx_buff, &meta) == -1) { LOG_ERROR("%s: unexpected raw packet, unable to get metadata, bypass !!!", LOG_TAG_PKTIO); marsio_send_burst(handle->dev_nf_interface.mr_path, thread_seq, &rx_buff, 1); return RAW_PKT_ERR_BYPASS; } struct session_node *node = session_table_search_by_id(thread->session_table, meta.session_id); if (node == NULL) { LOG_ERROR("%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); return RAW_PKT_ERR_BYPASS; } // update sids struct session_ctx *s_ctx = (struct session_ctx *)node->val_data; if (meta.dir_is_e2i) { sids_write_once(&(s_ctx->raw_pkt_e2i_sids), &(meta.sids)); if (!strlen(s_ctx->raw_pkt_e2i_route_ctx)) { memcpy(s_ctx->raw_pkt_e2i_route_ctx, meta.route_ctx, sizeof(s_ctx->raw_pkt_e2i_route_ctx)); } } else { sids_write_once(&(s_ctx->raw_pkt_i2e_sids), &(meta.sids)); if (!strlen(s_ctx->raw_pkt_i2e_route_ctx)) { memcpy(s_ctx->raw_pkt_i2e_route_ctx, meta.route_ctx, sizeof(s_ctx->raw_pkt_i2e_route_ctx)); } } // search chaining struct selected_chaining *chaining = s_ctx->chaining; if (chaining == NULL) { LOG_ERROR("%s: unexpected raw packet, session %lu %s misses policy, bypass !!!", LOG_TAG_PKTIO, s_ctx->session_id, s_ctx->first_ctrl_pkt.addr_string); marsio_send_burst(handle->dev_nf_interface.mr_path, thread_seq, &rx_buff, 1); return RAW_PKT_ERR_BYPASS; } for (int i = 0; i < chaining->chaining_size; i++) { struct selected_sf *node = &(chaining->chaining[i]); LOG_INFO("%s: session %lu %s execute policy: %d -> sff_profile_id %d -> sf_profile_id %d -> sf_need_skip %d sf_action_reason : %d", LOG_TAG_PKTIO, s_ctx->session_id, s_ctx->first_ctrl_pkt.addr_string, node->policy_id, node->sff_profile_id, node->sf_profile_id, node->sf_need_skip, node->sf_action_reason); if (node->sf_need_skip) { continue; } switch (node->sf_action) { case SESSION_ACTION_BYPASS: // BYPASS CURRENT SF continue; case SESSION_ACTION_BLOCK: // BLOCK ALL SF marsio_buff_free(handle->instance, &rx_buff, 1, 0, thread_seq); return RAW_PKT_HIT_BLOCK; case SESSION_ACTION_FORWARD: if (node->sf_connectivity.method != PACKAGE_METHOD_VXLAN_G) { LOG_ERROR("%s: processing raw packets, session %lu %s requires encapsulation format not supported, bypass !!!", LOG_TAG_PKTIO, s_ctx->session_id, s_ctx->first_ctrl_pkt.addr_string); marsio_send_burst(handle->dev_nf_interface.mr_path, thread_seq, &rx_buff, 1); return RAW_PKT_ERR_BYPASS; } if (forward_packet_to_sf(handle, rx_buff, &meta, i + 1, thread_seq, ctx) == 0) { throughput_metrics_inc(&node->tx, 1, meta.raw_len); return RAW_PKT_HIT_FORWARD; } else { LOG_ERROR("%s: processing raw packet, session %lu %s forwarding packet to service function failed, bypass !!!", LOG_TAG_PKTIO, s_ctx->session_id, s_ctx->first_ctrl_pkt.addr_string); // TODO marsio_send_burst(handle->dev_nf_interface.mr_path, thread_seq, &rx_buff, 1); return RAW_PKT_ERR_BYPASS; } default: continue; } } // BYPASS ALL SF LOG_INFO("%s: session %lu %s bypass all service function", LOG_TAG_PKTIO, s_ctx->session_id, s_ctx->first_ctrl_pkt.addr_string); marsio_send_burst(handle->dev_nf_interface.mr_path, thread_seq, &rx_buff, 1); return RAW_PKT_HIT_BYPASS; } // return : INJT_PKT_ERR_DROP // return : INJT_PKT_HIT_BLOCK // return : INJT_PKT_HIT_FWD2SF // return : INJT_PKT_HIT_FWD2NF static enum inject_pkt_action handle_inject_packet(struct packet_io *handle, marsio_buff_t *rx_buff, int thread_seq, void *ctx) { struct thread_ctx *thread = (struct thread_ctx *)ctx; struct g_vxlan *g_vxlan_hdr = NULL; int raw_len = marsio_buff_datalen(rx_buff); char *raw_data = marsio_buff_mtod(rx_buff); if (g_vxlan_decode(&g_vxlan_hdr, raw_data, raw_len) == -1) { LOG_ERROR("%s: unexpected inject packet, not a vxlan-encapsulated packet, drop !!!", LOG_TAG_PKTIO); marsio_buff_free(handle->instance, &rx_buff, 1, 0, thread_seq); return INJT_PKT_ERR_DROP; } struct metadata meta; memset(&meta, 0, sizeof(struct metadata)); meta.raw_data = (char *)g_vxlan_hdr + sizeof(struct g_vxlan); meta.raw_len = raw_len - sizeof(struct ethhdr) - sizeof(struct ip) - sizeof(struct udphdr) - sizeof(struct g_vxlan); meta.dir_is_e2i = g_vxlan_get_packet_dir(g_vxlan_hdr); meta.traffic_is_decrypted = g_vxlan_get_traffic_type(g_vxlan_hdr); meta.is_ctrl_pkt = 0; meta.l7_offset = 0; // meta.session_id set later // meta.sids set later int chain_index = g_vxlan_get_chain_index(g_vxlan_hdr); 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 *)meta.raw_data, meta.raw_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); LOG_ERROR("%s: unexpected inject packet, unable to find session %s from session table, drop !!!", LOG_TAG_PKTIO, addr_string); free(addr_string); marsio_buff_free(handle->instance, &rx_buff, 1, 0, thread_seq); return INJT_PKT_ERR_DROP; } struct session_ctx *s_ctx = (struct session_ctx *)node->val_data; meta.session_id = s_ctx->session_id; if (meta.dir_is_e2i) { sids_copy(&meta.sids, &s_ctx->raw_pkt_e2i_sids); memcpy(meta.route_ctx, s_ctx->raw_pkt_e2i_route_ctx, sizeof(s_ctx->raw_pkt_e2i_route_ctx)); } else { sids_copy(&meta.sids, &s_ctx->raw_pkt_i2e_sids); memcpy(meta.route_ctx, s_ctx->raw_pkt_i2e_route_ctx, sizeof(s_ctx->raw_pkt_i2e_route_ctx)); } struct selected_chaining *chaining = s_ctx->chaining; if (chaining == NULL || chain_index < 1 || chain_index > chaining->chaining_size) { LOG_ERROR("%s: unexpected inject packet, session %lu %s misses chaining index, drop !!!", LOG_TAG_PKTIO, s_ctx->session_id, s_ctx->first_ctrl_pkt.addr_string); marsio_buff_free(handle->instance, &rx_buff, 1, 0, thread_seq); return INJT_PKT_ERR_DROP; } throughput_metrics_inc(&chaining->chaining[chain_index - 1].rx, 1, meta.raw_len); for (int i = chain_index; i < chaining->chaining_size; i++) { struct selected_sf *node = &(chaining->chaining[i]); LOG_INFO("%s: session %lu %s execute policy: %d -> sff_profile_id %d -> sf_profile_id %d -> sf_need_skip %d sf_action_reason : %d", LOG_TAG_PKTIO, s_ctx->session_id, s_ctx->first_ctrl_pkt.addr_string, node->policy_id, node->sff_profile_id, node->sf_profile_id, node->sf_need_skip, node->sf_action_reason); if (node->sf_need_skip) { continue; } switch (node->sf_action) { case SESSION_ACTION_BYPASS: // BYPASS CURRENT SF continue; case SESSION_ACTION_BLOCK: // BLOCK ALL SF marsio_buff_free(handle->instance, &rx_buff, 1, 0, thread_seq); return INJT_PKT_HIT_BLOCK; case SESSION_ACTION_FORWARD: if (node->sf_connectivity.method != PACKAGE_METHOD_VXLAN_G) { LOG_ERROR("%s: processing inject packets, session %lu %s requires encapsulation format not supported, drop !!!", LOG_TAG_PKTIO, s_ctx->session_id, s_ctx->first_ctrl_pkt.addr_string); marsio_buff_free(handle->instance, &rx_buff, 1, 0, thread_seq); return INJT_PKT_ERR_DROP; } if (forward_packet_to_sf(handle, rx_buff, &meta, i + 1, thread_seq, ctx) == 0) { throughput_metrics_inc(&node->tx, 1, meta.raw_len); return INJT_PKT_HIT_FWD2SF; } else { LOG_ERROR("%s: processing inject packet, session %lu %s forwarding packet to service function failed, drop !!!", LOG_TAG_PKTIO, s_ctx->session_id, s_ctx->first_ctrl_pkt.addr_string); marsio_buff_free(handle->instance, &rx_buff, 1, 0, thread_seq); return INJT_PKT_ERR_DROP; } default: continue; } } if (chain_index != chaining->chaining_size) { LOG_ERROR("%s: unexpected inject packet, session %lu %s using invalid chaining index, drop !!!", LOG_TAG_PKTIO, s_ctx->session_id, s_ctx->first_ctrl_pkt.addr_string); marsio_buff_free(handle->instance, &rx_buff, 1, 0, thread_seq); return INJT_PKT_ERR_DROP; } if (forward_packet_to_nf(handle, rx_buff, &meta, -1, thread_seq, ctx) == -1) { LOG_ERROR("%s: processing inject packet, session %lu %s forwarding packet to network function failed, drop !!!", LOG_TAG_PKTIO, s_ctx->session_id, s_ctx->first_ctrl_pkt.addr_string); marsio_buff_free(handle->instance, &rx_buff, 1, 0, thread_seq); return INJT_PKT_ERR_DROP; } else { return INJT_PKT_HIT_FWD2NF; } } // return 0 : success // return -1 : error static int forward_packet_to_sf(struct packet_io *handle, marsio_buff_t *rx_buff, struct metadata *meta, int chain_index, int thread_seq, void *ctx) { // TODO get dst ip frome policy // TODO get dst mac by dst ip marsio_buff_reset(rx_buff); struct ethhdr *eth_hdr = (struct ethhdr *)marsio_buff_prepend(rx_buff, sizeof(struct ethhdr) + sizeof(struct ip) + sizeof(struct udphdr) + sizeof(struct g_vxlan)); struct ip *ip_hdr = (struct ip *)((char *)eth_hdr + sizeof(struct ethhdr)); struct udphdr *udp_hdr = (struct udphdr *)((char *)ip_hdr + sizeof(struct ip)); struct g_vxlan *g_vxlan_hdr = (struct g_vxlan *)((char *)udp_hdr + sizeof(struct udphdr)); memset(g_vxlan_hdr, 0, sizeof(struct g_vxlan)); g_vxlan_set_packet_dir(g_vxlan_hdr, meta->dir_is_e2i); g_vxlan_set_chain_index(g_vxlan_hdr, chain_index); g_vxlan_set_traffic_type(g_vxlan_hdr, meta->traffic_is_decrypted); build_ether_header(eth_hdr, ETH_P_IP, handle->config.default_src_mac, handle->config.default_dst_mac); build_ip_header(ip_hdr, IPPROTO_UDP, handle->config.default_src_ip, handle->config.default_dst_ip, sizeof(struct udphdr) + sizeof(struct g_vxlan) + meta->raw_len); build_udp_header((const char *)&ip_hdr->ip_src, 8, udp_hdr, meta->session_id % (65535 - 49152) + 49152, 4789, sizeof(struct g_vxlan) + meta->raw_len); if (marsio_send_burst(handle->dev_endpoint.mr_path, thread_seq, &rx_buff, 1) != 0) { LOG_ERROR("%s: unable to send burst on device %s, thread_seq: %d", LOG_TAG_PKTIO, handle->config.dev_endpoint, thread_seq); return -1; } return 0; } // return 0 : success // return -1 : error static int forward_packet_to_nf(struct packet_io *handle, marsio_buff_t *rx_buff, struct metadata *meta, int chain_index, int thread_seq, void *ct) { marsio_buff_adj(rx_buff, marsio_buff_datalen(rx_buff) - meta->raw_len); marsio_buff_reset(rx_buff); if (packet_io_set_metadata(rx_buff, meta) != 0) { return -1; } if (marsio_send_burst(handle->dev_nf_interface.mr_path, thread_seq, &rx_buff, 1) != 0) { LOG_ERROR("%s: unable to send burst on device %s, thread_seq: %d", LOG_TAG_PKTIO, handle->config.dev_nf_interface, thread_seq); return -1; } return 0; } static void forward_all_nf_packet_to_sf(struct packet_io *handle, marsio_buff_t *rx_buff, int thread_seq, void *ctx) { struct thread_ctx *thread = (struct thread_ctx *)ctx; struct global_metrics *g_metrics = thread->ref_metrics; struct metadata meta; struct raw_pkt_parser raw_parser; struct addr_tuple4 inner_tuple4; struct session_ctx *s_ctx = NULL; struct session_node *node = NULL; const void *payload = NULL; // get metadata if (packet_io_get_metadata(rx_buff, &meta) == -1) { LOG_ERROR("%s: unexpected raw packet, unable to get metadata, drop !!!", LOG_TAG_PKTIO); marsio_buff_free(handle->instance, &rx_buff, 1, 0, thread_seq); return; } // search session id node = session_table_search_by_id(thread->session_table, meta.session_id); if (node) { goto forward; } // add to session table raw_packet_parser_init(&raw_parser, meta.session_id, LAYER_TYPE_ALL, 8); 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) { LOG_ERROR("%s: incorrect dataoffset in the control zone of session %lu", LOG_TAG_PKTIO, meta.session_id); } s_ctx = session_ctx_new(); s_ctx->session_id = meta.session_id; s_ctx->first_ctrl_pkt.dir_is_e2i = meta.dir_is_e2i; raw_packet_parser_get_most_inner_tuple4(&raw_parser, &inner_tuple4); s_ctx->first_ctrl_pkt.addr_string = addr_tuple4_to_str(&inner_tuple4); if (meta.dir_is_e2i) { sids_write_once(&(s_ctx->raw_pkt_e2i_sids), &(meta.sids)); if (!strlen(s_ctx->raw_pkt_e2i_route_ctx)) { memcpy(s_ctx->raw_pkt_e2i_route_ctx, meta.route_ctx, sizeof(s_ctx->raw_pkt_e2i_route_ctx)); } } else { sids_write_once(&(s_ctx->raw_pkt_i2e_sids), &(meta.sids)); if (!strlen(s_ctx->raw_pkt_i2e_route_ctx)) { memcpy(s_ctx->raw_pkt_i2e_route_ctx, meta.route_ctx, sizeof(s_ctx->raw_pkt_i2e_route_ctx)); } } LOG_INFO("%s: session %lu %s opening", LOG_TAG_PKTIO, s_ctx->session_id, s_ctx->first_ctrl_pkt.addr_string); // TODO only add session __atomic_fetch_add(&g_metrics->session_nums, 1, __ATOMIC_RELAXED); session_table_insert(thread->session_table, s_ctx->session_id, &inner_tuple4, s_ctx, session_value_free_cb); // forward data forward: if (forward_packet_to_sf(handle, rx_buff, &meta, 0, thread_seq, ctx) == 0) { LOG_ERROR("%s: processing raw packet, session %lu %s forwarding packet to service function failed, drop !!!", LOG_TAG_PKTIO, s_ctx->session_id, s_ctx->first_ctrl_pkt.addr_string); marsio_buff_free(handle->instance, &rx_buff, 1, 0, thread_seq); return; } } static void forward_all_sf_packet_to_nf(struct packet_io *handle, marsio_buff_t *rx_buff, int thread_seq, void *ctx) { struct thread_ctx *thread = (struct thread_ctx *)ctx; struct global_metrics *g_metrics = thread->ref_metrics; // vxlan decode struct g_vxlan *g_vxlan_hdr = NULL; int raw_len = marsio_buff_datalen(rx_buff); char *raw_data = marsio_buff_mtod(rx_buff); if (g_vxlan_decode(&g_vxlan_hdr, raw_data, raw_len) == -1) { LOG_ERROR("%s: unexpected inject packet, not a vxlan-encapsulated packet, drop !!!", LOG_TAG_PKTIO); marsio_buff_free(handle->instance, &rx_buff, 1, 0, thread_seq); return; } struct metadata meta; memset(&meta, 0, sizeof(struct metadata)); meta.raw_data = (char *)g_vxlan_hdr + sizeof(struct g_vxlan); meta.raw_len = raw_len - sizeof(struct ethhdr) - sizeof(struct ip) - sizeof(struct udphdr) - sizeof(struct g_vxlan); meta.dir_is_e2i = g_vxlan_get_packet_dir(g_vxlan_hdr); meta.traffic_is_decrypted = g_vxlan_get_traffic_type(g_vxlan_hdr); meta.is_ctrl_pkt = 0; meta.l7_offset = 0; // get inner tuple4 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 *)meta.raw_data, meta.raw_len); raw_packet_parser_get_most_inner_tuple4(&raw_parser, &inner_addr); // search session table 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); LOG_ERROR("%s: unexpected inject packet, unable to find session %s from session table, drop !!!", LOG_TAG_PKTIO, addr_string); free(addr_string); marsio_buff_free(handle->instance, &rx_buff, 1, 0, thread_seq); return; } // add meta data struct session_ctx *s_ctx = (struct session_ctx *)node->val_data; meta.session_id = s_ctx->session_id; if (meta.dir_is_e2i) { sids_copy(&meta.sids, &s_ctx->raw_pkt_e2i_sids); memcpy(meta.route_ctx, s_ctx->raw_pkt_e2i_route_ctx, sizeof(s_ctx->raw_pkt_e2i_route_ctx)); } else { sids_copy(&meta.sids, &s_ctx->raw_pkt_i2e_sids); memcpy(meta.route_ctx, s_ctx->raw_pkt_i2e_route_ctx, sizeof(s_ctx->raw_pkt_i2e_route_ctx)); } // sendto nf if (forward_packet_to_nf(handle, rx_buff, &meta, -1, thread_seq, ctx) == -1) { LOG_ERROR("%s: processing inject packet, session %lu %s forwarding packet to network function failed, drop !!!", LOG_TAG_PKTIO, s_ctx->session_id, s_ctx->first_ctrl_pkt.addr_string); marsio_buff_free(handle->instance, &rx_buff, 1, 0, thread_seq); return; } } // return 0 : success // return -1 : error static int handle_session_opening(struct metadata *meta, struct ctrl_pkt_parser *parser, int thread_seq, void *ctx) { struct thread_ctx *thread = (struct thread_ctx *)ctx; struct global_metrics *g_metrics = thread->ref_metrics; if (session_table_search_by_id(thread->session_table, meta->session_id)) { return -1; } 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) { LOG_ERROR("%s: incorrect dataoffset in the control zone of session %lu", LOG_TAG_PKTIO, meta->session_id); } struct session_ctx *s_ctx = session_ctx_new(); fixed_num_array_init(&s_ctx->policy_ids); s_ctx->session_id = meta->session_id; 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 = strndup(meta->raw_data, meta->l7_offset); s_ctx->first_ctrl_pkt.header_len = meta->l7_offset; s_ctx->chaining = selected_chaining_create(128); for (int i = 0; i < parser->policy_id_num; i++) { int new_policy_id = parser->policy_ids[i]; if (fixed_num_array_exist_elem(&s_ctx->policy_ids, new_policy_id)) { continue; } else { policy_enforce_select_chaining(s_ctx->chaining, thread->ref_enforcer, &raw_parser, new_policy_id, meta->dir_is_e2i); fixed_num_array_add_elem(&s_ctx->policy_ids, new_policy_id); } } LOG_INFO("%s: session %lu %s opening", LOG_TAG_PKTIO, s_ctx->session_id, s_ctx->first_ctrl_pkt.addr_string); __atomic_fetch_add(&g_metrics->session_nums, 1, __ATOMIC_RELAXED); session_table_insert(thread->session_table, s_ctx->session_id, &(s_ctx->first_ctrl_pkt.tuple4), s_ctx, session_value_free_cb); return 0; } // 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 thread_ctx *thread = (struct thread_ctx *)ctx; struct global_metrics *g_metrics = thread->ref_metrics; 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; LOG_INFO("%s: session %lu %s closing", LOG_TAG_PKTIO, s_ctx->session_id, s_ctx->first_ctrl_pkt.addr_string); // TODO send log to firewall __atomic_fetch_sub(&g_metrics->session_nums, 1, __ATOMIC_RELAXED); 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 thread_ctx *thread = (struct thread_ctx *)ctx; struct global_metrics *g_metrics = thread->ref_metrics; 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) { LOG_ERROR("%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; for (int i = 0; i < parser->policy_id_num; i++) { int new_policy_id = parser->policy_ids[i]; if (fixed_num_array_exist_elem(&s_ctx->policy_ids, new_policy_id)) { continue; } else { policy_enforce_select_chaining(s_ctx->chaining, thread->ref_enforcer, &raw_parser, new_policy_id, meta->dir_is_e2i); fixed_num_array_add_elem(&s_ctx->policy_ids, new_policy_id); } } } 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 thread_ctx *thread = (struct thread_ctx *)ctx; struct global_metrics *g_metrics = thread->ref_metrics; LOG_ERROR("%s: session %lu notification clears all session tables !!!", LOG_TAG_PKTIO, meta->session_id); __atomic_fetch_and(&g_metrics->session_nums, 0, __ATOMIC_RELAXED); // TODO reset all session_table return 0; } static void session_value_free_cb(void *ctx) { struct session_ctx *s_ctx = (struct session_ctx *)ctx; session_ctx_free(s_ctx); }