免责声明
本文仅用于技术讨论与学习，利用此文所提供的信息而造成的任何直接或者间接的后果及损失，均由使用者本人负责，文章作者及本公众号不为此承担任何责任。
DDoS探测
BCC作为一个流行的ebpf开发方案，提供了很多案例供开发者学习，其中/examples/tracing/dddos.py是BCC官方提供的简单探测DDoS攻击的案例。探测DDoS攻击对于网络防御很有帮助，值得学习。
内核态源码
1. 检测的基本思想是如果两个先后到达的时间差过短，就给计数器加一，当计数器达到阈值时告警，因此我们需要先给出指标：
2. 创建eBPF哈希表rcv_packets，用于存储数据包数量和时间戳：
BPF_HASH(rcv_packets);3. 定义数据结构detectionPackets，包含字段nb_ddos_packets，用于存储检测到的DDoS数据包总数：
struct detectionPackets {u64 nb_ddos_packets;};
4. 创建性能事件（perf events）的输出events，允许用户空间程序接收内核空间程序的信息：
BPF_PERF_OUTPUT(events);5. 定义函数detect_ddos：
int detect_ddos(struct pt_regs *ctx, void *skb){6. 定义和初始化变量，以跟踪接收到的数据包数和时间戳：
struct detectionPackets detectionPacket = {};// Used to count number of received packetsu64 rcv_packets_nb_index = 0, rcv_packets_nb_inter=1, *rcv_packets_nb_ptr;// Used to measure elapsed time between 2 successive received packetsu64 rcv_packets_ts_index = 1, rcv_packets_ts_inter=0, *rcv_packets_ts_ptr;
7. 通过lookup函数检查rcv_packets哈希表中索引为0和1的元素。若元素存在，计算时间戳差并更新包计数器：如果两个连续数据包的时间差小于LEGAL_DIFF_TIMESTAMP_PACKETS，计数器加一；否则重置计数器：
rcv_packets_nb_ptr = rcv_packets.lookup(&rcv_packets_nb_index);rcv_packets_ts_ptr = rcv_packets.lookup(&rcv_packets_ts_index);!= 0 && rcv_packets_ts_ptr != 0){rcv_packets_nb_inter = *rcv_packets_nb_ptr;rcv_packets_ts_inter = bpf_ktime_get_ns() - *rcv_packets_ts_ptr;< LEGAL_DIFF_TIMESTAMP_PACKETS){rcv_packets_nb_inter++;else {rcv_packets_nb_inter = 0;}
8. 如果计数器超过MAX_NB_PACKETS，说明检测到异常流量，此时将会向性能事件输出中写入信息：
if(rcv_packets_nb_inter > MAX_NB_PACKETS){detectionPacket.nb_ddos_packets = rcv_packets_nb_inter;events.perf_submit(ctx, &detectionPacket, sizeof(detectionPacket));}
9. 最后，更新哈希表rcv_packets的数据包计数和时间戳信息：
rcv_packets_ts_inter = bpf_ktime_get_ns();&rcv_packets_nb_inter);&rcv_packets_ts_inter);
用户态
1. 加载eBPF程序prog至内核，并在Python用户态创建一个BPF对象：
# Loads eBPF programb = BPF(text=prog)
2. 将内核函数ip_rcv使用kprobe机制附加到detect_ddos函数，ip_rcv是内核网络栈中处理接收IP数据包的函数：
b.attach_kprobe(event="ip_rcv", fn_name="detect_ddos")
3. 定义DetectionTimestamp结构体类映射内核事件数据结构，包含检测到的DDoS数据包数量：
class DetectionTimestamp(ct.Structure):_fields_ = [("nb_ddos_packets", ct.c_ulonglong)]
4. 定义trigger_alert_event函数作为回调来处理内核发出的性能事件，将数据强制类型转换为DetectionTimestamp结构体，并打印当前时间和检测到的DDoS数据包数量：
def trigger_alert_event(cpu, data, size):event = ct.cast(data, ct.POINTER(DetectionTimestamp)).contentsprint("%-26s %s %ld" % (datetime.datetime.now(),"DDOS Attack => nb of packets up to now : ", event.nb_ddos_packets))
5. 创建性能事件的缓冲区，并通过b["events"].open_perf_buffer(trigger_alert_event)侦听事件，设置回调函数：
# loop with callback to trigger_alert_eventb["events"].open_perf_buffer(trigger_alert_event)
while 1:try:b.perf_buffer_poll()except KeyboardInterrupt:exit()
运行效果
原文始发于微信公众号（赛博安全狗）：【eBPF】BCC简单探测DDoS攻击
- 左青龙
- 微信扫一扫
-
- 右白虎
- 微信扫一扫
-
评论