简介面向毕业设计与课程设计的网络入侵检测与防御项目以Python为基底整合实时流量分析、攻击检测、自动防御与可视化监控适合网络安全方向学生快速搭建课设原型也可作为毕设演示系统二次开发。资源共38个文件以14个Python源码文件为核心辅以9个编译后的pyc、4个HTML页面、3个JavaScript脚本及CSS、JSON、Dockerfile等配置压缩包整体约92KB目录分明代码内附注释便于定位与修改。系统基于Flask与Flask-SocketIO构建Web服务利用Scapy捕获解析网络数据包内置检测规则与告警响应机制可对恶意流量实施拦截前端采用Bootstrap与Chart.js以图表形式展示流量趋势、攻击统计与系统状态。容器化方面提供Docker Compose编排文件配合install脚本与requirements依赖清单能够快速完成环境部署。随包另附详细项目说明指南包含模块设计与运行说明帮助理解整体架构。目前已有172人学习下载适合需要完整参考实现并追求可读性的本科毕设与课程设计场景。1. 基于Python的网络入侵检测与防御系统毕设场景下它到底能干什么毕业设计撞上“网络入侵检测”这个题目很多人上来就想堆机器学习模型可真到演示节点才发现连实时流量都没地方抓、告警不知道往哪展示。这个Python项目换了一条务实的路线用Scapy实时捕获网卡流量规则引擎解析五元组和TCP特征识别端口扫描、SYN Flood、ICMP Flood这类常见攻击命中后自动执行封禁再通过Web仪表盘把流量趋势和攻击事件实时亮出来。做毕设、课程设计可以直接跟着运行指南把服务跑起来想搭IDS/IPS原型的从业者规则库、防御模块和MongoDB存储结构都是现成的改造点不用从零拼。2. 核心架构与模块职责Flask、Scapy、SocketIO、MongoDB各自扮演什么角色2.1 整体架构三层面怎么分工这个项目没有把抓包、检测、展示揉在同一个文件里而是拆成三层各跑各的线程靠队列和事件把链路串起来。抓包层由Scapy负责它绑定网卡做实时流量捕获抓到包后只做最简单的字段提取就丢进内存队列检测层是一个规则引擎消费队列里的数据包做协议解析和攻击判定展示层是Flask应用通过Flask-SocketIO把告警实时推到浏览器。数据层用MongoDB存攻击日志、封禁列表和系统配置Docker则负责把整套环境打包交付省去每台机器手动配Python环境的重复劳动。把一条数据包走完的链路写出来对后面理解踩坑很有帮助网卡收到一个SYN包Scapy回调提取源IP、目的IP、源端口、目的端口和TCP标志位写入packet_queue后台detector线程从队列取包更新滑动窗口统计判断是否超过阈值一旦命中SYN_FLOOD规则防御模块封禁源IP并写告警记录同时SocketIO把告警事件推给前端Chart.js页面上的流量曲线和告警表格同步刷新。链路里最容易被忽视的是抓包与检测必须解耦否则高流量下Scapy回调被数据库写操作卡住抓包本身就会丢包。Flask在这里的角色是“应用外壳”不管抓包细节只负责路由、模板渲染和静态资源。modules目录下按职责拆了sniffer和detector模块routes.py负责API和页面路由templates和static分别是Jinja2模板与前端资源。这种拆法对毕设答辩很友好评委问“模块之间怎么解耦”直接指出队列和SocketIO事件这两条边界就行比一锅粥的脚本清晰得多。2.2 源码结构解读哪些文件是你要改的收到压缩包后先别急着跑把目录结构过一遍搞清楚每个文件的作用后面改规则、加页面才不会找半天。典型结构如下路径职责你需要关注的改动点app.pyFlask应用入口启动Web服务和抓包线程端口、host、是否启动抓包app/init.py应用工厂注册路由与数据库连接MongoDB连接串、模块注册modules/sniffer.pyScapy抓包与队列写入iface网卡名、抓包过滤表达式modules/detector.py规则引擎与攻击判定阈值、时间窗口、规则逻辑modules/defense.py封禁、解封与黑名单管理封禁时长、iptables命令routes.py页面路由和API接口/api/alerts之类接口的返回字段templates/Jinja2页面模板仪表盘布局static/Chart.js、Bootstrap等前端资源图表刷新频率、告警颜色docker/Dockerfile与docker-compose.yml端口映射、卷挂载install.sh / install.py依赖安装与初始化Python版本、pip源requirements.txtPython依赖清单版本锁定app.py是整个系统的心脏初始化Flask应用、绑定SocketIO、启动抓包子线程。核心代码大概是这个骨架from flask import Flask from flask_socketio import SocketIO from modules.sniffer import start_sniffer from modules.detector import DetectionEngine from routes import register_routes app Flask(__name__) socketio SocketIO(app, cors_allowed_origins*, async_modethreading) detector DetectionEngine() register_routes(app, detector, socketio) if __name__ __main__: start_sniffer(detector, ifaceeth0) socketio.run(app, host0.0.0.0, port5000, debugFalse)这里有几个参数值得展开。host设成0.0.0.0是为了让Docker容器或局域网里其他机器能通过IP访问仪表盘如果只在本机调试改成127.0.0.1也行。debugTrue在我们这种带抓包线程的场景里要慎开因为Flask的reloader会重启进程把抓包子线程也带着重启容易出现端口占用或重复抓包。async_modethreading指的是SocketIO用线程模式跑不依赖eventlet和gevent兼容性最好代价是并发性能一般对毕设仪表盘完全够用。start_sniffer(detector, ifaceeth0)这一行是在Web服务启动前拉起抓包线程注意它是非阻塞的内部用daemon线程跑Scapy的sniff循环所以不会挡在socketio.run前面。iface名字在不同操作系统上差异很大Linux常叫eth0或ens33Windows常见WLAN或以太网macOS是en0写死之前先跑一遍Scapy的list_interfaces确认。2.3 容器化部署结构Docker Compose 编排与两个注意点压缩包里带了docker-compose.yml和Dockerfile说明项目作者考虑到不同机器跑Python环境容易翻车直接容器化交付。Compose文件大致长这样version: 3.8 services: app: build: . ports: - 5000:5000 depends_on: - mongo environment: MONGO_URI: mongodb://mongo:27017/ids_db volumes: - ./logs:/app/logs mongo: image: mongo:6.0 volumes: - mongo_data:/data/db volumes: mongo_data:这里面第一个坑是端口映射跟抓包的关系。app容器做端口映射后虽然外部能访问5000但容器内Scapy监听的是容器自己的虚拟网卡抓不到宿主机物理网卡的真实流量。我一般会用两种解法一是把app服务改成network_mode: hosthost模式下容器直接用宿主网络栈但这时ports配置要删掉访问直接用宿主机IP加5000二是让抓包模块支持通过参数关闭抓包把抓包进程单独跑在宿主机上容器只跑Web和检测。两种方式各有取舍具体看你的演示场景。第二个坑是depends_on只是控制启动顺序不代表MongoDB已经完全就绪。容器启动的瞬间MongoDB可能在初始化过程中Flask应用连库会报错。常见做法是在install.py或app启动代码里加一个重试逻辑或者用healthcheck检查mongo的健康状态再启动app不然第一次docker-compose up偶尔会看到MongoDB连接超时。提示容器部署时抓包权限和网络栈隔离是两件最容易出问题的事建议先在宿主机上把整套系统跑通再套容器这样排错时能把环境因素和代码因素分开。2.4 选型理由为什么是 Scapy、MongoDB 而不是别的方案选型不是拍脑袋这个技术栈明显是为“教学可演示、开发速度快、改起来直观”服务的。抓包层面Python生态里最常用的是Scapy和dpkt。Scapy不仅能解析还能构造和发送数据包调试攻击脚本和验证防御效果时不用换另一套工具链这是它最大的优势。dpkt解析性能更好但只有解析没有构造想模拟SYN Flood还要额外找发包库。Suricata和Snort性能、规则库都强但要先学一套规则语法和配置体系对课程设计来说学习成本偏高。下面是面向毕设场景的对比方案性能上手成本发包能力适合场景Scapy一般低支持教学、原型验证、快速开发dpkt较好中不支持纯解析类工具Suricata高高不支持生产级入侵检测Snort高高不支持生产级入侵检测存储层选MongoDB而不是MySQL是因为攻击日志和流量统计都是半结构化数据字段随时可能扩展比如今天要加一个user_agent字段用MySQL就得改表结构MongoDB直接往文档里塞就行了。另外黑名单、检测规则这些配置以JSON文档形式存跟Python的字典结构天然匹配不用做ORM映射。对毕设项目答辩时解释“为什么文档型数据库更适合日志类数据”本身就是个加分项。3. 实时流量分析与攻击检测抓包参数、规则定义与检测链路3.1 抓包入口sniff参数到底怎么调Scapy的sniff是系统流量分析的入口几乎所有实时数据都从这函数出来。很多人第一次跑起来发现“啥都抓不到”或者“一抓就卡死”基本都是参数没吃透。先看一段最核心的抓包代码from scapy.all import sniff, IP, TCP, UDP, ICMP, list_interfaces iface eth0 # Windows上先跑 list_interfaces() 确认实际网卡名 print(list_interfaces()) def handle_packet(pkt): if IP in pkt: src pkt[IP].src dst pkt[IP].dst proto pkt[IP].proto size len(pkt) print(f[capture] {src} - {dst} proto{proto} size{size}) sniff(ifaceiface, prnhandle_packet, storeFalse, count0, timeout60)参数逐个说清楚iface指定监听的网卡写错就抓不到所以前面先打印list_interfaces把所有可用网卡列出来prn是每抓到一包就会被调用的回调函数这是数据处理的主入口storeFalse很关键sniff默认会把所有包缓存在内存里长时间抓包内存会持续上涨关掉存储只做实时处理才能跑长任务count0表示不限数量一直抓timeout60是分段停止配合循环可以做定时任务。注意抓包需要root权限Linux下要用sudo运行Windows下要装Npcap驱动。回调函数里我通常只做最轻量的字段提取不做数据库写入不做复杂运算。原因很简单sniff的回调是同步执行的如果回调里耗时太长后面的包就要排队高流量下直接丢包而且CPU会被拖满。这个“回调里只干轻活”的原则后续小节会用队列来落实。3.2 规则定义从抓包到攻击事件有了原始包下一步是从协议字段里提取特征跟规则库比对。规则引擎的核心是五元组加TCP标志位源IP、目的IP、源端口、目的端口、协议再加上包大小和时间戳这八个字段足以识别出绝大多数常见攻击。项目里预置的规则大致是这么一张表规则名特征条件判定动作端口扫描同一源IP在60秒内访问超过20个不同目的端口且均为SYN包告警并封禁源IPSYN Flood同一源IP在60秒内SYN包数量超过30告警并封禁源IPICMP Flood同一源IP在60秒内ICMP包数量超过100告警并限制速率Ping of DeathICMP包长度异常或分片偏移异常告警并丢弃规则引擎的代码实现可以写得很简洁核心就是一个滑动窗口计数器import time from scapy.all import IP, TCP THRESHOLD 30 WINDOW 60 stats {} def detect_attack(pkt): if IP in pkt and TCP in pkt: flags pkt[TCP].flags if flags 0x02: # SYN标志位 now time.time() src pkt[IP].src stats.setdefault(src, []).append(now) stats[src] [t for t in stats[src] if now - t WINDOW] if len(stats[src]) THRESHOLD: return SYN_FLOOD, src return None参数说明都写在这里THRESHOLD30表示一个源IP在WINDOW60秒内SYN包累计超过30次就判定为攻击。这里不是简单累加而是用列表存储每个包的时间戳然后过滤掉窗口之外的老记录这样计数器就能滑动不会因为“两小时前发过攻击”这种历史数据一直触发告警。flags 0x02是位运算检查TCP头部SYN标志位是否为1比直接比较flags 0x02更严谨因为SYN包可能同时带其他标志位。阈值怎么定这是整个系统里最玄学的部分。30这个数字在Demo环境里很灵敏几行扫描脚本就能触发但如果你在自己电脑上开着浏览器访问外网正常连接产生的SYN可能也会到几十次容易误报。我一般会把检测分成两步先用低阈值做“可疑标记”再要求连续多个窗口都命中才判定攻击这个思路在后面进阶章节会细说。3.3 检测链路与队列解耦别在回调里做重活流量分析和攻击检测两个阶段要分开跑靠queue连接。抓包线程只负责把包放队列检测线程负责消费和判定这样两边速度不一致时不会互相拖累。工程上的标准做法import queue import threading packet_queue queue.Queue(maxsize2000) alert_queue queue.Queue() def on_packet(pkt): try: packet_queue.put_nowait(pkt) except queue.Full: pass # 队列满了就丢优先保证抓包线程不被阻塞 def detection_worker(): while True: pkt packet_queue.get() event detect_attack(pkt) if event: alert_queue.put(event) threading.Thread(targetdetection_worker, daemonTrue).start()put_nowait配合queue.Full的处理是防止队列无限积压的兜底策略当检测速度跟不上抓包速度时新的包会被丢弃而不是撑爆内存这在实时监控里是“丢新不丢旧”的取舍。detection_worker放在daemon线程里主程序退出它跟着结束不会残留僵尸线程。alert_queue是给下一阶段防御模块用的检测模块只负责“发现问题”并把事件交给队列不直接执行封禁保持模块间单向依赖。队列解耦之后有一个附带的好处流量统计和告警展示可以各自消费数据。前端需要每秒刷新一次统计数字直接读stats字典就行告警推送只消费alert_queue。四层链路各干各的定位问题的时候也能很快判断是抓包死了还是检测逻辑写错了。另外一个实用技巧是给sniff加BPF过滤表达式比如只关心TCP流量就写sniff(ifaceiface, filtertcp, ...)能显著降低无效包的数量让CPU省下来干活。4. 自动防御与告警响应拦截机制、黑名单策略与可视化监控4.1 自动防御从命中到封禁的完整动作检测到攻击只是第一步这个项目的亮点是“自动防御”也就是命中规则后立刻执行拦截策略。项目里最直接的做法是通过iptables封禁来源IP模块里用一个ban_ip函数封装整条命令import subprocess import time from pymongo import MongoClient client MongoClient(mongodb://localhost:27017) db client.ids_db BAN_DURATION 300 # 封禁时长单位秒 def save_alert(event): db.alert_logs.insert_one({ time: int(time.time()), type: event[0], src: event[1], action: ban }) def ban_ip(ip, durationBAN_DURATION): if is_whitelisted(ip): return False db.blacklist.insert_one({ ip: ip, created_at: int(time.time()), expire_at: int(time.time()) duration }) cmd [iptables, -I, INPUT, -s, ip, -j, DROP] result subprocess.run(cmd, capture_outputTrue, textTrue) return result.returncode 0逻辑说明先查白名单命中的IP直接放行避免把自己或学校网关封掉然后写MongoDB黑名单记录封禁时间和过期时间为后面的自动解封留依据最后用subprocess调用iptables添加拒绝规则。-I INPUT表示插入到INPUT链最前面优先匹配-s指定源IP-j DROP直接丢包比REJECT更省资源也不给对方明确反馈。save_alert把攻击事件落到alert_logs集合这是MongoDB在项目里的主要数据来源。这里有个权限前提执行iptables需要root权限。如果你用sudo python app.py启动整套系统subprocess调用iptables时是继承sudo身份的如果用普通用户启动这里会报权限不足需要在defense.py里做一次权限检测并给出明确提示。Docker容器里跑iptables还需要容器具备NET_ADMIN能力否则同样失败。4.2 黑名单策略与自动解封别把自己封出局封禁IP做到一半就不管了演示久了会发现黑名单越积越长内网里大家全被封掉。项目里的解封逻辑是这样做的后台开一个定时任务每30秒扫描一次blacklist集合检查过期记录调用iptables删除对应规则def sweep_expired_rules(): now int(time.time()) expired db.blacklist.find({expire_at: {$lt: now}}) for doc in expired: ip doc[ip] subprocess.run([iptables, -D, INPUT, -s, ip, -j, DROP]) db.blacklist.delete_one({_id: doc[_id]}) threading.Timer(30, sweep_expired_rules).start()delete_one用_id定位删除避免删错记录-D参数跟-I对应表示删除指定规则。之所以做成30秒扫一次而不是每次封禁时顺带检查是为了避免频繁遍历iptables规则列表毕竟iptables -L在规则多的时候也不算快。封禁时长BAN_DURATION默认300秒也就是5分钟这个值对演示刚好攻击脚本跑完去Web面板截图5分钟后恢复连通不影响后续测试。防御策略上还应该预留白名单机制项目里黑名单与白名单通常是两个集合白名单优先级高于黑名单。判断顺序一定要是“先白名单后黑名单”我在实际调试时有过一次血泪教训把自己电脑的IP写进黑名单规则后SSH直接断了只能去机房手动清规则。从那之后我强制要求所有封禁动作前必须过一遍白名单。4.3 实时监控与告警推送SocketIO怎么把事件推给浏览器防御动作做完可视化监听是这套系统最出效果的部分。后端通过routes.py暴露统计接口同时用SocketIO把新告警实时推给前端。后端接口大致提供了以下数据接口返回内容前端用途/api/stats总包数、告警数、封禁IP数、当前速率仪表盘顶部数字卡片/api/alerts最新告警列表告警表格初始渲染/api/blacklist当前黑名单及过期时间黑名单管理页面/api/rules规则列表与阈值配置系统设置页面前端JavaScript监听告警事件插到表格顶部并刷新图表var socket io(); socket.on(alert_event, function(data) { $(#alert-table).prepend( trtd data.time /tdtd data.type /tdtd data.src /tdtd已封禁/td/tr ); updateChart(data.time, data.type); }); function updateChart(time, type) { if (typeof attackChart ! undefined) { attackChart.data.labels.push(time); attackChart.data.datasets[0].data.push(1); attackChart.update(); } }SocketIO事件的数据结构由后端决定通常包含time、type、src三个字段分别对应攻击时间、攻击类型、来源IP。prepend把最新告警插到表格第一行比append更像实时监控的交互习惯。Chart.js的attackChart实例在页面初始化时创建updateChart每次推送都往里追加一个点形成攻击频率的时间序列。后端推送的逻辑其实很薄检测线程命中规则后defense模块把事件写入alert_queue然后由推送线程通过socketio.emit(alert_event, event)广播给所有前端页面。这里顺带解释为什么选Flask-SocketIO而不是前端轮询API轮询每秒请求一次接口流量大、延迟高而且浏览器要一直维持HTTP请求SocketIO建立长连接后服务端主动往客户端推浏览器页面打开就能收到实时告警演示体验明显更流畅。5. 避坑与常见问题从环境搭建到跑通的五条踩坑记录这套系统看着不复杂但真从头搭一遍容易踩的坑比想象中多。下面五条都是我在实际跑项目时真实触发过的问题每条按“现象、原因、解决”写清楚遇到类似情况可以直接对着排。5.1 抓包环境起不来或收不到包现象启动后直接抛OSError: Permission denied连list_interfaces都能正常输出一到sniff就崩或者程序跑起来但一行流量都打不出来回调函数从不执行。原因两类情况混在一起。一是抓包权限不对Scapy要开原始套接字普通用户没这个权限二是监听的网卡选错了Windows上常见的是WLAN或以太网代码里写死eth0自然抓不到另外sniff里的filtertcp这类BPF表达式也会把UDP、ICMP都过滤掉造成“什么都没发生”的假象。解决Linux、macOS下用sudo python app.py启动Windows下先装Npcap装的时候勾选WinPcap API兼容模式。网卡名不要靠猜在代码开头跑一次scapy.all.list_interfaces()对照系统网络设置里的名字再填。测试阶段先把filter参数去掉用手机往电脑发消息验证有数据进来再逐步加过滤这样能快速确认抓包链路是通的。5.2 MongoDB连接超时与启动顺序问题现象Flask启动后报pymongo.errors.ServerSelectionTimeoutError: No servers found yetWeb服务起来了但所有依赖数据库的接口都在报错。原因Mongo服务没启动或者连接串写错。本地开发时mongod没跑起来后端代码直接连localhost:27017自然连不上容器部署时depends_on只是控制启动顺序MongoDB容器在初始化完成之前Flask就开始连库了。解决本地先命令行跑mongod --dbpath验证数据库能独立启动再用mongosh连接确认没问题后再启动Python应用。容器环境给mongo服务加healthcheck或者在应用启动代码里做三次重试、每次间隔两秒。连接串注意区分环境容器内部用mongodb://mongo:27017宿主机直连用mongodb://localhost:27017别混着填。5.3 Docker容器里iptables防御失效现象Web面板显示“已封禁”来源IP进了黑名单但攻击流量还在源源不断进来防御形同虚设。原因Docker容器内的iptables跟宿主机网络栈是隔离的容器里执行iptables -I INPUT封的是容器自己的INPUT链宿主机物理网卡收到的流量根本不经过这条链。另外有的defense模块只往数据库写了黑名单没校验iptables命令是否真的执行成功前端看到“已封禁”其实是误报。解决容器部署时把app服务设置成network_mode: host让容器共享宿主机网络栈封禁规则才能作用到真实流量或者把封禁逻辑从容器里挪出来改成宿主机上的独立进程执行。defense模块里必须检查subprocess.run的返回码非零就记录错误日志并回滚前端状态数据库里有没有记录不能当作封禁成功的依据。5.4 CPU飙到100%与页面卡死现象系统运行一段时间后CPU占用接近100%Web页面刷新极慢抓包结果也出现明显延迟。原因Scapy的回调函数里干了重活比如在回调里直接写MongoDB或者做复杂规则计算导致抓包线程被阻塞sniff没有设置storeFalse所有包都被缓存在内存里越积越多stats字典里滑动窗口的过期数据没清理内存和计算量都在涨。解决回调里只做字段提取和入队把检测逻辑交给detection_worker线程sniff显式传storeFalse统计字典每次写入时顺手把窗口之外的时间戳删掉。如果流量确实很大给sniff加BPF过滤表达式只保留TCP或指定协议减少无效包。5.5 前端SocketIO始终连不上现象页面能打开但浏览器控制台一直报WebSocket连接失败实时图表完全不更新只有手动刷新才能看到新数据。原因Flask-SocketIO的async_mode没显式指定机器上装了eventlet就优先用eventlet没装就退回threading两套异步行为不一致如果前面还挂了网关或负载均衡网关没开启WebSocket协议升级长连接建立不起来。解决创建SocketIO时强制指定async_modethreading去掉对eventlet和gevent的隐式依赖本地演示直接访问5000端口不套额外网关层能少一层变量。确认后端代码里用了socketio.emit而不是普通request上下文里的写法后者在实际部署时也常导致推送失效。6. 进阶验证用Scapy模拟攻击流量校准检测灵敏度6.1 自己给自己发攻击包验证系统是否真能命中系统跑通之后最需要证明的是“它真的能检测到攻击”而不是界面上一直显示零告警。安全起见只在自己实验网段里测试别拿公网IP练手。用Scapy写一个最小的SYN Flood模拟器from scapy.all import IP, TCP, send target_ip 192.168.1.100 # 换成你跑检测系统的机器 for i in range(50): pkt IP(dsttarget_ip)/TCP(flagsS, sport10000i, dport80) send(pkt, verboseFalse)这段脚本发起50个源端口递增的SYN包模拟一次小规模握手洪泛。跑完去Web面板看告警日志正常应该看到SYN_FLOOD事件来源IP是执行脚本的机器状态为已封禁。端口扫描的验证同理把目的端口改成1到100连续扫一遍触发端口扫描规则。如果发完包面板没反应优先怀疑检测线程没有消费队列或者规则阈值太高先用小数量发包确认抓包链路是通的再逐步加大数量。6.2 灵敏度校准把阈值从“乱封”调到“可用”我最早把SYN阈值设成10结果开个浏览器、刷个网页后台就封了一片IP连自己路由器都封进去页面直接打不开。后来改了思路用滑动窗口连续两次命中才触发封禁第一次命中只记可疑不执行动作。也就是把“一次超标”升级为“持续超标”误报率立刻降下来。参数上我最终用的是窗口60秒、单窗口SYN阈值30、连续命中2次、封禁300秒、白名单包含网关和本机IP。这套值在单机演示和课程设计场景下很省心。如果你要在更真实的网络里跑把阈值拉高到50到80连续命中次数保持2到3次。从那以后我每次改规则都会先跑一遍仿真流量再放行确认检测逻辑确实命中、封禁确实生效然后才放心做后续测试希望帮到你。本文还有配套的精品资源点击获取