简介一款面向毕业设计与课程设计的基于Python的网络入侵检测与防御系统源码包针对实时流量分析、攻击检测、自动防御与可视化监控等需求提供可直接运行的完整项目实现适合网络安全方向学生二次开发与功能演示。包体共38个文件包含14个Python源码文件、9个编译后的pyc文件以及HTML/CSS/JS前端页面、JSON配置、dockerfile与docker-compose.yml容器编排、shell安装脚本和说明文档等整体仅92KB结构紧凑。项目后端采用Flask、Flask-SocketIO与Scapy前端使用Bootstrap和Chart.js结合MongoDB存储日志可实现流量实时捕获、异常识别、恶意流量拦截、自动告警响应及Web端攻击统计与系统配置。附带的详细运行指南zip内包含操作说明能帮助理解部署与使用流程。已有172人学习下载适合作为毕业设计选题参考或网络安全实践入门项目。1. 基于Python的网络入侵检测与防御系统毕设开题前先看清楚这三件事拿到“基于Python的网络入侵检测与防御系统”这个题目很多人的第一反应是“用Scapy抓包画几条曲线”结果开题答辩就被问住你的“攻击检测”到底检测了什么你的“自动防御”是防御了个寂寞这套系统的核心不是抓包而是把实时流量分析、攻击检测、自动防御、可视化监控串成一个闭环——抓到的包要能判断出“谁是攻击者”判断出来之后要能“立刻阻断”同时还要在网页上让评委看到整个过程。本文按一个可复现的毕业设计方案来拆解技术栈是Python 3.8 Scapy Flask iptables只要你有一台Linux机器按步骤就能把最小可用系统跑起来答辩论点也足够扎实。2. 系统架构与实时流量分析先把抓包和特征提取跑通2.1 为什么选Scapy而不是纯socket解析半天跑通和一周跑通的区别做实时流量分析第一个选择是抓包库。如果你从零开始用socket读原始报文然后自己解析以太网头、IP头、TCP头光处理IPv4选项、TCP flag、分片重组就够写几天而且这还是在整个毕设的“前菜”阶段。常见做法是直接用Scapy它把IP、TCP、UDP、ICMP的解析封装好了提供sniff接口过滤语法和tcpdump的BPF一致开发效率高很多。Scapy是纯Python实现解析性能不算强但毕设实验环境流量小完全扛得住。真正并发几十万PPS的生产环境也不会用Python那是DPDK和C的天下这个区别在答辩时反而是加分项。系统的整体结构是一个生产者-消费者模型抓包线程用sniff从网卡收包把包放进有界队列一个或多个检测线程从队列取包提取特征后跑攻击检测规则检测到攻击就把事件抛给防御线程由防御线程执行iptables封禁同时Flask后端从共享指标里读数据供前端ECharts轮询展示。这个分层的好处是每一块都能单独测试后面替换检测算法、换掉可视化框架都不会动到其他模块。2.2 实时抓包的最小实现sniff加队列别在主线程里干重活先把抓包这层跑通。以下是一个最小可运行的抓包线程注意不要把sniff放在主线程里否则后面加载检测规则或者页面卡顿会直接扔包。import threading import queue from scapy.all import sniff, IP, TCP, UDP # 有界队列避免抓包速度大于处理速度时内存无限上涨 packet_queue queue.Queue(maxsize1000) def packet_callback(pkt): 收到一个包后的回调只关心IP/TCP/UDP过滤ARP等链路层噪声 if IP in pkt and (TCP in pkt or UDP in pkt): packet_queue.put(pkt) def start_capture(interface, bpf_filtertcp or udp): 启动抓包线程sniff会阻塞当前线程 def _run(): sniff(ifaceinterface, filterbpf_filter, prnpacket_callback, storeFalse) t threading.Thread(target_run, daemonTrue) t.start() return t if __name__ __main__: # 网卡名不要猜先执行 python -c from scapy.all import get_if_list; print(get_if_list()) start_capture(eth0, tcp or udp) print(抓包线程已启动当前队列积压, packet_queue.qsize())这段代码里sniff的filter参数用BPF语法只放行TCP和UDP过滤掉ARP、DHCP等与攻击检测关系不大的报文能明显降低后续处理压力。storeFalse表示抓到的包不做内存缓存否则抓一小时就把内存占爆。队列的maxsize1000是背压阀值如果检测线程处理不过来队列满后抓包线程会暂时阻塞这是保护机制不是bug。用daemonTrue启动线程主程序CtrlC退出时不会留下僵尸线程。实际使用时把eth0替换成你机器的网卡名。在虚拟机里往往是ens33、ens192用前面注释里的命令列出所有网卡再选有IP的那个。如果服务器只有一块网卡抓包网卡和被检测流量网卡是同一块那么本机对外发起的正常SSH流量也会被抓到设计检测规则时要注意这一点。2.3 特征提取把报文转成检测模块认识的统一字典抓包只是搬砖真正的分析在后续。为了不让每个检测函数都去重复解析Scapy包我在包一进队就把它转成普通dict这一步叫特征提取。字段统一后检测模块和抓包库完全解耦以后换dpkt、换离线pcap解析都只需要改这一个函数。from scapy.all import IP, TCP, UDP def extract_features(pkt): 从Scapy报文提取关键字段返回dict。 没有的字段置None统一格式方便检测规则直接取值。 feat { src: None, dst: None, sport: None, dport: None, proto: None, flags: None, len: len(pkt) } if IP in pkt: feat[src] pkt[IP].src feat[dst] pkt[IP].dst feat[proto] pkt[IP].proto # 6TCP, 17UDP if TCP in pkt: feat[sport] pkt[TCP].sport feat[dport] pkt[TCP].dport feat[flags] int(pkt[TCP].flags) # SYN0x02, ACK0x10 if UDP in pkt: feat[sport] pkt[UDP].sport feat[dport] pkt[UDP].dport return feat这里把TCP flags转成整型是个关键细节。Scapy的flags对象拿到的是FlagValue不能直接用等号比较。转成int之后判断SYN包就是flags 0x02 ! 0判断ACK包就是flags 0x10 ! 0后续写规则会非常顺手。len字段记录包总长度DDoS检测里能用来算字节速率虽然本文没用到先留着不亏。2.4 队列消费线程单线程检测还是批量处理有了特征提取函数消费队列的检测线程可以这样写while True: pkt packet_queue.get(timeout1) feat extract_features(pkt) # 下一章的攻击检测入口 integrate_detection(feat)最常见的翻车是把检测函数直接写在回调里sniff回调本身在Scapy的接收线程里跑如果检测逻辑里有sleep或者复杂计算Sniff线程被卡住内核缓冲区会先满然后开始丢包表现是“流量一大就漏报”。用队列解耦之后即使检测线程偶尔卡顿sniff线程也只是往队列里放包队列满了才阻塞。我习惯消费者线程同时跑三四个每个线程消费同一个队列检测性能不够时这是最便宜的扩展方式。3. 攻击检测模块用阈值规则识别端口扫描、暴力破解和DDoS3.1 为什么先做规则检测而不是机器学习毕设答辩最怕被问“你的检测模型为什么这么选”。机器学习要做的话你得自己标注数据集或者用现成的NSL-KDD但那和实时流量分析的结合需要额外写特征工程和模型部署管道工作量直接翻倍。我建议第一版用阈值规则引擎它有三点优势第一可解释性强评委问“为什么判断这是扫描”你可以直接回答“滑动窗口内不同目标端口数超过50”第二规则参数可以现场调测试的时候改一个数字就能看到效果变化演示效果非常直观第三模型升级路径清晰规则产生的特征可以作为机器学习的输入还能用规则结果做伪标签。标题里没提机器学习那规则检测就是最贴合题目的方案。检测目标选“端口扫描、暴力破解、DDoS”三类是因为它们特征差异大、覆盖了从主机行为到网络流量的不同层次而且每一类都能用简单统计量表达。端口扫描看的是“同源IP请求的端口多样性”暴力破解看的是“同源IP到同目标端口的频率”DDoS看的是“目标IP收到的聚合速率”。三个维度不重叠展示时不至于糊成一团。3.2 端口扫描检测滑动窗口内不同端口数超过阈值就告警端口扫描的典型行为是一个源IP在短时间内探测大量不同端口意图发现开放服务。下面用滑动窗口做统计只关注TCP SYN包。import time from collections import defaultdict SCAN_WINDOW 5 # 统计窗口单位秒 SCAN_PORT_THRESHOLD 50 # 窗口内不同目标端口数 scan_records defaultdict(list) def is_port_scan(feat): 判断当前包是否属于一次端口扫描。 只统计src-dport的SYN包窗口内去重端口数达到阈值即告警。 # 非TCP或不是纯SYN直接跳过 if feat[proto] ! 6 or feat[flags] ! 0x02: return False now time.time() src feat[src] scan_records[src].append((now, feat[dport])) # 清理窗口外记录避免列表无限膨胀 scan_records[src] [(t, p) for t, p in scan_records[src] if now - t SCAN_WINDOW] if len(scan_records[src]) SCAN_PORT_THRESHOLD: distinct_ports {p for _, p in scan_records[src]} if len(distinct_ports) SCAN_PORT_THRESHOLD: # 告警后清空防止同一IP连续触发告警风暴 scan_records[src].clear() return True return False这里有几个参数值得细化。flags ! 0x02是要求“纯SYN包”因为正常TCP连接也是从SYN开始的但正常主机在几秒内不会对50个不同端口发起纯SYN连接如果只做三次握手会有大量ACK回来把阈值逻辑搞混。distinct_ports去重是必要的因为扫描器可能对同一个端口发多个重传SYN不去重会把重传也算成端口数误报率会高。告警后立刻clear()这个源IP的记录是为了防止同一个慢速扫描每隔几秒就触发一次告警刷屏到无法看日志。参数怎么调演示环境建议把SCAN_PORT_THRESHOLD调到20否则在校园网环境里别人一次正常爬虫抓取也可能触发50个端口。生产环境至少80起步。SCAN_WINDOW调大能发现慢速扫描但记录积压和内存占用也会增长一般5到10秒就够。3.3 暴力破解检测同源IP对同目标端口的高频连接暴力破解的特征更直接一个源IP不停尝试连同一个目标端口往往是SSH的22、MySQL的3306、Web后台的8080。这类检测如果只按源IP统计会把正常的多端口访问也算进去所以要按“源IP-目标IP-目标端口”三元组来统计。from collections import defaultdict import time BRUTE_WINDOW 10 # 统计窗口单位秒 BRUTE_ATTEMPTS 30 # 窗口内最大连接尝试次数 brute_records defaultdict(list) def is_brute_force(feat): 检测同源IP对同目标端口的暴力破解行为。 只统计TCP包因为破解必然要建立连接。 if feat[proto] ! 6: return False now time.time() key (feat[src], feat[dst], feat[dport]) brute_records[key].append(now) brute_records[key] [t for t in brute_records[key] if now - t BRUTE_WINDOW] if len(brute_records[key]) BRUTE_ATTEMPTS: brute_records[key].clear() return True return False这里没有区分SYN还是其他flag因为暴力破解无论成功失败都会产生大量TCP包。时间戳列表的清理方式和端口扫描一样超过窗口的记录直接过滤掉。调参时注意BRUTE_ATTEMPTS30在10秒窗口内对SSH破解来说已经算很高了因为自动化工具每秒可以发几十次尝试。如果你的实验环境里有一台内网机器每秒钟掉线重连几十次也会误报这个时候不是单纯调高阈值而是要加白名单后面避坑章节会细说。3.4 DDoS检测从全局包速率和单目标包速率两个角度判断DDoS比前两种更难用单一阈值定死因为正常业务在高峰期也可能有明显流量抬升。常见做法是同时维护两个速率全局PPS和单目标PPS。当单目标速率超过阈值且明显高于全局基线时说明这个目标正被集中攻击。import time from collections import defaultdict DDOS_WINDOW 5 GLOBAL_PPS_THRESHOLD 1000 # 全局每秒包数 TARGET_PPS_THRESHOLD 200 # 单个目标每秒包数 packet_times [] target_records defaultdict(list) def check_ddos(feat): 判断是否DDoS全局速率过高或单一目标IP速率过高。 返回True表示当前包参与了一次DDoS事件。 now time.time() # 全局PPS packet_times.append(now) packet_times[:] [t for t in packet_times if now - t DDOS_WINDOW] global_pps len(packet_times) / DDOS_WINDOW # 单目标PPS dst feat[dst] target_records[dst].append(now) target_records[dst] [t for t in target_records[dst] if now - t DDOS_WINDOW] target_pps len(target_records[dst]) / DDOS_WINDOW if target_pps TARGET_PPS_THRESHOLD and global_pps GLOBAL_PPS_THRESHOLD: target_records[dst].clear() return True return Falsepacket_times[:] ...这种写法是原地修改列表而不是重新赋值。因为检测线程可能同时被多个消费者调用重新赋值会让另一个线程读到旧引用。全局阈值和单目标阈值同时满足时才告警能避免某台机器被爬虫集中访问导致的单目标速率波动。实际部署时这两个阈值要根据日常流量基线来设。我通常先让系统跑一天统计平均PPS和P95 PPS再把阈值设为P95的2到3倍。没有基线数据就盲目设数字不是“防DDoS”是“随机报警器”。3.5 检测模块的统一入口和告警事件三个检测函数写好后需要一个入口把它们组合起来并且输出结构化的告警事件def integrate_detection(feat): 统一分析入口按顺序跑规则命中则产生告警事件 if is_port_scan(feat): publish_event(port_scan, feat) if is_brute_force(feat): publish_event(brute_force, feat) if check_ddos(feat): publish_event(ddos, feat) def publish_event(event_type, feat): 把事件交给防御模块的队列同时打印日志 print(f[告警] {event_type} | {feat[src]} - {feat[dst]}:{feat[dport]}) defense_queue.put((event_type, feat[src]))注意这里每个if都是独立判断不用elif。因为一个扫描行为可能同时伴随暴力破解比如同一个IP在探测完端口后紧接着对某端口做密码尝试。用elif会让后续规则永远没机会执行。告警事件放入defense_queue之后检测线程的工作就结束了封禁动作由专门的防御线程去做避免检测线程被iptables子进程阻塞。4. 自动防御联动iptables做短时封禁别直接拉黑4.1 自动防御的闭环检测、封禁、解封、记录自动防御是“基于Python的网络入侵检测与防御系统”里最容易做过头的一环。我见过有同学检测到扫描就直接把IP加进iptables并且不设置解封时间第二天实验室里有一半机器被他封了实验都做不下去。真实运维里“封禁”永远是临时的因为源IP可能被伪造也可能是一个NAT网关里的合法用户。正确的闭环是检测到攻击后封禁一段时间时间到自动解封同时在内存里维护一份封禁记录避免重复封禁。还有一条原则是先查白名单。内网的网关、DNS服务器、你自己的远程管理IP永远不应该被自动防御封住。否则一次误报可能直接切断所有机器和外网的连接。白名单不是一个“可选优化”而是自动防御能安全演示的前提。4.2 用iptables封禁IP的Python封装自动防御动作在Linux上最常见的做法是调用iptables命令向INPUT链插入一条DROP规则。下面封装了封禁、解封、判重复三个逻辑。import subprocess import time # 白名单至少包含本机回环和管理IP建议从配置文件读取 WHITE_LIST {127.0.0.1, 192.168.1.1} # 记录每个IP的解封时间 ban_expire {} def block_ip(ip, duration300): 封禁指定IPduration秒后自动解封。 返回True表示这次调用真的执行了封禁。 if ip in WHITE_LIST: print(f[防御] 跳过白名单IP{ip}) return False now time.time() if ip in ban_expire and ban_expire[ip] now: return False # 已封禁且未到期 ret subprocess.run( [iptables, -A, INPUT, -s, ip, -j, DROP], capture_outputTrue, textTrue ) if ret.returncode 0: ban_expire[ip] now duration print(f[防御] 已封禁 {ip}{duration} 秒后解封) return True else: print(f[防御] 封禁失败{ret.stderr.strip()}) return False def unban_expired(): 遍历封禁记录到期则删除iptables规则。 这个函数需要定时调用而不是只调一次。 now time.time() expired_ips [ip for ip, expire in ban_expire.items() if expire now] for ip in expired_ips: subprocess.run( [iptables, -D, INPUT, -s, ip, -j, DROP], capture_outputTrue, textTrue ) del ban_expire[ip] print(f[防御] 解封 {ip})block_ip里先判断白名单再判断是否已经在封禁期内避免同一个IP被多个告警事件重复封禁。subprocess.run每次执行iptables都会fork一个子进程性能开销不小但防御动作本来就不是高频操作每秒最多处理几十个IP完全够用。capture_outputTrue是为了把stderr拿回来打印否则权限不足时你连报错都看不到只能干瞪眼。解封函数单独拆出来是因为封禁和解封要放在同一个循环里调度。expired_ips先收集再删除不能在遍历列表的同时修改列表这是Python里常见的易错点。duration默认300秒演示时可以改成60秒方便在同一个IP上反复测试。4.3 防御线程和检测线程的联动队列解耦避免互相阻塞检测线程发现攻击后如果直接调用subprocess的iptables那么一次封禁操作耗时可能几十毫秒这期间检测线程无法处理后续报文在高流量下就是灾难。所以防御动作要放到独立线程里用队列把两个线程解耦。前面的defense_queue在这里消费import queue import time # 全局告警队列检测模块往这里写 defense_queue queue.Queue() def defense_loop(): 防御工作循环从队列取告警执行封禁每秒钟顺带解封到期IP。 在python主线程里单独启动。 while True: try: event_type, ip defense_queue.get(timeout1) block_ip(ip, duration60) except queue.Empty: pass # 每秒检查一次过期封禁 unban_expired() time.sleep(1)defense_queue.get(timeout1)让循环在队列空时不会死等能每隔一秒走到unban_expired。如果把get设为无超时阻塞那么解封逻辑只有在收到下一个攻击事件时才会执行到期IP会被一直封着。这个线程启动后检测模块只需往队列里丢事件整个系统才是“实时检测-自动防御-自动解封”的真正闭环。4.4 演示逃不过的三件事root权限、规则顺序、规则持久化第一iptables需要root权限Python脚本在普通用户下运行会得到Permission denied。演示时直接sudo python main.py不要用setcap那种花活容易把规则搞乱。第二-A INPUT是把新规则追加到输入链末尾如果链里已有更早的DROP规则拦住了同源IP新规则不会立刻生效。需要在封禁前执行iptables -F INPUT清空旧规则或者用-I INPUT把新规则插到最前面。我写的时候用-A是考虑到要维护多条封禁规则互不干扰但演示前要先确认链是干净的。第三iptables规则重启后消失答辩演示如果中途重启了机器规则全没了系统看起来像是“没有防御”。如果需要持久化用iptables-save导出到文件但这部分不是必做项。5. 可视化监控用Flask和ECharts把流量与告警画出来5.1 为什么可视化监控不能省标题里的“可视化监控”是硬性要求也是答辩时最直观的展示点。评委走进来看不到实时变化的界面光听你讲算法很难建立信任。常见做法是Flask后端提供一个JSON接口前端ECharts轮询接口把包速率、告警次数、封禁IP数画成实时刷新的折线图和列表。不选Grafana是因为它需要额外安装服务而且Grafana的仪表盘一看就是现成模板显得你在做集成而不是做开发。自己用Flask写还能顺手演示一下后端API设计能力。5.2 后端API设计从检测模块取实时指标先写一个最小可运行的Flask服务接口返回当前指标。from flask import Flask, jsonify, render_template import time app Flask(__name__) # 实际项目中这些数据从检测模块的共享内存、Redis或sqlite读取 # 这里用函数包一层方便后续替换 def get_metrics(): return { timestamp: time.time(), packet_rate: 123.4, # 每秒包数由抓包模块统计 alert_count: 7, # 累计告警数 banned_ips: 3 # 当前封禁的IP数量 } app.route(/api/metrics) def metrics(): resp jsonify(get_metrics()) # 禁止浏览器缓存否则轮询接口可能拿到旧数据 resp.headers[Cache-Control] no-store return resp app.route(/) def index(): return render_template(index.html) if __name__ __main__: app.run(host0.0.0.0, port5000, debugFalse)get_metrics里现在是写死的数字你接检测模块时需要维护一个全局指标对象抓包线程每收一个包就计数器加一检测线程每次发布告警就告警计数加一然后在这里读出来。Cache-Control: no-store这个头很重要浏览器默认会缓存GET请求尤其是Vercel或Nginx反代环境不加这个头你会发现曲线隔十几秒才跳一下以为是程序卡了其实是缓存。debugFalse必须设死Flask的debug模式会启动重载器从而把抓包线程重复拉起产生“拦截两次”的假象。5.3 前端仪表盘ECharts折线图轮询后端接口前端页面放在templates/index.html下用ECharts画一条滚动流量曲线。以下是可以直接复用的模板!DOCTYPE html html head meta charsetutf-8 title入侵检测与防御监控/title script srchttps://cdn.jsdelivr.net/npm/echarts5/dist/echarts.min.js/script /head body div idchart stylewidth:100%;height:400px;/div script var chart echarts.init(document.getElementById(chart)); var option { title: { text: 实时流量包速率 }, xAxis: { type: category }, yAxis: { type: value }, series: [{ type: line, data: [], showSymbol: false }] }; chart.setOption(option); function fetchData() { fetch(/api/metrics) .then(r r.json()) .then(data { var now new Date(data.timestamp * 1000).toLocaleTimeString(); option.xAxis.data.push(now); option.series[0].data.push(data.packet_rate.toFixed(1)); // 只保留最近30个点避免曲线越来越挤 if (option.xAxis.data.length 30) { option.xAxis.data.shift(); option.series[0].data.shift(); } chart.setOption(option); }) .catch(err console.error(err)); } setInterval(fetchData, 1000); /script /body /htmlECharts从CDN加载如果是无网络环境演示要下载echarts.min.js到static目录并把script标签的src改成相对路径。前端只保留30个数据点形成滚动效果避免内存泄漏。轮询间隔1000毫秒如果后端处理压力大调到2000毫秒也可以。实际演示时可以先在终端看到检测日志再切到浏览器看曲线变化两边一致就会很有说服力。5.4 常见问题排查与避坑一套系统做完调试时总会踩到几个典型的坑。我把高频问题按“现象-原因-解决”整理如下。第一个抓包程序运行后完全没有输出。现象sniff回调什么都没打印队列一直为空。原因要么是非root用户没有抓包权限要么是网卡名写错了。解决先执行sudo python -c from scapy.all import sniff; print(ok)确认权限再用python -c from scapy.all import get_if_list; print(get_if_list())列出真实网卡名不要猜eth0。如果连Scapy都装不上回过去把Python安装教程里的pip和虚拟环境配置再走一遍这个基本功省不了。第二个误报率高到没法看。现象内网一台机器正常做业务被判定为端口扫描或暴力破解。原因阈值设置太低或者没有配置白名单。解决先调高阈值观察业务流量的正常基线再把内网网关、数据库服务器、办公系统的IP加进白名单。注意白名单应该覆盖检测模块和防御模块两层检测层白名单可以跳过分析防御层白名单避免误封。第三个iptables封禁命令返回权限错误。现象调用block_ip后终端打印“封禁失败”stderr是Permission denied。原因Python进程不是root或iptables链不存在。解决用sudo运行程序用iptables -L INPUT确认链存在。同时检查有没有别的安全模块拦截了iptables调用比如云服务器默认的firewalld和iptables冲突要先停掉firewalld。第四个可视化页面接口有数据但曲线不更新。现象打开页面能看到第一帧后面一直不动。原因浏览器缓存了/api/metrics的GET响应。解决在Flask响应头加Cache-Control: no-store并确认没有经过带缓存的Nginx。第五个流量稍微一大程序就卡死。现象CPU占用不高但检测吞吐骤降队列积压到满。原因检测线程在同步执行耗时操作比如subprocess调用iptables或打印大量日志。解决把防御动作丢给独立线程用队列消费检测线程只做特征计算和规则判断日志异步写或用logging的QueueHandler。如果queue满了sniff回调会阻塞在put上导致内核丢包这种情况下只能升级硬件或减少抓包范围。6. 进阶怎么验证检测效果并完成性能调优写完规则和防御链路下一步是让别人相信你的系统真的能检测到攻击。不推荐用真实攻击工具做演示原因有二一是可能触发你所在网络的管理告警二是攻击工具的沙箱环境兼容性很麻烦。我用的做法是用Scapy构造特征明显的测试包直接喂给检测函数做单元测试这样不依赖真实网卡也便于在答辩现场重复跑。def test_port_scan_detection(): # 模拟1秒内向10.0.0.1的不同端口发SYN包 for dport in list(range(20, 70)): feat { src: 10.0.0.9, dst: 10.0.0.1, sport: 1000, dport: dport, proto: 6, flags: 0x02 } is_port_scan(feat) # 50个不同端口已经超过阈值is_port_scan返回True这段测试代码把feature字典直接传给is_port_scan跑完检查返回值。同样的方法可以给暴力破解和DDoS各写一个测试函数。这样做的好处是不需要真实流量任何阶段都能跑。更进一步可以把已抓取的pcap文件重放用rdpcap读取后逐包调用extract_features和integrate_detection验证规则是不是在真实历史流量上能命中。调优方面我推荐先用cProfile找出热点函数。如果发现extract_features占用高考虑裁剪字段比如不解析UDP字段如果规则统计里的列表过滤占用高改用collections.deque代替list做滑动窗口时间复杂度和内存占用都更友好。封禁逻辑里subprocess.run是最明显的性能瓶颈可以加一个批量封禁接口把多条iptables规则合成一条链比如先创建自定义链然后统一跳转但这属于锦上添花。最后说我自己的教训。第一版自动防御我写的是永久封禁没有解封流程结果自己测试的时候把虚拟机的网关封掉了整个实验环境断网最后只能进物理终端手动清规则。后来改成短时封禁加白名单再也不敢用永久封禁做默认行为。毕设系统可以写得“狠”但一定要留后退的余地这就是防御模块里那句“duration300”的真正价值。希望这篇围绕实时流量分析、攻击检测、自动防御、可视化监控的拆解能帮你在答辩前少走这段弯路。本文还有配套的精品资源点击获取 基于Python的网络入侵检测与防御系统：Scapy+Flask实战拆解
简介一款面向毕业设计与课程设计的基于Python的网络入侵检测与防御系统源码包针对实时流量分析、攻击检测、自动防御与可视化监控等需求提供可直接运行的完整项目实现适合网络安全方向学生二次开发与功能演示。包体共38个文件包含14个Python源码文件、9个编译后的pyc文件以及HTML/CSS/JS前端页面、JSON配置、dockerfile与docker-compose.yml容器编排、shell安装脚本和说明文档等整体仅92KB结构紧凑。项目后端采用Flask、Flask-SocketIO与Scapy前端使用Bootstrap和Chart.js结合MongoDB存储日志可实现流量实时捕获、异常识别、恶意流量拦截、自动告警响应及Web端攻击统计与系统配置。附带的详细运行指南zip内包含操作说明能帮助理解部署与使用流程。已有172人学习下载适合作为毕业设计选题参考或网络安全实践入门项目。1. 基于Python的网络入侵检测与防御系统毕设开题前先看清楚这三件事拿到“基于Python的网络入侵检测与防御系统”这个题目很多人的第一反应是“用Scapy抓包画几条曲线”结果开题答辩就被问住你的“攻击检测”到底检测了什么你的“自动防御”是防御了个寂寞这套系统的核心不是抓包而是把实时流量分析、攻击检测、自动防御、可视化监控串成一个闭环——抓到的包要能判断出“谁是攻击者”判断出来之后要能“立刻阻断”同时还要在网页上让评委看到整个过程。本文按一个可复现的毕业设计方案来拆解技术栈是Python 3.8 Scapy Flask iptables只要你有一台Linux机器按步骤就能把最小可用系统跑起来答辩论点也足够扎实。2. 系统架构与实时流量分析先把抓包和特征提取跑通2.1 为什么选Scapy而不是纯socket解析半天跑通和一周跑通的区别做实时流量分析第一个选择是抓包库。如果你从零开始用socket读原始报文然后自己解析以太网头、IP头、TCP头光处理IPv4选项、TCP flag、分片重组就够写几天而且这还是在整个毕设的“前菜”阶段。常见做法是直接用Scapy它把IP、TCP、UDP、ICMP的解析封装好了提供sniff接口过滤语法和tcpdump的BPF一致开发效率高很多。Scapy是纯Python实现解析性能不算强但毕设实验环境流量小完全扛得住。真正并发几十万PPS的生产环境也不会用Python那是DPDK和C的天下这个区别在答辩时反而是加分项。系统的整体结构是一个生产者-消费者模型抓包线程用sniff从网卡收包把包放进有界队列一个或多个检测线程从队列取包提取特征后跑攻击检测规则检测到攻击就把事件抛给防御线程由防御线程执行iptables封禁同时Flask后端从共享指标里读数据供前端ECharts轮询展示。这个分层的好处是每一块都能单独测试后面替换检测算法、换掉可视化框架都不会动到其他模块。2.2 实时抓包的最小实现sniff加队列别在主线程里干重活先把抓包这层跑通。以下是一个最小可运行的抓包线程注意不要把sniff放在主线程里否则后面加载检测规则或者页面卡顿会直接扔包。import threading import queue from scapy.all import sniff, IP, TCP, UDP # 有界队列避免抓包速度大于处理速度时内存无限上涨 packet_queue queue.Queue(maxsize1000) def packet_callback(pkt): 收到一个包后的回调只关心IP/TCP/UDP过滤ARP等链路层噪声 if IP in pkt and (TCP in pkt or UDP in pkt): packet_queue.put(pkt) def start_capture(interface, bpf_filtertcp or udp): 启动抓包线程sniff会阻塞当前线程 def _run(): sniff(ifaceinterface, filterbpf_filter, prnpacket_callback, storeFalse) t threading.Thread(target_run, daemonTrue) t.start() return t if __name__ __main__: # 网卡名不要猜先执行 python -c from scapy.all import get_if_list; print(get_if_list()) start_capture(eth0, tcp or udp) print(抓包线程已启动当前队列积压, packet_queue.qsize())这段代码里sniff的filter参数用BPF语法只放行TCP和UDP过滤掉ARP、DHCP等与攻击检测关系不大的报文能明显降低后续处理压力。storeFalse表示抓到的包不做内存缓存否则抓一小时就把内存占爆。队列的maxsize1000是背压阀值如果检测线程处理不过来队列满后抓包线程会暂时阻塞这是保护机制不是bug。用daemonTrue启动线程主程序CtrlC退出时不会留下僵尸线程。实际使用时把eth0替换成你机器的网卡名。在虚拟机里往往是ens33、ens192用前面注释里的命令列出所有网卡再选有IP的那个。如果服务器只有一块网卡抓包网卡和被检测流量网卡是同一块那么本机对外发起的正常SSH流量也会被抓到设计检测规则时要注意这一点。2.3 特征提取把报文转成检测模块认识的统一字典抓包只是搬砖真正的分析在后续。为了不让每个检测函数都去重复解析Scapy包我在包一进队就把它转成普通dict这一步叫特征提取。字段统一后检测模块和抓包库完全解耦以后换dpkt、换离线pcap解析都只需要改这一个函数。from scapy.all import IP, TCP, UDP def extract_features(pkt): 从Scapy报文提取关键字段返回dict。 没有的字段置None统一格式方便检测规则直接取值。 feat { src: None, dst: None, sport: None, dport: None, proto: None, flags: None, len: len(pkt) } if IP in pkt: feat[src] pkt[IP].src feat[dst] pkt[IP].dst feat[proto] pkt[IP].proto # 6TCP, 17UDP if TCP in pkt: feat[sport] pkt[TCP].sport feat[dport] pkt[TCP].dport feat[flags] int(pkt[TCP].flags) # SYN0x02, ACK0x10 if UDP in pkt: feat[sport] pkt[UDP].sport feat[dport] pkt[UDP].dport return feat这里把TCP flags转成整型是个关键细节。Scapy的flags对象拿到的是FlagValue不能直接用等号比较。转成int之后判断SYN包就是flags 0x02 ! 0判断ACK包就是flags 0x10 ! 0后续写规则会非常顺手。len字段记录包总长度DDoS检测里能用来算字节速率虽然本文没用到先留着不亏。2.4 队列消费线程单线程检测还是批量处理有了特征提取函数消费队列的检测线程可以这样写while True: pkt packet_queue.get(timeout1) feat extract_features(pkt) # 下一章的攻击检测入口 integrate_detection(feat)最常见的翻车是把检测函数直接写在回调里sniff回调本身在Scapy的接收线程里跑如果检测逻辑里有sleep或者复杂计算Sniff线程被卡住内核缓冲区会先满然后开始丢包表现是“流量一大就漏报”。用队列解耦之后即使检测线程偶尔卡顿sniff线程也只是往队列里放包队列满了才阻塞。我习惯消费者线程同时跑三四个每个线程消费同一个队列检测性能不够时这是最便宜的扩展方式。3. 攻击检测模块用阈值规则识别端口扫描、暴力破解和DDoS3.1 为什么先做规则检测而不是机器学习毕设答辩最怕被问“你的检测模型为什么这么选”。机器学习要做的话你得自己标注数据集或者用现成的NSL-KDD但那和实时流量分析的结合需要额外写特征工程和模型部署管道工作量直接翻倍。我建议第一版用阈值规则引擎它有三点优势第一可解释性强评委问“为什么判断这是扫描”你可以直接回答“滑动窗口内不同目标端口数超过50”第二规则参数可以现场调测试的时候改一个数字就能看到效果变化演示效果非常直观第三模型升级路径清晰规则产生的特征可以作为机器学习的输入还能用规则结果做伪标签。标题里没提机器学习那规则检测就是最贴合题目的方案。检测目标选“端口扫描、暴力破解、DDoS”三类是因为它们特征差异大、覆盖了从主机行为到网络流量的不同层次而且每一类都能用简单统计量表达。端口扫描看的是“同源IP请求的端口多样性”暴力破解看的是“同源IP到同目标端口的频率”DDoS看的是“目标IP收到的聚合速率”。三个维度不重叠展示时不至于糊成一团。3.2 端口扫描检测滑动窗口内不同端口数超过阈值就告警端口扫描的典型行为是一个源IP在短时间内探测大量不同端口意图发现开放服务。下面用滑动窗口做统计只关注TCP SYN包。import time from collections import defaultdict SCAN_WINDOW 5 # 统计窗口单位秒 SCAN_PORT_THRESHOLD 50 # 窗口内不同目标端口数 scan_records defaultdict(list) def is_port_scan(feat): 判断当前包是否属于一次端口扫描。 只统计src-dport的SYN包窗口内去重端口数达到阈值即告警。 # 非TCP或不是纯SYN直接跳过 if feat[proto] ! 6 or feat[flags] ! 0x02: return False now time.time() src feat[src] scan_records[src].append((now, feat[dport])) # 清理窗口外记录避免列表无限膨胀 scan_records[src] [(t, p) for t, p in scan_records[src] if now - t SCAN_WINDOW] if len(scan_records[src]) SCAN_PORT_THRESHOLD: distinct_ports {p for _, p in scan_records[src]} if len(distinct_ports) SCAN_PORT_THRESHOLD: # 告警后清空防止同一IP连续触发告警风暴 scan_records[src].clear() return True return False这里有几个参数值得细化。flags ! 0x02是要求“纯SYN包”因为正常TCP连接也是从SYN开始的但正常主机在几秒内不会对50个不同端口发起纯SYN连接如果只做三次握手会有大量ACK回来把阈值逻辑搞混。distinct_ports去重是必要的因为扫描器可能对同一个端口发多个重传SYN不去重会把重传也算成端口数误报率会高。告警后立刻clear()这个源IP的记录是为了防止同一个慢速扫描每隔几秒就触发一次告警刷屏到无法看日志。参数怎么调演示环境建议把SCAN_PORT_THRESHOLD调到20否则在校园网环境里别人一次正常爬虫抓取也可能触发50个端口。生产环境至少80起步。SCAN_WINDOW调大能发现慢速扫描但记录积压和内存占用也会增长一般5到10秒就够。3.3 暴力破解检测同源IP对同目标端口的高频连接暴力破解的特征更直接一个源IP不停尝试连同一个目标端口往往是SSH的22、MySQL的3306、Web后台的8080。这类检测如果只按源IP统计会把正常的多端口访问也算进去所以要按“源IP-目标IP-目标端口”三元组来统计。from collections import defaultdict import time BRUTE_WINDOW 10 # 统计窗口单位秒 BRUTE_ATTEMPTS 30 # 窗口内最大连接尝试次数 brute_records defaultdict(list) def is_brute_force(feat): 检测同源IP对同目标端口的暴力破解行为。 只统计TCP包因为破解必然要建立连接。 if feat[proto] ! 6: return False now time.time() key (feat[src], feat[dst], feat[dport]) brute_records[key].append(now) brute_records[key] [t for t in brute_records[key] if now - t BRUTE_WINDOW] if len(brute_records[key]) BRUTE_ATTEMPTS: brute_records[key].clear() return True return False这里没有区分SYN还是其他flag因为暴力破解无论成功失败都会产生大量TCP包。时间戳列表的清理方式和端口扫描一样超过窗口的记录直接过滤掉。调参时注意BRUTE_ATTEMPTS30在10秒窗口内对SSH破解来说已经算很高了因为自动化工具每秒可以发几十次尝试。如果你的实验环境里有一台内网机器每秒钟掉线重连几十次也会误报这个时候不是单纯调高阈值而是要加白名单后面避坑章节会细说。3.4 DDoS检测从全局包速率和单目标包速率两个角度判断DDoS比前两种更难用单一阈值定死因为正常业务在高峰期也可能有明显流量抬升。常见做法是同时维护两个速率全局PPS和单目标PPS。当单目标速率超过阈值且明显高于全局基线时说明这个目标正被集中攻击。import time from collections import defaultdict DDOS_WINDOW 5 GLOBAL_PPS_THRESHOLD 1000 # 全局每秒包数 TARGET_PPS_THRESHOLD 200 # 单个目标每秒包数 packet_times [] target_records defaultdict(list) def check_ddos(feat): 判断是否DDoS全局速率过高或单一目标IP速率过高。 返回True表示当前包参与了一次DDoS事件。 now time.time() # 全局PPS packet_times.append(now) packet_times[:] [t for t in packet_times if now - t DDOS_WINDOW] global_pps len(packet_times) / DDOS_WINDOW # 单目标PPS dst feat[dst] target_records[dst].append(now) target_records[dst] [t for t in target_records[dst] if now - t DDOS_WINDOW] target_pps len(target_records[dst]) / DDOS_WINDOW if target_pps TARGET_PPS_THRESHOLD and global_pps GLOBAL_PPS_THRESHOLD: target_records[dst].clear() return True return Falsepacket_times[:] ...这种写法是原地修改列表而不是重新赋值。因为检测线程可能同时被多个消费者调用重新赋值会让另一个线程读到旧引用。全局阈值和单目标阈值同时满足时才告警能避免某台机器被爬虫集中访问导致的单目标速率波动。实际部署时这两个阈值要根据日常流量基线来设。我通常先让系统跑一天统计平均PPS和P95 PPS再把阈值设为P95的2到3倍。没有基线数据就盲目设数字不是“防DDoS”是“随机报警器”。3.5 检测模块的统一入口和告警事件三个检测函数写好后需要一个入口把它们组合起来并且输出结构化的告警事件def integrate_detection(feat): 统一分析入口按顺序跑规则命中则产生告警事件 if is_port_scan(feat): publish_event(port_scan, feat) if is_brute_force(feat): publish_event(brute_force, feat) if check_ddos(feat): publish_event(ddos, feat) def publish_event(event_type, feat): 把事件交给防御模块的队列同时打印日志 print(f[告警] {event_type} | {feat[src]} - {feat[dst]}:{feat[dport]}) defense_queue.put((event_type, feat[src]))注意这里每个if都是独立判断不用elif。因为一个扫描行为可能同时伴随暴力破解比如同一个IP在探测完端口后紧接着对某端口做密码尝试。用elif会让后续规则永远没机会执行。告警事件放入defense_queue之后检测线程的工作就结束了封禁动作由专门的防御线程去做避免检测线程被iptables子进程阻塞。4. 自动防御联动iptables做短时封禁别直接拉黑4.1 自动防御的闭环检测、封禁、解封、记录自动防御是“基于Python的网络入侵检测与防御系统”里最容易做过头的一环。我见过有同学检测到扫描就直接把IP加进iptables并且不设置解封时间第二天实验室里有一半机器被他封了实验都做不下去。真实运维里“封禁”永远是临时的因为源IP可能被伪造也可能是一个NAT网关里的合法用户。正确的闭环是检测到攻击后封禁一段时间时间到自动解封同时在内存里维护一份封禁记录避免重复封禁。还有一条原则是先查白名单。内网的网关、DNS服务器、你自己的远程管理IP永远不应该被自动防御封住。否则一次误报可能直接切断所有机器和外网的连接。白名单不是一个“可选优化”而是自动防御能安全演示的前提。4.2 用iptables封禁IP的Python封装自动防御动作在Linux上最常见的做法是调用iptables命令向INPUT链插入一条DROP规则。下面封装了封禁、解封、判重复三个逻辑。import subprocess import time # 白名单至少包含本机回环和管理IP建议从配置文件读取 WHITE_LIST {127.0.0.1, 192.168.1.1} # 记录每个IP的解封时间 ban_expire {} def block_ip(ip, duration300): 封禁指定IPduration秒后自动解封。 返回True表示这次调用真的执行了封禁。 if ip in WHITE_LIST: print(f[防御] 跳过白名单IP{ip}) return False now time.time() if ip in ban_expire and ban_expire[ip] now: return False # 已封禁且未到期 ret subprocess.run( [iptables, -A, INPUT, -s, ip, -j, DROP], capture_outputTrue, textTrue ) if ret.returncode 0: ban_expire[ip] now duration print(f[防御] 已封禁 {ip}{duration} 秒后解封) return True else: print(f[防御] 封禁失败{ret.stderr.strip()}) return False def unban_expired(): 遍历封禁记录到期则删除iptables规则。 这个函数需要定时调用而不是只调一次。 now time.time() expired_ips [ip for ip, expire in ban_expire.items() if expire now] for ip in expired_ips: subprocess.run( [iptables, -D, INPUT, -s, ip, -j, DROP], capture_outputTrue, textTrue ) del ban_expire[ip] print(f[防御] 解封 {ip})block_ip里先判断白名单再判断是否已经在封禁期内避免同一个IP被多个告警事件重复封禁。subprocess.run每次执行iptables都会fork一个子进程性能开销不小但防御动作本来就不是高频操作每秒最多处理几十个IP完全够用。capture_outputTrue是为了把stderr拿回来打印否则权限不足时你连报错都看不到只能干瞪眼。解封函数单独拆出来是因为封禁和解封要放在同一个循环里调度。expired_ips先收集再删除不能在遍历列表的同时修改列表这是Python里常见的易错点。duration默认300秒演示时可以改成60秒方便在同一个IP上反复测试。4.3 防御线程和检测线程的联动队列解耦避免互相阻塞检测线程发现攻击后如果直接调用subprocess的iptables那么一次封禁操作耗时可能几十毫秒这期间检测线程无法处理后续报文在高流量下就是灾难。所以防御动作要放到独立线程里用队列把两个线程解耦。前面的defense_queue在这里消费import queue import time # 全局告警队列检测模块往这里写 defense_queue queue.Queue() def defense_loop(): 防御工作循环从队列取告警执行封禁每秒钟顺带解封到期IP。 在python主线程里单独启动。 while True: try: event_type, ip defense_queue.get(timeout1) block_ip(ip, duration60) except queue.Empty: pass # 每秒检查一次过期封禁 unban_expired() time.sleep(1)defense_queue.get(timeout1)让循环在队列空时不会死等能每隔一秒走到unban_expired。如果把get设为无超时阻塞那么解封逻辑只有在收到下一个攻击事件时才会执行到期IP会被一直封着。这个线程启动后检测模块只需往队列里丢事件整个系统才是“实时检测-自动防御-自动解封”的真正闭环。4.4 演示逃不过的三件事root权限、规则顺序、规则持久化第一iptables需要root权限Python脚本在普通用户下运行会得到Permission denied。演示时直接sudo python main.py不要用setcap那种花活容易把规则搞乱。第二-A INPUT是把新规则追加到输入链末尾如果链里已有更早的DROP规则拦住了同源IP新规则不会立刻生效。需要在封禁前执行iptables -F INPUT清空旧规则或者用-I INPUT把新规则插到最前面。我写的时候用-A是考虑到要维护多条封禁规则互不干扰但演示前要先确认链是干净的。第三iptables规则重启后消失答辩演示如果中途重启了机器规则全没了系统看起来像是“没有防御”。如果需要持久化用iptables-save导出到文件但这部分不是必做项。5. 可视化监控用Flask和ECharts把流量与告警画出来5.1 为什么可视化监控不能省标题里的“可视化监控”是硬性要求也是答辩时最直观的展示点。评委走进来看不到实时变化的界面光听你讲算法很难建立信任。常见做法是Flask后端提供一个JSON接口前端ECharts轮询接口把包速率、告警次数、封禁IP数画成实时刷新的折线图和列表。不选Grafana是因为它需要额外安装服务而且Grafana的仪表盘一看就是现成模板显得你在做集成而不是做开发。自己用Flask写还能顺手演示一下后端API设计能力。5.2 后端API设计从检测模块取实时指标先写一个最小可运行的Flask服务接口返回当前指标。from flask import Flask, jsonify, render_template import time app Flask(__name__) # 实际项目中这些数据从检测模块的共享内存、Redis或sqlite读取 # 这里用函数包一层方便后续替换 def get_metrics(): return { timestamp: time.time(), packet_rate: 123.4, # 每秒包数由抓包模块统计 alert_count: 7, # 累计告警数 banned_ips: 3 # 当前封禁的IP数量 } app.route(/api/metrics) def metrics(): resp jsonify(get_metrics()) # 禁止浏览器缓存否则轮询接口可能拿到旧数据 resp.headers[Cache-Control] no-store return resp app.route(/) def index(): return render_template(index.html) if __name__ __main__: app.run(host0.0.0.0, port5000, debugFalse)get_metrics里现在是写死的数字你接检测模块时需要维护一个全局指标对象抓包线程每收一个包就计数器加一检测线程每次发布告警就告警计数加一然后在这里读出来。Cache-Control: no-store这个头很重要浏览器默认会缓存GET请求尤其是Vercel或Nginx反代环境不加这个头你会发现曲线隔十几秒才跳一下以为是程序卡了其实是缓存。debugFalse必须设死Flask的debug模式会启动重载器从而把抓包线程重复拉起产生“拦截两次”的假象。5.3 前端仪表盘ECharts折线图轮询后端接口前端页面放在templates/index.html下用ECharts画一条滚动流量曲线。以下是可以直接复用的模板!DOCTYPE html html head meta charsetutf-8 title入侵检测与防御监控/title script srchttps://cdn.jsdelivr.net/npm/echarts5/dist/echarts.min.js/script /head body div idchart stylewidth:100%;height:400px;/div script var chart echarts.init(document.getElementById(chart)); var option { title: { text: 实时流量包速率 }, xAxis: { type: category }, yAxis: { type: value }, series: [{ type: line, data: [], showSymbol: false }] }; chart.setOption(option); function fetchData() { fetch(/api/metrics) .then(r r.json()) .then(data { var now new Date(data.timestamp * 1000).toLocaleTimeString(); option.xAxis.data.push(now); option.series[0].data.push(data.packet_rate.toFixed(1)); // 只保留最近30个点避免曲线越来越挤 if (option.xAxis.data.length 30) { option.xAxis.data.shift(); option.series[0].data.shift(); } chart.setOption(option); }) .catch(err console.error(err)); } setInterval(fetchData, 1000); /script /body /htmlECharts从CDN加载如果是无网络环境演示要下载echarts.min.js到static目录并把script标签的src改成相对路径。前端只保留30个数据点形成滚动效果避免内存泄漏。轮询间隔1000毫秒如果后端处理压力大调到2000毫秒也可以。实际演示时可以先在终端看到检测日志再切到浏览器看曲线变化两边一致就会很有说服力。5.4 常见问题排查与避坑一套系统做完调试时总会踩到几个典型的坑。我把高频问题按“现象-原因-解决”整理如下。第一个抓包程序运行后完全没有输出。现象sniff回调什么都没打印队列一直为空。原因要么是非root用户没有抓包权限要么是网卡名写错了。解决先执行sudo python -c from scapy.all import sniff; print(ok)确认权限再用python -c from scapy.all import get_if_list; print(get_if_list())列出真实网卡名不要猜eth0。如果连Scapy都装不上回过去把Python安装教程里的pip和虚拟环境配置再走一遍这个基本功省不了。第二个误报率高到没法看。现象内网一台机器正常做业务被判定为端口扫描或暴力破解。原因阈值设置太低或者没有配置白名单。解决先调高阈值观察业务流量的正常基线再把内网网关、数据库服务器、办公系统的IP加进白名单。注意白名单应该覆盖检测模块和防御模块两层检测层白名单可以跳过分析防御层白名单避免误封。第三个iptables封禁命令返回权限错误。现象调用block_ip后终端打印“封禁失败”stderr是Permission denied。原因Python进程不是root或iptables链不存在。解决用sudo运行程序用iptables -L INPUT确认链存在。同时检查有没有别的安全模块拦截了iptables调用比如云服务器默认的firewalld和iptables冲突要先停掉firewalld。第四个可视化页面接口有数据但曲线不更新。现象打开页面能看到第一帧后面一直不动。原因浏览器缓存了/api/metrics的GET响应。解决在Flask响应头加Cache-Control: no-store并确认没有经过带缓存的Nginx。第五个流量稍微一大程序就卡死。现象CPU占用不高但检测吞吐骤降队列积压到满。原因检测线程在同步执行耗时操作比如subprocess调用iptables或打印大量日志。解决把防御动作丢给独立线程用队列消费检测线程只做特征计算和规则判断日志异步写或用logging的QueueHandler。如果queue满了sniff回调会阻塞在put上导致内核丢包这种情况下只能升级硬件或减少抓包范围。6. 进阶怎么验证检测效果并完成性能调优写完规则和防御链路下一步是让别人相信你的系统真的能检测到攻击。不推荐用真实攻击工具做演示原因有二一是可能触发你所在网络的管理告警二是攻击工具的沙箱环境兼容性很麻烦。我用的做法是用Scapy构造特征明显的测试包直接喂给检测函数做单元测试这样不依赖真实网卡也便于在答辩现场重复跑。def test_port_scan_detection(): # 模拟1秒内向10.0.0.1的不同端口发SYN包 for dport in list(range(20, 70)): feat { src: 10.0.0.9, dst: 10.0.0.1, sport: 1000, dport: dport, proto: 6, flags: 0x02 } is_port_scan(feat) # 50个不同端口已经超过阈值is_port_scan返回True这段测试代码把feature字典直接传给is_port_scan跑完检查返回值。同样的方法可以给暴力破解和DDoS各写一个测试函数。这样做的好处是不需要真实流量任何阶段都能跑。更进一步可以把已抓取的pcap文件重放用rdpcap读取后逐包调用extract_features和integrate_detection验证规则是不是在真实历史流量上能命中。调优方面我推荐先用cProfile找出热点函数。如果发现extract_features占用高考虑裁剪字段比如不解析UDP字段如果规则统计里的列表过滤占用高改用collections.deque代替list做滑动窗口时间复杂度和内存占用都更友好。封禁逻辑里subprocess.run是最明显的性能瓶颈可以加一个批量封禁接口把多条iptables规则合成一条链比如先创建自定义链然后统一跳转但这属于锦上添花。最后说我自己的教训。第一版自动防御我写的是永久封禁没有解封流程结果自己测试的时候把虚拟机的网关封掉了整个实验环境断网最后只能进物理终端手动清规则。后来改成短时封禁加白名单再也不敢用永久封禁做默认行为。毕设系统可以写得“狠”但一定要留后退的余地这就是防御模块里那句“duration300”的真正价值。希望这篇围绕实时流量分析、攻击检测、自动防御、可视化监控的拆解能帮你在答辩前少走这段弯路。本文还有配套的精品资源点击获取