☰
PyQt6+PyShark+DNN:实时网络入侵检测系统实践
2026/10/26 13:52:52 网站建设 项目流程

简介:基于PyQt6图形界面框架与深度学习DNN模型实现的网络安全入侵检测及实时抓包分析系统项目包,适合安全运维人员、高校网络安全专业学生及流量分析开发者。系统以CIC-IDS2017数据集训练二分类模型,配合PyShark抓包引擎完成实时网络异常识别,可部署于企业内网、金融机构等安全敏感环境。资源共31个文件、约7.7MB,核心包括5个Python脚本(主程序、抓包视图、分析视图、训练与预测)、5张PNG界面截图、TensorFlow模型文件(pb/pyc)、CSV数据集与测试流量包pcap,另附README和说明文档辅助上手。已有52人学习下载。内容从模型构建、图形界面设计到实时抓包分析形成完整链路,既可直接复跑演示,也便于按需二次开发,可作为网络安全实战项目或毕设参考。

1. 先抓流量再判异常:PyQt6 + DNN + PyShark 这套组合到底能干什么

如果你在运维或安全岗,大概率经历过这样的下午:内网流量突然暴涨,交换机面板闪个不停,手工抓包能看到一堆 TCP 重传和异常连接,但就是说不清这是入侵还是业务抖动。这套基于 PyQt6、深度学习 DNN、PyShark 的网络安全入侵检测工程,就是把“手工抓包 + 手工看特征”变成一台能实时出结论的桌面工具。它用 PyShark 做实时抓包,把流量聚合成 CIC-IDS2017 数据集同口径的特征,交给二分类模型给出正常/异常概率,最后在 PyQt6 界面里展示。适合安全运维、攻防研究、毕业论文,以及想从“会调包”往“会搭流程”走的开发。它的边界也清楚:二分类模型只能给出异常概率,不负责告诉你具体攻击类型,也不擅长加密流量里的隐藏行为。

2. PyQt6 还是 PySide6:选型理由和 DNN 二分类的边界条件

2.1 为什么选 PyQt6 而不是 PySide6:授权、打包、接口差异一次说清

讨论区里“pyqt6 和 pyside6 哪个强”几乎是每天都有的问题。就纯界面能力来说,两个库同源于 Qt 封装,真实差距很小。落到项目上,差别主要在三处。授权上,PyQt6 采用 GPL/商业双授权,PySide6 是 LGPL,前者对闭源交付或给客户做安全服务时会有法务风险,后者更宽松;打包体量上,PyQt6 的 wheel 依赖更重,但换来的是对 Qt6 小版本跟进更快;接口细节上,两个库的信号槽和枚举命名有差异,比如pyqtSignal与Signal、Qt.AlignmentFlag.AlignCenter这类写法不完全一致。

本工程选 PyQt6,并不是因为它在功能上碾压 PySide6,而是三个现实理由:第一,PyQt6 的社区示例多,抓包、流量表格、仪表盘这类界面项目大量是 PyQt6 写的,踩坑时有更大概率搜到同款报错;第二,从 PyQt5 迁到 PyQt6 的人多,exec()、setContentsMargins这些差异早就被整理透了,照着改成本低;第三,在 Qt6 长期支持版本上,PyQt6 的绑定发布节奏比 PySide6 更跟手。如果你后续决定换 PySide6,迁移成本其实不高,核心是把工程结构和线程边界写对,界面框架换掉没有想象中那么伤筋动骨。

另一种更值得注意的问题是环境兼容。PyQt6 依赖 Qt6 运行库,Qt6 对 OpenGL 的依赖在远程桌面场景比 Qt5 敏感。我上一版代码就遇过本地界面正常、远程桌面一打开就黑屏的情况,排查到最后是渲染引擎选择问题,设置QT_OPENGL=software环境变量才解决。这一条先记着,后面避坑部分还会提到类似场景。选 PyQt6 不是选“最好的框架”,而是选“在当前团队、当前交付环境里最不容易踩坑的框架”,这两件事在工程落地阶段经常不是同一个答案。

2.2 系统分层:抓包层、特征层、推断层、界面层各管一件事

很多初版工程会把 PyShark 的每个包对象直接塞进界面线程,抓到十个包界面就开始假死。这个资源拆包时我看到的第一眼,是它把整个系统分成四层,各管一件事。

抓包层只做一件事:用 PyShark 的 LiveCapture 从网卡读包,配合display_filter做粗过滤,只留 IP 流量,把 ARP、MDNS 这类广播噪声挡在前面。特征层负责把包的统计字段抽出来,按五元组聚合成一行 CIC 风格特征,这一层是纯 Python 计算,不碰任何界面对象。推断层加载训练好的 DNN 模型,对特征向量做预测,返回概率值和判定结果,它不知道界面长什么样。界面层只接收基本类型数据,比如字符串、浮点数、时间戳,再负责刷新表格和弹告警。

这个分层不是摆样子。PyShark 底层调的是 tshark 命令行,每抓一批包就有一次进程通信开销;如果让界面线程直接干这件事,Qt 事件循环会被 tshark 的输出堵死。把抓包放在子线程,把模型推断也放在子线程,主线程只参与信号槽收结果,这是整套工程“界面不卡”的前提。后面第 4 章给出的CaptureWorker和FlowClassifierWorker两个线程类,就是围绕这四层边界设计的。

2.3 二分类模型能识别什么,不该让它背什么锅

CIC-IDS2017 数据集里标注了 Benign 与 DoS、PortScan、DDoS、Bot 等多种攻击,常见的处理是把非 Benign 全部标成 Attack 做二分类。这样模型学到的其实是“流量统计分布正常与异常”的边界:比如连接时长极短、单流包数和字节数偏离基线、SYN 标志位占比过高。它能发现端口扫描、暴力破解、部分 DoS/DDoS,甚至某些外联心跳型异常。

但它分辨不了攻击类型,更看不到载荷内容。二分类模型天然不擅长加密流量里的内部威胁,因为加密后的统计分布可以看起来完全正常;也不擅长单包级别的注入攻击,因为单包统计特征不明显。所以界面里的判定结果我建议写成“疑似异常”,而不是“检测到 XX 攻击”,否则一个误报会放大成一片警报。

模型训练用的特征依赖 CIC-IDS2017 的统计口径,真实网络流不会被严格切成固定大小窗口。工程里必须有一个流聚合器,把实时包按滑动窗口聚合成“当前状态”,这个简化是第 4 章实时管道要解决的关键问题。换句话讲,这套工程的正确用法是把它当成流量异常感知的前置过滤器,替人工盯住那些统计上不正常的东西,再把可疑窗口交给日志和人工研判,而不是让它当一个全知全能的检测大脑。

3. CIC-IDS2017 数据集加工:从原始 pcap 到可训练的二分类特征表

3.1 数据准备:选 CSV 还是 pcap,怎么切出训练窗口

CIC-IDS2017 发布时有完整 pcap 和已经抽好的 CSV 特征表两种形态。第一次训练我建议直接用官方 CSV 特征表,不要动 pcap:每一行已经按流聚合好,最后一列 Label 是分类标签,pandas 读进来就能开始做特征筛选,这样能在最短时间内跑通基线。动 pcap 是后面要做“从抓包到模型”全流程复现时才需要的事。

读 CSV 有个很坑的细节:官方文件的列名和标签里自带空格,比如" Label",而且文件行数多,pandas 默认按块推断类型容易把数值列读成 object。我一般第一版脚本先做最小读取,确认能正常解析再往下走:

import pandas as pd df = pd.read_csv("Friday-WorkingHours.csv", low_memory=False) print(df.shape) print(df.columns.tolist()[:10]) print(df[" Label"].value_counts())

low_memory=False是为了避免 pandas 分块读取时把数值列猜成 object,否则后面特征筛选全部要再做一次字符串转 float,纯属给自己加活。" Label"前面的空格是官方 CSV 自带的,或者漏了一个,建议统一strip()之后再统计类别,否则 Benign 和一个带空格的 Benign 会被当成两个类。

读取之后做三件清洗:去掉Infinity和NaN行,把负数值置 0 或直接删除,再把标签二值化。CIC 官方特征表里有少量Infinity值,不处理的话训练时会被当成缺失值,Keras 不会直接告诉你哪一行出了问题,只会让 loss 变成 NaN:

import numpy as np df[" Label"] = df[" Label"].str.strip() df[" label_binary"] = (df[" Label"] != "Benign").astype(int) df = df.replace([np.inf, -np.inf], np.nan).dropna()

标签二值化之后,攻击类是 1、正常类是 0。这里注意类别不平衡:CIC 数据集中正常流量样本量通常远大于攻击样本,后面训练时要靠class_weight或欠采样平衡。

3.2 特征裁剪与标签二值化:让 DNN 只面对它该面对的字段

CIC 原始特征表有 80 多列,经验做法是只保留基础流量统计列,不要全喂进去。DNN 在这个场景里不需要特别宽的特征面,特征过宽反而容易记住训练集里的特定噪点。我保留的核心特征如下,这份列表也是后面在线抓包特征提取要复用的字段定义:

特征列含义对异常的意义
Flow Duration流时长(微秒)扫描流极短,大流量 DoS 也短
Total Length of Fwd Packets正向包总长度方向性吞吐偏差
Total Length of Bwd Packets反向包总长度反射型攻击反向异常
Fwd Packet Length Mean正向平均包长扫描常见小包
Bwd Packet Length Mean反向平均包长同上
SYN Flag CountSYN 标志位计数扫描和 SYN Flood 核心特征
Label标签二分类目标

实际训练时不要只留 6 列,我一般会把输入控制在 20 列以内,在上面基础上再加 Packet Length Std、Flow IAT Mean、Active Mean 这类方差敏感特征。关键是训练和推理必须共用同一份特征列定义,顺序不能乱,这一点第 5 章会专门讲。

3.3 训练脚本与关键超参:Adam、BatchNorm、Dropout 和早停怎么配合

DNN 模型结构不需要很复杂,三层隐藏层在这个数据规模上已经够用。我用的是 128/64/32 递减结构,每层 Dense 之后接 BatchNorm 和 Dropout:

import tensorflow as tf from tensorflow.keras import layers, models model = models.Sequential([ layers.Input(shape=(feature_dim,)), layers.Dense(128, activation="relu"), layers.BatchNormalization(), layers.Dropout(0.3), layers.Dense(64, activation="relu"), layers.BatchNormalization(), layers.Dropout(0.3), layers.Dense(32, activation="relu"), layers.Dense(1, activation="sigmoid") ]) model.compile( optimizer=tf.keras.optimizers.Adam(learning_rate=1e-3), loss="binary_crossentropy", metrics=["accuracy", tf.keras.metrics.AUC(name="auc")] )

BatchNorm 放在 ReLU 后面,是为了把上一层输出重新拉回稳定分布,流量特征的方差动辄差几个数量级,没有 BN 前面几轮 loss 会跳得很厉害。Dropout 取 0.3,防止模型死记某几条流的细节特征。Adam 用 1e-3,这是流量二分类里最常见的一组经验值,不需要一开始就去调。

训练时早停是必须的,不然很容易跑满 30 轮把验证集跑崩:

callbacks = [ tf.keras.callbacks.EarlyStopping( monitor="val_loss", patience=5, restore_best_weights=True ), tf.keras.callbacks.ReduceLROnPlateau( monitor="val_loss", factor=0.5, patience=3 ) ] history = model.fit( X_train, y_train, validation_data=(X_val, y_val), batch_size=64, epochs=30, callbacks=callbacks, class_weight={0: 1.0, 1: 1.5} )

class_weight给攻击类 1.5 的权重,是因为 CIC 二分类里攻击样本相对少,不加权的话模型会倾向把所有样本都判成 Benign,准确率看着很高,实际一点用没有。评估时也不要只盯准确率,这个数据集天生不平衡,要同时看精确率、召回率和 F1。如果召回率上不去,先把阈值从 0.5 往下调,再去看特征是不是选少了。

4. PyShark 抓包引擎与模型推断:实时管道怎么搭才不卡界面

4.1 PyShark 的 LiveCapture 与 Qt 线程边界

很多教程会把 PyShark 的sniff_continuously直接写进按钮点击回调,这是一个很容易翻车的写法。PyShark 内部是调用 tshark 命令行并读取它的 stdout,每个包的获取都涉及进程通信和 JSON 解析,必然卡住主线程。我最早一版就是直接在 PyQt6 窗口的按钮回调里跑sniff_continuously,界面在前几个包之后就开始无响应,根源不是 PyShark 慢,而是把阻塞操作放进了 Qt 事件循环。

正确的做法是把抓包挪到 QThread 子类里:

import pyshark from PyQt6.QtCore import QThread, pyqtSignal class CaptureWorker(QThread): on_statistic = pyqtSignal(dict, str) def __init__(self, interface: str = "eth0"): super().__init__() self.interface = interface self._running = True def run(self): cap = pyshark.LiveCapture( interface=self.interface, display_filter="ip", use_json=True ) for packet in cap.sniff_continuously(): if not self._running: break statistics = extract_flow_statistics(packet) if statistics is not None: self.on_statistic.emit(statistics, packet.ip.src)

display_filter="ip"是粗过滤,只让 IP 包进入 Python 层,ARP、MDNS 这些噪声在前端就被 tshark 挡掉了。use_json=True让 tshark 输出 JSON 而不是原始文本,PyShark 解析速度更快,CPU 占用也更低。_running标志用于在主窗口关闭时通知线程退出,否则抓包线程会一直阻塞在sniff_continuously里,进程退不干净。

4.2 在线特征提取与实时推断:线程和队列把抓、判、显分开

抓包线程拿到包之后,下一步是特征提取和模型推断。这里我建议用一个 worker 串行做“抓包→特征→预测”,不要把特征提取和模型推断拆成两个线程,因为流量顺序本身就是天然队列,拆开反而要处理锁和同步。

在线推理的骨架大致是这样:

class FlowClassifierWorker(QThread): result = pyqtSignal(str, str, float) def __init__(self, capture_conf, feature_cols, model): super().__init__() self.capture_conf = capture_conf self.feature_cols = feature_cols self.model = model self._running = True def run(self): cap = pyshark.LiveCapture(**self.capture_conf) for packet in cap.sniff_continuously(): if not self._running: break vec = build_feature_vector(packet, self.feature_cols) if vec is None: continue prob = self.model.predict(vec.reshape(1, -1), verbose=0)[0][0] self.result.emit(packet.ip.src, packet.ip.dst, float(prob))

build_feature_vector是特征层的核心函数。它维护一个按五元组分键的滑动窗口字典,窗口内累计包数、字节数、SYN 计数,每隔固定时间或固定包数刷新一次特征向量。这个设计有一个重要取舍:CIC 特征是在流结束后统计的,但实时检测不能等流结束,只能计算“窗口内当前状态”,所以在线特征和离线训练特征的分布天然存在偏差。这是正常的,要在工程里接受它,而不是试图让模型去完美对齐。

4.3 告警更新节奏:界面只消费结果,不接触包对象

PyQt6 的信号槽在跨线程传递时,只保证基本类型安全;pyshark 的 Packet 对象不要试图直接 emit,否则会触发元类型转换错误,或者在最不该出问题的时候变成隐式崩溃。上面代码里只传字符串和浮点数,这是刻意为之。

界面侧接收信号的写法:

def on_result(self, src, dst, score): self.table_model.append_row(src, dst, f"{score:.4f}") if score > 0.85: self.tray.showMessage("流量异常", f"{src} -> {dst}")

这里有三个注意点。第一,不要用定时器轮询抓包线程里的全局变量,用信号驱动刷新,避免 Qt 定时器高频轮询和线程变量竞争。第二,表格刷新要控制节奏,比如每 100 条结果刷新一次视图,不要每秒插入几十行直接把 QTableView 拖垮。第三,告警阈值 0.85 是经验值,不是固定值,可以拿一周真实流量跑出分数分布后再调。实时流量噪声比离线数据集大,阈值太低会变成“每秒都报警”,阈值太高又失去检测意义。

5. PyQt6 加 PyShark 加 DNN:实时抓包集成里绕不开的五个坑

5.1 现象:PyShark 过滤条件写对了,却一直抓不到包

原因:PyShark 本身不实现抓包协议,它只是 tshark 的封装。如果系统没装 Wireshark,或者 Wireshark 安装目录没进 PATH,LiveCapture会静默返回空结果,而不是抛一个明确的“找不到 tshark”异常。

解决:先打印路径确认后端存在:

python -c "import pyshark; print(pyshark.get_tshark_path())"

如果输出为空,去装 Wireshark 并把安装目录加进 PATH。另外 PyShark 和 tshark 的版本有兼容关系,遇到tshark: The installed version of Wireshark is not compatible这类报错时,不要硬凑,直接调 PyShark 版本或 Wireshark 版本,把验证过的一组版本写进工程 README,避免后来接手的人在这上面浪费半天。

5.2 现象:离线模型指标很高,一接实时流量就狂报警

原因:在线抓包拿到的不是完整流,而是滑动窗口里的中间态统计值。模型训练时用的是 CIC 数据集的完整流聚合特征,分布错位导致误报放大。这不是模型坏了,是训练和推理的特征口径没对齐。

解决:把实时特征提取改成和训练口径对齐,训练和推理共用同一份FEATURE_COLUMNS常量,不要各写一套字段顺序。同时把告警阈值从 0.5 提到 0.85 左右。实时流量里正常业务抖动也会造成统计偏移,阈值适当提高能压掉大量伪警报,代价是漏报增加,这个阈值要在具体网络环境里反复试。

5.3 现象:Qt 界面在抓包线程启动后会不定时崩溃

原因:最常见的两个。一是把 pyshark 的 Packet 对象或自定义特征字典直接传给pyqtSignal,元类型无法在跨线程队列里正确转换;二是窗口关闭时抓包线程还在阻塞等待包,线程对象被销毁后代码还在跑。

解决:信号只发基本类型,字符串、浮点数、整型都安全;在closeEvent里先置worker._running = False,再调用worker.wait(2000)等线程退出,超时再强杀。还有前面提到的远程桌面黑屏问题,设QT_OPENGL=software可以绕开 Qt6 在远程环境下的渲染选择异常,这个也归在环境坑里。

5.4 现象:CIC 数据集随机切分后,训练集 0.99、验证集直接崩

原因:随机train_test_split会把同一条流的前后包切进训练集和验证集,模型靠记住同源分布而不是学习异常逻辑,验证集一旦遇到新时段流量就露馅。CIC 是时间序列数据集,时间相关性很强,不能当时序无关的 i.i.d 数据切。

解决:按时间排序后,取前 70% 做训练、后 30% 做验证;或者先按五元组或 flow ID 分组,再对组编号切分。不要只看random_state=42就当作没问题,关键是保证任何一条流的记录不会跨集合存在。

5.5 现象:同一个包,离线特征和在线抓包特征对不上

原因:列顺序不一致,或者数值规范化方式不一致。离线 CSV 用 pandas 读,列顺序取决于表头;在线抓包用字典解析,可能出现 key 顺序不同。另一个更隐蔽的问题是 scaler:有的代码对整批在线流量重新计算均值和方差,这会把在线分布污染进标准化参数,导致推理输入漂移。

解决:训练结束后,把feature_names列表和训练集上拟合好的 scaler 一起用 pickle 保存:

import pickle with open("model_package.pkl", "wb") as f: pickle.dump({ "model": model, "feature_names": FEATURE_COLUMNS, "scaler": scaler }, f)

推理时只从 pickle 里加载 scaler,对每一条特征向量做独立transform,绝不重新计算均值和方差。这样离线特征和在线特征至少能保证在字段和标准化上说的是同一套口径。

6. 进阶用法:把二分类扩成多分类,并导出可复核的特征记录

6.1 从二分类扩多分类:把 label 换成攻击类型

如果不想只停留在“正常/异常”的二元判断,可以把label_binary换回原始Label,输出层改成 softmax,输出类别就是数据集里的攻击类型。隐藏层结构不需要动,DNN 前面 128/64/32 的特征提取能力可以复用。需要改动的只有三个位置:输出层Dense(num_classes, activation="softmax"),loss 换成categorical_crossentropy,标签换成 one-hot。真正费工夫的是类别不平衡,CIC 里 PortScan 样本远多于其他攻击类型,需要按类别设置权重或者对少数类做过采样。

6.2 可复核记录:每次告警都留一份特征哈希

比多分类更实用的一个习惯是给每次告警留可复核记录。把窗口首包时间、五元组、模型版本、告警阈值和特征向量哈希写进 JSONL 文件,存放在 worker 线程里,不碰界面层。这样事后可以说清楚“这条报警当时模型是怎么判的、用了哪些特征、阈值是多少”,而不是留下一张只有时间戳的截图。对交付审计和误报排查,这个文件的价值比模型本身更高。

我以前做过一版把抓包直接塞进界面主线程的工程,结果就是界面假死、线程退出不干净、误报无处追溯三件事同时爆炸。后来每次改完抓包逻辑,我都强制先拿一段离线 pcap 回放一遍,确认特征口径和训练时一致,再切实时接口,这个习惯帮我挡掉了至少三次半夜误报。希望帮到你。

本文还有配套的精品资源,点击获取

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询