基于Python的网络入侵检测与防御系统：实时流量分析与自动封禁

简介面向Python网络方向的毕业设计或课程设计场景这份资源提供一套可运行的网络入侵检测与防御系统解决从实时流量捕获、异常行为识别到自动拦截阻断与可视化监控的完整实现问题。系统基于Python语言、Flask框架与Scapy抓包库构建具备实时流量分析、攻击检测、自动防御、告警响应及Web管理界面可帮助网络空间安全或计算机相关专业学生快速搭建毕设原型也便于理解入侵检测系统的典型工作流程。资源压缩包共包含38个文件整体大小仅92KB以14个Python源码文件和9个编译后的pyc文件为主体另含4个HTML页面、3个JavaScript脚本、JSON配置文件、Dockerfile与docker-compose部署编排文件以及安装脚本和依赖清单覆盖后端逻辑、前端展示、容器化部署、环境安装等多个环节目录结构清晰便于按模块阅读和定位修改。目前已有172人学习/下载适合需要快速完成毕业设计、课程设计或安全类实验的读者。随包附带的详细运行指南和项目说明能有效降低环境配置与启动门槛使使用者尽快跑通系统并在此基础上扩展检测规则、调整防御策略或集成到现有网络监控方案中。1. 基于Python的网络入侵检测与防御系统是什么把实时流量分析与自动防御做成毕设“基于Python的网络入侵检测与防御系统”这个标题看着像课程作业但真正做过的同学会告诉你难点不在“检测”两个字上。抓包要权限解析要特征封禁要联动系统防火墙可视化还要把告警实时推到前端。这套系统能落地的关键是先跑通一条从网卡到告警再到封禁的数据链路而不是一上来就训练模型。我见过不少毕设动辄上机器学习结果训练数据全是从网上找的离线PCAP上线后连本机流量都看得不全。这个项目老老实实走规则检测和阈值统计用Scapy做实时流量分析把SYN扫描、高频请求这类行为抓出来再调用iptables做自动防御最后用Flask-SocketIO把结果可视化。技术栈干净每一层都能讲清楚原理适合网络方向毕设也适合准备安全开发岗位的在校生。如果你手上只有一台普通笔记本或虚拟机这套方案同样能跑。下面按“采集、检测、防御、监控、调试、验证”的顺序把我的做法完整过一遍。2. 实时流量分析模块用Scapy把抓包、特征解析与数据落库跑通整个系统里最先要打通的是数据链路而这条链路的起点不是检测算法是抓包。很多毕设翻车的共同点是先写检测代码再回头补采集结果流量根本进不来阈值全部空转。所以第2章先把实时流量分析链路走通从网卡到特征字典再到数据库。2.1 解析方案选型为什么是Scapy而不是dpkt、pyshark或Snort实时流量分析首先要选一个抓包解析方案。常见选项是Scapy、dpkt、pyshark外加Snort、Suricata这类成熟引擎。Snort功能确实完整但对毕设来说太重规则语法和内部架构要解释半天容易把自己绕进去dpkt解析效率最高但要把以太网、IP、TCP逐层手工解开代码量直接翻倍pyshark解析起来方便可前提是机器上装了tshark部署环境多一个外部依赖。我一般选Scapy。它底层走libpcap抓包能力有保障上层提供sniff()回调和现成的包对象可以直接提取五元组、TCP标志位、载荷长度。实验室环境的包速率不大Scapy的解析性能完全够用而且代码量小答辩时每一行都能讲清楚。方案优点明显代价适合场景Scapy抓包解析一体代码量小包处理性能一般毕设、安全工具原型dpkt解析效率最高每层协议都要手写解析追求吞吐量的工具pyshark展示字段丰富依赖系统tshark离线PCAP分析Snort/Suricata规则和检测能力完备配置文件多学习曲线陡生产环境部署判断标准很简单包速率不高但解析字段要灵活回调里拿到包就能转成特征字典。Scapy在这两条上恰好都占。2.2 抓包与特征提取代码一个队列缓冲实时流量先看采集部分的完整代码。这个模块只做一件事把网卡上的IP包转成扁平字典塞进队列不在这里写检测逻辑。from queue import Queue from scapy.all import sniff, IP, TCP, UDP, Raw # 抓包线程和生产队列只放特征不放整个包对象 feature_queue Queue(maxsize2000) def packet_handler(pkt): # 只要IP层非IP流量直接丢弃 if not pkt.haslayer(IP): return ip pkt[IP] item { ts: pkt.time, src: ip.src, dst: ip.dst, proto: ip.proto, } if pkt.haslayer(TCP): tcp pkt[TCP] item[sport] tcp.sport item[dport] tcp.dport item[syn] bool(tcp.flags.S) item[ack] bool(tcp.flags.A) item[payload_len] len(tcp.payload) elif pkt.haslayer(UDP): udp pkt[UDP] item[sport] udp.sport item[dport] udp.dport item[payload_len] len(udp.payload) if pkt.haslayer(Raw): # 只保留载荷前64字节的hex用来回溯攻击特征 item[payload_head] pkt[Raw].load[:64].hex() try: feature_queue.put_nowait(item) except Exception: # 队列满直接丢包优先保实时性 pass # 监听网卡只放行IP流量storeFalse表示不留原始包 sniff(ifaceens33, prnpacket_handler, storeFalse, filterip)几个关键参数要注意。iface必须填实际网卡名不填的话Scapy会走默认网卡同一台机器有多个网卡时容易抓到不想要的包storeFalse表示不在内存里堆积原始包这个参数必须带上否则跑一段时间内存就满了filterip是BPF语法只放行IP包把ARP等链路层噪音挡在外面。packet_handler里提取的特征是后续检测模块直接消费的字段。tcp.flags.S和tcp.flags.A分别判断SYN、ACK标志位这是识别端口扫描和TCP握手状态的原材料。载荷前64字节转hex是为了给告警留证据又不至于把整包打印出来刷屏。这里还有一个重要的实战细节如果这台机器同时用来SSH远程管理建议把过滤器写成filterip and not tcp port 22把管理流量排除在采集范围外避免检测模块把你自己的SSH会话也当成异常流量。这个动作能省掉后面大半的误封麻烦。2.3 数据落库用SQLite批量写入替代逐条insert流量解析成字典之后不能每条都立刻写数据库否则高频流量下SQLite的commit会成为瓶颈。常见做法是采集线程只负责投递单独开一个线程做批量落库。我用SQLitets字段存unixepoch秒后面按时间聚合查询非常方便。import sqlite3 import threading import time conn sqlite3.connect(traffic.db, check_same_threadFalse) conn.execute( CREATE TABLE IF NOT EXISTS packet_features ( id INTEGER PRIMARY KEY AUTOINCREMENT, ts REAL, src TEXT, dst TEXT, proto INTEGER, sport INTEGER, dport INTEGER, flags TEXT, payload_len INTEGER, payload_head TEXT )) BUFFER [] BUFFER_LOCK threading.Lock() FLUSH_INTERVAL 5 # 每5秒批量写一次 def append_buffer(item): with BUFFER_LOCK: BUFFER.append(item) def flush_worker(): while True: time.sleep(FLUSH_INTERVAL) with BUFFER_LOCK: if not BUFFER: continue rows [ (i[ts], i[src], i[dst], i[proto], i.get(sport), i.get(dport), , i.get(payload_len), i.get(payload_head)) for i in BUFFER ] BUFFER.clear() # 注意clear在锁内executemany在锁外 conn.executemany( INSERT INTO packet_features (ts, src, dst, proto, sport, dport, flags, payload_len, payload_head) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?), rows ) conn.commit()check_same_threadFalse让采集线程和落库线程可以共用同一个连接前提是同一时刻只有一个线程真正执行execute和commit。我这里的写法是把BUFFER的读写都放进锁里而数据库操作在锁外避免长时间持锁卡住抓包线程。FLUSH_INTERVAL设成5秒意思是最多丢5秒的流量数据。实验室环境这个值足够。如果追求更低的时延改成1秒也可以但磁盘写入会明显频繁整机性能会往下掉。2.4 自检方法用一次ping验证整条链路写检测模块之前先跑一次最小链路验证这一步能筛掉八成后续问题。流程是这样的# 终端1启动采集脚本 sudo python3 collector.py # 终端2发3个ICMP包 ping -c 3 192.168.1.1 # 终端3查看库里有没有记录 sqlite3 traffic.db SELECT src, dst, proto FROM packet_features ORDER BY id DESC LIMIT 5;如果查询结果里出现了本机IP и目标IP说明抓包、解析、落库这条实时流量分析链路是通的。ping包属于ICMP协议proto字段会显示1能查ICMP说明BPF过滤器没把非TCP流量误过滤掉。确认这一步后面检测模块才有数据可吃。3. 攻击检测与自动防御把SYN扫描、高频请求与iptables封禁做成闭环流量采下来之后攻击检测模块要解决的问题是“哪条流量可疑”。这块不要贪多做两个行为检测就足够撑起整个系统端口扫描/SYN洪泛以及同一源IP高频新建TCP连接。防御动作做到自动封禁IP、到期自动解封并把告警推送出去。3.1 检测规则设计先想清楚要抓什么行为写检测代码前先定义异常行为。我见过太多代码把阈值写得像随机数自己都解释不清为什么是20而不是50。检测规则的三要素是时间窗口、阈值、白名单。端口扫描和SYN洪泛有一个共同特征短时间内同一个源IP发出大量SYN包但没有完成TCP握手。正常浏览器访问网页也会建TCP连接但密度完全达不到攻击水平。高频请求则反过来表现为同一个源IP在几秒内完成大量TCP握手这种更像被滥用的代理池或者刷接口脚本。设计规则时把这两类行为分开后续调参才不糊涂。3.2 滑动窗口计数器识别SYN洪泛与高频建连检测模块的核心代码我推荐用滑动窗口列表而不是一个简单计数器。原因是计数器只统计总数时间窗口一滚动旧数据还留在计数里误报率会随运行时长不断升高。看下面的实现from collections import defaultdict SYN_WINDOW 10 # 统计窗口10秒 SYN_THRESHOLD 20 # 窗口内SYN包个数阈值 syn_memo defaultdict(list) def detect_syn_flood(item): # 只统计纯SYN包带ACK的TCP握手包不算 if not (item.get(syn) and not item.get(ack)): return None src item[src] syn_memo[src].append(item[ts]) # 只保留当前窗口内的记录 syn_memo[src] [t for t in syn_memo[src] if item[ts] - t SYN_WINDOW] if len(syn_memo[src]) SYN_THRESHOLD: # 命中后清掉该源IP的窗口避免同一波攻击刷屏告警 syn_memo[src].clear() return { alert: syn_flood, src: src, count: SYN_THRESHOLD 1, window: SYN_WINDOW, } return None这段代码有两个容易被忽视的细节。第一判断条件是synTrue且ackFalse只统计扫描或洪泛特征的SYN包不把正常握手的第二个包算进来。第二命中后立即clear()清楚这个源IP的窗口否则10秒内后续每个SYN包都会再触发一次告警告警数量会被同一波攻击刷爆。高频连接检测的逻辑类似但要换成对ACK包的统计。一次完整TCP连接建立的标志是收到SYNACK所以判断条件是synTrue且ackTrueCONN_INTERVAL 5 # 统计窗口5秒 CONN_THRESHOLD 60 # 窗口内新建连接阈值 conn_memo defaultdict(list) def detect_high_conn(item): # 只统计TCP握手的第二个包SYNACK if not (item.get(syn) and item.get(ack)): return None src item[src] conn_memo[src].append(item[ts]) recent [t for t in conn_memo[src] if item[ts] - t CONN_INTERVAL] conn_memo[src] recent if len(recent) CONN_THRESHOLD: # 高频建连不清空窗口因为这是连续行为 return { alert: high_conn, src: src, count: len(recent), interval: CONN_INTERVAL, } return None高频建连我没有清空窗口因为这类攻击往往是持续性的清空会导致下一秒又重新计数告警节奏反而不对。具体阈值怎么调第5章会展开。3.3 自动防御iptables封禁、白名单与自动解封检测命中之后必须接上防御动作否则检测模块只是纸上谈兵。我在Linux下用的是iptables封禁源IP核心操作就两条-I INPUT在规则表最前面插入一条DROP规则-D INPUT删除这条规则实现解封。import subprocess import time import threading BAN_SECONDS 300 # 封禁时长300秒演示时可调成120 WHITELIST {127.0.0.1} # 本机和内网管理地址必须放行 ban_records {} # ip - 解封时间戳 def do_ban(ip, alert_type): if ip in WHITELIST: return False # 已经封禁且未到期不重复操作 if ip in ban_records and ban_records[ip] time.time(): return False result subprocess.run( [iptables, -I, INPUT, -s, ip, -j, DROP], capture_outputTrue, textTrue, checkFalse, ) if result.returncode 0: ban_records[ip] time.time() BAN_SECONDS save_alert(ip, alert_type, int(ban_records[ip])) return True return False def unban_loop(): while True: time.sleep(10) now time.time() for ip, expire in list(ban_records.items()): if now expire: subprocess.run( [iptables, -D, INPUT, -s, ip, -j, DROP], capture_outputTrue, checkFalse, ) del ban_records[ip]do_ban里用了checkFalse并捕获stderr而不是直接让子进程抛异常。因为iptables在权限不足时返回非0退出码如果你用checkTrue脚本会当场崩溃你只会在Traceback里看到一行Permission denied连日志都没有。现在这个写法可以拿到报错信息方便排查。unban_loop每10秒扫一次封禁记录到期的IP自动解封这样即使写错了也不会造成永久封禁。这里要特别强调白名单我的习惯是把127.0.0.1写死再把本机所在网段的网关地址加进去否则误封会把你自己也关在门外。具体踩坑记录在第5章。3.4 告警去重与削峰不要让抓包回调去做防御检测和防御动作不能直接写在packet_handler里。抓包回调是高频热点一个扫描过程可能触发几十上百个SYN包如果每个包都去执行一次subprocess调iptables抓包线程直接卡死监控页面也会跟着假死。正确的做法是生产消费模型from queue import Queue alert_queue Queue() def detector_loop(): while True: item feature_queue.get() alert detect_syn_flood(item) or detect_high_conn(item) if alert: alert_queue.put(alert) def defense_worker(): while True: alert alert_queue.get() if do_ban(alert[src], alert[alert]): push_alert(alert)detector_loop从流量队列里取特征做检测命中后放进告警队列。defense_worker是唯一负责执行封禁和解封的线程它拿到告警后调用do_ban只有第一次真正执行封禁时才推送告警。这样同一波攻击不会重复刷屏前端监控也不会被海量重复消息淹没。4. 可视化监控用Flask-SocketIO把检测结果实时推到Dashboard检测和防御都接上了最后把结果用Web页面展示出来。可视化监控这一层做得太重会被说是凑字数完全不做又体现不出“监控”。合理的目标是一张实时告警列表、一个最近30分钟攻击趋势图加上一张历史告警表。4.1 监控数据链路三个线程各管一件事整个系统的运行结构是三条线程。抓包线程负责采集流量放进feature_queue检测线程消费特征产生告警后放进alert_queue并执行封禁Web服务线程负责把告警推给浏览器并提供历史查询接口。三个线程之间靠队列解耦任何一环阻塞都不会拖垮其他环节。如果只用一个进程跑全流程Web服务要选一个支持异步的框架。普通Flask开发服务器在处理长轮询时表现并不好所以我选了Flask-SocketIO用WebSocket把新告警直接推到前端。实时流量分析的结果能不能“实时”看到关键就在这一层。4.2 Flask-SocketIO服务端新告警主动推送下面这段是监控服务端的骨架。它只做两件事客户端连上来时发一条确认消息告警产生时广播到所有已连接页面。from flask import Flask, render_template, jsonify from flask_socketio import SocketIO, emit app Flask(__name__) app.config[SECRET_KEY] monitor # async_modethreading 是最稳的运维方式 socketio SocketIO(app, async_modethreading) MONITOR_NS /monitor socketio.on(connect, namespaceMONITOR_NS) def handle_connect(): emit(pong, {msg: connected}, namespaceMONITOR_NS) def push_alert(alert): socketio.emit(alert, alert, namespaceMONITOR_NS, broadcastTrue)async_mode有三个可选值threading、eventlet、gevent。我推荐直接用threading它不依赖额外安装高性能并发库兼容性最好。eventlet和gevent在Windows上经常装出问题折腾半天只是为了让性能更好但毕设场景根本用不到那个量级的并发。socketio.emit的broadcastTrue表示推送给所有在/monitor命名空间下的客户端。命名空间的作用是隔离监控页面和其他页面如果以后加一个配置管理页两边的WebSocket事件不会互相干扰。4.3 前端接收事件与历史聚合查询前端的核心是接收“alert”事件并即时渲染。页面用ECharts画趋势图数据来自SQLite按分钟聚合的查询const socket io(/monitor); socket.on(alert, function (data) { const li document.createElement(li); li.textContent [${data.timestamp}] ${data.src} - ${data.dst} | ${data.alert}; document.getElementById(alert-list).prepend(li); refreshChart(); }); function refreshChart() { fetch(api/trend?minutes30) .then((resp) resp.json()) .then((rows) myChart.setOption({ xAxis: { data: rows.map((r) r.minute) }, series: [{ data: rows.map((r) r.cnt) }], })); }对应后端聚合查询的SQL语句是SELECT strftime(%H:%M, datetime(ts, unixepoch, 8 hours)) AS minute, COUNT(*) AS cnt FROM alerts WHERE ts strftime(%s, now) - 1800 GROUP BY minute ORDER BY minute;8 hours这个写法是固定的东八区偏移如果你在别的时区就改成对应偏移。更通用的写法是datetime(ts, unixepoch, localtime)让SQLite按系统本地时区转换。WHERE ts strftime(%s,now) - 1800表示只取最近30分钟的告警防止图表数据量越积越大导致页面卡顿。4.4 线程调度经验抓包线程不能和Web服务挤在一起一个必须提前避开的坑是不要把sniff()写在Flask的启动代码里也不要放在和socketio.run()同一个线程里。sniff()是阻塞调用一旦进入监听循环后面的代码就不执行了Web服务根本起不来。更隐蔽的问题是Scapy的回调函数运行在Scapy自己的线程里你在回调里调socketio.emit()没问题但如果直接在回调里执行SQLite写库和iptables命令就会把抓包线程拖死。正确做法是保持第3章的队列模型Web服务只消费告警队列和查询数据库不碰抓包逻辑。整个系统的实时流量分析、攻击检测、自动防御、可视化监控四个模块通过两条队列串成一条完整的流水线。5. 调试与避坑5条踩坑记录和3个必调参数这个标题听起来简单真正跑起来后问题主要集中在环境和权限。拿到源码包后别急着安装依赖先把第2章的采集脚本跑通否则后面所有检测和防御代码都处于没有数据可吃的状态。这一章把我的血泪经验按售后问题记录的方式写出来每一条都是现象、原因、解决三件套。5.1 抓包环境和权限类问题踩坑1虚拟机里抓不到任何外部流量现象宿主机访问Web应用虚拟机的采集脚本一包未收但ping虚拟机的网关地址却能收到包。原因默认NAT模式下虚拟网卡只能看到自己的虚拟网络流量。外部流量进不了这块虚拟网卡自然抓不到。解决在虚拟机软件里把网卡改成桥接模式让虚拟机直接拿到物理网段的地址或者把演示场景收敛到局域网内部用同一网段的两台机器互发流量。跑采集脚本必须用sudo python3 collector.py普通用户没有libpcap的权限。踩坑2Scapy报“assert _ip is not None”或“kernel arp filter failed”现象启动sniff()后秒退控制台输出assert错误BPF过滤器不生效。原因libpcap版本与当前内核驱动不匹配或者iface参数填成了不存在的网卡名。解决先执行sudo python3 -c from scapy.all import conf; print(conf.ifaces)看有哪些接口名把iface参数改成实际存在的名字。如果问题依旧重新安装libpcap-dev库再升级scapy到新版本再次启动前用sudo ldconfig刷新动态链接库。5.2 检测与自动防御联动类问题踩坑3iptables把SSH管理会话误封自己把自己踢下线现象检测到高频连接告警随后本机SSH连接中断再也登录不上去。原因检测规则把所有访问本机的流量都算进去把自己正在使用的管理来源IP也封禁了。do_ban()没有先查白名单。解决把管理来源IP放进WHITELIST并且在启动防御功能前先执行放行命令iptables -I INPUT -p tcp --dport 22 -j ACCEPT把它放在DROP规则之前。这样即使误封SSH管理通道也始终可用。踩坑4iptables封禁规则在重启后全部消失现象重启服务器后告警还在但攻击源IP已经能访问了检查发现iptables -L里没有任何DROP规则。原因iptables规则只存在于运行时内存重启即清空。没有做规则持久化。解决用iptables-save /etc/iptables/rules.v4导出规则在开机脚本里执行iptables-restore /etc/iptables/rules.v4恢复。如果只是毕设演示不需要坚持重启持久化但要提前想清楚演示时不能重启服务器或者让unban线程在每次启动时先清一遍旧规则。踩坑5SYN阈值设太低正常网页浏览被误判为攻击现象打开一个门户首页几十秒前端弹出告警IP被自动封禁。原因一个大型页面会同时发起几十个TCP连接SYN_THRESHOLD设成10自然会把正常浏览当攻击。解决把SYN_THRESHOLD提到30以上CONN_THRESHOLD提到60以上窗口时间缩短到5秒。阈值调整没有公式可用唯一可靠的方法是录制一段正常访问流量做测试再看告警日志里有没有误报。5.3 三个必调参数下面这张表是我跑这个系统时认为最需要提前调好、也最影响演示效果的一组参数。参数值不是越小越灵敏而是要和你的演示网络环境匹配。参数建议初始值调整依据iface/filter实际网卡名 ip and not tcp port 22网卡选错抓不到包不过滤管理端口容易误封SYN_WINDOW/SYN_THRESHOLD10秒 / 30次正常页面打开不会在10秒内发30个纯SYN包CONN_INTERVAL/CONN_THRESHOLD5秒 / 60次正常浏览器访问达不到这个建连密度BAN_SECONDS120秒演示 / 300秒正式太短看不到防御效果太长误封后恢复太慢参数改完不要直接跑主程序先跑一段小流量采集把告警日志打出来看一眼。这类阈值是纯经验值不试不知道。6. 验证与进阶用自测流量逼出告警再从计数特征升到状态机特征做完上面五章系统已经能跑但你要在答辩前证明它真的“检测”到了攻击而不是纸上谈兵。最直接的做法是构造一段SYN扫描流量去打自己看系统能不能在几秒内产生告警并执行封禁。from scapy.all import IP, TCP, send # 模拟一个扫描源发30个不同目标端口的SYN包 pkt IP(src10.0.0.66, dst192.168.1.20, ttl64) / TCP(flagsS) send(pkt, count30, inter0.01)这段代码用Scapy构造30个SYN包目标是被检测主机源IP伪造为10.0.0.66。跑完之后去监控页面的告警列表和iptables规则里确认10.0.0.66应该被拉黑。这个验证有两个作用一是确认检测逻辑没写错二是提前给演示录制一段可控的自测视频。再进一步单一计数阈值误报率高可以往状态机方向升级。现在的检测只看“SYN包个数”更精细的规则应该把TCP三次握手状态考虑进来比如统计“发了SYN但没有收到SYNACK”的组合或者统计连续N个SYN包都发往不同端口。把特征从单纯的计数扩展到“目标端口去重数、包长分布、重传间隔”后攻击检测会可靠很多。我现在的习惯是每加一条规则先用自动脚本造十条正常流量和十条攻击流量去对照跑让规则基于数据说话而不是急着上模型。希望帮到你。本文还有配套的精品资源点击获取