简介这份源码资源面向安全运维人员、后端开发者及网络安全爱好者围绕在线教育平台遭遇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 # 依赖bccpip 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(textbpf_text) # 挂到网卡入口eth0 按实际网卡名替换 b.attach_raw_socket(fn_namecount_pkt, ifaceeth0) try: while True: time.sleep(5) print(---- 5s 窗口包速率 TOP ----) items sorted(b[pkt_count].items(), keylambda x: x[1].value, reverseTrue) 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, n1): 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 一个桶rate100/scapacity200 buckets {} def check(ip): if ip not in buckets: buckets[ip] TokenBucket(rate100, capacity200) 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, interval5): self.path path self.interval interval self.whitelist set() self.blacklist set() self._load() threading.Thread(targetself._watch, daemonTrue).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两个数组。interval5表示每 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 秒内生效内存是否稳定看进程 RSS10 分钟内无持续增长这个清单不复杂但坚持做能省掉大量救火时间。我自己的教训是有一次改限速参数忘了验证内存结果桶字典没清理跑了半小时 OOM 了业务跟着挂。从那以后任何规则变更我都先跑影子模式再看内存曲线最后才切全量。防御这件事没有一劳永逸攻击手法在变你的规则也得跟着迭代。把采集做细、把规则版本化、把回滚练熟比追求某个「最强防御算法」实在得多。希望帮到你。本文还有配套的精品资源点击获取