简介这是一套面向高校计算机、网络安全相关专业学生及课程设计学习者的网络入侵检测与防御系统完整源码可作为毕业设计或课程设计参考方案帮助解决实时流量分析、攻击检测与自动防御等实践需求。资源包共38个文件约92KB以Python源码为主体包含14个py文件与9个pyc文件另有4个html、3个js及css等前端页面资源配合json配置、yml与Dockerfile容器化文件、sh安装脚本和txt说明覆盖后端逻辑、Web界面与部署环境。项目基于Flask、Flask-SocketIO与Scapy实现流量捕获和威胁识别前端采用Bootstrap与Chart.js完成可视化监控MongoDB存储日志与配置并通过Docker Compose简化运行。已有172人学习关注读者可获取实时流量检测、入侵拦截、告警响应、攻击统计与系统设置等模块的完整实现思路以及详细运行指南便于快速理解项目结构与二次开发。1. 从一份毕设源码说起Python 网络入侵检测与防御系统能跑出什么很多安全方向的毕设选题最后都卡在同一个地方论文框架搭得挺漂亮真到演示环节要么只能拿离线 pcap 文件回放要么抓包脚本一跑就报权限错误答辩老师问一句「实时性怎么体现」就答不上来。这份基于 Python 的网络入侵检测与防御系统源码解决的正是这个尴尬——它把实时流量分析、攻击检测、自动防御、可视化监控串成了一条能当场演示的链路Web 界面里流量曲线和攻击日志是动态刷新的不是截图糊弄。技术栈上它选了 Flask Flask-SocketIO 做后端推送Scapy 负责抓包和构造响应MongoDB 存日志和配置前端用 Bootstrap Chart.js 画图另外配了 Docker Compose 方便一键起环境。适合谁正在做网络空间安全、计算机相关毕设或课程设计的学生以及想快速搭一个 IDS/IPS 演示原型的初级安全从业者。下面我按「这东西怎么落地 → 参数怎么调 → 哪里会翻车」的顺序拆一遍都是能照着复现的细节。2. 环境搭建与依赖安装把 Flask、Scapy、MongoDB 三件套跑通2.1 为什么这套组合值得选而不是自己从零写先说选型逻辑。做入侵检测抓包层绕不开 libpcapPython 里封装得最成熟的就是 Scapy它能让你用几行代码解析 IP/TCP/UDP 甚至自定义 payload比直接调 socket 原始套接字省太多事。后端选 Flask 而不是 Django是因为这个项目本质是个轻量监控面板路由不多Flask 加 Flask-SocketIO 就能把实时推送做掉没必要上重型框架。数据库用 MongoDB 而不是 MySQL关键在于攻击日志的字段不固定——今天记的是 SYN 洪水明天可能是端口扫描文档型数据库存这种半结构化日志更顺手不用频繁改表结构。这套组合的边界也要清楚Scapy 在高流量场景下性能一般千兆线速抓包会丢包所以它定位是教学演示和中小流量监控不是生产级 IDS。明白这一点后面调参和压测时就不会有不切实际的期待。2.2 依赖清单与安装步骤项目根目录给了requirements.txt、install.py、install.sh说明作者考虑了跨平台安装。我一般先看 requirements 里锁了哪些版本再决定用不用虚拟环境。核心依赖大致是这些# 建议先建虚拟环境避免污染系统 Python python -m venv venv source venv/bin/activate # Windows 用 venv\Scripts\activate # 安装项目依赖 pip install -r requirements.txt # 如果 install.sh 存在也可以直接跑它做一键安装 bash install.shrequirements.txt里通常包含 flask、flask-socketio、scapy、pymongo、eventlet 或 gevent 这类异步库。这里有个参数要留意Flask-SocketIO 的异步模式如果用 eventlet启动方式会和普通 Flask 不一样得用socketio.run(app)而不是app.run()否则 WebSocket 推送会失效页面上的实时流量图就变成死的。安装完先别急着跑主程序用pip list核对一下 scapy 和 pymongo 是否真的装上了这两个是最容易因为系统缺少底层库而静默失败的。2.3 MongoDB 的启动与连接配置MongoDB 不在 pip 依赖里得单独装。Linux 上用包管理器Windows 上建议直接下官方安装包或走 Docker。项目给了docker/docker-compose.yml最省事的做法是把数据库也交给 Compose# docker-compose.yml 里数据库服务的典型结构 services: mongo: image: mongo:6 ports: - 27017:27017 # 默认端口本地开发直接映射 volumes: - mongo_data:/data/db volumes: mongo_data:启动后确认连接串。项目里连接 MongoDB 的地址一般写在配置或app/__init__.py里默认是mongodb://localhost:27017/。如果你的 MongoDB 开了认证就得改成带用户名密码的 URI否则程序启动时会卡在连接超时日志里报ServerSelectionTimeoutError。这一步是新手最容易忽略的装完数据库以为万事大吉结果程序连不上。3. 实时流量分析与攻击检测Scapy 抓包到规则匹配的完整链路3.1 抓包权限与网卡选择第一道坎Scapy 抓包需要原始套接字权限。Linux 下要么用 root 跑要么给 Python 解释器加CAP_NET_RAW能力Windows 下需要装 Npcap 并勾选 WinPcap 兼容模式。很多人第一次跑就报PermissionError或者抓不到任何包八成是权限或网卡选错了。抓包时要指定网卡不能让它自己猜。常见做法是先列出所有网卡再在配置里写死要监听的那个from scapy.all import get_if_list, sniff # 列出本机所有可用网卡确认要监听哪个 print(get_if_list()) def packet_handler(pkt): # 这里做协议解析和特征提取 if pkt.haslayer(IP): src pkt[IP].src dst pkt[IP].dst # 后续交给检测模块判断 # iface 参数必须显式指定否则可能监听到错误的网卡 sniff(ifaceeth0, prnpacket_handler, storeFalse)storeFalse这个参数很关键它让 Scapy 不把抓到的包全存内存里否则跑几分钟内存就爆了。prn是每抓一个包回调一次检测逻辑就挂在这里。如果你在虚拟机里跑网卡名可能是ens33或enp0s3别照抄eth0。3.2 检测规则怎么落地从阈值到特征匹配这个项目的攻击检测本质是两类方法结合一类是基于阈值的统计检测比如单位时间内同一源 IP 的 SYN 包数量超过阈值就判定为 SYN 洪水另一类是基于特征的匹配比如检测 payload 里有没有特定攻击字符串。阈值检测实现简单但参数敏感阈值设低了误报设高了漏报。我一般会把阈值做成可配置项放在 MongoDB 的配置集合里Web 界面能改。典型参数包括时间窗口比如 10 秒、SYN 包计数上限、单 IP 连接数上限、端口扫描的端口命中数阈值。下面是一个简化的检测逻辑from collections import defaultdict import time # 记录每个源 IP 在时间窗口内的 SYN 包数量 syn_counter defaultdict(list) WINDOW 10 # 时间窗口单位秒 SYN_THRESHOLD 100 # 窗口内 SYN 包超过这个数就告警 def detect_syn_flood(pkt): if pkt.haslayer(TCP) and pkt[TCP].flags S: src pkt[IP].src now time.time() # 只保留窗口内的记录 syn_counter[src] [t for t in syn_counter[src] if now - t WINDOW] syn_counter[src].append(now) if len(syn_counter[src]) SYN_THRESHOLD: return {type: SYN Flood, src: src, count: len(syn_counter[src])} return NoneWINDOW和SYN_THRESHOLD这两个参数要根据你的演示环境调。在本地测试流量本来就小阈值设 100 可能永远触发不了演示时可以把阈值降到 20 左右方便当场看到告警。但要注意阈值太低会把正常的并发连接误判成攻击这就是典型的误报答辩时被问到要有话说。3.3 检测结果如何进入告警与日志检测函数返回告警字典后要写进 MongoDB 并推送到前端。写库用 pymongo推送用 Flask-SocketIO 的emit。这里有个顺序问题先写库再推送还是先推送再写库我建议先写库保证日志不丢推送失败最多是前端晚一点刷新不影响数据完整性。from flask_socketio import emit from pymongo import MongoClient client MongoClient(mongodb://localhost:27017/) db client[ids_db] def handle_alert(alert): # 先落库保证可追溯 db.alerts.insert_one(alert) # 再推送到前端事件名要和前端监听的一致 emit(new_alert, alert, broadcastTrue)emit的事件名new_alert必须和前端 JavaScript 里socket.on(new_alert, ...)完全对应大小写都不能错否则前端收不到页面上的攻击日志永远是空的。这是调试实时功能时的高频坑。4. 自动防御与可视化监控拦截动作和前端图表的联动4.1 自动防御的几种实现方式与风险自动防御听起来很酷但实现上要克制。常见做法有三种一是调用系统防火墙iptables/firewalld封禁源 IP二是用 Scapy 构造 TCP RST 包断开连接三是直接在应用层丢弃后续包。第一种最彻底但需要 root 权限第二种轻量但只能断当前连接第三种依赖程序一直运行。以 iptables 封禁为例检测到攻击后执行import subprocess def block_ip(ip): # 注意这条命令需要 root 权限且规则是临时的 cmd [iptables, -A, INPUT, -s, ip, -j, DROP] try: subprocess.run(cmd, checkTrue) return True except subprocess.CalledProcessError as e: # 权限不足或规则冲突时会走到这里 print(f封禁失败: {e}) return False这里必须提醒自动封禁有误封风险。如果阈值调得太激进可能把正常用户的 IP 封了演示时如果封的是自己所在网段的网关整个环境就断网了。我一般会在演示前把封禁动作改成「只记录不执行」或者加一个白名单把本机 IP 和网关排除掉。项目里如果有modules目录防御逻辑大概率在里面改之前先备份。4.2 可视化监控Chart.js 数据从哪来前端用 Chart.js 画流量曲线和攻击统计数据来源是后端定时推送或前端定时拉取。实时性要求高就用 SocketIO 推要求不高就前端setInterval调 REST 接口。项目里routes.py应该定义了获取统计数据的接口templates下的 HTML 负责渲染。一个典型的流量统计接口返回结构是这样的# routes.py 里常见的统计接口 app.route(/api/stats) def stats(): # 聚合最近一分钟的包数量按秒分组 pipeline [ {$match: {timestamp: {$gte: time.time() - 60}}}, {$group: {_id: {$floor: $timestamp}, count: {$sum: 1}}} ] data list(db.traffic.aggregate(pipeline)) return jsonify(data)前端拿到数据后更新 Chart.js 的 dataset调用chart.update()刷新。如果图表不动先看浏览器控制台有没有报错再看 Network 面板里接口有没有正常返回数据最后才怀疑后端。这个排查顺序能省很多时间。4.3 Docker 部署与本地运行的取舍项目带了 Dockerfile 和 docker-compose.yml理论上docker-compose up就能起。但抓包这件事在容器里有个天然限制容器默认拿不到宿主机的网卡需要加network_mode: host或者挂载网卡设备。如果 Compose 里没配容器里的 Scapy 抓不到任何包页面数据全是零。# 抓包服务需要 host 网络模式才能看到宿主机流量 services: ids: build: . network_mode: host # 关键否则抓不到宿主机网卡 depends_on: - mongonetwork_mode: host在 Linux 上没问题在 Windows 和 macOS 的 Docker Desktop 上支持有限所以这两个平台我建议直接本地跑 Python别折腾容器抓包。本地跑虽然环境配置麻烦点但抓包权限和网卡选择都更直观。5. 避坑与常见问题排查那些让演示翻车的细节5.1 抓不到包页面数据全为零现象程序启动正常Web 界面能打开但流量统计一直是 0攻击日志也不更新。原因通常是三个之一网卡名写错、权限不足、或者跑在容器里没开 host 网络。解决先用get_if_list()确认网卡名Linux 下用sudo跑或加 capabilities容器部署改成network_mode: host。我踩过一次网卡名在虚拟机里是ens33配置里写的eth0排查了半小时才发现。5.2 WebSocket 连不上实时推送失效现象页面能加载但攻击告警不实时刷新要手动刷新页面才看到新数据。原因多半是 Flask-SocketIO 用了不匹配的异步模式或者启动时用了app.run()而不是socketio.run(app)。解决确认requirements.txt里装的是 eventlet 还是 gevent启动入口改成对应的socketio.run()并检查前端 socket.io 客户端版本和后端是否兼容。5.3 MongoDB 连接超时程序启动即退出现象启动时报ServerSelectionTimeoutError或ConnectionRefusedError。原因MongoDB 没启动、端口不对、或者开了认证但连接串没带账号密码。解决先systemctl status mongod或docker ps确认数据库在跑再用mongosh手动连一下验证最后核对程序里的连接串。这一步别偷懒手动连一次能排除大半问题。5.4 自动封禁把自己封了环境断网现象演示到一半本机突然上不了网或者整个局域网访问异常。原因自动防御模块把网关或本机 IP 加进了 iptables DROP 规则。解决演示前把本机 IP、网关、常用测试机加进白名单或者把封禁动作改成只记录日志不实际执行万一被封用iptables -D INPUT -s ip -j DROP删掉规则恢复。5.5 阈值调不好误报漏报两头堵现象要么正常访问被当成攻击疯狂告警要么真发起攻击了系统没反应。原因检测阈值和实际流量特征不匹配。解决先抓一段正常流量做基线统计单位时间内的平均包数和连接数把阈值设在基线的 3 到 5 倍演示攻击时用工具集中发包确保能超过阈值。参数没有万能值得按环境调。6. 进阶技巧把检测规则做成可热更新的配置跑通基础流程后真正拉开差距的是检测规则的可维护性。硬编码在 Python 里的阈值每次改都要重启服务演示时很尴尬。我的习惯是把规则抽到 MongoDB 的rules集合里程序定时拉取或监听变更做到热更新。# 规则文档结构示例存进 MongoDB { name: syn_flood, enabled: True, window: 10, threshold: 100, action: alert # 可选 alert / block } # 程序侧定时刷新规则避免重启 def load_rules(): return {r[name]: r for r in db.rules.find({enabled: True})}这样在 Web 界面加一个规则编辑页改完阈值点保存后端下一轮检测就用新参数不用重启。action字段控制是只告警还是直接封禁演示时可以临时把block改成alert避免误封。验证检测是否真的生效别只看页面。我一般会开两个终端一个跑系统一个用 Scapy 手动构造攻击包打过去看日志里有没有对应记录# 手动发一批 SYN 包验证检测规则是否触发 from scapy.all import IP, TCP, send for i in range(150): send(IP(dst127.0.0.1)/TCP(dport80, flagsS), verboseFalse)发完立刻去查 MongoDB 的alerts集合或者看页面告警列表。如果没触发先确认阈值是不是设高了再确认抓包网卡是不是回环口lo——打本机流量得监听lo监听eth0是抓不到的这个细节坑过不少人。从那以后我每次演示前都强制走一遍「确认网卡 → 确认权限 → 确认数据库 → 手动打一发测试包」的流程四步都过了才敢上台。希望这份拆解能帮到你少走几个我踩过的弯路。本文还有配套的精品资源点击获取
