diff --git a/worker/include/dpdk_filter/proc_packets.h b/worker/include/dpdk_filter/proc_packets.h index 3d12e2a..3d8281a 100644 --- a/worker/include/dpdk_filter/proc_packets.h +++ b/worker/include/dpdk_filter/proc_packets.h @@ -12,8 +12,14 @@ #include #include +void forward_packet_with_rewrite(struct rte_mbuf *pkt, + struct net_port *in_port, + struct net_port *out_port, + uint16_t queue_number); + void package_sending_decision(bool solution_is_send, struct rte_mbuf *pkt, - struct net_port *port_out, uint16_t queue_number); + struct net_port *port_in, struct net_port *port_out, + uint16_t queue_number); bool check_is_exception(uint16_t *port); diff --git a/worker/include/dpdk_filter/types.h b/worker/include/dpdk_filter/types.h index edc405a..c4590d5 100644 --- a/worker/include/dpdk_filter/types.h +++ b/worker/include/dpdk_filter/types.h @@ -4,6 +4,7 @@ #include "constants.h" #include #include +#include #ifdef DEBUG #define LOG_INFO(info, ...) \ @@ -41,6 +42,9 @@ struct net_port { char dev_name[64]; char dev_args[256]; struct rte_mempool *mbuf_pool; + struct rte_ether_addr mac_addr; + struct rte_ether_addr neighbor_mac; + bool neighbor_learned; }; struct info_of_pakage { diff --git a/worker/src/dpdk_filter/ip_cache.c b/worker/src/dpdk_filter/ip_cache.c index 2359a6d..07d3127 100644 --- a/worker/src/dpdk_filter/ip_cache.c +++ b/worker/src/dpdk_filter/ip_cache.c @@ -58,7 +58,7 @@ static int ip_str_to_key(const char *ip_str, struct ip_key *key) { static int load_ip_categories(const char *ip_str, struct node_cache_ip *node_ip) { const char *sql_cat = - "SELECT certain_category FROM categories_table WHERE ip_str = ?;"; + "SELECT certain_category FROM ip_categories_table WHERE ip_str = ?;"; sqlite3_stmt *stmt_cat = NULL; int rc_cat = sqlite3_prepare_v2(ip_cache_table, sql_cat, -1, &stmt_cat, NULL); if (rc_cat != SQLITE_OK) { @@ -243,7 +243,7 @@ static int insert_ip_main_record(const char *ip_str, } static int delete_ip_categories(const char *ip_str) { - const char *sql = "DELETE FROM categories_table WHERE ip_str = ?"; + const char *sql = "DELETE FROM ip_categories_table WHERE ip_str = ?"; sqlite3_stmt *stmt = NULL; int ret = sqlite3_prepare_v2(ip_cache_table, sql, -1, &stmt, NULL); if (ret != SQLITE_OK) { @@ -260,7 +260,7 @@ static int delete_ip_categories(const char *ip_str) { static int insert_ip_categories(const char *ip_str, struct node_cache_ip *node) { - const char *sql = "INSERT INTO categories_table (ip_str, " + const char *sql = "INSERT INTO ip_categories_table (ip_str, " "certain_category) VALUES (?, ?)"; sqlite3_stmt *stmt = NULL; int ret = sqlite3_prepare_v2(ip_cache_table, sql, -1, &stmt, NULL); diff --git a/worker/src/dpdk_filter/net_port.c b/worker/src/dpdk_filter/net_port.c index 21390ed..bad906a 100644 --- a/worker/src/dpdk_filter/net_port.c +++ b/worker/src/dpdk_filter/net_port.c @@ -69,9 +69,15 @@ struct net_port *init_struct_af_xdp_port(const char *iface_name, return NULL; } + #ifdef WITH_AF_PACKET + snprintf(port->dev_name, sizeof(port->dev_name), "eth_af_packet_%s", iface_name); snprintf(port->dev_args, sizeof(port->dev_args), - "iface=%s,start_queue=0,queue_count=1", iface_name); + "iface=%s,qpairs=1,blocksz=16384,framesz=2048,framecnt=4096", iface_name); + #else + snprintf(port->dev_args, sizeof(port->dev_args), "iface=%s,start_queue=0,queue_count=1", iface_name); snprintf(port->dev_name, sizeof(port->dev_name), "net_af_xdp_%s", iface_name); + #endif + strncpy(port->iface_name, iface_name, sizeof(port->iface_name) - 1); port->iface_name[sizeof(port->iface_name) - 1] = '\0'; port->mbuf_pool = mbuf_pool; @@ -134,8 +140,17 @@ int net_port_init(struct net_port *port) { return ret; } - LOG_INFO("Port %u initialized", port_id); - return 0; + ret = rte_eth_macaddr_get(port_id, &port->mac_addr); + if (ret < 0) { + LOG_ERROR("Failed to macaddr get: %s", strerror(-ret)); + rte_vdev_uninit(dev_name); + return ret; + } + port->neighbor_learned = false; + LOG_INFO("Port %u initialized, MAC=%02x:%02x:%02x:%02x:%02x:%02x", port_id, + port->mac_addr.addr_bytes[0], port->mac_addr.addr_bytes[1], + port->mac_addr.addr_bytes[2], port->mac_addr.addr_bytes[3], + port->mac_addr.addr_bytes[4], port->mac_addr.addr_bytes[5]); return 0; } int net_port_start(uint16_t port_id) { diff --git a/worker/src/dpdk_filter/proc_packets.c b/worker/src/dpdk_filter/proc_packets.c index 72c1508..f9f4cc6 100644 --- a/worker/src/dpdk_filter/proc_packets.c +++ b/worker/src/dpdk_filter/proc_packets.c @@ -14,22 +14,75 @@ extern bool worker_classify(const char *type, const char *target, const uint16_t LIST_EXCEPTION_PORTS[LEN_LIST_EXCEPTION_PORTS] = {22}; -void package_sending_decision(bool solution_is_send, struct rte_mbuf *pkt, - struct net_port *port_out, - uint16_t queue_number) { - if (solution_is_send) { - struct rte_mbuf *tx_pkt[1] = {pkt}; - uint16_t ret = rte_eth_tx_burst(port_out->port_id, queue_number, tx_pkt, 1); +void learn_neighbor_mac(struct net_port *port, struct rte_mbuf *pkt, struct net_port *port_2) { + struct rte_ether_hdr *eth = rte_pktmbuf_mtod(pkt, struct rte_ether_hdr *); + if (!port->neighbor_learned) { + struct rte_ether_addr *neigh_mac = ð->src_addr; + if (!rte_is_same_ether_addr(neigh_mac, &port_2->mac_addr) && !rte_is_same_ether_addr(neigh_mac, &port->mac_addr)){ + rte_ether_addr_copy(neigh_mac, &port->neighbor_mac); + port->neighbor_learned = true; + LOG_INFO("Learned neighbor MAC on %s: %02x:%02x:%02x:%02x:%02x:%02x", + port->iface_name, port->neighbor_mac.addr_bytes[0], + port->neighbor_mac.addr_bytes[1], + port->neighbor_mac.addr_bytes[2], + port->neighbor_mac.addr_bytes[3], + port->neighbor_mac.addr_bytes[4], + port->neighbor_mac.addr_bytes[5]); + } + } +} - if (ret < 1) { - LOG_ERROR("Failed to send packet"); - record_packet_dropped(); - rte_pktmbuf_free(pkt); - return; +void forward_packet_with_rewrite(struct rte_mbuf *pkt, + struct net_port *in_port, + struct net_port *out_port, + uint16_t queue_number) { + struct rte_ether_hdr *eth = rte_pktmbuf_mtod(pkt, struct rte_ether_hdr *); + + learn_neighbor_mac(in_port, pkt, out_port); + + rte_ether_addr_copy(&out_port->mac_addr, ð->src_addr); + + if (out_port->neighbor_learned) { + rte_ether_addr_copy(&out_port->neighbor_mac, ð->dst_addr); + } else { + struct rte_ether_addr broadcast = { .addr_bytes = {0xff, 0xff, 0xff, 0xff, 0xff, 0xff} }; + rte_ether_addr_copy(&broadcast, ð->dst_addr); + LOG_WARNING("Neighbor MAC not learned yet on %s, using broadcast", out_port->iface_name); } + if (eth->ether_type == rte_cpu_to_be_16(RTE_ETHER_TYPE_IPV4)) { + struct rte_ipv4_hdr *ip = (struct rte_ipv4_hdr *)((uint8_t *)eth + sizeof(struct rte_ether_hdr)); + + if (ip->time_to_live > 1) { + ip->time_to_live--; + ip->hdr_checksum = 0; + ip->hdr_checksum = rte_ipv4_cksum(ip); + } else { + rte_pktmbuf_free(pkt); + return; + } + } + + struct rte_mbuf *tx_pkt[1] = {pkt}; + printf("pkt_len=%u data_len=%u nb_segs=%u ol_flags=%lx\n", + pkt->pkt_len, + pkt->data_len, + pkt->nb_segs, + pkt->ol_flags); + uint16_t ret = rte_eth_tx_burst(out_port->port_id, queue_number, tx_pkt, 1); + if (ret < 1) { + LOG_ERROR("Failed to send packet"); + record_packet_dropped(); + rte_pktmbuf_free(pkt); + } record_packet_passed(); +} +void package_sending_decision(bool solution_is_send, struct rte_mbuf *pkt, + struct net_port *port_in, struct net_port *port_out, + uint16_t queue_number) { + if (solution_is_send) { + forward_packet_with_rewrite(pkt, port_in, port_out, queue_number); return; } @@ -62,7 +115,7 @@ void pakage_processing(struct net_port *port_in, struct net_port *port_out, if (atomic_load(&filtring_is_turned_off)) { for (int i = 0; i < nb_rx; i++) { record_packet_received(); - package_sending_decision(true, pkts[i], port_out, queue_number); + package_sending_decision(true, pkts[i], port_in, port_out, queue_number); } return; } @@ -84,12 +137,12 @@ void pakage_processing(struct net_port *port_in, struct net_port *port_out, if (check_is_exception(&info_pac.number_port) == true) { LOG_INFO("Exception port %hu, forwarding to exception port", ntohs(info_pac.number_port)); - package_sending_decision(true, pkts[i], port_exception, queue_number); + package_sending_decision(true, pkts[i], port_in, port_exception, queue_number); continue; } int ret; - struct ip_key key; + struct ip_key key = {0}; if (info_pac.ip_version == IP_4) { key.version = 4; key.addr.ip4 = info_pac.ip4_dist; @@ -103,7 +156,7 @@ void pakage_processing(struct net_port *port_in, struct net_port *port_out, if (ret >= 0 && cached_node_ip) { LOG_INFO("IP cache hit, decision: %s", cached_node_ip->solution_is_send ? "send" : "drop"); - package_sending_decision(cached_node_ip->solution_is_send, pkts[i], + package_sending_decision(cached_node_ip->solution_is_send, pkts[i], port_in, port_out, queue_number); } else if (ret == -ENOENT) { LOG_INFO("IP cache miss, applying filter"); @@ -132,7 +185,7 @@ void pakage_processing(struct net_port *port_in, struct net_port *port_out, LOG_WARNING("Classification failed for IP %s", ip_str); } - package_sending_decision(solution_is_send, pkts[i], port_out, + package_sending_decision(solution_is_send, pkts[i], port_in, port_out, queue_number); struct node_cache_ip *new_node = @@ -145,7 +198,7 @@ void pakage_processing(struct net_port *port_in, struct net_port *port_out, new_node->solution_is_send = solution_is_send; - struct ip_key key; + struct ip_key key = {0}; if (info_pac.ip_version == IP_4) { key.version = 4; key.addr.ip4 = info_pac.ip4_dist; @@ -169,7 +222,7 @@ void pakage_processing(struct net_port *port_in, struct net_port *port_out, if (check_is_exception(&info_pac.number_port) == true) { LOG_INFO("Exception port %hu, forwarding to exception port", ntohs(info_pac.number_port)); - package_sending_decision(true, pkts[i], port_exception, queue_number); + package_sending_decision(true, pkts[i], port_in, port_exception, queue_number); continue; } @@ -178,7 +231,7 @@ void pakage_processing(struct net_port *port_in, struct net_port *port_out, if (ret >= 0 && cached_node_domain) { LOG_INFO("Domain cache hit for '%s', decision: %s", info_pac.domain, cached_node_domain->solution_is_send ? "send" : "drop"); - package_sending_decision(cached_node_domain->solution_is_send, pkts[i], + package_sending_decision(cached_node_domain->solution_is_send, pkts[i], port_in, port_out, queue_number); } else if (ret == -ENOENT) { LOG_INFO("Domain cache miss for '%s', applying filter", @@ -198,7 +251,7 @@ void pakage_processing(struct net_port *port_in, struct net_port *port_out, LOG_WARNING("Classification failed for %s", info_pac.domain); } - package_sending_decision(solution_is_send, pkts[i], port_out, + package_sending_decision(solution_is_send, pkts[i], port_in, port_out, queue_number); struct node_cache_domain *new_node = diff --git a/worker/src/worker.cpp b/worker/src/worker.cpp index c6d4526..938fd17 100644 --- a/worker/src/worker.cpp +++ b/worker/src/worker.cpp @@ -145,12 +145,8 @@ void Worker::forward_to_out(struct net_port *incoming_port, uint16_t nb_tap = rte_eth_rx_burst(incoming_port->port_id, queue_number, tap_pkts, 32); for (int i = 0; i < nb_tap; i++) { - int ret = - rte_eth_tx_burst(outgoing_port->port_id, queue_number, &tap_pkts[i], 1); - if (ret < 1) { - spdlog::warn("Failed to send packet"); - rte_pktmbuf_free(tap_pkts[i]); - } + forward_packet_with_rewrite(tap_pkts[i], incoming_port, outgoing_port, queue_number); + } }