简介这是一套面向网络安全方向学生与开发者的网络异常流量检测系统源码适合用作毕业设计案例、课程研究或安全防护实践。系统融合多种算法与模型可实时监测并分析网络数据流定位潜在威胁与异常行为模块化设计便于按需定制功能。压缩包共170个文件约3.41MB以78个css样式文件、34个go后端源码、10个py脚本为主另含xml配置、js交互、proto接口定义、csv数据集及pkl模型文件等覆盖前端界面、服务端逻辑与算法模型多个层面。目前已有91人学习下载。读者可从中获取完整的检测系统实现方案理解流量采集、特征处理与异常判定流程并参考目录结构快速定位前后端模块适合作为学习网络流量分析与异常检测核心技术的实践材料仅供学习交流使用。1. 网络异常流量检测系统新版源码从抓包到告警一套能跑通的落地路径线上服务被打挂的那天运维群里最先炸的往往不是 CPU 或内存告警而是带宽曲线突然拉成一条直线。等你登上机器ss -s里一堆SYN-RECV日志里全是同一个 IP 段的高频请求——这就是典型的网络异常流量。很多团队的第一反应是买设备、上云清洗但真到自己动手做一套「网络异常流量检测系统」才发现难点不在算法而在数据怎么采、特征怎么算、阈值怎么定、告警怎么不误报。这份新版源码标题背后其实是一整套从网卡抓包、流表聚合、特征提取到规则与模型双引擎告警的工程链路。它适合有 Linux 和 Python 基础、想在自己内网或测试环境里搭一套可观测流量系统的后端、运维和安全方向工程师。下面我按自己实际搭过一遍的顺序把这条链路拆开讲清楚包括参数怎么设、哪里容易翻车。2. 抓包与流表聚合检测系统的数据底座怎么搭2.1 为什么不能直接对原始包做检测新手最容易犯的错是拿tcpdump抓一堆 pcap然后逐包跑检测逻辑。单机小流量下能跑一旦上到千兆网卡包速率轻松到几十万 ppsPython 逐包处理直接跪。真实系统里检测的输入单位不是「包」而是「流」——把五元组源 IP、源端口、目的 IP、目的端口、协议相同的包聚合成一条流记录再在流级别算特征。这样数据量能降一到两个数量级检测逻辑也从「看每个包」变成「看每条流的行为」。常见做法是用抓包库做采集用哈希表做流聚合超时或收到 FIN/RST 就输出一条流记录。流记录里至少要有起止时间、包数、字节数、平均包长、TCP 标志位统计、每秒包数。这些字段就是后面检测引擎的原料。2.2 用 Python 搭一个最小可用的流聚合器下面这段代码用 scapy 做抓包和流聚合跑在测试机上足够验证逻辑。生产环境建议换成基于 libpcap 的 C 扩展或直接用 eBPF但思路一致。from scapy.all import sniff, IP, TCP, UDP import time from collections import defaultdict # 流表key 为五元组value 为流统计 flow_table defaultdict(lambda: { start: None, last: None, pkts: 0, bytes: 0, syn: 0, fin: 0, rst: 0, sport: 0, dport: 0 }) FLOW_TIMEOUT 30 # 秒超过这个时间没新包就认为流结束 def flow_key(pkt): if IP in pkt and TCP in pkt: return (pkt[IP].src, pkt[TCP].sport, pkt[IP].dst, pkt[TCP].dport, TCP) if IP in pkt and UDP in pkt: return (pkt[IP].src, pkt[UDP].sport, pkt[IP].dst, pkt[UDP].dport, UDP) return None def on_packet(pkt): key flow_key(pkt) if key is None: return now time.time() f flow_table[key] if f[start] is None: f[start] now f[last] now f[pkts] 1 f[bytes] len(pkt) if TCP in pkt: flags pkt[TCP].flags f[syn] 1 if flags 0x02 else 0 f[fin] 1 if flags 0x01 else 0 f[rst] 1 if flags 0x04 else 0 def flush_expired(): now time.time() expired [k for k, v in flow_table.items() if v[last] and now - v[last] FLOW_TIMEOUT] for k in expired: v flow_table.pop(k) duration max(v[last] - v[start], 0.001) print({ key: k, pkts: v[pkts], bytes: v[bytes], pps: round(v[pkts] / duration, 2), bps: round(v[bytes] * 8 / duration, 2), syn: v[syn], rst: v[rst] }) # 抓包每 100 个包检查一次超时流 sniff(prnon_packet, storeFalse, count0)逻辑说明flow_table用五元组做 key每个包进来先判断协议再更新对应流的计数。FLOW_TIMEOUT是关键参数设太小会把一条长连接拆成多条流设太大内存会涨。内网服务建议 30 到 60 秒公网入口可以放到 120 秒。flush_expired负责把超时流输出实际系统里应该用独立线程定时跑而不是等抓包回调。参数上还有两个坑一是sniff默认不设过滤器会把所有包都抓进来生产环境一定要加filtertcp or udp并限定网段二是 scapy 在高流量下丢包严重验证逻辑可以压测必须换方案。2.3 流记录字段怎么选才有检测价值流记录字段不是越多越好要围绕「异常行为」来选。我一般保留这几类字段类别具体字段检测用途规模类包数、字节数、持续时长识别大流量、慢速连接速率类pps、bps、平均包长识别扫描、洪水标志类SYN/FIN/RST 计数识别半开连接、异常断开分布类目的端口数、目的 IP 数识别扫描、横向移动目的端口数和目的 IP 数需要额外维护一个集合不能只靠单条流。常见做法是在流聚合时对同一个源 IP 维护一个「目的端口集合」流输出时带上集合大小。这个字段对检测端口扫描特别有用——正常业务一条流只连一个端口扫描器一条流会连几百个端口。3. 检测引擎规则和模型怎么配合才不误报3.1 规则引擎负责已知威胁模型负责未知异常检测引擎分两层是行业共识。规则层处理「已知的坏」比如单 IP 每秒新建连接超过 200 次、单流 SYN 包占比超过 80%、目的端口数超过 100。这些规则确定性强、可解释、误报低。模型层处理「不像正常」用历史流量训练一个基线偏离基线的流打高分。两层输出合并后按分数排序高分告警中分记录低分丢弃。规则引擎的实现重点是「滑动窗口」。不能只看单条流要看一个源 IP 在最近 10 秒内的聚合行为。下面是一个滑动窗口计数器的实现import time from collections import defaultdict, deque class SlidingWindow: def __init__(self, window_sec10): self.window window_sec self.events defaultdict(deque) # key - deque of timestamps def add(self, key): now time.time() dq self.events[key] dq.append(now) # 清理窗口外的旧事件 while dq and now - dq[0] self.window: dq.popleft() return len(dq) def count(self, key): now time.time() dq self.events[key] while dq and now - dq[0] self.window: dq.popleft() return len(dq) # 用法统计每个源 IP 在 10 秒内的新建连接数 syn_window SlidingWindow(window_sec10) def check_syn_flood(src_ip): cnt syn_window.add(src_ip) if cnt 200: return {alert: SYN_FLOOD, src: src_ip, count: cnt} return None逻辑说明SlidingWindow用 deque 存时间戳每次 add 时清理过期事件保证窗口内计数准确。window_sec设 10 秒是经验值太短会漏掉慢速攻击太长会延迟告警。阈值 200 需要按自己业务调——正常网关单 IP 新建连接一般不超过 50超过 200 基本可以确认异常。3.2 基线模型用统计方法做轻量异常检测不是每个团队都有标注数据训练深度学习模型。更现实的做法是用统计基线对每个源 IP 的历史流量算均值和标准差当前值超过均值加 3 倍标准差就标记异常。这个方法简单、可解释、不需要训练框架。import numpy as np from collections import defaultdict class BaselineDetector: def __init__(self, history_len100): self.history defaultdict(list) self.history_len history_len def update(self, key, value): h self.history[key] h.append(value) if len(h) self.history_len: h.pop(0) def score(self, key, value): h self.history[key] if len(h) 20: # 样本太少不判断 return 0.0 mu np.mean(h) sigma np.std(h) 1e-6 z (value - mu) / sigma return z # z 3 认为异常 # 用法对每个源 IP 的 pps 做基线 detector BaselineDetector() def check_anomaly(src_ip, pps): z detector.score(src_ip, pps) detector.update(src_ip, pps) if z 3: return {alert: PPS_ANOMALY, src: src_ip, z: round(z, 2)} return None逻辑说明history_len决定基线窗口100 个采样点大约覆盖几分钟到几十分钟按采集频率定。score返回 z 分数大于 3 触发告警。这里有个坑如果攻击持续很久基线会被污染均值被拉高后续反而检测不到。解决办法是基线只用「非告警时段」的数据更新或者用中位数和 MAD 代替均值和标准差抗污染能力更强。3.3 告警合并与降噪别让运维被淹没检测引擎跑起来后最大的问题不是漏报是告警太多。一次扫描可能触发几百条规则每条都发告警运维直接麻木。必须做告警合并同一个源 IP 在 5 分钟内的同类告警只发一条附带触发次数。合并逻辑用一个字典记录「源 IP 告警类型」的最后发送时间超过冷却期才再发。import time alert_cooldown {} # (src, type) - last_alert_time COOLDOWN_SEC 300 def should_alert(src, alert_type): key (src, alert_type) now time.time() last alert_cooldown.get(key, 0) if now - last COOLDOWN_SEC: alert_cooldown[key] now return True return FalseCOOLDOWN_SEC设 300 秒是折中值核心业务可以缩短到 60 秒边缘业务可以拉长到 600 秒。这个机制能砍掉 80% 以上的重复告警是让系统「能用」的关键一步。4. 避坑与排查这套系统最容易翻车的五个地方4.1 抓包丢包导致检测失效现象压测时检测引擎完全没反应但手动tcpdump能看到攻击流量。原因scapy 或用户态抓包库在高 pps 下丢包流表根本没收到包。解决生产环境改用 AF_PACKET 的 PACKET_MMAP 模式或者直接上 eBPF/XDP 在内核态做流聚合用户态只读聚合结果。验证方法是对比网卡计数和流表计数差距超过 1% 就说明在丢包。4.2 流超时参数设错导致特征失真现象慢速攻击检测不到因为每条流都被拆成很多短流pps 看起来很低。原因FLOW_TIMEOUT设太小攻击者用慢速发包绕过。解决对 TCP 流用「空闲超时 最大时长」双限制空闲超时 60 秒最大时长 300 秒超时后强制输出。同时增加「流内包间隔方差」特征慢速攻击的间隔方差通常很大。4.3 基线被污染导致漏报现象系统运行一周后之前能检测到的异常现在不告警了。原因基线模型把攻击期间的数据也学进去了均值被拉高。解决基线更新时排除告警时段的数据或者改用中位数 MAD。更稳妥的做法是维护两套基线短期基线最近 1 小时和长期基线最近 7 天两者偏差大时以短期为准。4.4 告警风暴拖垮告警通道现象一次扫描触发几千条告警邮件和 webhook 被打爆运维手机响个不停。原因没有做告警合并和限流。解决按「源 IP 告警类型」做冷却冷却期内只记录不发送同时设置全局告警速率上限比如每分钟最多 50 条超过的降级为摘要。告警通道本身也要做异步队列不能阻塞检测主流程。4.5 误封正常业务流量现象检测系统联动防火墙封了某个 IP结果那是合作方的回调地址业务中断。原因规则阈值太激进或者没有白名单机制。解决所有自动封禁动作先走「观察模式」只记录不执行运行一周后人工确认规则准确率再开启自动封禁。同时维护 IP 白名单和端口白名单白名单流量只记录不告警。5. 进阶技巧用流量回放验证检测规则的真实命中率系统搭完只是开始真正决定它能不能上生产的是「规则命中率」和「误报率」这两个数字。我一般用流量回放来验证把历史 pcap 按时间顺序重放观察检测引擎的输出和人工标注的异常时间段对比。具体做法分三步。第一步准备两份 pcap一份是正常业务流量一份是已知攻击流量可以用 hping3、nmap 在隔离环境里自己打。第二步写一个回放脚本按原始时间戳间隔发包避免瞬间灌入导致丢包。from scapy.all import rdpcap, sendp import time def replay(pcap_file, iface, speed1.0): pkts rdpcap(pcap_file) if not pkts: return base pkts[0].time start time.time() for pkt in pkts: target start (pkt.time - base) / speed now time.time() if target now: time.sleep(target - now) sendp(pkt, ifaceiface, verboseFalse) # speed1.0 原速speed10 加速 10 倍 replay(attack.pcap, eth1, speed1.0)逻辑说明rdpcap读入 pcap按原始时间戳的相对间隔发送。speed参数控制回放速度验证检测延迟时用 1.0压测吞吐时用 10 以上。注意sendp是二层发送需要指定网卡且要在隔离环境里跑别往生产网打。第三步对比检测输出和标注。正常流量回放时告警数应该接近零攻击流量回放时告警应该覆盖攻击时间段。如果正常流量也大量告警说明阈值太松如果攻击流量漏报说明特征没选对。这个对比表我一般会记录回放场景总流数告警数命中攻击段误报数正常业务 1 小时120000303SYN 洪水 5 分钟800012120端口扫描 2 分钟3000880误报数不为零时逐条看告警的流记录判断是阈值问题还是特征问题。我踩过的最大坑是「正常业务里的健康检查流量」被当成扫描——健康检查会连很多 IP 的同一个端口目的 IP 数很高。后来加了「目的端口固定且为已知服务端口」的白名单才解决。这套系统值不值得做取决于你的流量规模和团队能力。日活十万以下、没有专职安全团队的话用云厂商的流量清洗加开源 IDS 组合更省事但如果你有内网东西向流量要监控、有合规要求、或者想深入理解流量行为自己搭一套的收益是长期的——规则和基线都能按自己业务定制误报率能压到比通用方案低一个量级。我的习惯是每季度用新抓的正常流量回放一次重新校准阈值因为业务在变流量基线也在变没有一劳永逸的参数。希望帮到你。本文还有配套的精品资源点击获取