简介这份资源是面向高校计算机、网络安全相关专业学生及课程设计学习者的完整项目源码包围绕网络入侵检测与防御系统展开可用于毕业设计、课程设计或安全方向练手。项目基于Python构建后端采用Flask、Flask-SocketIO与Scapy实现实时流量捕获与分析前端借助HTML5、Bootstrap与Chart.js完成可视化监控数据由MongoDB存储并通过Docker Compose实现容器化部署覆盖流量检测、入侵防御、告警响应与网络监控等核心模块。压缩包共38个文件约92KB包含14个py源码、9个pyc编译文件、4个html页面、3个js脚本及json、yml、Dockerfile、sh等配置与部署文件结构清晰便于按模块阅读与二次开发。目前已有172人学习关注。资源附带详细运行指南读者可据此快速搭建环境、理解检测与防御逻辑并参考Web界面与日志设计完成自己的安全类项目。1. 从一份毕设源码说起Python 网络入侵检测与防御系统到底在做什么很多人第一次接触「网络入侵检测与防御系统」是在毕设选题表上看到「Python 实现、实时流量分析、攻击检测、自动防御、可视化监控」这几个词觉得高大上真拿到源码却发现跑不起来抓包没权限、依赖装不上、Web 面板一片空白。这篇笔记就围绕这套典型架构把从环境搭建到实时流量分析、攻击检测规则、自动防御联动、可视化监控的完整链路拆开讲清楚让你既能照着复现也能看懂每一层在干什么、参数怎么调、哪里最容易翻车。它解决的核心问题是在单机或小型内网里用 Python 把「抓包 → 特征提取 → 规则/阈值判定 → 触发防御动作 → 前端展示」串成一条实时流水线。适合做毕设的学生、想入门安全开发的后端工程师以及需要给内网加一层轻量监控的运维。下面按落地顺序推进先讲清抓包与流量分析再讲检测、防御和可视化最后给一套可验证的进阶技巧。2. 实时流量分析用 Scapy 抓包并提取五元组特征实时流量分析是整个系统的地基。没有稳定的数据源后面的攻击检测和可视化都是空中楼阁。这一章先把抓包环境跑通再把原始数据包变成可判定的结构化特征。2.1 抓包选型Scapy 与原始套接字的取舍Python 抓包常见三条路Scapy、pcapy/winpcap 绑定、以及直接读/proc/net或调用系统命令。做毕设最稳的是 Scapy原因是它纯 Python、跨平台、能直接解析到 TCP/UDP/ICMP 各层字段改起来不用碰 C 扩展。代价是性能一般千兆满速下会丢包但对毕设演示和中小流量监控完全够用。抓包前必须解决权限问题。Linux 下普通用户没有原始套接字权限直接跑会报PermissionError。常见做法是给 Python 解释器加CAP_NET_RAW能力或者干脆用 root 跑Windows 下则需要装 Npcap 并勾选 WinPcap 兼容模式。这一步不解决后面所有代码都是白搭。# Linux给 python 解释器授予抓包能力避免每次 sudo sudo setcap cap_net_raw,cap_net_admineip $(readlink -f $(which python3)) # 验证是否生效 getcap $(readlink -f $(which python3))setcap给的是二进制能力eip分别代表有效、可继承、允许提升。执行后普通用户即可抓包。注意如果你用的是虚拟环境which python3要指向虚拟环境里的解释器否则能力加到了系统 Python 上跑起来仍然报权限错误。2.2 用 Scapy 写一个最小可用的抓包与特征提取脚本下面这段代码是整套系统的数据入口持续抓包按五元组源 IP、目的 IP、源端口、目的端口、协议聚合并统计单位时间内的包数和字节数。这两个统计量是后面检测 SYN Flood、端口扫描的基础。from scapy.all import sniff, IP, TCP, UDP from collections import defaultdict import time # 五元组 - [包数, 字节数, 首次时间, 末次时间] flow_stats defaultdict(lambda: [0, 0, 0.0, 0.0]) WINDOW 10 # 统计窗口单位秒 def extract(pkt): if IP not in pkt: return proto TCP if TCP in pkt else (UDP if UDP in pkt else OTHER) sport pkt[TCP].sport if TCP in pkt else (pkt[UDP].sport if UDP in pkt else 0) dport pkt[TCP].dport if TCP in pkt else (pkt[UDP].dport if UDP in pkt else 0) key (pkt[IP].src, pkt[IP].dst, sport, dport, proto) now time.time() rec flow_stats[key] if rec[0] 0: rec[2] now rec[0] 1 rec[1] len(pkt) rec[3] now def report(): now time.time() for key, rec in list(flow_stats.items()): if now - rec[3] WINDOW: # 窗口内无新包输出并清理 pps rec[0] / max(rec[3] - rec[2], 1e-6) print(f{key} 包数{rec[0]} 字节{rec[1]} pps{pps:.1f}) del flow_stats[key] if __name__ __main__: # store0 表示不把包留在内存避免长时间运行 OOM sniff(prnextract, store0, filterip)逻辑说明sniff的prn回调每来一个包就执行一次extract把包归入对应五元组。store0是关键参数默认 Scapy 会把所有包存进内存列表跑几分钟就吃满内存这是新手最常见的翻车点。filterip用 BPF 语法在底层过滤只把 IP 包交给 Python能显著降低 CPU 占用。参数说明WINDOW决定统计粒度设太小会频繁输出、噪声大设太大则攻击响应迟钝一般 510 秒。pps每秒包数是后续阈值判定的核心指标正常业务流量通常个位数到几百SYN Flood 能轻松上千。2.3 把特征落到可查询的结构里抓到的数据要能被检测模块和前端同时消费常见做法是写进一个带时间戳的队列或轻量数据库。毕设里我一般用queue.Queue做进程内缓冲再用 SQLite 落盘历史记录前端轮询最近 N 条。这样检测线程和 Web 线程解耦不会因为前端卡顿拖慢抓包。import queue, sqlite3, threading pkt_queue queue.Queue(maxsize10000) # 有界队列防止内存无限增长 def init_db(pathids.db): conn sqlite3.connect(path) conn.execute(CREATE TABLE IF NOT EXISTS flows( ts REAL, src TEXT, dst TEXT, sport INT, dport INT, proto TEXT, pkts INT, bytes INT)) conn.commit() return connmaxsize必须设否则生产快于消费时队列会撑爆内存。消费端用get(timeout1)配合queue.Empty异常避免线程永久阻塞。SQLite 写入建议批量提交每条都 commit 会让磁盘 IO 成为瓶颈。3. 攻击检测规则引擎与阈值判定怎么落地有了流量特征下一步是判定「这是不是攻击」。这一章讲三类最常见攻击的检测逻辑SYN Flood、端口扫描、ICMP 洪水并给出可调参数和误报控制思路。3.1 三类攻击的特征与判定条件不同攻击在流量特征上留下的痕迹不一样硬套一个阈值必然误报。下面这张表是我实际调参后总结的判定条件可以直接作为规则引擎的初始配置。攻击类型核心特征初始阈值误报风险点SYN Flood大量 SYN 包、无对应 ACK半连接堆积单源 pps 500 且 SYN 占比 80%压测、秒杀场景会误判端口扫描同一源 IP 短时间内访问大量不同目的端口10 秒内不同 dport 50爬虫、健康检查会触发ICMP 洪水ICMP 包速率异常高单源 pps 200内网 ping 巡检会误判判定条件要组合多个维度单看 pps 太粗糙。比如 SYN Flood 必须同时满足「速率高」和「SYN 占比高」否则正常的高并发业务也会被拦。端口扫描则要看「目的端口去重数量」而不是总包数。3.2 规则引擎的代码实现下面是一个可扩展的规则判定函数输入是聚合后的流特征输出是命中的攻击类型。用字典注册规则新增攻击类型只需加一个函数不用改主流程。from collections import defaultdict # 记录每个源 IP 在窗口内访问过的目的端口集合 port_scan_tracker defaultdict(set) def detect_syn_flood(flow, syn_ratio, pps): # 速率高 SYN 占比高两个条件同时满足才判定 return pps 500 and syn_ratio 0.8 def detect_port_scan(src_ip, dport, window10): port_scan_tracker[src_ip].add(dport) # 简化版超过阈值即判定实际应带时间窗口清理 return len(port_scan_tracker[src_ip]) 50 def detect_icmp_flood(flow, pps): return flow[4] OTHER and pps 200 RULES { SYN_FLOOD: detect_syn_flood, PORT_SCAN: detect_port_scan, ICMP_FLOOD: detect_icmp_flood, }逻辑说明detect_syn_flood接收速率和 SYN 占比两个参数避免单一指标误判。port_scan_tracker用字典记录每个源 IP 访问过的端口实际生产里要加时间窗口和定期清理否则字典会无限增长——这是内存泄漏的经典来源。RULES字典让规则可插拔检测主循环只需遍历它。参数说明syn_ratio需要在抓包层额外统计 SYN 标志位数量不能只靠五元组。window参数控制端口扫描的观察周期设太短会漏判慢速扫描设太长会误伤正常的多端口访问。3.3 误报控制白名单与阈值自适应规则引擎上线后最大的敌人是误报。血泪经验是一定要有白名单机制把网关、监控服务器、已知压测机的 IP 排除。其次阈值不要写死可以按基线动态调整——统计过去一小时的正常 pps 均值超过均值 5 倍才告警。WHITELIST {192.168.1.1, 192.168.1.100} def is_whitelisted(src_ip): return src_ip in WHITELIST # 基线自适应维护滑动均值 from collections import deque baseline deque(maxlen360) # 一小时每 10 秒一个点 def adaptive_threshold(current_pps): baseline.append(current_pps) avg sum(baseline) / len(baseline) return max(avg * 5, 100) # 至少 100避免基线过低导致误报maxlen让 deque 自动淘汰旧数据不用手动清理。max(avg*5, 100)里的下限很重要凌晨流量接近零时avg*5可能只有个位数任何正常访问都会触发告警。这个下限值要根据你的网络规模调内网一般 100 起步。4. 自动防御从检测到封禁的联动链路检测出攻击只是第一步自动防御才是「防御系统」区别于「检测系统」的地方。这一章讲怎么把检测结果变成实际的封禁动作以及怎么保证封禁本身不出事。4.1 防御动作的三种实现方式自动防御常见三种手段调用系统防火墙iptables/firewalld、修改应用层黑名单、发送告警通知。毕设里最直观的是调 iptables 直接封 IP效果立竿见影但风险也最大——封错了可能把自己关在门外。# 封禁一个源 IP限制其新建连接 sudo iptables -I INPUT -s 192.168.1.200 -j DROP # 查看当前封禁列表 sudo iptables -L INPUT -n --line-numbers # 解封按行号删除 sudo iptables -D INPUT 1-I是插入到链首优先级最高-D按规则或行号删除。注意iptables 规则重启后会丢失生产环境要配合iptables-persistent或自己写恢复脚本。毕设演示时建议加一个「封禁时长」机制到期自动解封避免演示完忘了清理。4.2 用 Python 封装防御动作并加安全阀直接让检测线程调subprocess执行 iptables 很危险一旦规则写错或 IP 解析出错可能封掉整个网段。我一般会封装一层加三个安全阀白名单校验、封禁频率限制、自动解封定时器。import subprocess, time, threading WHITELIST {192.168.1.1, 192.168.1.100} BAN_DURATION 300 # 封禁 5 分钟后自动解封 ban_history {} # ip - 解封时间戳 def ban_ip(ip): if ip in WHITELIST: print(f[跳过] {ip} 在白名单中) return False if ip in ban_history and ban_history[ip] time.time(): return False # 已在封禁中避免重复加规则 # 校验 IP 格式防止命令注入 if not all(p.isdigit() and 0 int(p) 255 for p in ip.split(.)): return False subprocess.run([iptables, -I, INPUT, -s, ip, -j, DROP], checkTrue) ban_history[ip] time.time() BAN_DURATION threading.Timer(BAN_DURATION, unban_ip, args[ip]).start() return True def unban_ip(ip): subprocess.run([iptables, -D, INPUT, -s, ip, -j, DROP], checkFalse) ban_history.pop(ip, None)逻辑说明ban_ip先过白名单再查是否已在封禁期然后做 IP 格式校验——这一步是防命令注入的关键绝不能把未校验的字符串拼进 shell 命令。subprocess.run用列表传参而不是shellTrue同样是为了安全。threading.Timer实现自动解封ban_history防止同一 IP 被重复加规则导致 iptables 链膨胀。参数说明BAN_DURATION是封禁时长演示场景 300 秒够用生产环境可以按攻击严重程度分级比如扫描封 10 分钟、洪水封 1 小时。checkTrue让 iptables 执行失败时抛异常便于日志记录解封时用checkFalse因为规则可能已被手动删除。4.3 防御动作的可观测性封禁必须留痕否则出了问题无法回溯。每次封禁/解封都要写日志记录时间、IP、触发规则、操作结果。前端监控面板上要能看到「当前封禁列表」和「历史封禁记录」这是毕设答辩时的加分项也是实际运维的刚需。import logging logging.basicConfig(filenamedefense.log, levellogging.INFO, format%(asctime)s %(levelname)s %(message)s) def log_ban(ip, rule, success): logging.info(fban ip{ip} rule{rule} success{success})日志格式里带上规则名方便统计哪类攻击最频繁。日志文件要配轮转否则长期运行会撑满磁盘可以用logging.handlers.RotatingFileHandler。5. 可视化监控用 Flask ECharts 做实时面板可视化是毕设的门面也是把前面所有模块串起来的窗口。这一章讲怎么用 Flask 提供数据接口用 ECharts 做实时刷新以及怎么避免前端拖垮后端。5.1 后端接口设计轮询还是推送实时监控有两种数据通道前端定时轮询 REST 接口或者用 WebSocket 推送。毕设里轮询实现简单、调试方便推荐先用轮询刷新间隔 23 秒。WebSocket 虽然实时性更好但 Flask 原生不支持要引入 Flask-SocketIO复杂度上升除非答辩明确要求否则没必要。from flask import Flask, jsonify import sqlite3 app Flask(__name__) app.route(/api/recent_flows) def recent_flows(): conn sqlite3.connect(ids.db) rows conn.execute( SELECT ts, src, dst, proto, pkts FROM flows ORDER BY ts DESC LIMIT 50 ).fetchall() conn.close() return jsonify([{ts: r[0], src: r[1], dst: r[2], proto: r[3], pkts: r[4]} for r in rows]) app.route(/api/ban_list) def ban_list(): return jsonify(list(ban_history.keys()))逻辑说明接口只返回最近 50 条避免一次传太多数据拖慢前端。jsonify自动处理中文和序列化。数据库连接每次请求新建、用完关闭简单可靠高并发场景才需要连接池毕设规模用不上。参数说明LIMIT 50控制返回量前端表格一屏能显示的行数有限传太多没意义。刷新间隔由前端setInterval控制建议 20003000 毫秒太快会给后端和数据库压力。5.2 前端实时图表与表格前端用 ECharts 画流量趋势折线图用表格展示最近流量和封禁列表。关键是每次刷新只更新数据不重建图表实例否则会有内存泄漏和闪烁。// 初始化一次图表实例 const chart echarts.init(document.getElementById(traffic)); const option { xAxis: { type: category, data: [] }, yAxis: { type: value }, series: [{ type: line, data: [], smooth: true }] }; chart.setOption(option); // 定时拉取数据并更新不重建实例 setInterval(async () { const res await fetch(/api/recent_flows); const data await res.json(); const times data.map(d new Date(d.ts * 1000).toLocaleTimeString()); const pkts data.map(d d.pkts); chart.setOption({ xAxis: { data: times }, series: [{ data: pkts }] }); }, 3000);逻辑说明echarts.init只调一次后续用setOption增量更新这是 ECharts 的正确用法。fetch拉数据后直接映射成坐标轴和序列数据。时间戳乘 1000 是因为 ECharts 和 JS 用毫秒而 Python 的time.time()是秒。参数说明smooth: true让折线平滑视觉更好但会轻微失真追求精确可以关掉。刷新间隔 3000 毫秒和前端轮询保持一致避免请求堆积。5.3 面板要展示哪些指标一个能打的监控面板至少要有四块实时流量趋势、攻击告警列表、当前封禁 IP、系统运行状态抓包速率、队列长度。队列长度尤其重要如果pkt_queue持续接近maxsize说明消费跟不上生产要么加消费线程要么降低抓包过滤粒度。面板模块数据来源刷新频率异常判断流量趋势flows 表聚合3 秒pps 突增 5 倍告警列表检测模块日志3 秒出现新攻击类型封禁列表ban_history5 秒封禁数持续增长运行状态队列长度、线程状态5 秒队列 80% 容量6. 避坑与排查这套系统最容易翻车的五个地方前面讲的都是「应该怎么做」这一章专门讲「实际会怎么坏」。以下五条都是我在复现这类系统时真实踩过的坑按「现象 → 原因 → 解决」写照着排查能省大量时间。坑一抓包脚本跑几分钟就内存爆满。现象是进程 RSS 持续上涨直到被 OOM Killer 杀掉。原因是sniff默认store1把所有包存在内存里。解决是显式传store0并且聚合用的字典要加时间窗口清理不能只增不减。坑二iptables 封禁后自己 SSH 断连。现象是执行封禁脚本后终端卡死再也连不上。原因是白名单没配或者封禁规则误伤了管理 IP。解决是封禁前强制校验白名单并且永远保留一条允许管理网段的规则在最前面。演示环境建议先在虚拟机里试。坑三Flask 面板打开一片空白控制台报跨域。现象是接口单独访问正常页面里请求失败。原因是前端页面和接口不同源浏览器拦截。解决是 Flask 加flask-cors扩展或者把前端页面直接由 Flask 托管同源就没问题。坑四检测规则频繁误报正常业务被当成攻击。现象是告警列表刷屏封禁列表里全是正常 IP。原因是阈值写死且偏低没有基线自适应。解决是引入滑动均值做动态阈值并给阈值设下限同时把已知的正常高频 IP 加白名单。坑五SQLite 写入报database is locked。现象是抓包线程和 Web 线程同时写库时随机报错。原因是 SQLite 默认单写锁多线程并发写会冲突。解决是写操作集中到一个线程或者用WAL模式提升并发最简单的办法是给写操作加线程锁。提示以上五个坑里内存和权限问题占了大半。上线前先在低流量环境跑够 30 分钟观察内存曲线和日志比直接上生产靠谱得多。7. 进阶技巧用基线对比验证检测效果系统跑起来只是及格能证明它「真的检测到了」才算过关。这一章给一个可复现的验证方法用基线流量对比攻击流量量化检测的准确率和响应时间。这也是答辩时最有说服力的一页。7.1 构造可控的攻击流量做验证不要用真实攻击工具去打别人的网络在自己搭的隔离环境里用 Python 构造流量即可。下面这段代码模拟一次 SYN 扫描用来验证端口扫描规则是否触发。from scapy.all import IP, TCP, send import random target 192.168.1.50 # 向目标的不同端口发送 SYN模拟端口扫描 for port in random.sample(range(1, 1000), 80): pkt IP(dsttarget) / TCP(dportport, flagsS) send(pkt, verbose0) print(扫描流量发送完成)逻辑说明random.sample从 11000 里取 80 个不重复端口超过规则里 50 的阈值应该触发PORT_SCAN。flagsS只发 SYN不完成握手符合扫描特征。verbose0关掉 Scapy 的发送日志输出干净。参数说明端口数量要略高于规则阈值才能验证「刚好触发」的边界。发送速率用send的inter参数控制默认不限速验证慢速扫描时可以设inter0.1。7.2 用基线对比量化误报和漏报验证要成对做先跑一段纯正常流量记录基线再叠加攻击流量看系统是否只在攻击时段告警。下面这张表是我验证时用的记录模板把每次测试的结果填进去准确率和响应时间一目了然。测试场景基线 pps攻击 pps是否告警响应延迟误报数纯正常流量3030否-0SYN Flood30800是2.1s0端口扫描30120是1.8s0正常压测30600否白名单-0响应延迟从攻击开始到封禁生效用日志时间戳相减即可。误报数统计非攻击时段触发的告警。如果「正常压测」这一行出现告警说明白名单或阈值需要调整。7.3 我自己的习惯我做完这类系统一定会做一件事把检测规则和阈值单独抽成一个配置文件不写死在代码里。因为调参是反复的过程每次改阈值都重新改代码、重启服务效率太低。配置文件用 YAML 或 JSON启动时加载改完重启即可生效。另一个习惯是给每个封禁动作加一个「干跑模式」只记日志不真正执行 iptables先在干跑模式下观察一天确认没有误封再开真封禁。这个后悔药机制救过我很多次希望你也能用上。希望帮到你。本文还有配套的精品资源点击获取
