使用eBPF监控特定端口流量并按源IP统计的实现方案

本文将介绍如何使用eBPF技术来监控特定网络端口的流量，并按照源IP地址进行统计，找出流量最大的IP地址。我们将从eBPF程序的编写、部署到用户态程序的实现，一步步地讲解如何实现这个功能。

1. eBPF程序设计

首先，我们需要编写一个eBPF程序，该程序需要完成以下几个任务：

• 过滤指定端口的数据包： 只捕获目标端口的网络数据包，减少处理的数据量。
• 提取源IP地址： 从网络数据包中提取源IP地址，用于后续的统计。
• 统计流量： 使用eBPF map数据结构，以源IP地址为key，流量大小为value，进行流量统计。

下面是一个简单的eBPF程序示例（使用C语言编写）：

#include <linux/bpf.h>
#include <linux/pkt_cls.h>
#include <linux/ip.h>
#include <linux/tcp.h>
#include <linux/udp.h>
 
#include <bpf/bpf_helpers.h>
#include <bpf/bpf_endian.h>
 
#define DEST_PORT 8080 // 要监控的目标端口
 
struct {  // 定义eBPF map，用于存储IP地址和流量统计信息
    __uint(type, BPF_MAP_TYPE_HASH);
    __uint(key_size, 4);   // IPv4地址为4字节
    __uint(value_size, 8); // 流量大小为8字节
    __uint(max_entries, 65536);
} ip_流量统计 SEC(".maps");
 
 
SEC("socket")
int socket_handler(struct __sk_buff *skb) {
    // 加载以太网头部
    void *data = (void *)(long)skb->data;
    void *data_end = (void *)(long)skb->data_end;
 
    struct ethhdr *eth = data;
    if (data + sizeof(struct ethhdr) > data_end) {
        return TC_ACT_OK; // 包太小，直接放行
    }
 
    // 检查是否为IP包
    if (bpf_ntohs(eth->h_proto) != ETH_P_IP) {
        return TC_ACT_OK; // 不是IP包，直接放行
    }
 
    // 加载IP头部
    struct iphdr *iph = data + sizeof(struct ethhdr);
    if (data + sizeof(struct ethhdr) + sizeof(struct iphdr) > data_end) {
        return TC_ACT_OK; // 包太小，直接放行
    }
 
    // 获取源IP地址
    __u32 src_ip = iph->saddr;
 
    // 检查是否为TCP或UDP包
    if (iph->protocol == IPPROTO_TCP) {
        struct tcphdr *tcph = (void*)iph + sizeof(struct iphdr);
        if ((void*)tcph + sizeof(struct tcphdr) > (void*)data_end) {
            return TC_ACT_OK;
        }
        // 过滤目标端口
        if (bpf_ntohs(tcph->dest) == DEST_PORT) {
            // 统计流量
            __u64 pkt_len = skb->len;
            __u64 *value = bpf_map_lookup_elem(&ip_流量统计, &src_ip);
            if (value) {
                *value += pkt_len;
            } else {
                __u64 init_value = pkt_len;
                bpf_map_update_elem(&ip_流量统计, &src_ip, &init_value, BPF_ANY);
            }
        }
    } else if (iph->protocol == IPPROTO_UDP) {
        struct udphdr *udph = (void*)iph + sizeof(struct iphdr);
        if ((void*)udph + sizeof(struct udphdr) > (void*)data_end) {
            return TC_ACT_OK;
        }
         // 过滤目标端口
        if (bpf_ntohs(udph->dest) == DEST_PORT) {
            // 统计流量
            __u64 pkt_len = skb->len;
            __u64 *value = bpf_map_lookup_elem(&ip_流量统计, &src_ip);
            if (value) {
                *value += pkt_len;
            } else {
                __u64 init_value = pkt_len;
                bpf_map_update_elem(&ip_流量统计, &src_ip, &init_value, BPF_ANY);
            }
        }
    }
 
    return TC_ACT_OK; // 放行数据包
}
 
char _license[] SEC("license") = "GPL";

代码解释：

• DEST_PORT: 定义要监控的目标端口，可以根据实际情况修改。
• ip_流量统计: 定义了一个eBPF map，类型为BPF_MAP_TYPE_HASH，用于存储源IP地址和对应的流量大小。Key为4字节的IP地址，Value为8字节的流量大小。
• socket_handler: 这是eBPF程序的入口函数，当网络数据包到达时，该函数会被调用。
• 函数首先解析以太网头部、IP头部，然后判断是否为TCP或UDP包，并进一步判断目标端口是否为DEST_PORT。
• 如果是目标端口的数据包，则提取源IP地址，并在ip_流量统计 map中更新该IP地址对应的流量大小。

编译eBPF程序：

可以使用clang和llvm工具链来编译eBPF程序。例如：

clang -target bpf -D__TARGET_ARCH_x86_64 -O2 -Wall -Werror -c ebpf_program.c -o ebpf_program.o
llvm-strip -x ebpf_program.o

2. 加载和运行eBPF程序

编译完成后，我们需要将eBPF程序加载到内核中并运行。可以使用bpftool工具来完成这个任务。

首先，创建一个socket类型的eBPF程序：

bpftool prog load ebpf_program.o /sys/fs/bpf/my_ebpf_program

然后，将该程序attach到指定的socket上。你需要找到你想监控的socket的文件描述符。这通常涉及到一些编程技巧，例如遍历/proc/net/tcp或/proc/net/udp文件，找到对应端口的socket。

假设你找到了socket的文件描述符为fd，那么可以使用以下命令attach eBPF程序：

bpftool prog attach /sys/fs/bpf/my_ebpf_program attach_type sock_ops fd $fd

3. 用户态程序设计

现在，eBPF程序已经在内核中运行，并统计着流量数据。我们需要编写一个用户态程序，来读取eBPF map中的数据，并找出流量最大的IP地址。

以下是一个简单的Python程序示例：

from bcc import BPF
import socket
import struct
 
# 加载eBPF程序
b = BPF(src_file="ebpf_program.c") # 假设你直接使用bcc编译，也可以加载.o文件
ip_流量统计 = b["ip_流量统计"]
 
 
def int_to_ip(int_ip):
    return socket.inet_ntoa(struct.pack('I', int_ip))
 
# 循环读取map中的数据
while True:
    for k, v in ip_流量统计.items():
        ip_address = int_to_ip(k.value)
        traffic = v.value
        print(f"IP: {ip_address}, Traffic: {traffic} bytes")
 
    # 找出流量最大的IP地址
    max_ip = None
    max_traffic = 0
    for k, v in ip_流量统计.items():
        if v.value > max_traffic:
            max_traffic = v.value
            max_ip = int_to_ip(k.value)
 
    if max_ip:
        print(f"\nTop IP: {max_ip}, Traffic: {max_traffic} bytes")
 
    # 清空map，重新统计
    ip_流量统计.clear()
 
    # 等待一段时间
    import time
    time.sleep(5)

代码解释：

• 使用bcc库来加载eBPF程序和访问eBPF map。
• int_to_ip函数用于将整数IP地址转换为字符串格式。
• 程序循环读取ip_流量统计 map中的数据，并打印每个IP地址对应的流量大小。
• 同时，程序会找出流量最大的IP地址，并打印出来。
• 每次循环结束后，程序会清空ip_流量统计 map，以便重新统计。

4. 注意事项

• eBPF程序的安全性： eBPF程序运行在内核中，因此必须保证其安全性。eBPF程序在加载前会经过内核的验证器进行验证，以防止程序崩溃或恶意行为。编写eBPF程序时，应该尽量避免复杂的循环和递归，以及访问非法内存。
• eBPF程序的性能： eBPF程序的性能直接影响系统的性能。应该尽量减少eBPF程序的执行时间，避免过度消耗CPU资源。可以使用bpf_printk函数来打印调试信息，但应该在生产环境中禁用该函数，因为它会影响性能。
• eBPF map的大小： eBPF map的大小应该根据实际情况进行调整。如果map太小，可能会导致数据丢失。如果map太大，可能会浪费内存资源。
• 权限问题： 加载和运行eBPF程序需要root权限。
• 内核版本兼容性： 不同的内核版本可能对eBPF的支持有所不同。应该选择合适的eBPF程序和工具链，以保证程序的兼容性。

5. 总结

本文介绍了如何使用eBPF技术来监控特定网络端口的流量，并按照源IP地址进行统计。通过编写eBPF程序、加载和运行eBPF程序、以及编写用户态程序，我们可以实现对网络流量的实时监控和分析。eBPF技术在网络监控、安全分析等方面具有广泛的应用前景。