Fix: last version xdp и packet

worker/include/dpdk_filter/proc_packets.h

@@ -12,8 +12,14 @@
 #include <stdint.h>
 #include <stdlib.h>
 
+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);
 

worker/include/dpdk_filter/types.h

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

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

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) {

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 = &eth->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, &eth->src_addr);
+
+    if (out_port->neighbor_learned) {
+        rte_ether_addr_copy(&out_port->neighbor_mac, &eth->dst_addr);
+    } else {
+        struct rte_ether_addr broadcast = { .addr_bytes = {0xff, 0xff, 0xff, 0xff, 0xff, 0xff} };
+        rte_ether_addr_copy(&broadcast, &eth->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 =

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