恶意加密流量检测毕设实战：从数据到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 跑通全流程，因为它的 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, inplace=True) df.fillna(0, inplace=True) # 标签列重命名，假设原始标签在 'label' 列 if 'label' in df.columns: df['label'] = df['label'].apply(lambda x: 1 if x != 'BENIGN' else 0) df.to_csv(output_csv, index=False) 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', axis=1) y = df['label'] # 先划分训练集和测试集，stratify 保证比例一致 X_train, X_test, y_train, y_test = train_test_split( X, y, test_size=0.2, random_state=42, stratify=y ) # 只在训练集上做 SMOTE smote = SMOTE(random_state=42, k_neighbors=5) X_train_res, y_train_res = smote.fit_resample(X_train, y_train) print(f'原始训练集: {y_train.value_counts().to_dict()}') print(f'SMOTE 后: {y_train_res.value_counts().to_dict()}')

k_neighbors=5是 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, activation='relu', padding='same'), layers.BatchNormalization(), layers.MaxPooling1D(2), layers.Conv1D(128, 3, activation='relu', padding='same'), layers.BatchNormalization(), layers.GlobalAveragePooling1D(), layers.Dense(64, activation='relu'), layers.Dropout(0.5), layers.Dense(1, activation='sigmoid') ]) model.compile(optimizer='adam', loss='binary_crossentropy', metrics=['accuracy']) return model model = build_1d_cnn(X_train_res.shape[1]) model.summary()

这里Conv1D的卷积核大小设为 3，表示每次看 3 个相邻特征。padding='same'保持输出维度不变。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_len=50): 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_len=50表示每条流只看前 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_size=0.2, random_state=42) 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=['正常', '恶意']))

threshold=0.5是默认阈值，如果漏报太多，可以调低到 0.3，牺牲一点误报率换召回率。classification_report会输出每个类别的精确率、召回率和 F1。我一般要求恶意类的召回率不低于 0.95，否则模型在实际场景中不可用。

4.3 用 TensorBoard 看训练过程

训练时不要只盯着最后的结果，要看损失曲线。如果训练损失下降但验证损失上升，说明过拟合，需要加 Dropout 或早停。下面是一个带早停和 TensorBoard 的回调配置。

from tensorflow.keras.callbacks import EarlyStopping, TensorBoard callbacks = [ EarlyStopping(monitor='val_loss', patience=5, restore_best_weights=True), TensorBoard(log_dir='./logs', histogram_freq=1) ] history = model.fit( X_train_res, y_train_res, validation_split=0.2, epochs=50, batch_size=256, callbacks=callbacks )

patience=5表示验证损失连续 5 轮不下降就停止。restore_best_weights=True会恢复最佳轮次的权重，避免最后一轮过拟合。batch_size=256是我在 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 降到 50，batch_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, inplace=True) df.fillna(0, inplace=True) # 确保特征列与训练时一致 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(host='0.0.0.0', port=5000)

这段代码的关键点：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 "pcap=@test.pcap" http://localhost:5000/detect就能测试。

我自己的习惯是：每次改完模型，先用一条已知的恶意 pcap 和一条正常 pcap 跑一遍接口，确认输出符合预期，再去看批量测试的指标。这个“单条验证”步骤帮我省了很多后悔药。希望帮到你。

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