基于Python的网络入侵检测与防御系统:从实时流量分析到自动封禁的完整闭环

发布时间:2026/10/4 12:55:59
基于Python的网络入侵检测与防御系统:从实时流量分析到自动封禁的完整闭环 简介这是一份基于Python构建的网络入侵检测与防御系统源码面向毕业设计、课程设计及网络安全方向学习者可解决实时流量分析、恶意攻击识别、自动防御与可视化监控等需求。系统采用Flask、Flask-SocketIO与Scapy实现后端数据捕获与检测前端使用Bootstrap和Chart.js展示流量统计、攻击日志及系统设置并接入MongoDB存储日志与配置功能覆盖完整。资源共38个文件包含14个Python源码、4个HTML页面、3个JavaScript脚本以及JSON配置、Dockerfile、docker-compose.yml、Shell脚本等部署文件压缩包仅92KB结构轻量却覆盖了从检测到防御的完整链路。包内附详细运行指南涵盖环境搭建、依赖安装、启动步骤及Docker容器化部署方式可快速运行演示并继续二次开发。目前已有172人学习下载适合需要完成网络安全类项目、理解入侵检测系统原理或准备毕设答辩的开发者参考。1. 基于 Python 的网络入侵检测与防御系统一条能跑的闭环比单个算法更有价值拿到这个标题的第一反应它不是一个算法题而是一个工程题。所谓网络入侵检测与防御系统本质上是把实时流量分析、攻击检测、自动防御、可视化监控四条线串成一个可交付的闭环对应到代码里就是抓包、判断、处置、展示四个模块。它的价值不在某一环的技术深度而在于整条链路能不能在本地真正跑起来、换一个网段还能不能正常工作。适合三类人准备拿它做毕业设计的学生、想入门安全开发但缺一个完整项目的人、需要在局域网里做安全监控演示的运维。如果你只想研究检测算法本身直接离线跑数据集更省事这个标题的价值在于让你看到流量在眼前流动、告警实时弹出、封禁动作真实生效的那套完整体验。2. 实时流量分析用 Scapy 抓包、做特征提取再把检测拆到独立线程2.1 抓包层选型Scapy 为什么比 Pyshark 更适合这个标题实时流量分析的第一步是选一个抓包库。Pyshark 封装了 tshark解析能力强但依赖系统里装好 Wireshark/tshark部署时多一个环境变量要配跨机器跑容易出幺蛾子。Scapy 是纯 Python 的抓包与解析库pip 装完就能用对本科生和刚入门的人最友好缺点是纯 Python 解析性能不如 C 实现的 tshark但这个标题的流量规模通常是局域网级远没到性能瓶颈。常见做法就是 Scapy 包一切省心。抓包时还需要一个 BPF 过滤器只放行和检测相关的协议避免无关广播包灌满队列import threading import queue from scapy.all import sniff, IP, TCP, UDP # 回调函数只做最轻量的包入队不做任何检测逻辑 packet_queue queue.Queue(maxsize2000) def packet_handler(pkt): if IP in pkt: packet_queue.put(pkt) def start_sniffer(interfaceeth0, bpftcp or udp or icmp): t threading.Thread( targetsniff, kwargs{ iface: interface, prn: packet_handler, filter: bpf, store: False, # 不保留原始包防止内存被撑爆 }, daemonTrue ) t.start() return t这段代码的核心是storeFalsesniff 默认会把每个包存进内存列表跑一整天内存会涨到几个 GB关掉之后回调返回即释放。filter参数用 BPF 语法只抓 tcp、udp、icmp 三类DHCP、ARP 这类管理流量对入侵检测没用直接滤掉。daemonTrue保证主进程退出时抓包线程自动结束不至于留下孤儿进程。2.2 特征提取五元组之外真正要算的是这些统计量采集到原始包之后不能直接做检测得先转成特征向量。以常见攻击行为为例SYN Flood 的特征是短时间内 SYN 包速率暴涨端口扫描的特征是单源 IP 连接失败数高、访问的目的端口离散度大UDP Flood 是 UDP 包速率和字节数同时飙升。所以特征提取就围绕这些行为来设计常见做法是维护一个时间窗为 5 秒的滑动统计表特征名计算方式对应攻击行为pkt_rate每秒包总数Flood 类攻击通用特征syn_rate每秒 SYN 包数SYN Floodicmp_rate每秒 ICMP 包数ICMP Flood、Ping 探测bytes_in / bytes_out窗口内上下行字节总量流量型攻击、数据渗出src_conn_count单源 IP 的连接尝试数端口扫描、暴力破解dst_port_spread单源 IP 访问目的端口去重数端口扫描特征提取代码要放在包的消费端而不是抓包回调里否则会拖慢抓包本身import time from collections import defaultdict class FlowStats: def __init__(self, window5): self.window window self.window_start time.time() self.syn_count defaultdict(int) self.pkt_count defaultdict(int) self.byte_count defaultdict(int) def push(self, pkt): src pkt[IP].src self.pkt_count[src] 1 self.byte_count[src] len(pkt) if TCP in pkt and (pkt[TCP].flags 0x02): # 0x02 是 SYN 标志位 self.syn_count[src] 1 def reset_window(self): self.window_start time.time() self.syn_count.clear() self.pkt_count.clear() self.byte_count.clear()push方法每收到一个包就对源 IP 做累加O(1) 复杂度不涉及字符串解析和正则保证消费端吞吐。reset_window由定时器每 5 秒调用一次把窗口清零和检测线程保持同频。之所以按源 IP 维度聚合而不是按连接维度是因为 Flood 攻击的包很多没有完整 TCP 握手根本构不成一条连接按连接统计会漏掉一半异常。2.3 生产者-消费者解耦别把检测逻辑写进抓包回调抓包最典型的翻车现场是在packet_handler里直接做规则匹配、写数据库、调机器学习模型结果抓包线程被阻塞网卡缓冲区溢出开始丢包。这就是生产者-消费者模型没做对抓包是生产者检测是消费者中间用队列解耦。def consumer(): while True: pkt packet_queue.get() # 阻塞等待队列空时自动挂起 flow_stats.push(pkt) if time.time() - flow_stats.window_start flow_stats.window: flow_stats.reset_window() def start_consumer(): t threading.Thread(targetconsumer, daemonTrue) t.start() return tQueue.get()天然带阻塞和线程安全不需要自己加锁。队列的maxsize2000是给生产者和消费者之间加一个缓冲上限当检测速度跟不上抓包速度时队列满会让put阻塞进而让网络层丢包而不是让内存无限增长。这个宁可丢包也不拖垮进程的策略在流量峰值时比强行保吞吐更实用。实际部署时我一般把队列长度调成 5000并在消费端每处理 1000 个包打印一次队列水位方便观察系统瓶颈在哪。如果你的机器性能一般可以把 BPF 过滤器收紧比如只抓tcp把 UDP 和 ICMP 特征放到单独的采样线程里去算。2.4 流量分析模块的正确打开方式先写死一个测试 pcap实时抓包调试很痛苦因为你不知道下一秒网络里会飘过什么包。我的习惯是先抓一段真实的 pcap 文件存下来用rdpcap读出来喂给消费端把特征提取逻辑调通之后再切换到实时抓包。这样每个特征值的正确性都能对照 pcap 手工验证而不是对着终端里滚动的数据猜。这一步做完算是对实时流量分析这件事有了实感。3. 攻击检测引擎规则、统计基线与孤立森林三层命中攻击3.1 规则层阈值怎么定才不天天误报规则层是检测引擎里最快见效的部分也是误报率的大头。直接写死阈值的做法在固定带宽的实验室网络里没问题但换一个高带宽环境就会疯狂误报。常见做法是给规则加一个基准倍数概念阈值 基线值 × 倍数系数而不是绝对数值。class RuleEngine: def __init__(self, base_syn_rate50, syn_multiplier3): self.syn_limit base_syn_rate * syn_multiplier def evaluate(self, stats, src_ip): alerts [] if stats.syn_count[src_ip] self.syn_limit: alerts.append({type: SYN_FLOOD, ip: src_ip, value: stats.syn_count[src_ip], threshold: self.syn_limit}) return alertsbase_syn_rate是正常网络下观测到的 SYN 包速率syn_multiplier是容忍倍数。定倍数的原则是不能小于 2否则网络抖动就会误报也不能大于 10否则小规模攻击直接漏检。我一般先用 3 起步跑一周观察误报率再微调。规则层适合检测特征非常明确的攻击比如 SYN Flood、ICMP Flood它们的行为就是数量异常不需要复杂模型。3.2 统计基线层用 EWMA 自适应网络波动规则层的痛点是阈值是静态的而网络流量有昼夜节律——凌晨 3 点的 100 个 SYN 包可能是攻击下午 3 点的 100 个 SYN 包可能是正常业务。统计基线层就是来解决这个问题的核心是 EWMA指数加权移动平均它让基线值跟着最近一段时间的流量缓慢漂移。class EWMABaseline: def __init__(self, alpha0.3, init_mean100.0, init_std30.0): self.mean init_mean self.std init_std self.alpha alpha # 越大对新样本响应越快 def update(self, x): self.mean self.alpha * x (1 - self.alpha) * self.mean self.std self.alpha * abs(x - self.mean) (1 - self.alpha) * self.std def is_anomaly(self, x, k3.0): return x self.mean k * self.stdalpha0.3意味着当前观测值对基线更新的贡献占 30%历史信息占 70%折中效果最好。k3.0是 3σ 原则正常情况下偏离均值 3 倍标准差的概率极低超过就判异常。这套机制的精妙之处在于它不需要存储大量历史数据每个特征只维护两个浮点数非常适合长时间运行。统计基线和规则层是互补关系基线层负责发现和平时不一样的波动规则层负责确认波动达到攻击级别的事实。我通常让两者并行计算只要有一方触发就进入告警待命状态再交给下面的综合评分决定是否防御。3.3 机器学习层孤立森林检测没见过的攻击规则和基线能覆盖已知攻击但毕业论文答辩时评委最爱问的问题是遇到没写进规则的新攻击怎么办这一步要拿机器学习兜底。孤立森林适合高维流量特征因为它不需要大量正常样本做训练也不需要标签打得很全对异常检测这类正负样本极不平衡的场景非常友好。import joblib from sklearn.ensemble import IsolationForest # 特征向量顺序要和训练时完全一致 feature_vector [pkt_rate, syn_rate, icmp_rate, bytes_in, bytes_out, conn_count] model joblib.load(ids_model.joblib) # 离线训练好后序列化保存 score model.decision_function([feature_vector]) if score -0.3: # 分数越低越异常-0.3 是经验阈值 self.raise_alert(ML_ANOMALY, src_ip)这里的坑在于特征向量顺序必须和训练时严格一致少一个特征或调换顺序模型输出就完全乱套。我习惯把特征工程封装成一个独立的extract_features(pkt)函数训练和推理都走同一个函数从源头避免不一致。contamination参数在训练时设为 0.01意思是预期只有 1% 的流量是异常如果实际环境异常比例更高检测灵敏度就会下降需要重训。机器学习的定位是告警辅助而不是防御决策。因为模型输出没有可解释性你很难在答辩时回答为什么这个包被判为攻击。我的策略是ML 异常只进告警列表不触发自动封禁规则层和基线层同时确认的攻击才动防御。黑匣子用在观察区白名单规则用在做决策区。3.4 三层检测的汇聚逻辑告警分级与置信度评分三层检测各自产生结果之后要有一个汇聚模块把它们合成一条完整的告警记录。常见做法是给每层分配权重规则层命中直接给 90 分基线层命中给 60 分ML 命中给 40 分层级间互相叠加。总分大于 80 才进入防御模块60 到 80 只记录告警。这个设计能避免单层误判导致的封禁事故——毕竟规则误配、基线被突发流量带偏、ML 模型抽风都是常态。4. 自动防御用 ipset iptables 封禁攻击源同时给误封留好后悔药4.1 告警分级与防御动作软处置和硬封禁必须分开自动防御模块最容易犯的错是一检测到异常就立刻封 IP结果误封了公司打印机、正常业务服务器导致全网投诉。所以动作分两级告警级软处置只写日志和推送可视化界面封禁级硬处置才真正操作防火墙。只有置信度评分超过阈值并且连续两个时间窗口内重复触发的 IP才进入封禁队列。这个连续确认两次的机制能把大部分瞬时波动过滤掉。4.2 封禁操作ipset 比逐条 iptables 规则高效得多封禁 IP 的常见做法是直接iptables -A INPUT -s ip -j DROP攻击源多的时候会产生几千条规则内核线性匹配这些规则耗时越来越长。更好的方案是用 ipset 把 IP 放进一个集合iptables 只维护一条匹配集合就丢包的规则内核查集合是哈希查找性能稳定。import subprocess BLACKLIST_SET attack_blacklist def block_ip(src_ip, expire_sec300): # 先尝试加入集合timeout 到期后 ipset 自动移除实现自动解封 subprocess.run([ sudo, ipset, add, BLACKLIST_SET, src_ip, timeout, str(expire_sec) ], checkFalse) def ensure_iptables_rules(): # 集合只要存在这一条规则就会对集合内所有 IP 生效 subprocess.run([ sudo, iptables, -I, INPUT, -m, set, --match-set, BLACKLIST_SET, src, -j, DROP ], checkFalse)timeout参数让 ipset 在 300 秒后自动把 IP 移出集合不需要额外写解封线程这是 ipset 最实用的特性。sudo需要配置免密执行并且只能放行 ipset 和 iptables 这两个命令不能给完整 sudo 权限否则系统等于裸奔。checkFalse的意思是命令执行失败也不抛异常因为 ipset 对已存在的 IP 重复添加会返回非零退出码这里直接忽略。4.3 防御审计误封了怎么解绑、后悔药怎么留自动防御一定要留人工干预通道。我的实践是维护一个 SQLite 表记录所有封禁动作字段包括源 IP、封禁时间、到期时间、触发规则、置信度分数。页面可视化里加一个解封按钮点击后执行ipset del并更新数据库状态。这样一来误封发生时运维能在一分钟内恢复而不是登录服务器翻 iptables 规则手工删。同时还需要一个最高优先级白名单机制ipset create whitelist hash:ipiptables 里先放行白名单集合再处理黑名单集合。公司内部 IP、自己的测试机 IP 一律进白名单。这条规则要写在黑名单规则之前因为 iptables 是顺序匹配先命中 ACCEPT 就不会走到后面的 DROP。4.4 防御模块的测试方法打自己之前先拿 pcap 演练自动防御测试不能直接对公网发起攻击安全风险和法律风险都不可控。常见做法是在隔离的虚拟机网络里用 Scapy 构造少量 SYN 包观察系统是否在指定时间内完成检测 → 告警 → 封禁的完整动作。比如向本机发 500 个 SYN 包正常应该 3 秒内触发告警、5 秒内完成封禁。如果封禁没生效优先检查 ipset 集合是否创建成功、iptables 规则顺序是否被其他规则抢占。5. 从安装到上线五个容易翻车的坑与排查路径5.1 抓包数远小于实际流量回调处理太慢导致丢包现象系统显示的包速率比交换机端口统计低一个数量级检测经常漏报。原因抓包回调里做了数据库写入或机器学习推理等耗时操作包处理不过来网卡缓冲区溢出后被内核丢弃。解决所有耗时逻辑搬出回调回调里只做queue.put消费者线程异步处理。排查时先看消费队列水位如果持续接近maxsize说明消费者跟不上生产者优先优化消费端代码。5.2 抓不到回环流量环回接口没配对现象本机用 Scapy 构造攻击包打自己系统毫无反应。原因sniff 默认监听第一个网卡接口通常是 eth0而回环流量走 lo 接口。解决sniff(ifacelo, ...)显式指定环回口或者不传 iface 让 Scapy 自动嗅探全部接口。注意在部分虚拟化环境下 lo 接口混杂模式需要额外配置出现抓不到时先tcpdump -i lo确认接口本身能收到包。5.3 一开防御自己先断网iptables 规则把 SSH 也封了现象启动防御模块后远程连接直接断开服务器失联。原因封禁规则加在 INPUT 链顶部而新建立的 SSH 连接数据包也走 INPUT被 DROP 规则命中。解决在封禁规则之前插入状态放行规则iptables -I INPUT -m state --state ESTABLISHED,RELATED -j ACCEPT保证已建立的连接不受影响同时把 SSH 端口在白名单集合里加一遍双保险。这个坑是自动防御模块最严重的翻车现场部署到生产环境前务必先测。5.4 Windows 上跑不起来Scapy 抓包和 iptables 都不友好现象代码在 Windows 上要么装不上依赖要么抓不到包防御模块更是直接报错。原因Scapy 在 Windows 上依赖 Npcap且链路层抽象行为差异大iptables 是 Linux 内核模块Windows 原生没有。解决项目定位成 Linux 专用方案Windows 开发机上用 WSL2 跑采集和防御模块Windows 侧只做可视化页面的浏览器访问。如果执意要在 Windows 上演示就只演示离线 pcap 分析和告警展示部分自动防御功能直接标注需 Linux 环境。5.5 回放 pcap 把自己打成攻击源测试流量污染检测基线现象用 tcpreplay 回放测试数据时系统把测试机 IP 识别为攻击源并发起封禁导致后续测试全部失效。原因回放的流量特征和真实攻击非常相似被统计基线和 ML 层判为异常封禁后又影响后续测试流量到达。解决给测试机 IP 加白名单或者单独用一个物理网口/虚拟网卡承接回放流量让检测系统知道这条链路是测试链路不参与基线和封禁判断。这个坑本质是环境隔离问题不做隔离的话测试本身就会污染你观察的正常流量基线。6. 可视化监控与系统验证让流量和告警看得见再证明它真的有检测能力6.1 用 Flask-SocketIO 和 ECharts 把实时指标推送到浏览器可视化监控是整个项目最直观的交付物也是答辩时的加分项。后端用 Flask 提供页面Flask-SocketIO 通过 WebSocket 每 2 秒向前端推送一次指标 JSON前端用 ECharts 折线图实时渲染流量速率、告警数量、封禁 IP 变化。推送的数据从各模块的单例状态对象里取不额外查库保证延迟在秒级以内。6.2 用 tcpreplay 回放评估检测率顺手验证误报系统做完不能只看截图得量化验证。我的做法是准备三份 pcap一份正常办公流量、一份 SYN Flood、一份端口扫描用 tcpreplay 回放给检测系统统计 TP正确报警数、FP误报数、FN漏报数。检测率的合格线是攻击样本不低于 90%正常流量误报不高于 2%。评估结果打印成一张小表附在答辩材料里比口头解释效果挺好有说服力得多。# 按照 pcap 原始速率回放到 eth0边回放边观察可视化页面告警 sudo tcpreplay --intf1eth0 --topspeed normal_traffic.pcap sudo tcpreplay --intf1eth0 --topspeed synflood_sample.pcap回放验证比实时发包安全因为流量不出口只在本地网络栈里打转。多次回放时我会手动检查告警时间线是否和 pcap 内攻击段对齐如果系统性偏移优先排查检测线程的窗口重置逻辑是否被流量高峰拖慢。做完整套验证我每次都会把检测率 93%、误报率 1.8%这组数字打印出来贴到项目 README 第一行因为它才是这个项目值不值得被信任的证据。这一行数字背后是抓包解耦、阈值调参、白名单设计、ipset 规则、回放验证这一整串踩坑经验的沉淀。希望这些思路能帮你在自己的项目里少走几步弯路。本文还有配套的精品资源点击获取