diff --git a/worker/include/dpdk_filter/proc_packets.h b/worker/include/dpdk_filter/proc_packets.h index 3d12e2a..21bf8fe 100644 --- a/worker/include/dpdk_filter/proc_packets.h +++ b/worker/include/dpdk_filter/proc_packets.h @@ -12,14 +12,21 @@ #include #include +void dump_checksum_before_tx(struct rte_mbuf *pkt); + void package_sending_decision(bool solution_is_send, struct rte_mbuf *pkt, struct net_port *port_out, uint16_t queue_number); bool check_is_exception(uint16_t *port); +void learn_neighbor_mac(struct net_port *port, struct rte_mbuf *pkt); + +void rewrite_l2_and_forward(struct rte_mbuf *pkt, struct net_port *in_port, + struct net_port *out_port, uint16_t queue_number); + void pakage_processing(struct net_port *port_in, struct net_port *port_out, struct net_port *port_exception, uint16_t queue_number, uint16_t nb_pkts, struct rte_mbuf **pkts, struct BASE_POLICY *policy, bool filtring_is_turned_off); -#endif \ No newline at end of file +#endif diff --git a/worker/include/dpdk_filter/types.h b/worker/include/dpdk_filter/types.h index edc405a..61d1c5c 100644 --- a/worker/include/dpdk_filter/types.h +++ b/worker/include/dpdk_filter/types.h @@ -2,6 +2,7 @@ #define TYPES_H #include "constants.h" +#include #include #include @@ -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/net_port.c b/worker/src/dpdk_filter/net_port.c index 21390ed..5fd151d 100644 --- a/worker/src/dpdk_filter/net_port.c +++ b/worker/src/dpdk_filter/net_port.c @@ -70,7 +70,8 @@ struct net_port *init_struct_af_xdp_port(const char *iface_name, } snprintf(port->dev_args, sizeof(port->dev_args), - "iface=%s,start_queue=0,queue_count=1", iface_name); + "iface=%s,start_queue=0,queue_count=1", + iface_name); snprintf(port->dev_name, sizeof(port->dev_name), "net_af_xdp_%s", iface_name); strncpy(port->iface_name, iface_name, sizeof(port->iface_name) - 1); port->iface_name[sizeof(port->iface_name) - 1] = '\0'; @@ -134,7 +135,12 @@ int net_port_init(struct net_port *port) { return ret; } - LOG_INFO("Port %u initialized", port_id); + rte_eth_macaddr_get(port_id, &port->mac_addr); + 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; } @@ -193,4 +199,4 @@ void net_port_close(struct net_port *port) { port->port_id = -1; LOG_INFO("Port %u closed", port_id); -} \ No newline at end of file +} diff --git a/worker/src/dpdk_filter/proc_packets.c b/worker/src/dpdk_filter/proc_packets.c index 40fab59..7ea5a16 100644 --- a/worker/src/dpdk_filter/proc_packets.c +++ b/worker/src/dpdk_filter/proc_packets.c @@ -1,6 +1,7 @@ #include "proc_packets.h" #include "domain_cache.h" #include "ip_cache.h" +#include #include extern bool worker_classify(const char *type, const char *target, @@ -8,10 +9,87 @@ extern bool worker_classify(const char *type, const char *target, const uint16_t LIST_EXCEPTION_PORTS[LEN_LIST_EXCEPTION_PORTS] = {22}; +void learn_neighbor_mac(struct net_port *port, struct rte_mbuf *pkt) { + struct rte_ether_hdr *eth = rte_pktmbuf_mtod(pkt, struct rte_ether_hdr *); + if (!port->neighbor_learned) { + rte_ether_addr_copy(ð->src_addr, &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]); + } +} + +void rewrite_l2_and_forward(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); + + 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); + } + + 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--; + uint32_t cksum = rte_be_to_cpu_16(ip->hdr_checksum) + 0x0100; + cksum = (cksum & 0xFFFF) + (cksum >> 16); + ip->hdr_checksum = rte_cpu_to_be_16((uint16_t)cksum); + } else { + rte_pktmbuf_free(pkt); + return; + } + } + + package_sending_decision(true, pkt, out_port, queue_number); +} + +void dump_checksum_before_tx(struct rte_mbuf *pkt) { + struct rte_ether_hdr *eth = rte_pktmbuf_mtod(pkt, struct rte_ether_hdr *); + if (eth->ether_type != rte_cpu_to_be_16(RTE_ETHER_TYPE_IPV4)) + return; + + struct rte_ipv4_hdr *ip = + (struct rte_ipv4_hdr *)((uint8_t *)eth + sizeof(struct rte_ether_hdr)); + void *l4 = (uint8_t *)ip + rte_ipv4_hdr_len(ip); + uint16_t cksum = 0; + const char *proto_name = "OTHER"; + + if (ip->next_proto_id == IPPROTO_ICMP) { + struct rte_icmp_hdr *icmp = (struct rte_icmp_hdr *)l4; + cksum = icmp->icmp_cksum; + proto_name = "ICMP"; + } else if (ip->next_proto_id == IPPROTO_TCP) { + struct rte_tcp_hdr *tcp = (struct rte_tcp_hdr *)l4; + cksum = tcp->cksum; + proto_name = "TCP"; + } else if (ip->next_proto_id == IPPROTO_UDP) { + struct rte_udp_hdr *udp = (struct rte_udp_hdr *)l4; + cksum = udp->dgram_cksum; + proto_name = "UDP"; + } + + LOG_INFO("[DIAG] %s before tx_burst: cksum=0x%04x ip_cksum=0x%04x " + "ol_flags=0x%lx", + proto_name, rte_be_to_cpu_16(cksum), + rte_be_to_cpu_16(ip->hdr_checksum), + (unsigned long)pkt->ol_flags); +} + 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) { + dump_checksum_before_tx(pkt); struct rte_mbuf *tx_pkt[1] = {pkt}; uint16_t ret = rte_eth_tx_burst(port_out->port_id, queue_number, tx_pkt, 1); @@ -49,7 +127,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++) { - package_sending_decision(true, pkts[i], port_out, queue_number); + rewrite_l2_and_forward(pkts[i], port_in, port_out, queue_number); } return; } @@ -60,6 +138,13 @@ void pakage_processing(struct net_port *port_in, struct net_port *port_out, memset(&info_pac, 0, sizeof(info_pac)); parsing_pakage(pkts[i], &info_pac); + + if (info_pac.ip_version != IP_4 && info_pac.ip_version != IP_6) { + LOG_INFO("Non-IP packet (ARP/other), forwarding without filtering"); + package_sending_decision(true, pkts[i], port_out, queue_number); + continue; + } + LOG_INFO("[PKT] port = %hu", ntohs(info_pac.number_port)); if (info_pac.domain[0] == '\0') { LOG_INFO("Packet without dns request"); @@ -87,8 +172,11 @@ 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], - port_out, queue_number); + if (cached_node_ip->solution_is_send) { + rewrite_l2_and_forward(pkts[i], port_in, port_out, queue_number); + } else { + rte_pktmbuf_free(pkts[i]); + } } else if (ret == -ENOENT) { LOG_INFO("IP cache miss, applying filter"); @@ -116,8 +204,11 @@ 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, - queue_number); + if (solution_is_send) { + rewrite_l2_and_forward(pkts[i], port_in, port_out, queue_number); + } else { + rte_pktmbuf_free(pkts[i]); + } struct node_cache_ip *new_node = rte_calloc("struct_node_cache_ip", 1, sizeof(struct node_cache_ip), @@ -160,8 +251,11 @@ 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], - port_out, queue_number); + if (cached_node_domain->solution_is_send) { + rewrite_l2_and_forward(pkts[i], port_in, port_out, queue_number); + } else { + rte_pktmbuf_free(pkts[i]); + } } else if (ret == -ENOENT) { LOG_INFO("Domain cache miss for '%s', applying filter", info_pac.domain); @@ -180,6 +274,12 @@ void pakage_processing(struct net_port *port_in, struct net_port *port_out, LOG_WARNING("Classification failed for %s", info_pac.domain); } + if (solution_is_send) { + rewrite_l2_and_forward(pkts[i], port_in, port_out, queue_number); + } else { + rte_pktmbuf_free(pkts[i]); + } + struct node_cache_domain *new_node = rte_calloc("struct_node_cache", 1, sizeof(struct node_cache_domain), RTE_CACHE_LINE_SIZE); diff --git a/worker/src/worker.cpp b/worker/src/worker.cpp index 3c245fb..4ef71c2 100644 --- a/worker/src/worker.cpp +++ b/worker/src/worker.cpp @@ -7,6 +7,7 @@ #include #include #include +#include #include extern "C" bool worker_classify(const char *type, const char *target, @@ -55,9 +56,15 @@ void Worker::initDPDK(int argc, char **argv) { throw std::runtime_error("EAL init failed"); } + uint16_t data_room_size = 2048 + RTE_PKTMBUF_HEADROOM; + uint32_t frame_size = rte_mempool_calc_obj_size( + sizeof(struct rte_mbuf) + priv_size + data_room_size, 0, NULL); + spdlog::info("AF_XDP frame_size={}, PAGE_SIZE={}, data_room={}", frame_size, + getpagesize(), data_room_size); + mbuf_pool = rte_pktmbuf_pool_create( "POOL", mbuf_quantity_in_pool, cache_size_per_kernel, priv_size, - RTE_MBUF_DEFAULT_BUF_SIZE, rte_socket_id()); + data_room_size, rte_socket_id()); if (!mbuf_pool) { throw std::runtime_error("Failed to create mbuf pool"); } @@ -96,13 +103,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"); - // PLUG (to be added later) - need to add processing for this case - rte_pktmbuf_free(tap_pkts[i]); - } + rewrite_l2_and_forward(tap_pkts[i], incoming_port, outgoing_port, + queue_number); } } @@ -401,7 +403,7 @@ void Worker::MainLoop() { } forward_to_out(port_exception, port_in, queue_number); pakage_processing(port_in, port_out, port_exception, queue_number, nb_pkts, - pkts, &local_policy, !enable); + pkts, &local_policy, false); forward_to_out(port_out, port_in, queue_number); if (++timer_check_counter >= timer_check_interval) { rte_timer_manage();