fix: fix af xdp packets

worker/include/dpdk_filter/proc_packets.h

@@ -12,14 +12,21 @@
 #include <stdint.h>
 #include <stdlib.h>
 
+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
+#endif

worker/include/dpdk_filter/types.h

@@ -2,6 +2,7 @@
 #define TYPES_H
 
 #include "constants.h"
+#include <rte_ether.h>
 #include <stdbool.h>
 #include <stdint.h>
 
@@ -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 {

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);
-}
+}

worker/src/dpdk_filter/proc_packets.c

@@ -1,17 +1,95 @@
 #include "proc_packets.h"
 #include "domain_cache.h"
 #include "ip_cache.h"
+#include <rte_icmp.h>
 #include <stdatomic.h>
 
 extern bool worker_classify(const char *type, const char *target,
                             struct requested_classification *out_req);
 
 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(&eth->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, &eth->src_addr);
+  if (out_port->neighbor_learned) {
+    rte_ether_addr_copy(&out_port->neighbor_mac, &eth->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);

worker/src/worker.cpp

@@ -7,6 +7,7 @@
 #include <grpcpp/grpcpp.h>
 #include <signal.h>
 #include <spdlog/spdlog.h>
+#include <unistd.h>
 #include <thread>
 
 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();