简介：这份源码资源面向安全运维人员、后端开发者及网络安全爱好者，围绕在线教育平台遭遇DDoS攻击的真实案例，提供可落地的防御实战参考。内容涵盖带宽耗尽、协议资源耗尽与应用层耗尽三类攻击的原理剖析，并延伸至基础防护、高防IP与清洗服务、动态带宽调度、CDN流量分发、限流熔断降级及高可用架构等全链路防御方案，帮助读者建立系统化的抗DDoS思路。资源包共3个文件，以html页面为主体，辅以inscode配置与gitignore文件，整体约7KB，结构轻量，便于快速查阅与二次整理。目前已有298人学习下载，适合希望从案例中理解攻击过程、掌握防御策略并对照自身业务查漏补缺的读者参考。
1. DDoS 攻击防御实战：从被刷到 40Gbps 到把清洗规则压进 8 毫秒
凌晨两点被告警电话叫醒，打开监控一看，某台对外网关的入向流量从日常 300Mbps 直接拉到 40Gbps，业务端口全红。这不是电影桥段，是很多做在线服务的人迟早要碰上的场景。DDoS 攻击防御实战这件事，核心不是「买高防」三个字能概括的，它是一整套从流量采集、特征识别、清洗策略到回注验证的工程链路。这篇笔记面向的是手里有真实业务、需要自己搭一套可观测可干预的防御体系的工程师，不是讲 DDoS 是什么的科普。我会把源码级的实现思路拆开，包括采集层怎么写、限速桶怎么设计、黑白名单怎么热更新，以及那些只有被打过才懂的坑。看完你应该能自己搭一个最小可用的防御原型，并知道每一层该盯哪些参数。
2. 先搞清楚你面对的是哪种流量：攻击分类与采集层落地
在写任何一行防御代码之前，必须先能看见流量。很多团队一上来就买清洗设备，结果连攻击是从哪个 IP、哪个协议、哪个端口打进来的都说不清，清洗规则全靠猜，这就是典型的翻车起点。DDoS 攻击防御实战的第一步，是把采集层做扎实。
2.1 按协议层把攻击拆成三类可处理对象
从工程视角，我一般把攻击流量分成三类，因为它们的处理路径完全不同。
第一类是体积型攻击，典型是 UDP 反射和 ICMP 洪泛，特征是包大、无连接状态、源 IP 大量伪造。这类流量在采集层就能看到明显的包速率突增，处理手段以近源清洗和速率限制为主。
第二类是协议型攻击，比如 SYN 洪泛，特征是半连接队列被打满，服务端netstat里SYN_RECV数量暴涨。这类攻击包本身不大，但会耗尽内核连接跟踪表，处理要靠 SYN Cookie 和连接跟踪调优。
第三类是应用层攻击，比如 HTTP 慢速请求和高频接口调用，特征是 QPS 异常但流量不大，传统流量清洗根本看不见。这类必须靠应用层指纹和业务指标来识别。
把这三类分开，是因为它们的采集点不同：体积型在网卡层抓，协议型在内核连接跟踪层看，应用层在反向代理日志里分析。混在一起做，规则一定互相打架。
2.2 用 eBPF 在网卡层做无侵入采集
采集层我推荐用 eBPF，因为它不需要改业务代码，也不像 tcpdump 那样在高流量下把 CPU 吃满。下面是一个最小可用的采集脚本，统计每个源 IP 的包速率。
# collector.py # 依赖：bcc（pip install bcc），内核需支持 eBPF from bcc import BPF import time # eBPF 程序：按源 IP 统计包数，写入 hash map bpf_text = """ #include <uapi/linux/ptrace.h> #include <linux/ip.h> BPF_HASH(pkt_count, u32, u64); // key: 源IP, value: 包计数 int count_pkt(struct __sk_buff *skb) { void *data = (void *)(long)skb->data; void *data_end = (void *)(long)skb->data_end; struct iphdr *iph = data; if ((void *)(iph + 1) > data_end) return 0; u32 src = iph->saddr; u64 zero = 0, *val; val = pkt_count.lookup_or_init(&src, &zero); if (val) (*val)++; return 0; } """ b = BPF(text=bpf_text) # 挂到网卡入口，eth0 按实际网卡名替换 b.attach_raw_socket(fn_name="count_pkt", iface="eth0") try: while True: time.sleep(5) print("---- 5s 窗口包速率 TOP ----") items = sorted(b["pkt_count"].items(), key=lambda x: x[1].value, reverse=True) for k, v in items[:10]: ip = ".".join(str((k.value >> (8 * i)) & 0xFF) for i in range(4)) print(f"{ip}: {v.value} pkt/5s") b["pkt_count"].clear() except KeyboardInterrupt: pass这段代码的逻辑是：在内核态挂一个 eBPF 程序，每来一个包就按源 IP 累加计数，用户态每 5 秒读一次 map 并排序输出。关键参数有两个，iface要换成你实际对外网卡名，time.sleep(5)是统计窗口，窗口太短会抖动，太长会延迟发现攻击。注意 eBPF 需要内核 4.9 以上，且要 root 权限运行。
2.3 采集层必须同时记录的三个维度
只统计包速率是不够的。我一般要求采集层同时输出三个维度：源 IP 的包速率、目的端口的连接新建速率、以及每个源 IP 的平均包大小。为什么？因为体积型攻击包大且速率高，协议型攻击包小但连接新建快，应用层攻击包大小正常但请求频率高。三个维度交叉看，基本能在一分钟内判断出攻击类型。
采集数据落到本地时序库时，建议保留最近 15 分钟的秒级粒度，超过 15 分钟的降采样到分钟级。这样既能在攻击时快速回溯，又不会把磁盘写爆。很多团队采集做了但没做降采样，被打一次磁盘就满了，这是血泪经验。
3. 清洗策略怎么写：限速桶、黑白名单与热更新
采集层告诉你「谁在打」，清洗层负责「怎么挡」。这一层是 DDoS 攻击防御实战里最容易写错的地方，因为限速算法选错，正常用户会被误伤，攻击却没挡住。
3.1 令牌桶与漏桶的选择依据
限速算法常见两种：令牌桶和漏桶。令牌桶允许突发流量，只要桶里有令牌就放行，适合业务本身有波峰的场景；漏桶强制恒定速率输出，适合对下游有严格保护需求的场景。
我的经验是：对源 IP 做限速用令牌桶，因为正常用户的请求本身就是突发的；对下游服务做保护用漏桶，因为下游处理能力是恒定的。下面是一个令牌桶的实现。
# token_bucket.py import time import threading class TokenBucket: def __init__(self, rate, capacity): # rate: 每秒补充令牌数；capacity: 桶容量 self.rate = rate self.capacity = capacity self.tokens = capacity self.last_time = time.time() self.lock = threading.Lock() def allow(self, n=1): with self.lock: now = time.time() # 按时间差补充令牌 self.tokens = min( self.capacity, self.tokens + (now - self.last_time) * self.rate ) self.last_time = now if self.tokens >= n: self.tokens -= n return True return False # 每个源 IP 一个桶，rate=100/s，capacity=200 buckets = {} def check(ip): if ip not in buckets: buckets[ip] = TokenBucket(rate=100, capacity=200) return buckets[ip].allow()逻辑说明：rate决定长期平均放行速率，capacity决定能容忍多大的突发。这两个参数怎么定？我的做法是先看业务正常时单 IP 的 P99 请求速率，把rate设成 P99 的 1.5 倍，capacity设成 P99 的 3 倍。这样正常用户几乎不会触发限速，而攻击 IP 会在几秒内把桶打空然后被持续拒绝。注意buckets字典要加过期清理，否则被大量伪造源 IP 攻击时内存会涨爆，这是必踩的坑。
3.2 黑白名单的热更新机制
清洗规则不能写死在代码里，因为攻击特征几分钟就变。我一般用「本地文件 + 定时 reload」的方式做热更新，简单可靠。
# rule_loader.py import json import time import threading class RuleManager: def __init__(self, path, interval=5): self.path = path self.interval = interval self.whitelist = set() self.blacklist = set() self._load() threading.Thread(target=self._watch, daemon=True).start() def _load(self): try: with open(self.path) as f: data = json.load(f) self.whitelist = set(data.get("whitelist", [])) self.blacklist = set(data.get("blacklist", [])) except Exception as e: print(f"规则加载失败，保留旧规则: {e}") def _watch(self): while True: time.sleep(self.interval) self._load() def verdict(self, ip): if ip in self.whitelist: return "allow" if ip in self.blacklist: return "deny" return "limit"规则文件格式建议用 JSON，字段就whitelist和blacklist两个数组。interval=5表示每 5 秒重载一次，这个值不要太小，否则频繁读盘；也不要太大，否则封禁生效慢。加载失败时保留旧规则这个设计很关键，避免规则文件写坏导致防御整体失效。
3.3 把清洗决策落到转发路径上
规则算出来了，得让它真正生效。常见做法有三种：在 iptables 层加 DROP 规则、在 XDP 层直接丢包、在应用层反向代理里拒绝。三者的生效速度和影响范围不同。
| 方式 | 生效延迟 | 吞吐影响 | 适用场景 |
|---|---|---|---|
| iptables | 毫秒级 | 中等 | 中小流量、快速上线 |
| XDP | 微秒级 | 极低 | 大流量、网卡支持 |
| 应用层 | 十毫秒级 | 低 | 应用层攻击、精细规则 |
我一般先用 iptables 快速止血，同时把高频攻击 IP 同步到 XDP 做长期拦截。iptables 规则示例：
# 封禁单个 IP，限制每秒新建连接数 iptables -I INPUT -s 203.0.113.10 -j DROP # 对某端口做连接速率限制，超过 50/s 丢弃 iptables -I INPUT -p tcp --dport 443 -m connlimit --connlimit-above 50 -j DROPconnlimit-above 50表示单 IP 对 443 端口的并发连接超过 50 就丢，这个值要按业务调，设太小会误伤正常长连接用户。iptables 规则多了以后匹配性能会下降，所以只把最紧急的规则放这里，长期规则走 XDP。
4. 避坑与排查：被打时最容易做错的五件事
这一章是我自己踩过的坑，每一条都对应真实故障。DDoS 攻击防御实战里，防御代码写对只是及格，知道什么时候不该做什么才是关键。
4.1 现象：加了限速后正常用户大面积超时
原因：限速桶的 key 用了 NAT 出口 IP，导致整个办公网或运营商 NAT 后的用户共用一个桶，被少数高频用户打满。
解决：限速 key 不能只用源 IP，要结合业务会话标识。对 HTTP 场景，用源IP + User-Agent + 会话Cookie做复合 key；对 API 场景，用源IP + API Key。这样 NAT 后的不同用户不会互相影响。代价是内存占用上升，需要配合 LRU 淘汰。
4.2 现象：攻击流量明明被丢了，业务还是不可用
原因：攻击打的是带宽而不是服务，即使包被 DROP，流量依然占满了上行链路，正常请求的回包出不去。
解决：这种情况本地清洗无效，必须做近源清洗或流量牵引。判断方法很简单，看网卡入向流量是否接近链路带宽上限。如果是，本地任何软件手段都救不了，要提前和上游运营商或清洗中心对接。这个坑的教训是：防御方案要分「带宽型」和「资源型」两套，不能一套打天下。
4.3 现象：eBPF 采集程序跑一段时间后 CPU 飙高
原因：hash map 没有设置最大条目数，被伪造源 IP 攻击时 map 无限增长，查找和遍历开销越来越大。
解决：给 map 设置max_entries，并在用户态做定期清理。eBPF 里可以用BPF_HASH(pkt_count, u32, u64, 100000)限制十万条，超出后新 key 无法插入，但不会拖垮内核。同时用户态每轮读取后清理低频 key。
4.4 现象：黑名单封了攻击 IP，几分钟后攻击换 IP 继续
原因：只做了静态黑名单，没有做行为聚类。攻击者用 IP 池轮换，单个 IP 封禁意义不大。
解决：从「封 IP」升级到「封网段 + 封行为特征」。比如同一 /24 网段内多个 IP 同时触发限速，就把整个 /24 加黑；或者提取攻击包的 TTL、TCP 窗口大小等指纹，做指纹级拦截。这一步需要采集层保留更多字段，所以采集设计要提前留扩展位。
4.5 现象：清洗规则生效后，监控显示攻击停了，但业务指标没恢复
原因：攻击可能已经打到了应用层，流量层看不到，但后端数据库或缓存已经被打挂，恢复需要时间。
解决：防御体系必须和应用指标打通。我一般会在清洗层加一个「放行后验证」环节，对放行的请求做抽样，检查后端响应时间。如果放行流量正常但后端 P99 持续偏高，说明攻击已经穿透到应用层，要立刻切到应用层限流。这个环节很多团队没有，导致「流量正常但业务挂了」的诡异现象。
5. 进阶：把防御规则做成可回滚的版本化配置
前面讲的都是「怎么挡」，这一章讲「怎么安全地改规则」。防御规则本身也是代码，改错了比不挡还危险。我现在的习惯是：所有清洗规则走版本管理，每次变更都能一键回滚。
5.1 规则版本化的最小实现
规则文件不要直接覆盖，而是每次写入新版本文件，用一个current软链接指向生效版本。
# 规则目录结构 # rules/ # v20240101_120000.json # v20240101_123000.json # current -> v20240101_123000.json # 写入新规则并切换 cp new_rules.json rules/v$(date +%Y%m%d_%H%M%S).json ln -sfn rules/v$(date +%Y%m%d_%H%M%S).json rules/current # 回滚到上一版本 ln -sfn rules/v20240101_120000.json rules/currentln -sfn的-n参数很关键，它保证替换软链接时不会把目标当成目录。回滚只需要改软链接指向，规则加载器下一次 reload 就会读到旧版本，整个过程不超过 5 秒。
5.2 变更前的灰度验证方法
规则改完不要直接全量生效，先做影子模式：新规则只记录不拦截，对比新旧规则的判定差异。
# shadow.py def shadow_check(ip, old_verdict, new_verdict): if old_verdict != new_verdict: # 记录差异，但不执行新规则 log.warning(f"规则差异 ip={ip} old={old_verdict} new={new_verdict}") return old_verdict # 影子模式返回旧规则结果影子模式跑 10 分钟，如果差异率低于 1%，说明新规则和旧规则行为基本一致，可以切全量；如果差异率很高，说明新规则有误伤风险，先查清楚再切。这个方法帮我避免过好几次「新规则把正常爬虫全封了」的事故。
5.3 一个我常用的验证清单
每次规则变更后，我会按这个清单过一遍：
| 检查项 | 方法 | 通过标准 |
|---|---|---|
| 正常流量是否受影响 | 对比变更前后 5 分钟业务 QPS | 波动小于 5% |
| 攻击是否被有效拦截 | 看黑名单命中计数 | 命中数持续增长 |
| 规则加载是否成功 | 看加载日志 | 无异常报错 |
| 回滚是否可用 | 手动切一次软链接 | 5 秒内生效 |
| 内存是否稳定 | 看进程 RSS | 10 分钟内无持续增长 |
这个清单不复杂，但坚持做能省掉大量救火时间。我自己的教训是：有一次改限速参数忘了验证内存，结果桶字典没清理，跑了半小时 OOM 了，业务跟着挂。从那以后，任何规则变更我都先跑影子模式，再看内存曲线，最后才切全量。
防御这件事没有一劳永逸，攻击手法在变，你的规则也得跟着迭代。把采集做细、把规则版本化、把回滚练熟，比追求某个「最强防御算法」实在得多。希望帮到你。
本文还有配套的精品资源，点击获取
