深入浅出CNN+LSTM：从pcap到网络流量识别系统实战

简介基于CNN和LSTM的流量分析识别系统完整项目资源面向人工智能与深度学习方向的学生、安全分析人员及竞赛备赛者。系统将思博伦官方流量pcap解析为URL后训练模型通过CNN提取流量的空间特征、LSTM提取时序特征实现正常业务流量、恶意软件流量与网络攻击流量的实时分类识别并支持随时序变化可视化展示在官方测试流量包上达到93.5%的准确率。压缩包共24个文件大小约23.58MB。内容以6个Python源码文件为核心覆盖数据预处理、CNN与LSTM模型搭建、训练和测试等环节另含csv数据、pkl/vocab辅助文件、pb格式的TensorFlow模型文件以及PDF版设计报告并附有使用说明与依赖清单便于快速复现和二次开发。资源已包含训练好的模型可直接加载运行省去重复训练成本源码与报告有助于理解流量特征提取和时空建模思路。目前已有572人学习下载适合希望完成流量识别项目、课程设计或入门深度学习时序分析的读者。1. 流量分析识别系统为什么需要 CNN 和 LSTM 同时参与网络流量分析识别系统如今已经不满足于“看端口猜协议”而是要把 pcap 里的会话流当成一条带空间排列和时间节奏的序列来看。标题里的 CNN 和 LSTM 也正是为此出现的CNN 负责从每个包的字节载荷里提取局部指纹LSTM 负责把包与包之间的先后关系、间隔和长度变化纳入判断。一个常见的落地场景是加密流量分类payload 里没有明文协议头能用的只有包长、到达间隔和握手阶段的字节分布这些都是单分类器难以同时处理的信息。我一般会这样理解交付包里三个组成部分源码是数据清洗、模型构造和训练脚本的统称训练好的模型是state_dict加网络定义才能一起加载的权重文件PDF 报告则记录特征选择、实验指标和设计取舍。三者在工程上缺一不可只给权重而没给预处理源码换个 pcap 就无法复现只给源码而没给模型训练动辄数小时评估成本太高。这篇文章不适合只想要公式推导的读者适合要自己动手搭流量识别系统、又不想从零踩坑的工程师。下文会把“为什么这么设计”和“具体怎么复现”并在一起讲代码按 PyTorch 写读者可以在自己的数据集上替换路径后直接运行。2. CNN 与 LSTM 在流量识别中的分工与配合2.1 卷积网络负责什么样的“空间特征”提取流量不是图像但可以变成图像一条会话流取前N个包每个包取前M个字节就能得到一个N乘M的整数矩阵。字节值除以 255 归一化后矩阵里相邻的字节组合和相邻的包顺序都携带信息。CNN 的卷积核在这里做的是局部模式发现TLS 握手阶段固定位置的 0x16 0x03 0x01HTTP 请求行里的方法名DNS 响应里固定的头部偏移都会被同一个卷积核捕捉到。使用 Conv1D 还是 Conv2D取决于你希望卷积核在哪个维度上共享。常见做法是用Conv1d按 payload 位置卷积因为每个包本身没有明显的二维空间结构也有团队把一条流的所有字节直接拉平成一维图再由全连接层接手但这样会丢掉包边界。如果把模型结构画成一张 CNN 结构图最稳定的形态是“卷积层、批归一化、ReLU、最大池化”卷积核大小取 3 或 5输出通道数先设 64后面再根据验证集 loss 增减。流量识别与图像分类有个不同点输入矩阵的每一行是同一个包的 payload行间顺序有意义但行内字节跨度过大时卷积核不应跨行扫描。所以我的实现里Conv1d只作用于单包 payload池化也在单包上做完成再把每个包压缩为固定长度的特征向量。这一步把原始流量从“字节矩阵”变成了“包级特征序列”正好交给 LSTM。2.2 LSTM 在会话流里学习“时间上下文”CNN 的输出只能看到每个包本身但一个会话是否异常往往取决于多个包之间的关联。例如一种木马通信会先发送 3 个短包再间隔 10 秒发送一个携带固定字节的包单包卷积完全看不出模式必须把时间步展开。LSTM 通过三个门控结构维护一个细胞状态遗忘门决定上一时间步的状态保留多少输入门决定当前包特征写入多少输出门决定最终输出哪个部分。常被问到的“lstm遗忘门的输入是什么数据”答案就是当前时间步的输入特征x_t和上一时间步的隐藏状态h_{t-1}拼接后经过一个带 sigmoid 的线性层它的取值范围在 0 到 1 之间因此能缓解长会话中的梯度消失问题。在这个系统里LSTM 的时间步就是一条流里的包序号。每个时间步的输入来自 CNN 输出的包级特征也可以额外拼接包长、相对时间戳。如果做在线识别LSTM 必须是单向的因为当前包到来时看不到未来的包如果做离线分析或流量包分析 CTF 那种事后取证可以启用bidirectionalTrue效果往往更好但参数量和 CPU/GPU 推理延迟都会上升。hidden_size 一般选 64 到 256超过 256 在流量这种中等长度序列上收益很小。一个值得注意的点LSTM 对输入尺度很敏感。CNN 输出的激活值如果不做归一化可能达到 10 以上导致 LSTM 输入门饱和。因此我会在 CNN 与 LSTM 之间加LayerNorm把特征稳定在一个范围这一步对收敛速度的影响比改 LSTM 层数更明显。在 LSTM 时间序列预测任务里很多人直接把原始数值喂进去再调学习率效果往往不如先做一次归一化再进模型。2.3 串行融合是最容易复现的结构CNNLSTM 的融合方式有并行、串行和带注意力机制的三种。并行是两个分支分别提取特征后拼接理论上能同时保留 payload 指纹和会话时序但训练时梯度需要同时经过两条路容易出现其中一个分支收敛过快而另一个没学明白。带注意力权重的方式更合适做协议级解释但要额外训练 Attention 层数据量不足时容易过拟合。因此我一般先做串行也就是“CNN 先卷积每个包LSTM 再按包顺序学习”等 baseline 稳定后再改并行也不迟。下面是一段可直接运行的串行结构定义输入形状为(batch, seq_len, payload_len)其中seq_len是设定的最大包数payload_len是设定的单包字节数。import torch import torch.nn as nn class StreamNet(nn.Module): def __init__(self, seq_len20, payload_len256, num_classes12): super().__init__() # 每个包的 payload 视为单通道信号做一维卷积 self.cnn nn.Sequential( nn.Conv1d(1, 64, kernel_size3, padding1), nn.BatchNorm1d(64), nn.ReLU(), nn.AdaptiveMaxPool1d(64), # 把每个包压成 64 维特征 ) # 包级特征序列交给 LSTM self.lstm nn.LSTM( input_size64, hidden_size128, num_layers1, batch_firstTrue ) self.classifier nn.Linear(128, num_classes) def forward(self, x): b, t, p x.size() x x.view(b * t, 1, p) # 方便 Conv1d 按包卷积 x self.cnn(x) # (b*t, 64, 1) 或自己调整池化输出 x x.view(b, t, -1) x, _ self.lstm(x) # (b, t, 128) last x[:, -1, :] # 取最后一个时间步 return self.classifier(last)这段代码里有个容易踩的坑AdaptiveMaxPool1d(64)作用于长度 256 的信号输出的第三维可能是 1需要确认实际 shape。更稳妥的做法是在conv后插入一个Flatten或者把AdaptiveMaxPool1d改成nn.AdaptiveAvgPool1d(1)这样每个包输出维度固定为 64。上面为了演示可读性保留了AdaptiveMaxPool1d(64)实际使用时要根据output_size检查维度。参数解释seq_len取 20意味着超过 20 个包的流的后续包会被截断少于 20 包的流用零填充payload_len取 256 能在大多数协议下保留完整头部。AdaptiveMaxPool1d不改变 batch 和通道维度只把时间维度收敛。输出last代表整条流结束后的隐藏状态用它做分类相当于“读完整个会话再下结论”而不是每个包都输出一个预测。表格式的选型建议如下设计点常用取法一句话理由卷积核大小3 或 5太小抓不到协议字段太大会混入无关字节Conv1d 通道数64 起步在过拟合和表达力之间取平衡LSTM hidden_size128足够建模常见会话状态LSTM 层数1流量序列通常不深两层以上收益小融合方式CNN - LSTM 串行训练稳定易于 debug这一章没有贴完整训练代码只解决模型结构问题。下一章将把预处理和训练真正接起来。3. 核心实现从 pcap 到可复现的 CNNLSTM 训练闭环3.1 预处理流水线把 pcap 切成定长会话样本训练数据不是一张张图片而是 pcap 文件。第一步需要决定“一条样本”代表什么。常见做法是以四元组源 IP、目的 IP、源端口、目的端口为键把双向包聚合成一个会话流并按时间排序。这样做的好处是 LSTM 能同时看到请求和响应学到正常的握手顺序如果按五元组带协议号客户端到服务器和服务器到客户端会被拆成两个样本流量特征反而被切断。下面的代码用 Scapy 读取 pcap输出固定形状的 numpy 数组。这是整个流程里最容易写错的一环建议单独放到preprocess.py里。import numpy as np from scapy.all import rdpcap, IP, TCP, UDP def pcap_to_session(path, max_pkts20, payload_len256): packets rdpcap(path) sessions {} for pkt in packets: if IP not in pkt: continue if TCP not in pkt and UDP not in pkt: continue src pkt[IP].src dst pkt[IP].dst sport pkt[TCP].sport if TCP in pkt else pkt[UDP].sport dport pkt[TCP].dport if TCP in pkt else pkt[UDP].dport key (src, dst, sport, dport) payload bytes(pkt[TCP].payload) if TCP in pkt else bytes(pkt[UDP].payload) sessions.setdefault(key, []).append(payload or b\x00 * 4) X [] for key, pkts in sessions.items(): seq pkts[:max_pkts] matrix np.zeros((max_pkts, payload_len), dtypenp.float32) for i, seg in enumerate(seq): raw np.frombuffer(seg[:payload_len], dtypenp.uint8) matrix[i, :len(raw)] raw X.append(matrix / 255.0) return np.array(X)逻辑说明rdpcap一次性将整个 pcap 读入内存适合离线预处理。若 pcap 超过 2GB建议改用PcapReader流式读取。key里的四元组忽略了 IP 方向所以客户端发出的包和服务端返回的包会进同一个集合。payload取TCP或UDP层之后的字节不包含 IP 和 TCP 头部因为头部里的端口、地址会被卷积核当成“协议特征”而不是“内容特征”。最后除以 255 是为了让输入落在 0 到 1 之间避免nn.Conv1d后面接ReLU时整体偏移。参数说明max_pkts控制序列长度选 20 对大多数短连接足够对 P2P 长流则会截断尾部这也是损失信息的源头。payload_len选 256 字节时能覆盖 DNS、NTP、MQTT 等短协议的全部内容但无法覆盖完整的 TLS 证书。实际中我看到很多项目喜欢把payload_len设到 1024训练集指标会略降因为大部分包的尾部都是零填充模型要花更多容量学会忽略“空字节”。所以优先保持 256不要盲目加大。补零时用matrix[i, :len(raw)] raw比用pad快因为不需要生成额外的 padding 数组。不过补零的位置等同于真实字节0x00模型可能会学到“这个位置出现了空字节”这种伪特征。更严谨的做法是额外维护一个 mask在训练时告诉模型哪些位置是无效填充但初版系统不必做观察混淆矩阵如果没有明显受影响就不加 mask。3.2 模型定义用 PyTorch 把 CNN 和 LSTM 拼成一个可训练模块预处理得到(batch, seq_len, payload_len)张量后可以接入第二章的StreamNet。为了训练稳定我在 CNN 块之后加一个LayerNorm在 LSTM 输出之后加Dropout。下面这段代码可以直接放到model.py配合上一章的网络骨架升级class StreamNetV2(nn.Module): def __init__(self, seq_len20, payload_len256, num_classes12, encoded_size64, hidden_size128, dropout0.3): super().__init__() self.cnn nn.Sequential( nn.Conv1d(1, 64, kernel_size3, padding1), nn.BatchNorm1d(64), nn.ReLU(), nn.AdaptiveAvgPool1d(1), # 每个包输出 64 维 ) self.norm nn.LayerNorm(64) self.lstm nn.LSTM(input_size64, hidden_sizehidden_size, num_layers1, batch_firstTrue) self.dropout nn.Dropout(dropout) self.classifier nn.Linear(hidden_size, num_classes) def forward(self, x): b, t, p x.size() x x.view(b * t, 1, p) feats self.cnn(x).squeeze(-1).view(b, t, -1) feats self.norm(feats) out, _ self.lstm(feats) out self.dropout(out[:, -1, :]) return self.classifier(out)这里把AdaptiveMaxPool1d换成AdaptiveAvgPool1d(1)每个包输出64维squeeze(-1)后变成(b, t, 64)。LayerNorm在时间步上做归一化保证 LSTM 的输入分布稳定Dropout只加在分类前防止全连接层记住训练样本。encoded_size在本例中没有显式使用目的是提醒读者如果想把包长、到达间隔也喂给 LSTM应该在feats后面再拼一维统计特征而不是直接改变卷积输出。与 Keras 写法对比PyTorch 的nn.LSTM返回一个(output, (h_n, c_n))的元组batch_firstTrue表示输入形状是(batch, seq_len, input_size)。如果忘了设置这个参数PyTorch 默认把seq_len放在第 1 维很多人在DataLoader出报错就是因此。还有hidden_size和input_size如果设得过大显存占用按线性增长流量识别这种任务不必追求超大模型。3.3 训练循环配置交叉熵、Adam、早停和类别权重模型定义好后训练流程按标准做法来。流量数据集通常类别不平衡比如 DNS 包数量远超视频流包数量所以loss要用class_weight加权。优化器用 Adam初始学习率1e-3batch size 设为 32 或 64。下面的训练循环固定了随机种子同时保存验证集 loss 最低的模型。torch.manual_seed(0) model StreamNetV2(num_classes12) class_counts np.array([1000, 50, 20, 300, 80, 10, 500, 200, 40, 25, 15, 60]) class_weight 1.0 / class_counts class_weight torch.tensor(class_weight, dtypetorch.float32) criterion nn.CrossEntropyLoss(weightclass_weight) optimizer torch.optim.Adam(model.parameters(), lr1e-3) best_loss float(inf) patience 5 wait 0 for epoch in range(30): model.train() for xb, yb in train_loader: optimizer.zero_grad() out model(xb) loss criterion(out, yb) loss.backward() optimizer.step() v_loss evaluate(model, val_loader) if v_loss best_loss: best_loss v_loss torch.save(model.state_dict(), best_model.pth) wait 0 else: wait 1 if wait patience: print(fearly stop at epoch {epoch}) breakevaluate需要单独实现可以简单复用torch.no_grad()计算平均验证 loss。CrossEntropyLoss的weight参数是每个类别的权重数量少的类别获得更大的梯度能避免模型把样本少的类别全部预测为多数类。torch.manual_seed(0)只固定 PyTorch 的随机数还要分别设置 Pythonrandom和 NumPy 的种子否则数据打乱顺序不稳定具体放到第五章讲。训练超参数可以参照下表超参数推荐值调整方向学习率1e-3loss 震荡时降至 5e-4batch size32显存不足时降为 16max_pkts20长会话任务增大到 30 到 50payload_len256加密识别可增加到 512到这里已经具备从原始 pcap 到训练完成的完整链路。但只有best_model.pth还不够还需要一套评估脚本否则无法回答“模型到底在哪些流量上失效”这个问题。4. 加载训练好的模型把准确率拆成可行动的信息4.1 加载权重并复现评估指标交付物里的best_model.pth只是权重文件必须与网络定义配套使用。先torch.load加载到对应设备再load_state_dict。以下代码假设模型文件与model.py在同一目录import torch from model import StreamNetV2 device torch.device(cuda if torch.cuda.is_available() else cpu) model StreamNetV2(num_classes12) state torch.load(best_model.pth, map_locationdevice) model.load_state_dict(state) model.to(device).eval() def predict_loader(model, loader): preds, trues [], [] with torch.no_grad(): for xb, yb in loader: out model(xb.to(device)) preds.extend(out.argmax(dim1).cpu().tolist()) trues.extend(yb.tolist()) return preds, trues参数说明torch.load的map_location不能省略否则开发机上保存的模型会尝试加载到默认 GPU在无 GPU 环境下直接报错。model.eval()会关闭 Dropout 和 BatchNorm 的训练状态这一步容易漏漏掉后预测结果会有随机性。argmax(dim1)得到每个样本的类别序号返回的是 numpy 列表可以直接交给 sklearn 计算指标。评估指标建议同时看macro-F1和weighted-F1不能只看 accuracy。流量识别里多数类别样本占比很高一个全部预测为“正常流量”的模型 accuracy 也能到 90%但小类别完全失效。weighted-F1反映整体macro-F1对小类更公平两者差距大就说明类别不平衡处理不到位。4.2 用混淆矩阵和逐类别指标定位误判边界sklearn.metrics直接产出三项指标和混淆矩阵。在项目里我会额外保存一个classification_report.txt方便报告引用。from sklearn.metrics import classification_report, confusion_matrix print(classification_report(trues, preds, digits3)) cm confusion_matrix(trues, preds) print(cm)逐类别指标可以定位两处问题一是某个类别 recall 低而 precision 高说明模型把这类流量识别成其他类需要增加该类样本或检查它的包长分布二是两个类别互相混淆常见于 DNS 和 QUIC 都使用 UDP 短包或 HTTPS 与未知加密协议都呈现“大包后面跟着大量零填充”。此时需要在报告里写清楚是特征选择问题还是样本标注问题避免简单归因于模型容量。流量包分析 CTF 里常做的一件事也是看混淆拿到一个样本后用模型 predict 输出一个带概率的类别再回到 pcap 用 Wireshark 手工验证协议关键字节确认模型不是靠偶然字节分类。这个操作在真实系统里对应可解释性验证是最快发现标签错误的手段。4.3 低置信度拒绝给未知流量留一个出口CNNLSTM 的输出是logits经过 softmax 后得到每个类别的概率。网络流量种类比训练集多得多遇到没见过的协议时强行分类反而会造成误报。常见做法是设一个最低置信度阈值低于它就把样本标记为unknown交给后续规则引擎或人工分析。logits model(xb) prob torch.softmax(logits, dim1) max_prob, pred torch.max(prob, dim1) # 阈值低就拒绝不进入三类分类结果 pred_cls torch.where(max_prob 0.6, -1, pred)torch.max返回(values, indices)torch.where(condition, x, y)在条件为真时取-1否则取pred。阈值0.6是一个起点通常根据验证集上的误报率调如果系统允许“宁可不报、不可错报”可以提高到 0.8如果目标是所有流量必须落入已知类别则需要直接关闭拒绝机制。阈值调好后把max_prob的分布画出来也能发现模型对哪些样本普遍犹豫这些样本往往就是没出现在训练集里的新协议。阈值行为适用场景0.6低于则 unknown通用分类系统0.8严格拒绝安全检测偏保守0.0关闭拒绝离线分析必须全分类依靠拒绝策略系统能在真实网络中把未知流量隔离出来继续收集而不是强迫 LSTM 硬交代一个答案。这样后续增量训练才有高质量负样本。5. 交付前用版本锁和固定随机种子重跑报告指标5.1 不要把“能跑”当成“可复现”拿到一个训练好的模型最容易踩的坑是在自己机器上加载后 accuracy 比报告里低 10 个百分点。常见原因不是权重损坏而是以下三处不一致训练时的随机种子没固定导致数据打乱和初始化不同PyTorch 或 CUDA 版本变动导致卷积算子顺序不同预处理函数里某个参数从max_pkts20改成了30输入形状完全变掉。我在每个项目里都会在训练脚本开头写上完整的随机种子固定代码import random import numpy as np import torch def seed_everything(seed0): random.seed(seed) np.random.seed(seed) torch.manual_seed(seed) torch.cuda.manual_seed_all(seed) torch.backends.cudnn.benchmark False torch.backends.cudnn.deterministic True同时把pip freeze requirements.txt放进交付目录安装时以环境实际版本为准。cudnn.benchmark False关闭自动调优保证相同输入在多次运行下产生一致结果代价是可能降低一点吞吐评估阶段可以开回True。另一个关键是对齐预处理参数。加载模型前先看model.py里构造函数的seq_len和payload_len如果和训练时不一致load_state_dict会报形状不匹配。更隐蔽的问题是预处理时把max_pkts改小了模型输入还是相同的20但数据里有大量补零推理结果自然不同。我的习惯是在preprocess.py里打印训练集和测试集的seq_len、payload_len然后把这两个值写进报告。如果要正式外发我会把训练好的模型用torch.jit.trace导出为 TorchScript这样不依赖 Python 的模型类定义。但这个动作对源码不熟悉的人有风险所以作为进阶手段而非默认方案。本文还有配套的精品资源点击获取