97: filtering methods

worker/Makefile.test_af_xdp

@@ -0,0 +1,22 @@
+CC = gcc
+CFLAGS = -Iinclude -O2 -msse4.2 -mpclmul -maes
+LDFLAGS = -lrte_eal -lrte_ethdev -lrte_mempool -lrte_mbuf -lrte_bus_vdev -lpthread -lnuma -ldl
+
+SRCS = src/dpdk_filter/test.c src/dpdk_filter/af_xdp_port.c
+TARGET = test
+
+all: $(TARGET)
+
+$(TARGET): $(SRCS)
+	$(CC) $(CFLAGS) -o $(TARGET) $(SRCS) $(LDFLAGS)
+
+clean:
+	rm -f $(TARGET)
+
+run: $(TARGET)
+	sudo ./$(TARGET) eth0
+
+run-wlan: $(TARGET)
+	sudo ./$(TARGET) wlan0
+
+.PHONY: all clean run run-wlan

worker/include/dpdk_filter/af_xdp_port.h

@@ -0,0 +1,13 @@
+#ifndef AF_XDP_PORT_H
+#define AF_XDP_PORT_H
+
+#include <stdint.h>
+#include <rte_mempool.h>
+
+int af_xdp_port_init(const char *iface_name, uint16_t *port_id, struct rte_mempool *mbuf_pool);
+
+int af_xdp_port_start(uint16_t port_id);
+
+void af_xdp_port_close(uint16_t port_id);
+
+#endif

worker/include/dpdk_filter/dns_parser.h

@@ -0,0 +1,8 @@
+#ifndef DNS_PARSER_H
+#define DNS_PARSER_H
+
+#include <rte_mbuf.h>
+
+void extract_dns_domain(struct rte_mbuf *pkt, char *domain, int max_len);
+
+#endif

worker/src/dpdk_filter/af_xdp_port.c

@@ -0,0 +1,92 @@
+#include <rte_eal.h>
+#include <rte_ethdev.h>
+#include <rte_mempool.h>
+#include <rte_mbuf.h>
+#include <stdio.h>
+#include <string.h>
+#include <errno.h>
+#include <unistd.h>
+#include <rte_bus_vdev.h>
+
+
+#include "../../include/dpdk_filter/af_xdp_port.h"
+
+#define RX_RING_SIZE 1024
+#define TX_RING_SIZE 1024
+
+int af_xdp_port_init(const char *iface_name, uint16_t *port_id, struct rte_mempool *mbuf_pool)
+{
+    int ret;
+    char vdev_args[256];
+    struct rte_eth_conf port_conf = {0};
+
+    printf("[AF_XDP] Initializing on %s\n", iface_name);
+
+    snprintf(vdev_args, sizeof(vdev_args), "iface=%s,start_queue=0,queue_count=1", iface_name);
+    ret = rte_vdev_init("net_af_xdp", vdev_args);
+
+    if (ret < 0) {
+        printf("[AF_XDP ERROR] Failed to create vdev: %s\n", strerror(-ret));
+        return ret;
+    }
+
+    *port_id = rte_eth_dev_count_avail() - 1;
+
+    ret = rte_eth_dev_configure(*port_id, 1, 1, &port_conf);
+    if (ret < 0) {
+        printf("[AF_XDP ERROR] Failed to configure port: %s\n", strerror(-ret));
+        rte_vdev_uninit("net_af_xdp");
+        return ret;
+    }
+
+    ret = rte_eth_rx_queue_setup(*port_id, 0, RX_RING_SIZE, rte_eth_dev_socket_id(*port_id), NULL, mbuf_pool);
+    if (ret < 0) {
+        printf("[AF_XDP ERROR] Failed to setup RX queue: %s\n", strerror(-ret));
+        return ret;
+    }
+
+    ret = rte_eth_tx_queue_setup(*port_id, 0, TX_RING_SIZE, rte_eth_dev_socket_id(*port_id), NULL);
+
+    if (ret < 0) {
+        printf("[AF_XDP ERROR] Failed to setup TX queue: %s\n", strerror(-ret));
+        return ret;
+    }
+
+    printf("[AF_XDP] Port %u initialized\n", *port_id);
+    return 0;
+}
+
+
+
+
+int af_xdp_port_start(uint16_t port_id)
+{
+    int ret;
+
+    printf("[AF_XDP] Starting port %u\n", port_id);
+
+    ret = rte_eth_dev_start(port_id);
+    if (ret < 0) {
+        printf("[AF_XDP ERROR] Failed to start: %s\n", strerror(-ret));
+        return ret;
+    }
+
+    rte_eth_promiscuous_enable(port_id);
+
+    printf("[AF_XDP] Port %u started\n", port_id);
+    return 0;
+}
+
+
+
+
+void af_xdp_port_close(uint16_t port_id)
+{
+    printf("[AF_XDP] Closing port %u\n", port_id);
+
+    rte_eth_dev_stop(port_id);
+    rte_eth_dev_close(port_id);
+    rte_vdev_uninit("net_af_xdp");
+
+    printf("[AF_XDP] Port %u closed\n", port_id);
+}

worker/src/dpdk_filter/dns_parser.c

@@ -0,0 +1,79 @@
+#include <rte_mbuf.h>
+#include <rte_ether.h>
+#include <rte_ip.h>
+#include <rte_udp.h>
+#include <stdio.h>
+#include <string.h>
+#include <stdint.h>
+
+#include "../../include/dpdk_filter/dns_parser.h"
+
+static const char *blocked_domains[] = { // пока как заглушка вместо политики воркера
+    "facebook.com",
+    "youtube.com",
+    "instagram.com",
+};
+
+
+static int is_domain_blocked(const char *domain) {
+    for (int i = 0; blocked_domains[i] != NULL; i++) {
+        if (strstr(domain, blocked_domains[i]) != NULL) {
+            return 1;
+        }
+    }
+    return 0;
+}
+
+void extract_dns_domain(struct rte_mbuf *pkt, char *domain, int max_len) {
+    struct rte_ether_hdr *eth_hdr;
+    struct rte_ipv4_hdr *ip_hdr;
+    struct rte_udp_hdr *udp_hdr;
+    int dns_hdr = 12;
+    uint8_t *dns_data;
+
+    eth_hdr = rte_pktmbuf_mtod(pkt, struct rte_ether_hdr *);
+
+    if (rte_be_to_cpu_16(eth_hdr->ether_type) != RTE_ETHER_TYPE_IPV4) {
+        domain[0] = '\0';
+        return;
+    }
+
+    ip_hdr = (struct rte_ipv4_hdr *)(eth_hdr + 1);
+
+    if (ip_hdr->next_proto_id != IPPROTO_UDP) {
+        domain[0] = '\0';
+        return;
+    }
+
+    udp_hdr = (struct rte_udp_hdr *)((uint8_t *)ip_hdr + ((ip_hdr->version_ihl) & 0x0f) * 4);
+
+    if (rte_be_to_cpu_16(udp_hdr->dst_port) != 53) {
+        domain[0] = '\0';
+        return;
+    }
+
+    dns_data = (uint8_t *)(udp_hdr + 1);
+    uint8_t *qname = dns_data + dns_hdr;
+    int pos = 0;
+
+    while (*qname != 0 && pos < max_len - 1) {
+        uint8_t label_len = *qname;
+        qname++;
+
+        for (int i = 0; i < label_len && pos < max_len - 1; i++) {
+            domain[pos++] = *qname++;
+        }
+
+        if (*qname != 0 && pos < max_len - 1) {
+            domain[pos++] = '.';
+        }
+    }
+
+    domain[pos] = '\0';
+
+    if (is_domain_blocked(domain)) {
+        printf("\n block: %s\n", domain);
+    } else {
+        printf("\n allow: %s\n", domain);
+    }
+}

worker/src/dpdk_filter/test.c

@@ -0,0 +1,53 @@
+#include <rte_ethdev.h>
+
+#include "../../include/dpdk_filter/af_xdp_port.h"
+
+int main(int argc, char **argv)
+{
+    struct rte_mbuf *pkts[32];
+    uint64_t total_pkts = 0;
+    uint16_t port_id;
+    struct rte_mempool *mbuf_pool;
+
+    int ret = rte_eal_init(argc, argv);
+    if (ret < 0) {
+        printf("ERROR: EAL init failed\n");
+        return -1;
+    }
+
+    mbuf_pool = rte_pktmbuf_pool_create("MBUF_POOL", 8192, 250, 0, RTE_MBUF_DEFAULT_BUF_SIZE, rte_socket_id());
+    if (!mbuf_pool) {
+        printf("ERROR: Failed to create mbuf pool\n");
+        return -1;
+    }
+
+    if (af_xdp_port_init("eth0", &port_id, mbuf_pool) < 0) {
+        printf("ERROR: Failed to init port\n");
+        return -1;
+    }
+
+    if (af_xdp_port_start(port_id) < 0) {
+        af_xdp_port_close(port_id);
+        return -1;
+    }
+
+    printf("AF_XDP port %u is running. Press Ctrl+C to stop.\n", port_id);
+
+    while (1) {
+        uint16_t nb_rx = rte_eth_rx_burst(port_id, 0, pkts, 32);
+        if (nb_rx > 0) {
+            total_pkts += nb_rx;
+            printf("\rPackets received: %u", total_pkts);
+            fflush(stdout);
+            // parse_packet(pkts[0]);
+            for (int i = 0; i < nb_rx; i++) {
+                rte_pktmbuf_free(pkts[i]);
+            }
+        }
+        usleep(5);
+    }
+
+    printf("total pkts: %u", total_pkts);
+    af_xdp_port_close(port_id);
+    return 0;
+}