1. 为什么我用Python自己做DoS检测与防御1.1 先说结论这件事到底值不值得自己写如果你去搜DoS攻击防御搜出来的基本都是各种商业WAF、高防IP、流量清洗设备的广告或者让你买某云厂商的DDoS防护包。这些方案当然有效但有一个很现实的问题贵而且很多场景根本用不上那么重的东西。我说一个具体场景。某公司内部有个业务系统用户量不大但业务关键挂了就要出事。之前被人用脚本刷接口单台服务器CPU直接打满服务完全不可用。查了日志发现就是几个IP在疯狂请求登录接口并发也不高大概每秒几十个请求就把业务打死了。这种情况你上高防没必要。你装商业WAF一套下来一年几万块预算不批。所以我才花了两个周末用Python搭了一套轻量级的DoS检测与防御系统。它能实时监控流量特征、自动封禁异常IP、触发限流而且全部代码加起来不到500行部署也简单——一台普通服务器Python 3环境就能跑。这篇文章就把整套东西拆开讲清楚我是怎么设计检测指标的代码怎么写的踩过哪些坑以及最重要的——怎么避免把自己锁在门外。1.2 Python做这件事的天然优势选Python不是因为它最快恰恰相反Python在高并发流量处理上并不占优。但我依然选了它理由有三条。第一开发效率。整套系统从需求到上线核心代码不到500行如果用C或Go写一周都不一定调试完。而且网络安全本身就是对抗变化的场景攻击手段隔几个月就会翻新你需要频繁调整检测逻辑。脚本语言在这方面的优势是碾压级的。第二生态完整。Python在流量分析、网络抓包、系统监控方面有非常成熟的库psutil拿系统指标、scapy做包解析、FastAPI/Flask写个Web服务也不费劲甚至可以直接复用。检测逻辑用Python表达比用iptables规则硬写要直观得多。第三运维门槛低。团队成员不一定都熟悉Go或C但Python基本人人能看懂。出了问题维护的人不需要理解复杂的底层实现改改阈值参数就能上线这一点在中小团队里非常重要。当然它的局限我也得说清楚。Python做深度包检测DPI或者超大规模流量分析每秒上百万包是力不从心的那种场景需要C/DPDK或者专用硬件。但如果你只是保护一台或者几台业务服务器应对CC攻击、慢速连接耗尽、接口刷爆这类应用层DoSPython绰绰有余。2. 检测的核心思路你到底在测什么2.1 四个关键指标比流量异常四个字靠谱得多很多人一上来就想写检测流量突增的脚本但流量突增本身是个模糊概念。真正落地的时候我把它拆成了四个可量化的指标指标检测对象典型异常特征QPS每秒请求数单个IP或全局限流单IP请求频率飙升至基线的数倍活跃连接数服务器并发连接连接数陡增且长时间不释放SYN半开连接TCP握手状态SYN包大量涌入连接队列被塞满资源水位CPU、内存、带宽CPU持续打满、内存占用异常攀升这四个指标分别对应不同的攻击层次。QPS和连接数针对的是应用层攻击比如CC攻击、频繁请求SYN半开连接针对的是网络层SYN Flood资源水位则是兜底——不管对方用什么招只要最终打到了资源瓶颈系统必须能感知到。有一点容易被忽略单看任何一个指标都可能误判。比如QPS高了可能只是正常的业务高峰连接数多了也许是爬虫在正常抓取。所以我的策略是多指标联合判定后面代码部分会具体演示。2.2 主动轮询与被动旁路两种采样姿势检测系统怎么获取数据我实际用下来有两种主流方式各有取舍。第一种是主动轮询就是用psutil这类库定时去读系统指标。比如每2秒读一次/proc/net/tcp统计处于SYN_RECV状态的连接数每5秒读一次CPU和内存占用。这种方式实现简单性能开销小适合检测网络层攻击和资源异常。第二种是被动旁路通常用scapy抓包或者在nginx层挂日志分析脚本。被动方式能看到更细的请求特征URL、UA、请求头但开销更大、处理速度慢在高流量下容易成为瓶颈。我实际的做法是网络层异常用主动轮询应用层异常则依赖Web服务器日志的旁路分析。2.3 基线数据没有对比就没有异常这是整套系统里最关键、也最容易被新手跳过的一步——先建立基线再谈检测。我刚开始写防御脚本的时候犯过一个错拍脑子定阈值单IP每秒超过20次请求就封禁。结果上线第一天就把一个正常的API网关IP封了因为那个IP聚合了后端几十台机器的请求。后来我改成先收集一周的流量数据统计出每个指标的正常分布范围再用基线值×系数动态计算阈值。具体做法很简单把每个时间窗口比如5分钟的QPS、连接数、CPU水位记录下来求平均值和标准差。正常波动范围就是均值±3倍标准差超过这个范围才判定为异常。攻击来了指标会瞬间突破这个范围而正常业务高峰通常不会触发3σ以外的波动。这套思路来自监控领域的经典方法不用搞太复杂的机器学习效果就很好。3. 实操构建一个轻量级检测与自动防御系统3.1 项目结构总览我把整个系统拆成了四个模块各司其职dos_defense/ ├── metrics/ # 数据采集 │ ├── counter.py # QPS与连接数统计 │ ├── resource.py # CPU/内存/带宽监控 │ └── sniffer.py # SYN半开连接检测可选 ├── detector/ # 异常判定 │ ├── rules.py # 基线判定与告警规则 │ └── alerter.py # 告警通知邮件/钉钉/文件 ├── defender/ # 自动防御动作 │ ├── firewall.py # iptables封禁/解封 │ └── ratelimit.py # 应用层限流中间件 ├── config.yaml # 阈值、白名单、封禁时长 └── main.py # 主控调度设计的时候我特意把检测和防御拆开。检测模块只负责出报告、发告警防御模块才动手封禁。这样做的原因是检测模块出错了顶多误报防御模块出错了就是误杀两者混在一起调试的时候会非常痛苦。3.2 第一层基于请求频率的实时计数器这一层解决的是某个IP在短时间内请求过于频繁的问题。核心数据结构是滑动窗口计数器。我不用简单的计数器累加因为那没法处理1秒内100次请求和10秒内均匀分布100次请求的区别——两种模式对服务的伤害完全不同。滑动窗口的思路是把时间切成小块比如每1秒一个小格只保留最近30秒的数据再加上每秒的QPS计算。代码是这样的import time from collections import deque, defaultdict class SlidingWindowCounter: def __init__(self, window_size30, bucket_size1): self.window_size window_size self.bucket_size bucket_size self.buckets defaultdict(deque) # ip - deque[(timestamp, count)] self._lock threading.Lock() def record(self, ip, count1): now int(time.time()) with self._lock: bucket self.buckets[ip] if bucket and bucket[-1][0] now: # 还在当前秒内直接累加 ts, c bucket[-1] bucket[-1] (ts, c count) else: bucket.append((now, count)) # 清理超出时间窗口的旧数据 while bucket and bucket[0][0] now - self.window_size: bucket.popleft() if not bucket: del self.buckets[ip] def get_qps(self, ip): now int(time.time()) bucket self.buckets.get(ip, []) total sum(c for ts, c in bucket if ts now - self.window_size) return total / self.window_size这里有个细节值得说一下注意加锁。counter会被多线程并发调用我实际是在每个HTTP请求的入口中间件里调用record方法不加锁在低并发下看不出问题但流量一上来dict并发读写会直接抛异常或者丢数据。实际部署的时候我把这个计数器嵌进了FastAPI的中间件里app.middleware(http) async def dos_detect_middleware(request: Request, call_next): ip request.client.host counter.record(ip) qps counter.get_qps(ip) if qps config.trigger_threshold: defender.ban_ip(ip, reasonfHigh QPS: {qps:.1f}) return Response(status_code403, contentForbidden) return await call_next(request)3.3 第二层基于连接状态的深度检测QPS检测只能覆盖应用层面对SYN Flood这类网络层攻击就抓瞎了——因为攻击流量根本不会到达应用层你的中间件压根收不到请求。这时候要看TCP连接状态。我用的方案是解析系统的连接状态统计。Linux下可以直接读/proc/net/tcp也可以用psutil更优雅地获取import psutil import time class ConnectionMonitor: def __init__(self, syn_threshold500, conn_threshold2000): self.syn_threshold syn_threshold self.conn_threshold conn_threshold def check(self): # 统计各种TCP状态的数量 tcp_stats {} for conn in psutil.net_connections(kindtcp): state conn.status tcp_stats[state] tcp_stats.get(state, 0) 1 syn_recv tcp_stats.get(SYN_RECV, 0) established tcp_stats.get(ESTABLISHED, 0) severe False if syn_recv self.syn_threshold: severe True print(f[告警] SYN半开连接数异常: {syn_recv}) if established self.conn_threshold: severe True print(f[告警] 活跃连接数异常: {established}) return { syn_recv: syn_recv, established: established, severe: severe, }为什么SYN_RECV数量是关键解释一下原理。TCP握手要经过三次握手服务器收到客户端的SYN包后会进入SYN_RECV状态并等待客户端的ACK。正常的握手过程这个状态只持续几毫秒。但如果攻击者只发SYN包不回应服务器的半开连接队列就会被填满新来的正常连接无法建立这就是经典的SYN Flood。我实测的阈值参考一台4核8G的普通Web服务器正常业务下SYN_RECV通常是个位数到几十如果持续超过300基本可以断定在被打超过1000服务已经开始不可用了。这几个数大家可以根据自己的机器配置调整我给的只是参考值。3.4 第三层资源监控与自保护前两层都防不住的时候最后兜底的是系统资源监控。不管攻击手法多花哨最终都会体现在CPU、内存和带宽上。这一层不追求提前发现攻击它的核心目标是别让服务器彻底死掉留一条活路给自己运维。我写了一个简单的资源监控循环import psutil class ResourceMonitor: def __init__(self, cpu_limit85, mem_limit90): self.cpu_limit cpu_limit self.mem_limit mem_limit def check(self): cpu psutil.cpu_percent(interval1) mem psutil.virtual_memory().percent # 这里也可以加带宽统计psutil.net_io_counters() if cpu self.cpu_limit or mem self.mem_limit: # 记录当前最活跃的连接方便后续定位攻击源 self._dump_top_connections(20) return {alert: True, cpu: cpu, mem: mem} return {alert: False, cpu: cpu, mem: mem}有个小技巧值得分享当资源告警触发时我会顺手把当前连接数最多的20个IP打印出来。这条信息在人工介入排查时极有价值——它能直接告诉你谁在吃资源省掉翻半天日志的时间。def _dump_top_connections(self, top_n20): conn_count {} for conn in psutil.net_connections(kindtcp): if conn.raddr: ip conn.raddr.ip conn_count[ip] conn_count.get(ip, 0) 1 sorted_ips sorted(conn_count.items(), keylambda x: -x[1])[:top_n] for ip, cnt in sorted_ips: print(f {ip}: {cnt} connections)3.5 自动防御限流、封禁与恢复检测到异常之后怎么处理分寸感很重要。我把它分成三级第一级限流。对于QPS超标的IP不直接封先做限流。比如正常阈值是每秒20次超过后把该IP的请求速度限制到每秒5次。这样即使误判了影响也可控。限流我用的是令牌桶算法Python里可以直接用内置的rate limiter或者自己写个简单的import time import threading class TokenBucket: def __init__(self, rate5, capacity10): self.rate rate self.capacity capacity self.tokens capacity self.last time.time() self._lock threading.Lock() def acquire(self): with self._lock: now time.time() self.tokens min(self.capacity, self.tokens (now - self.last) * self.rate) self.last now if self.tokens 1: self.tokens - 1 return True return False第二级封禁。QPS持续超过红线或者SYN半开连接数异常就直接封IP。我通过调用iptables来实现封禁这是Linux上最直接高效的方式封禁后内核层面直接丢弃包不占应用资源。import subprocess class FirewallManager: def __init__(self, ban_duration600): self.ban_duration ban_duration self.banned {} # ip - 解封时间戳 def ban_ip(self, ip, reason): if ip in self.banned.keys(): return # 已封禁不重复操作 # 白名单检查 if ip in WHITE_LIST: return expire time.time() self.ban_duration self.banned[ip] expire subprocess.run([iptables, -A, INPUT, -s, ip, -j, DROP], checkFalse) print(f[防御] 已封禁 {ip}, 原因: {reason}, {self.ban_duration}s后自动解封) def unban_all(self): for ip, expire in list(self.banned.items()): if time.time() expire: subprocess.run([iptables, -D, INPUT, -s, ip, -j, DROP], checkFalse) del self.banned[ip]第三级恢复。封禁必须带自动解封不能永久封。我设定默认封禁10分钟超时后自动移除规则。理由很简单攻击者的IP经常是动态的一个IP封久了没意义而正常用户如果被误封等10分钟也能恢复访问不至于造成不可逆的影响。3.6 部署与告警整套系统我用一个主循环调度没有上复杂的任务队列import time import threading def monitor_loop(): conn_monitor ConnectionMonitor() res_monitor ResourceMonitor() while True: result conn_monitor.check() if result[severe]: # 这里只打印封禁动作由专门的防御线程做 pass res_monitor.check() time.sleep(2) def defense_loop(): while True: fw.unban_all() # 定期检查并解封到期的IP time.sleep(30) if __name__ __main__: threading.Thread(targetmonitor_loop, daemonTrue).start() threading.Thread(targetdefense_loop, daemonTrue).start() # 主进程继续跑业务服务告警这一块我做得比较简单但有效异常事件写入本地日志文件同时调用一个webhook地址发送到内部的告警群。如果你是在个人项目里用甚至可以写个脚本检测到攻击后自动把关键信息发到你手机实现「人不在电脑前也能知道服务器在挨打」。4. 防御策略的取舍检测到之后怎么办4.1 应用层限流与网络层封禁别混为一谈很多教程把限流和封禁当成一回事其实两者解决的层次完全不同。限流是应用层动作请求到达Web服务后才被处理能精细地控制每个来源的速率封禁是网络层动作在内核层面直接丢弃数据包效率极高但手段粗糙。我的经验是两个都要用但场景不同。对于攻击特征明显的比如SYN Flood、单IP瞬间上万请求直接封禁因为这时候已经不用讲道理了先保住服务再说。对于处于灰度阶段的异常比如某个IP请求量是基线的5倍但又不到红线优先限流给它一个改过自新的机会避免误杀。4.2 与现有基础设施的三种联动模式你的Python检测系统不应该是孤岛它得能和现有的基础设施配合。我实践过三种联动方式大家按自己环境选第一种nginx层联动。检测系统发现异常IP后不直接操作iptables而是把IP追加到nginx的封禁配置里然后reload。这种方式的优势是只针对HTTP协议生效不影响SSH等其他服务。def add_nginx_blacklist(ip): with open(/etc/nginx/conf.d/blacklist.conf, a) as f: f.write(fdeny {ip};\n) subprocess.run([nginx, -s, reload], checkFalse)第二种iptables直连。就是我上面演示的方式简单粗暴效果最好。但要注意如果跑的云服务器有些云平台的安全组规则可能和iptables冲突需要在控制台同时配置。第三种业务层限流。在Web框架的中间件里实现比如我在3.2节写的那个中间件。这种方式的粒度最细可以做到针对特定URL、特定用户限流但只保护当前服务对其他端口无能为力。我最终的方案是三管齐下中间件做精细化限流iptables做兜底封禁nginx层做HTTP流量管控。大家可以根据自己的场景裁剪。4.3 误报与误杀防御系统的副作用很多人忽略了防御系统本身的副作用——它误杀正常用户的风险。这一点我在第6节踩坑里还会详细说这里先给出三条设计原则。第一宁可漏杀不可错杀。检测阈值宁高勿低。业务挂了还能恢复但把正常用户全封了业务投诉就来了后来你连解释的机会都没有。这也是我为什么引入基线数据动态阈值的原因——拍脑袋设的阈值误杀率一定高。第二所有防御动作必须可逆。封禁要有超时时间限流要有恢复机制。我见过有人把攻击IP永久加入黑名单结果那个IP后面换给了正常用户用导致正常用户永远无法访问。封禁永远要带过期时间。第三白名单和灰度是保命符。内部监控系统、合作方的IP、自家运维的IP一定要有白名单机制。否则攻击来的时候你自动封禁可能顺手把自己监控系统的IP给封了那就真的瞎了。5. 实测中的坑与排查技巧5.1 高并发下Counter本身的性能问题第一次压测的时候我发现一个尴尬的问题防御系统自己先扛不住了。SlidingWindowCounter在单IP高频请求下还好但几十个IP同时高并发时Python的dict操作和高频加锁让接口延迟从2ms涨到了30ms。后来我做了三个优化每个IP的deque限制最大长度超出后抛弃最老的bucket避免内存无限增长加锁粒度从全局锁改成分片锁用ip的hash值分到不同的锁上减少锁竞争每N次写入批量合并降低record的调用频率def record(self, ip, count1): now int(time.time()) # 分片锁根据ip哈希取mod分散到8把锁上 lock_idx hash(ip) % 8 with self._locks[lock_idx]: # ... 原逻辑这是我实测中最典型的防御系统成了性能瓶颈问题值得大家重视。5.2 日志别写成磁盘炸弹检测系统上线后我收到过一次磁盘告警/var目录被打满了。查了半天发现是防御系统的日志文件疯狂增长——攻击发生的时候每秒写几十条告警日志一天下来就几个GB。后来我加了两道保险一是日志按天滚动保留7天超期自动清理二是同一IP的重复告警做个合并比如10秒内相同告警只记一条。class AlertLogger: def __init__(self): self._last_alert {} def log(self, ip, reason): now time.time() if self._last_alert.get(ip, 0) 10 now: return # 10秒内同一IP的告警只记一次 self._last_alert[ip] now with open(/var/log/dos_defense/alerts.log, a) as f: f.write(f{time.strftime(%Y-%m-%d %H:%M:%S)} {ip} {reason}\n)5.3 封禁策略怎样避免锁死自己这是我最想提醒大家的一个坑千万别把自己封了。有一次测试SYN Flood检测我故意从一台测试机发起攻击流量结果封禁规则一下发连我自己SSH登录服务器的连接也被干掉了。原因很简单测试机和运维机在同一个网段攻击源IP段和运维管理IP段重叠了。从那以后我的封禁模块强制加了三道保险# 第一道白名单检查 if ip in WHITE_LIST: return # 第二道管理网段保护 if ip.startswith((10.10., 192.168.1.)): return # 第三道防自封禁 my_ip get_my_ip() if ip my_ip: return顺便提一句如果你的服务器有远程管理卡或者云控制台建议提前确认在iptables全清的情况下还能不能登录。我见过同行做过一个最极端的自杀式封禁脚本把所有非白名单IP全封了结果远程登录断了云控制台的VNC又被安全组挡住最后只能提工单让机房重启。这种事故一次就能让你长记性。5.4 常见问题速查表现象可能原因排查方法正常用户被误封阈值设置过低或没有基线比对检查封禁原因日志调高阈值或引入基线统计攻击流量没被识别检测层次不匹配攻击在网络层但只检测了应用层增加SYN半开连接检测和资源监控封禁不生效iptables规则被清空或云平台安全组覆盖检查iptables规则是否还在配合云控制台策略防御系统拖慢业务锁竞争严重或日志写入频繁优化锁粒度合并告警异步写日志重启后封禁全失效iptables规则是临时的重启即清空将关键封禁规则持久化到iptables配置文件误封内部服务内部调用IP没进白名单完善白名单定期review白名单列表5.5 关于网络层攻击的补充说明上面的代码示例里SYN检测是识别网络层攻击的主要手段但如果你遇到的是大流量的UDP Flood或者ICMP Flood只靠psutil统计可能不够。这种情况我一般会用sysctl调整内核参数做基础加固效果立竿见影# 限制SYN半开连接队列长度超过即丢弃新包 sysctl -w net.ipv4.tcp_max_syn_backlog1024 # 开启SYN Cookie不依赖半开队列也能完成握手 sysctl -w net.ipv4.tcp_syncookies1 # 缩短SYN_RECV的超时时间快速回收无用连接 sysctl -w net.ipv4.tcp_synack_retries1这几个参数配合Python检测脚本能够在不依赖外部设备的情况下抵御中小规模的SYN Flood我自己测试大约每秒几万个SYN包的规模。再大一级的洪水攻击说实话个人脚本已经无能为力了那就该考虑运营商侧的高防清洗了这不是自己能解决的问题。6. 一点个人心得整套系统用了差不多两周时间落地实际运行下来我对这个方案的定位是它解决的是最后一公里的防御问题。商业高防和云WAF解决的是门口的大洪水但真正打到业务层的那些小规模、低成本攻击——一个脚本、几台肉鸡、刷一刷接口——其实更常见而且恰好是这套Python方案能扛住的场景。我在实际使用中还有一个切身体会防御系统本身的代码质量比业务代码要求更高。业务代码出了问题最坏是功能不可用防御代码出了问题轻则误杀正常用户重则把自己锁在服务器门外。所以写完防御脚本一定要做两轮自测一轮模拟攻击验证能识别另一轮模拟正常流量验证不误报。这两轮都通过了才算真的能上线。如果你打算在自己项目里复刻这套东西我建议从最小闭环开始先只做QPS检测和iptables封禁跑通之后再逐步加连接检测、资源监控、基线统计。防御系统最怕一上来就做得复杂复杂意味着不可控不可控的防御系统本身就是风险。最后再分享一个小技巧封禁和告警的日志我建议单独存一个文件不要和业务日志混在一起。出问题的时候直接tail防御日志就能还原攻击发生时的完整时间线排查效率能高一倍。这个习惯帮我省了不知道多少时间希望也能帮到你。