轻量级网络流量异常检测方案:用Scapy实时抓包+滑动窗口统计,动态基线(中位数±1.5×MAD)替代固定阈值,分级告警(日志/企微/邮件短信)并支持白名单与redis抑制,兼顾落地性、可调性与易维护性。

用python做网络流量异常检测并自动告警,核心不在写多复杂的模型,而在于快速落地、可调、易维护。下面是一个轻量但实用的方案,不依赖深度学习,适合中小规模出口或内网旁路镜像场景。
一、用Scapy实时抓包 + 滑动窗口统计
不用NetFlow或sFlow采集器,直接用Scapy监听网卡(如eth0),每10秒统计一次基础指标:
- 每秒新建TCP连接数(SYN包计数)
- UDP包占比突增(>60%且环比+200%)
- 单IP目的端口离散度(熵值<2.0视为扫描行为)
- http状态码5xx比例>15%(需解析Payload,加简单正则即可)
滑动窗口用deque维护最近5个周期数据,避免内存堆积。注意:开启混杂模式需root权限,生产建议用AF_PACKET + PF_RING加速。
二、阈值用动态基线替代固定数字
固定阈值在业务波动时误报高。改用“滚动中位数±1.5×MAD(中位数绝对偏差)”作为上下限:
立即学习“Python免费学习笔记(深入)”;
例如:正常SYN/sec中位数是42,MAD=8,则告警阈值≈42±12 → 超过54持续30秒就记为异常。
三、告警分级 + 多通道推送
不是所有异常都要发企业微信/邮件:
- 一级(低危):单IP扫描行为 → 写入本地日志 + 控制台打印
- 二级(中危):ddos特征(SYN洪泛、udp反射)→ 企业微信机器人@值班组
- 三级(高危):https证书异常+大量404+User-Agent为空 → 邮件+短信双触达
用requests调用Webhook,用smtplib发邮件,短信可用阿里云API。所有告警带时间、源IP、目标IP、指标快照、原始包摘要(前64字节hex)。
四、加一层“白名单+抑制”防骚扰
上线后第一周必有误报。提前配置两个过滤层:
白名单支持CIDR和正则(如red”>^192.168.d+.d+$),改完热加载,无需重启脚本。
基本上就这些。不复杂但容易忽略的是:抓包别用Filter=’tcp’硬过滤,否则漏掉伪造IP的UDP Flood;告警正文别只写“流量异常”,要带具体数值对比;还有——记得加信号捕获(SIGINT/SIGTERM),让Ctrl+C能优雅退出释放网卡。