恶意加密流量检测毕设实战：从数据到Flask演示系统

简介本资源为基于深度学习的恶意加密流量检测系统毕业设计完整资料包面向计算机、网络安全与人工智能方向的高校学生及需要完成课程设计、期末大作业的开发者。内容围绕加密流量特征提取与恶意流量识别展开涵盖DoH与CTU-13等数据集的Boruta特征筛选、相关性分析及模型训练结果可帮助读者理解从数据预处理到检测模型落地的完整链路。压缩包共217个文件约25.61MB包含4个Python源码文件、6个CSV特征与模型结果表、6个NPY数据文件、2个PCAP流量包以及14个HTML可视化页面、8张JPG与7张PNG图表和165个日志文件另附CSS、Markdown说明等辅助材料目录结构清晰便于按模块查阅。代码注释较为详尽新手也能对照理解。目前已有251人学习下载适合作为毕业设计、课程设计的高分参考方案也可用于快速部署与二次开发。1. 恶意加密流量检测从毕业设计选题到能跑通的系统做计算机毕业设计最怕选到一个“看起来能写、实际上跑不起来”的题目。恶意加密流量检测就是这样一个方向论文里概念清晰真动手时却发现公开数据集难找、加密流量特征难提取、模型训完指标虚高。我带过几届学生的毕设也自己复现过几个开源方案踩过的坑足够写一本小册子。这个方向的核心价值在于它同时覆盖了网络安全和深度学习两个热门领域答辩时既有技术深度又有应用场景而且用 Python 就能完成从数据处理到模型部署的全链路。适合有 Python 基础、学过一点深度学习课程、想在毕设里做出一个“能演示、能讲清楚”的系统的同学。接下来我会按实际做项目的顺序把数据、特征、模型、训练、排错这几个环节拆开讲每一步都给出可复现的代码和参数说明。2. 数据从哪来公开数据集选型与预处理流水线2.1 三个常用公开数据集的实际对比做恶意加密流量检测第一步不是写模型而是找到“带标签的加密流量”。我试过用爬虫自己抓结果标签全靠人工猜训出来的模型连自己都说服不了。后来固定用下面三个公开数据集各有各的脾气。数据集流量类型标签粒度样本量级适用场景CIC-IDS2017混合流量含加密与非加密二分类/多分类约 280 万条流入门首选特征已提取好USTC-TFC2016恶意软件产生的加密流量多分类10 恶意家族约 50 万条流恶意家族识别更贴近标题ISCX-VPN2016加密隧道流量应用分类约 20 万条流研究加密流量分类但注意合规使用我一般会先用 CIC-IDS2017 跑通全流程因为它的 CSV 文件里已经包含了 CICFlowMeter 提取的 80 多个流特征省去了自己写特征提取器的麻烦。但要注意这个数据集里的“恶意”标签包含大量非加密攻击直接拿来训加密流量检测模型会引入噪声。更贴近标题的做法是从 USTC-TFC2016 里取恶意软件加密流量作为正样本从 CIC-IDS2017 里取正常加密流量如 HTTPS作为负样本自己拼一个二分类数据集。2.2 用 Python 做流特征提取与清洗如果你拿到的原始数据是 pcap 包就需要自己提取流特征。常见做法是用cicflowmeter或scapy按五元组分流再计算统计特征。下面这段代码是我常用的预处理脚本把 pcap 转成 CSV并做基础清洗。import pandas as pd import numpy as np from cicflowmeter import extract_features # 需先安装 cicflowmeter def pcap_to_csv(pcap_path, output_csv): 将 pcap 文件转为流特征 CSV extract_features(pcap_path, output_csv) df pd.read_csv(output_csv) # 去掉全零列和方差极小的列 df df.loc[:, df.std() 1e-6] # 处理无穷值 df.replace([np.inf, -np.inf], np.nan, inplaceTrue) df.fillna(0, inplaceTrue) # 标签列重命名假设原始标签在 label 列 if label in df.columns: df[label] df[label].apply(lambda x: 1 if x ! BENIGN else 0) df.to_csv(output_csv, indexFalse) return df # 调用示例 df pcap_to_csv(malware.pcap, malware_flows.csv) print(df.shape, df[label].value_counts())这段代码的逻辑是先用cicflowmeter提取每条流的统计特征包长均值、方差、到达间隔等然后剔除方差过小的列这些列对分类没贡献把无穷值替换为 0最后把标签转成 0/1 二分类。参数上df.std() 1e-6这个阈值可以调整如果特征维度太高可以放宽到1e-5。标签映射那里如果你的数据集标签不是BENIGN需要按实际情况改。2.3 处理类别不平衡别让模型只会猜多数类恶意流量通常远少于正常流量直接训练会导致模型偏向多数类。我一般用 SMOTE 过采样少数类但要注意SMOTE 只能在训练集上做验证集和测试集必须保持原始分布。下面是一个划分数据集并做 SMOTE 的示例。from sklearn.model_selection import train_test_split from imblearn.over_sampling import SMOTE X df.drop(label, axis1) y df[label] # 先划分训练集和测试集stratify 保证比例一致 X_train, X_test, y_train, y_test train_test_split( X, y, test_size0.2, random_state42, stratifyy ) # 只在训练集上做 SMOTE smote SMOTE(random_state42, k_neighbors5) X_train_res, y_train_res smote.fit_resample(X_train, y_train) print(f原始训练集: {y_train.value_counts().to_dict()}) print(fSMOTE 后: {y_train_res.value_counts().to_dict()})k_neighbors5是 SMOTE 的默认近邻数如果少数类样本极少比如少于 10 条需要调小这个值否则会报错。另外SMOTE 对高维稀疏特征效果一般如果特征维度超过 100建议先做特征选择再采样。3. 模型选型CNN 和 LSTM 在流量特征上的真实表现3.1 为什么 1D-CNN 比 2D-CNN 更适合流特征很多同学一上来就用 2D-CNN把流量特征当成图片处理结果训练慢、效果差。流特征是一维向量每个特征代表一个统计量如包长均值、流持续时间特征之间没有空间邻域关系。用 1D-CNN 在特征维度上做卷积相当于自动学习局部特征组合效果更稳。下面是我常用的 1D-CNN 结构。import tensorflow as tf from tensorflow.keras import layers, models def build_1d_cnn(input_dim): model models.Sequential([ layers.Reshape((input_dim, 1), input_shape(input_dim,)), layers.Conv1D(64, 3, activationrelu, paddingsame), layers.BatchNormalization(), layers.MaxPooling1D(2), layers.Conv1D(128, 3, activationrelu, paddingsame), layers.BatchNormalization(), layers.GlobalAveragePooling1D(), layers.Dense(64, activationrelu), layers.Dropout(0.5), layers.Dense(1, activationsigmoid) ]) model.compile(optimizeradam, lossbinary_crossentropy, metrics[accuracy]) return model model build_1d_cnn(X_train_res.shape[1]) model.summary()这里Conv1D的卷积核大小设为 3表示每次看 3 个相邻特征。paddingsame保持输出维度不变。GlobalAveragePooling1D替代Flatten减少参数量降低过拟合风险。Dropout(0.5)在全连接层前丢弃一半神经元这是防止过拟合的关键。输入维度input_dim就是特征数量比如 CICFlowMeter 提取了 78 个特征就传 78。3.2 LSTM 处理流特征序列的坑有些同学想用 LSTM 捕捉流内数据包的时序关系思路是对的但直接把流统计特征当成序列输入 LSTM 就错了。流统计特征没有时间顺序LSTM 学不到东西。正确做法是用原始数据包序列每个包的长度、方向、到达间隔作为 LSTM 输入。这需要你从 pcap 里提取包级序列而不是流级统计。下面是一个简化的包序列提取示例。from scapy.all import rdpcap, IP, TCP, UDP def extract_packet_sequence(pcap_path, max_len50): packets rdpcap(pcap_path) seq [] for pkt in packets[:max_len]: if IP in pkt: length len(pkt) direction 1 if pkt[IP].src pkt[IP].dst else -1 if TCP in pkt: proto 6 elif UDP in pkt: proto 17 else: proto 0 seq.append([length, direction, proto]) # 补齐到 max_len while len(seq) max_len: seq.append([0, 0, 0]) return np.array(seq) # 生成包序列数据 X_seq np.array([extract_packet_sequence(p) for p in pcap_list])max_len50表示每条流只看前 50 个包这是速度和信息的折中。direction用源 IP 和目标 IP 的字符串比较来判断简单但有效。proto用数字编码协议类型。这个序列可以直接输入 LSTM 或 Transformer。但要注意包级序列提取比流级统计慢很多如果数据集有几十万条流建议先并行提取再训练。3.3 混合模型CNN 提特征 LSTM 学时序如果既想用流统计特征又想用包序列可以搭一个双输入模型。一个分支用 1D-CNN 处理流特征另一个分支用 LSTM 处理包序列最后拼接分类。这种结构在论文里好看实际效果取决于数据质量。我试过在 USTC-TFC2016 上混合模型比单 CNN 高 1-2 个百分点但训练时间翻倍。如果毕设时间紧建议先把单 CNN 调好。4. 训练与评估让指标不再“虚高”4.1 划分数据集时必须按时间或流 ID 划分很多同学随机划分数据集导致同一条流的不同片段同时出现在训练集和测试集指标虚高到 99%。正确做法是按流 ID 或时间划分同一会话的流只能出现在一个集合里。如果数据集没有流 ID可以按 pcap 文件划分比如 80% 的文件做训练20% 做测试。# 假设 df 里有 flow_id 列 unique_flows df[flow_id].unique() train_flows, test_flows train_test_split(unique_flows, test_size0.2, random_state42) train_df df[df[flow_id].isin(train_flows)] test_df df[df[flow_id].isin(test_flows)]如果没有flow_id可以用五元组源 IP、源端口、目的 IP、目的端口、协议生成一个唯一标识。这一步不做后面所有指标都是自欺欺人。4.2 评估指标别只看准确率恶意流量检测中漏报把恶意判成正常比误报更危险。所以要看召回率Recall和 F1 分数。下面是一个评估代码示例。from sklearn.metrics import classification_report, confusion_matrix y_pred (model.predict(X_test) 0.5).astype(int) print(confusion_matrix(y_test, y_pred)) print(classification_report(y_test, y_pred, target_names[正常, 恶意]))threshold0.5是默认阈值如果漏报太多可以调低到 0.3牺牲一点误报率换召回率。classification_report会输出每个类别的精确率、召回率和 F1。我一般要求恶意类的召回率不低于 0.95否则模型在实际场景中不可用。4.3 用 TensorBoard 看训练过程训练时不要只盯着最后的结果要看损失曲线。如果训练损失下降但验证损失上升说明过拟合需要加 Dropout 或早停。下面是一个带早停和 TensorBoard 的回调配置。from tensorflow.keras.callbacks import EarlyStopping, TensorBoard callbacks [ EarlyStopping(monitorval_loss, patience5, restore_best_weightsTrue), TensorBoard(log_dir./logs, histogram_freq1) ] history model.fit( X_train_res, y_train_res, validation_split0.2, epochs50, batch_size256, callbackscallbacks )patience5表示验证损失连续 5 轮不下降就停止。restore_best_weightsTrue会恢复最佳轮次的权重避免最后一轮过拟合。batch_size256是我在 16G 内存下的常用值如果显存不够可以降到 128。5. 避坑与排查那些让模型“翻车”的细节5.1 现象训练准确率 99%测试准确率 60%原因随机划分数据集导致同一条流的不同片段泄漏到测试集。解决按流 ID 或 pcap 文件划分确保同一会话只出现在一个集合。5.2 现象模型把几乎所有样本判为正常原因类别不平衡正常样本远多于恶意样本模型学到“全判正常”就能拿高准确率。解决用 SMOTE 过采样或在损失函数里加类别权重。class_weight{0:1, 1:10}可以让模型更关注恶意类。5.3 现象特征里有大量无穷值或 NaN原因CICFlowMeter 在计算流持续时间或包间隔时遇到零除或空流会生成inf或NaN。解决在预处理阶段用df.replace([np.inf, -np.inf], np.nan)然后fillna(0)。不要直接删行会丢失恶意样本。5.4 现象LSTM 训练极慢一个 epoch 要几小时原因包序列长度设得太大或者 batch size 太小。解决把max_len从 200 降到 50batch_size从 32 提到 128。如果还慢改用 1D-CNN 处理流特征放弃包序列。5.5 现象模型在验证集上 F1 很高但实际抓包测试时完全不能用原因训练数据和实际流量分布不一致。公开数据集是几年前的实际网络协议和流量模式已经变了。解决在毕设里加一个“在线测试”模块用 scapy 实时抓包并提取特征观察模型输出。如果偏差大说明模型过时了需要在论文里讨论局限性。6. 从毕设到可演示系统用 Flask 搭一个检测接口最后一章讲怎么把训练好的模型变成一个能演示的系统。答辩时老师不会只看代码他们想看到“输入一个 pcap输出检测结果”。我用 Flask 搭一个最小接口前端上传 pcap后端提取特征、跑模型、返回 JSON。from flask import Flask, request, jsonify import joblib import numpy as np from cicflowmeter import extract_features import os app Flask(__name__) model joblib.load(cnn_model.pkl) # 假设已保存 scaler joblib.load(scaler.pkl) app.route(/detect, methods[POST]) def detect(): file request.files[pcap] pcap_path temp.pcap file.save(pcap_path) # 提取特征 csv_path temp.csv extract_features(pcap_path, csv_path) import pandas as pd df pd.read_csv(csv_path) df.replace([np.inf, -np.inf], np.nan, inplaceTrue) df.fillna(0, inplaceTrue) # 确保特征列与训练时一致 feature_cols joblib.load(feature_cols.pkl) X df[feature_cols].values X scaler.transform(X) X X.reshape(X.shape[0], X.shape[1], 1) preds (model.predict(X) 0.5).astype(int).flatten() result { total_flows: len(preds), malicious_flows: int(preds.sum()), details: preds.tolist()[:20] # 只返回前20条 } os.remove(pcap_path) os.remove(csv_path) return jsonify(result) if __name__ __main__: app.run(host0.0.0.0, port5000)这段代码的关键点feature_cols.pkl保存了训练时的特征列顺序预测时必须按同样顺序取列否则模型输出全是乱码。scaler.pkl是训练时用的标准化器预测时也要用同一个。X.reshape把二维特征转成 1D-CNN 需要的三维输入。返回结果只给前 20 条详情避免响应过大。部署时用gunicorn -w 4 -b 0.0.0.0:5000 app:app启动比 Flask 自带服务器稳定。如果要在论文里展示可以再加一个简单的 HTML 上传页面用curl -F pcaptest.pcap http://localhost:5000/detect就能测试。我自己的习惯是每次改完模型先用一条已知的恶意 pcap 和一条正常 pcap 跑一遍接口确认输出符合预期再去看批量测试的指标。这个“单条验证”步骤帮我省了很多后悔药。希望帮到你。本文还有配套的精品资源点击获取