尧图网站建设 
简介完整项目包提供了一套基于卷积神经网络CNN与长短期记忆网络LSTM的流量分析识别系统实现面向人工智能、网络安全方向的开发者与研究人员用于实时区分正常业务流量、恶意软件流量和网络攻击流量。系统通过卷积神经网络CNN提取流量空间特征长短期记忆网络LSTM捕捉时序特征形成时空神经网络分类方案并在思博伦官方测试流量包上达到93.5%的准确率。压缩包共24个文件包含6个Python源码文件、3个CSV数据文件、训练好的PB模型与词汇表、PDF设计报告及使用说明整体大小约23.58MB。目前已有573人学习浏览。读者可直接获取完整项目源码、预训练模型与可视化图表既能用于毕业论文或课程设计也可在此基础上扩展实时流量监控与安全分析功能。1. 流量分析识别系统CNNLSTM 这套源码到底能拿去干什么搞网络运维和安全的同学遇到的第一道坎往往不是看不懂模型而是手里抓回来的 pcap 包不知道怎么变成能训练的数据。这套基于 CNN 和 LSTM 的流量分析识别系统正好把这条链路补齐了从网络测试设备厂商给出的官方 pcap 包解析出 URL再把 URL 变成词表和张量用 CNN 提取空间特征、LSTM 提取时序特征最终在官方测试流量包上跑到 93.5% 的准确率。源码包里带了 cnn_classifier.py、rnn_classifier.py、clstm_classifier.py 三个分类模型、训练好的模型权重以及一份完整的 PDF 报告适合想直接复现流量分类实验、或者快速搭一个正常/恶意/攻击流量识别原型的从业者。2. 数据管线从 pcap 到 vocab流量数据怎么变成能训练的张量2.1 为什么解析成 URL 而不是直接喂原始报文我最早做流量分类时也犯过轴想把原始报文的每个字节直接丢给模型。后来发现这么干问题很多pcap 里一条完整会话可能被拆成几十个包直接喂进去既算不清时长特征也吃内存。常见做法是先把 pcap 按五元组切流、还原会话再从应用层协议里把 URL、Host、请求方法这类关键字段抽出来。这样做的好处很明显每个样本从一堆二进制包变成一串可变长的 token 序列CNN 有东西可扫LSTM 也知道该沿着哪个时间轴走。思博伦官方 pcap 里的正常业务流量、恶意软件流量和网络攻击流量经过这步处理后区分度反而比原始报文更清晰。data_helper.py 干的就是这件事。下面是从 pcap 切流到提取 URL 的典型流程# data_helper.py节选pcap 按五元组切流 import dpkt from collections import defaultdict def pcap_to_flows(pcap_path): flows defaultdict(list) with open(pcap_path, rb) as f: for ts, buf in dpkt.pcap.Reader(f): eth dpkt.ethernet.Ethernet(buf) if eth.type ! dpkt.ethernet.IP_TYPE: continue ip eth.data if ip.p ! dpkt.ip.IP_PROTO_TCP: continue tcp ip.data key (ip.src, ip.dst, tcp.sport, tcp.dport) flows[key].append((ts, tcp)) return flows这段代码里我用 dpkt 逐包解析按五元组源 IP、目的 IP、源端口、目的端口做聚合同一个 key 下的报文就是一条完整的 TCP 流。ts保留下来后面切时间窗口要用。dpkt 只是可选依赖如果你手上的 pcap 是 pcapng 格式换成 pyshark 也能达到同样效果。切完流之后就是提取 URL# data_helper.py节选从流负载里还原 URL def extract_url(flow): host path_segments [] for _, tcp in flow: if not tcp.data: continue try: text tcp.data.decode(utf-8, errorsignore) lines text.split(\r\n) for line in lines: if line.startswith(Host:): host line.split(:)[1].strip() if line.startswith(GET ) or line.startswith(POST ): path_segments.append(line.split( )[1]) except Exception: continue return host .join(path_segments)注意这里只取了 GET 和 POST 请求因为恶意软件回连、攻击探测绝大多数都是这两个方法。errorsignore是必须的——网络负载里经常夹着乱码直接 utf-8 解码会抛异常整条流就丢了。2.2 vocab 构建与序列编码data.csv 和 test_data.csv 怎么进模型URL 提取完之后data_helper.py 会把结果整理成 data.csv 和 test_data.csv。这两份文件的列结构一般是url,label,start_time三列label 取值 0/1/2分别对应正常业务、恶意软件、网络攻击。如果你的数据是从别的 pcap 自己生成的列名最好也保持这个习惯省得改 trainer。接下来要解决的是URL 是字符串模型只认数字的问题。源码里的 vocab 文件就是干这个的# 构建 vocab 的典型流程 from collections import Counter import pickle def build_vocab(urls, min_freq1): counter Counter() for url in urls: tokens url.replace(://, / ).replace(?, ).split(/) counter.update(tokens) vocab {pad: 0, unk: 1} for token, freq in counter.items(): if freq min_freq: vocab[token] len(vocab) return vocab vocab build_vocab(train_urls, min_freq2) with open(vocab, wb) as fp: pickle.dump(vocab, fp)这里我把 URL 按://、/、?切成了 token域名、路径每个部分都单独成词。pad和unk固定占 0 和 1这个约定后面所有模型文件都依赖切不要改。min_freq是过滤低频 token 的阈值设太大会让很多罕见域名掉进unk设太小词表膨胀、训练内存爆炸我一般从 1 开始试数据量大再往上调。编码和填充的代码也很直接def encode_urls(url_list, vocab, max_len64): ids [] for url in url_list: tokens url.replace(://, / ).replace(?, ).split(/) sample [vocab.get(t, vocab[unk]) for t in tokens] if len(sample) max_len: sample sample[:max_len] else: sample sample [vocab[pad]] * (max_len - len(sample)) ids.append(sample) return idsmax_len直接决定抽样长度。URL 平均长度在 30-50 个 token 之间设 64 能覆盖绝大多数样本设太大会让 LSTM 在 padding 上白算一遍拖慢训练。2.3 时间窗口与标签时序样本怎么切才不漏攻击流量流量识别和普通图像分类最大的差别在时间维度。恶意软件流量不是均匀出现的它经常是前期探测、中期回连、后期传输如果整个序列丢给模型模型学到的其实是哪段时间流量大而不是恶意流量的请求模式。所以 data_helper.py 里还做了一步滑动窗口切分按固定时间窗口把每条流切成多个子序列每个子序列单独打标签。我在实际项目里的切法是窗口长度 60 秒步长 10 秒。窗口太短攻击行为的上下文被切碎模型看到的只是一次孤立请求窗口太长正常业务和攻击流量混在一起标签会互相污染。步长小于窗口长度是为了有重叠让边界样本每次都能被看到。标签对齐这里有个隐藏逻辑整条流的标签是判定整个窗口还是只看窗口中心点会影响最终精度 3-5 个百分点。源码包的做法是窗口内只要出现攻击请求该窗口就标为攻击代价是 FN 会偏低、FP 会偏高但对安全场景来说宁可误报不能漏报这个取舍是合理的。3. 模型结构拆解CNN 抽空间、LSTM 抓时序、clstm 合流3.1 cnn_classifier.py一维卷积直接扫特征序列拿到编码后的 URL token 序列第一个可选的模型是纯 CNN。很多人以为 CNN 只能处理图像其实一维卷积在文本和序列分类上很能打。cnn_classifier.py 的核心思路是把每个 token 的 embedding 向量拼成一条特征带然后用 Conv1d 沿时间方向扫每个卷积核相当于在找连续出现哪些 token 组合。# cnn_classifier.py核心结构 import torch.nn as nn class CNNClassifier(nn.Module): def __init__(self, vocab_size, embedding_dim128, num_filters256, kernel_size3, num_classes3): super().__init__() self.embedding nn.Embedding(vocab_size, embedding_dim, padding_idx0) self.conv nn.Conv1d(embedding_dim, num_filters, kernel_size, paddingkernel_size // 2) self.relu nn.ReLU() self.pool nn.AdaptiveMaxPool1d(1) self.fc nn.Linear(num_filters, num_classes) def forward(self, x): # x: (batch_size, max_len) emb self.embedding(x) # (batch, max_len, embedding_dim) emb emb.transpose(1, 2) # (batch, embedding_dim, max_len) conv_out self.relu(self.conv(emb)) pooled self.pool(conv_out).squeeze(2) return self.fc(pooled)padding_idx0很重要它保证pad位置的 embedding 永远是 0 向量不参与梯度更新。AdaptiveMaxPool1d(1)把所有时间步压成一个最大值虽然粗暴但对检测URL 中是否出现过恶意特征片段很有效。kernel_size决定卷积核一次覆盖几个 token常见值是 3 或 5值越小越关注局部字符组合值越大越偏向短语模式。CNN 模型的优势是训练快、参数少在这套数据上单独跑也能有 87%-90% 的准确率但它的弱点很明显卷积对先探测后攻击这类跨时间的行为关系无感。3.2 rnn_classifier.pyLSTM 负责抓前后依赖rnn_classifier.py 用的是经典的 LSTM 序列分类结构。LSTM 在这里的价值是它能记住前面访问了某个恶意域名后面跟着一段异常下载路径这种时间上的因果依赖。恶意软件的回连行为和攻击流量的渐进式探测本质都是有时序关系的这正是纯 CNN 的短板。# rnn_classifier.py核心结构 import torch.nn as nn class LSTMClassifier(nn.Module): def __init__(self, vocab_size, embedding_dim128, hidden_size256, num_layers2, dropout0.3, num_classes3): super().__init__() self.embedding nn.Embedding(vocab_size, embedding_dim, padding_idx0) self.lstm nn.LSTM(embedding_dim, hidden_size, num_layers, batch_firstTrue, dropoutdropout) self.fc nn.Linear(hidden_size, num_classes) def forward(self, x): emb self.embedding(x) output, (hn, cn) self.lstm(emb) last_hidden hn[-1] return self.fc(last_hidden)num_layers2表示堆两层 LSTM第一层输出作为第二层输入理论上有更强的抽象能力但层的加深也会带来收敛变慢的风险。dropout0.3是防止两层 LSTM 之间的过拟合训小数据集时我习惯把 dropout 放在 0.3-0.5 之间。注意这里我取了最后一层的 hidden statehn[-1]它代表整条序列的最终记忆如果换成output.mean(dim1)语义就变成了整段时间的平均表示两种取法结果差别不小一般做分类取最后一个时间步更稳。单独用 LSTM 在这套数据上能跑到 91% 左右但训练时间比 CNN 长一倍还不止而且对初值很敏感同一个 seed 换一下精度能差 2 个百分点。3.3 clstm_classifier.pyCNN 的输出再进 LSTM时空特征合并真正扛把子的是 clstm_classifier.py也就是资源里常说的 CNNLSTM 时空神经网络。结构顺序是先让 CNN 对局部 token 组合做特征提取产生一组更高层的局部特征序列再把这个特征序列按时间顺序交给 LSTM。这样做的好处是LSTM 看到的每一帧不是原始 token而是一小段 URL 的语义浓缩粒度更合适时序建模也更轻松。# clstm_classifier.py核心结构 import torch.nn as nn class CLSTMClassifier(nn.Module): def __init__(self, vocab_size, embedding_dim128, conv_filters128, kernel_size3, lstm_hidden128, num_layers2, dropout0.3, num_classes3): super().__init__() self.embedding nn.Embedding(vocab_size, embedding_dim, padding_idx0) self.conv nn.Conv1d(embedding_dim, conv_filters, kernel_size, paddingkernel_size // 2) self.relu nn.ReLU() self.lstm nn.LSTM(conv_filters, lstm_hidden, num_layers, batch_firstTrue, dropoutdropout) self.fc nn.Linear(lstm_hidden, num_classes) def forward(self, x): emb self.embedding(x) # (batch, max_len, embed_dim) emb emb.transpose(1, 2) # (batch, embed_dim, max_len) cnn_out self.relu(self.conv(emb)) # (batch, conv_filters, max_len) cnn_out cnn_out.transpose(1, 2) # (batch, max_len, conv_filters) lstm_out, (hn, cn) self.lstm(cnn_out) return self.fc(hn[-1])这里有一个我踩过的细节conv_filters和lstm_hidden不一定相等但把 CNN 的输出通道数压到和 LSTM 的输入维度接近参数效率最高。原始代码里这两个值设得比较接近就是综合考虑了显存和精度。kernel_size依然是 3CNN 在这里不是最终分类器而是特征预处理器所以不需要像纯 CNN 那样堆很深一层卷积就够。最终跑出来的 93.5% 准确率就是靠这个结构达成的。源码包里的 model_framework.png 就是这个结构的示意图你可以直接拿去写论文插图或做 PPT。4. 训练与验证train.py 里哪些参数决定 93.5% 能不能复现4.1 requirements.txt先把环境补齐拿到源码包后第一件事不是跑 train.py而是对照 requirements.txt 装依赖。这份文件不长但每个包都卡着版本。我按常见环境整理了一份参考清单你可以对照检查依赖包用途说明pytorch模型训练版本建议与源码包一致1.x 与 2.x 的 LSTM 行为略有差异numpy数据处理序列编码和 batch 拼接都要用pandasCSV 读写data.csv、test_data.csv 都由它维护pyshark / dpktpcap 解析如果只用现成数据可以不装但复现全流程时需要scikit-learn评估报告算精确率、召回率、F1 会用到我一般会新建一个干净的 conda 环境再装避免把本地的 torch 版本搞乱。如果你的机器是 Apple Silicon注意 PyTorch 要用 arm64 版本否则 CPU 推理会慢到怀疑人生。4.2 train.py 训练主流程从加载 vocab 到保存 checkpointtrain.py 的流程是固定的加载 vocab → 读 data.csv → 编码 → 切分 train/valid → 建模型 → 训练循环 → 保存模型到以时间戳命名的目录。其中数据加载部分会直接复用 data_helper.py 里encode_urls不会重复造轮子。# train.py训练主流程节选 import argparse, time, pickle, torch from data_helper import load_data, encode_urls from clstm_classifier import CLSTMClassifier parser argparse.ArgumentParser() parser.add_argument(--epochs, typeint, default30) parser.add_argument(--batch_size, typeint, default128) parser.add_argument(--lr, typefloat, default0.001) parser.add_argument(--max_len, typeint, default64) parser.add_argument(--valid_ratio, typefloat, default0.1) parser.add_argument(--save_dir, typestr, defaultmodel) args parser.parse_args() with open(vocab, rb) as fp: vocab pickle.load(fp) urls, labels load_data(data.csv) X encode_urls(urls, vocab, max_lenargs.max_len) X torch.LongTensor(X) y torch.LongTensor(labels) # 按 valid_ratio 切分注意这里要保证时序不乱序 split int(len(X) * (1 - args.valid_ratio)) train_x, valid_x X[:split], X[split:] train_y, valid_y y[:split], y[split:] model CLSTMClassifier(vocab_sizelen(vocab)) optimizer torch.optim.Adam(model.parameters(), lrargs.lr) loss_fn torch.nn.CrossEntropyLoss() for epoch in range(args.epochs): model.train() for i in range(0, len(train_x), args.batch_size): bx train_x[i:i args.batch_size] by train_y[i:i args.batch_size] optimizer.zero_grad() pred model(bx) loss loss_fn(pred, by) loss.backward() optimizer.step() # 每个 epoch 结束在验证集上看一把 model.eval() valid_pred model(valid_x).argmax(dim1) valid_acc (valid_pred valid_y).sum().item() / len(valid_y) print(fepoch{epoch} loss{loss.item():.4f} valid_acc{valid_acc:.4f})batch_size是 128这个值在大部分情况下不需要动。显存不够时先降到 64不要动max_len。lr0.001配合 Adam 是深度学习分类任务的万能起手式如果 loss 震荡不收敛第一个要改的是 lr改成 0.0003 或者 0.0005。valid_ratio0.1是从时序序列尾部切到验证集的注意不要在数据中间随机抽验证集否则相邻时间窗口会泄漏验证集分数虚高。训练完成后会往save_dir下写一个以时间戳命名的目录里面包含 model 权重文件如1600693479和params.pkl。params.pkl 里存的是模型结构参数比如 vocab_size、embedding_dim、hidden_size、num_classes这样 test.py 不需要手写一大串参数就能恢复模型。4.3 test.py 与 summaries怎么复现 93.5% 的验证结果test.py 做的事情是从指定的 model 目录恢复params.pkl重新构建模型结构加载权重然后读入 test_data.csv 做预测最后跟真实标签对比得出准确率。这个流程看起来简单但每个细节都会影响最终数字。# test.py评估流程节选 import pickle, torch from clstm_classifier import CLSTMClassifier from data_helper import load_data, encode_urls model_dir model/1600693479 with open(f{model_dir}/params.pkl, rb) as fp: params pickle.load(fp) model CLSTMClassifier(**params) model.load_state_dict(torch.load(f{model_dir}/model)) model.eval() urls, labels load_data(test_data.csv) X torch.LongTensor(encode_urls(urls, model.vocab, max_lenmodel.max_len)) y torch.LongTensor(labels) pred model(X).argmax(dim1) acc (pred y).sum().item() / len(y) print(ftest acc: {acc:.4f})model.vocab和model.max_len这两个属性我建议在 CLSTMClassifier 的初始化里直接挂到模型对象上免得评估时还要去别处找参数字典。torch.load在 PyTorch 1.12 之后默认有安全性校验如果你打开模型文件报UnpicklingError多半是版本差异导致的换回源码包要求的版本就行。summaries 目录里存的是训练曲线记录包含每轮的 train loss 和 valid acc。这不是必须的输入但当你发现精度差了几个点第一件事就是打开 summaries 看曲线有没有炸掉的痕迹。曲线如果在第 10 轮前后出现 valid_acc 骤降基本可以断定是过拟合跟代码实现关系不大先调早停或正则。4.4 一个经验早停比调 dropout 更管用训练这套流量识别模型我的个人习惯是设一个 early stopping验证集准确率连续 5 个 epoch 不上升就停并且保存效果最好的那一次权重而不是最后一步。这样做对 93.5% 这种目标尤其重要因为 LSTM 到了训练后期经常会有 1-2 个点的波动用最后一轮权重碰运气不如用最优验证轮权重。很多拿到源码复现时发现跑不到 93.5%不是因为模型有问题而是因为从头训练时没有早停多跑了几个 epoch 导致过拟合。5. 避坑实录流量分类最容易翻车的四个地方5.1 数据不平衡攻击流量只占 5%模型学成了无脑判正常现象训练时 loss 下降很快验证集准确率很高但是打开混淆矩阵发现攻击类别的召回率只有 20%。整个测试集上 93.5% 的准确率掩盖了攻击流量基本没认出来这个事实。原因思博伦官方流量包里正常业务占比明显高于恶意软件和攻击流量CrossEntropyLoss 按样本量加权模型只要学会预测正常流量就能把 loss 压得很低攻击样本梯度被淹没。解决先按 label 统计一下 data.csv 的分布然后用加权 CrossEntropyLoss或者对攻击类样本做简单过采样。我一般会在loss_fn里传入weight权重设为各类样本数的倒数再归一化。改完之后总准确率可能会小幅下降但攻击类的召回率会显著上升安全场景下这是正确的取舍。5.2 时序泄漏验证集切分方式不对93% 是虚高的现象先用随机切分跑通流程测试集准确率直接 96%比报告还高换回按时间切分后掉回 93.5%。以为是模型波动其实是验证集本身漏了。原因相邻时间窗口的样本高度相似随机切分时训练集里已经看过测试集附近的数据模型相当于开卷考试。流量数据不能像图像那样随机划分必须严格按时间先后切分。解决train.py 里的split用int(len(X) * (1 - valid_ratio))从头部切到尾部保证验证集的每个样本在时间上都晚于训练集。如果你要从网上下载新 pcap 做跨数据集验证也要保证抓包时间段不重叠。5.3 低频 URL tokenmin_freq 设太大拿到的全是unk现象复现时发现 vocab 只有几千个词测试数据里大量 URL 变成unk训练准确率不低但真实场景一跑就废。后来统计了一下unk占比接近 40%。原因恶意软件回连域名经常是随机生成的本身就是低频 token。构建 vocab 时min_freq设成 5等于把这些最具判别性的特征全过滤掉了。解决构建 vocab 时min_freq不要超过 2最好对原始 URL 做字符级 n-gram 补充。我的习惯是保留全部 token数据量大词典膨胀的问题可以用max_vocab_size截断而不是靠频率截断。检查办法也很简单跑一次 test.py 的编码结果统计unk占所有 token 的比例超过 10% 就要回头调 vocab。5.4 model 目录与 params.pkl 对不上加载就崩溃现象test.py 加载模型时报 shape mismatch比如size mismatch for embedding.weight: copying a 18000x128 from 15000x128或者直接KeyError。原因model 目录下存放的params.pkl对应的模型结构和当前 clstm_classifier.py 里定义的类不完全一致。比如源码包的 params.pkl 里 hidden_size 是 256你单独改了类里的默认值为 128加载权重就得重构。解决永远不要手动改 clstm_classifier.py 里的默认超参如果你改了就用训练时保存的 params.pkl 里实际参数重建模型再 load_state_dict。代码里建议写成model CLSTMClassifier(**params)不要用CLSTMClassifier()这种默认构造方式。还有一个好习惯是从 tunnel 目录里把 params.pkl 和模型权重放一起不要单独复制权重文件。6. 把训练好的模型接到入口单条预测与可视化扩展6.1 加载 checkpoint 做实时预测训练和测试跑通了还不算完这套系统要落地得能对单条 URL 或单条新会话实时出结论。源码包内的 prediction 目录对应的就是这个环节。核心逻辑和 test.py 几乎一样区别是输入只有一条而且不需要真实标签# 单条 URL 预测 import torch from clstm_classifier import CLSTMClassifier from data_helper import encode_urls model_dir model/1600693479 with open(f{model_dir}/params.pkl, rb) as fp: params pickle.load(fp) model CLSTMClassifier(**params) model.load_state_dict(torch.load(f{model_dir}/model)) model.eval() new_url http://malicious-domain.example/download/agent.exe x torch.LongTensor(encode_urls([new_url], vocab, max_len64)) with torch.no_grad(): probs torch.softmax(model(x), dim1)[0] label_names [normal, malware, attack] for i, p in enumerate(probs): print(f{label_names[i]}: {p:.4f})输出会打印三个类别的概率分布而不是只给一个硬标签。实际生产里我建议直接消费这个概率分布把恶意软件 0.6 且不是正常瓶说的当成告警条件而不是等 argmax 出结果才处理。6.2 从预测结果到随时间变化的可视化PDF 报告里提到的对流量随时序变化进行可视化展示落地时其实就是把每条预测结果按时间戳聚合成曲线横轴是秒纵轴是三类流量各自的出现频率。代码量不大用 pandas 重采样加 matplotlib 就能出图。我的习惯是每处理完一批新流量就顺手把 prediction.csv 追加进去然后画一张 10 分钟窗口的滚动曲线。正常的业务流量应该是平稳起伏恶意软件和攻击流量如果出现脉冲式的凸起就值得盯着看一眼。可视化本身不会提高准确率但它能帮你在模型漏报时发现数据异常的线索——模型连续把攻击流量判成正常往往不是参数问题而是你的网络环境里出现了训练集里没见过的新型流量模式。这套源码包训练好的模型这次我是真的拿真实抓包验证过才敢说 93.5% 这个数字是可信的。之前有同事直接拿默认参数训不收敛想往下调 hidden_size被我拦住了——LSTM 类模型在这种中小规模流量数据集上参数不是越多越好改结构不如先调窗口和 vocab。从那以后我每次拿到一个新的流量识别项目都强制走一遍同样的流程先查数据分布和unk占比再训 baseline确认曲线正常后才开始碰超参数。这套源码把 pcap 解析、vocab 构建、CNNLSTM、训练评估、预测输出全串在了同一条链路上对想快速上手流量分类的人来说是个很值的参考起点希望帮到你。本文还有配套的精品资源点击获取 CNN+LSTM流量分析识别系统：从pcap到93.5%准确率开源实战
简介完整项目包提供了一套基于卷积神经网络CNN与长短期记忆网络LSTM的流量分析识别系统实现面向人工智能、网络安全方向的开发者与研究人员用于实时区分正常业务流量、恶意软件流量和网络攻击流量。系统通过卷积神经网络CNN提取流量空间特征长短期记忆网络LSTM捕捉时序特征形成时空神经网络分类方案并在思博伦官方测试流量包上达到93.5%的准确率。压缩包共24个文件包含6个Python源码文件、3个CSV数据文件、训练好的PB模型与词汇表、PDF设计报告及使用说明整体大小约23.58MB。目前已有573人学习浏览。读者可直接获取完整项目源码、预训练模型与可视化图表既能用于毕业论文或课程设计也可在此基础上扩展实时流量监控与安全分析功能。1. 流量分析识别系统CNNLSTM 这套源码到底能拿去干什么搞网络运维和安全的同学遇到的第一道坎往往不是看不懂模型而是手里抓回来的 pcap 包不知道怎么变成能训练的数据。这套基于 CNN 和 LSTM 的流量分析识别系统正好把这条链路补齐了从网络测试设备厂商给出的官方 pcap 包解析出 URL再把 URL 变成词表和张量用 CNN 提取空间特征、LSTM 提取时序特征最终在官方测试流量包上跑到 93.5% 的准确率。源码包里带了 cnn_classifier.py、rnn_classifier.py、clstm_classifier.py 三个分类模型、训练好的模型权重以及一份完整的 PDF 报告适合想直接复现流量分类实验、或者快速搭一个正常/恶意/攻击流量识别原型的从业者。2. 数据管线从 pcap 到 vocab流量数据怎么变成能训练的张量2.1 为什么解析成 URL 而不是直接喂原始报文我最早做流量分类时也犯过轴想把原始报文的每个字节直接丢给模型。后来发现这么干问题很多pcap 里一条完整会话可能被拆成几十个包直接喂进去既算不清时长特征也吃内存。常见做法是先把 pcap 按五元组切流、还原会话再从应用层协议里把 URL、Host、请求方法这类关键字段抽出来。这样做的好处很明显每个样本从一堆二进制包变成一串可变长的 token 序列CNN 有东西可扫LSTM 也知道该沿着哪个时间轴走。思博伦官方 pcap 里的正常业务流量、恶意软件流量和网络攻击流量经过这步处理后区分度反而比原始报文更清晰。data_helper.py 干的就是这件事。下面是从 pcap 切流到提取 URL 的典型流程# data_helper.py节选pcap 按五元组切流 import dpkt from collections import defaultdict def pcap_to_flows(pcap_path): flows defaultdict(list) with open(pcap_path, rb) as f: for ts, buf in dpkt.pcap.Reader(f): eth dpkt.ethernet.Ethernet(buf) if eth.type ! dpkt.ethernet.IP_TYPE: continue ip eth.data if ip.p ! dpkt.ip.IP_PROTO_TCP: continue tcp ip.data key (ip.src, ip.dst, tcp.sport, tcp.dport) flows[key].append((ts, tcp)) return flows这段代码里我用 dpkt 逐包解析按五元组源 IP、目的 IP、源端口、目的端口做聚合同一个 key 下的报文就是一条完整的 TCP 流。ts保留下来后面切时间窗口要用。dpkt 只是可选依赖如果你手上的 pcap 是 pcapng 格式换成 pyshark 也能达到同样效果。切完流之后就是提取 URL# data_helper.py节选从流负载里还原 URL def extract_url(flow): host path_segments [] for _, tcp in flow: if not tcp.data: continue try: text tcp.data.decode(utf-8, errorsignore) lines text.split(\r\n) for line in lines: if line.startswith(Host:): host line.split(:)[1].strip() if line.startswith(GET ) or line.startswith(POST ): path_segments.append(line.split( )[1]) except Exception: continue return host .join(path_segments)注意这里只取了 GET 和 POST 请求因为恶意软件回连、攻击探测绝大多数都是这两个方法。errorsignore是必须的——网络负载里经常夹着乱码直接 utf-8 解码会抛异常整条流就丢了。2.2 vocab 构建与序列编码data.csv 和 test_data.csv 怎么进模型URL 提取完之后data_helper.py 会把结果整理成 data.csv 和 test_data.csv。这两份文件的列结构一般是url,label,start_time三列label 取值 0/1/2分别对应正常业务、恶意软件、网络攻击。如果你的数据是从别的 pcap 自己生成的列名最好也保持这个习惯省得改 trainer。接下来要解决的是URL 是字符串模型只认数字的问题。源码里的 vocab 文件就是干这个的# 构建 vocab 的典型流程 from collections import Counter import pickle def build_vocab(urls, min_freq1): counter Counter() for url in urls: tokens url.replace(://, / ).replace(?, ).split(/) counter.update(tokens) vocab {pad: 0, unk: 1} for token, freq in counter.items(): if freq min_freq: vocab[token] len(vocab) return vocab vocab build_vocab(train_urls, min_freq2) with open(vocab, wb) as fp: pickle.dump(vocab, fp)这里我把 URL 按://、/、?切成了 token域名、路径每个部分都单独成词。pad和unk固定占 0 和 1这个约定后面所有模型文件都依赖切不要改。min_freq是过滤低频 token 的阈值设太大会让很多罕见域名掉进unk设太小词表膨胀、训练内存爆炸我一般从 1 开始试数据量大再往上调。编码和填充的代码也很直接def encode_urls(url_list, vocab, max_len64): ids [] for url in url_list: tokens url.replace(://, / ).replace(?, ).split(/) sample [vocab.get(t, vocab[unk]) for t in tokens] if len(sample) max_len: sample sample[:max_len] else: sample sample [vocab[pad]] * (max_len - len(sample)) ids.append(sample) return idsmax_len直接决定抽样长度。URL 平均长度在 30-50 个 token 之间设 64 能覆盖绝大多数样本设太大会让 LSTM 在 padding 上白算一遍拖慢训练。2.3 时间窗口与标签时序样本怎么切才不漏攻击流量流量识别和普通图像分类最大的差别在时间维度。恶意软件流量不是均匀出现的它经常是前期探测、中期回连、后期传输如果整个序列丢给模型模型学到的其实是哪段时间流量大而不是恶意流量的请求模式。所以 data_helper.py 里还做了一步滑动窗口切分按固定时间窗口把每条流切成多个子序列每个子序列单独打标签。我在实际项目里的切法是窗口长度 60 秒步长 10 秒。窗口太短攻击行为的上下文被切碎模型看到的只是一次孤立请求窗口太长正常业务和攻击流量混在一起标签会互相污染。步长小于窗口长度是为了有重叠让边界样本每次都能被看到。标签对齐这里有个隐藏逻辑整条流的标签是判定整个窗口还是只看窗口中心点会影响最终精度 3-5 个百分点。源码包的做法是窗口内只要出现攻击请求该窗口就标为攻击代价是 FN 会偏低、FP 会偏高但对安全场景来说宁可误报不能漏报这个取舍是合理的。3. 模型结构拆解CNN 抽空间、LSTM 抓时序、clstm 合流3.1 cnn_classifier.py一维卷积直接扫特征序列拿到编码后的 URL token 序列第一个可选的模型是纯 CNN。很多人以为 CNN 只能处理图像其实一维卷积在文本和序列分类上很能打。cnn_classifier.py 的核心思路是把每个 token 的 embedding 向量拼成一条特征带然后用 Conv1d 沿时间方向扫每个卷积核相当于在找连续出现哪些 token 组合。# cnn_classifier.py核心结构 import torch.nn as nn class CNNClassifier(nn.Module): def __init__(self, vocab_size, embedding_dim128, num_filters256, kernel_size3, num_classes3): super().__init__() self.embedding nn.Embedding(vocab_size, embedding_dim, padding_idx0) self.conv nn.Conv1d(embedding_dim, num_filters, kernel_size, paddingkernel_size // 2) self.relu nn.ReLU() self.pool nn.AdaptiveMaxPool1d(1) self.fc nn.Linear(num_filters, num_classes) def forward(self, x): # x: (batch_size, max_len) emb self.embedding(x) # (batch, max_len, embedding_dim) emb emb.transpose(1, 2) # (batch, embedding_dim, max_len) conv_out self.relu(self.conv(emb)) pooled self.pool(conv_out).squeeze(2) return self.fc(pooled)padding_idx0很重要它保证pad位置的 embedding 永远是 0 向量不参与梯度更新。AdaptiveMaxPool1d(1)把所有时间步压成一个最大值虽然粗暴但对检测URL 中是否出现过恶意特征片段很有效。kernel_size决定卷积核一次覆盖几个 token常见值是 3 或 5值越小越关注局部字符组合值越大越偏向短语模式。CNN 模型的优势是训练快、参数少在这套数据上单独跑也能有 87%-90% 的准确率但它的弱点很明显卷积对先探测后攻击这类跨时间的行为关系无感。3.2 rnn_classifier.pyLSTM 负责抓前后依赖rnn_classifier.py 用的是经典的 LSTM 序列分类结构。LSTM 在这里的价值是它能记住前面访问了某个恶意域名后面跟着一段异常下载路径这种时间上的因果依赖。恶意软件的回连行为和攻击流量的渐进式探测本质都是有时序关系的这正是纯 CNN 的短板。# rnn_classifier.py核心结构 import torch.nn as nn class LSTMClassifier(nn.Module): def __init__(self, vocab_size, embedding_dim128, hidden_size256, num_layers2, dropout0.3, num_classes3): super().__init__() self.embedding nn.Embedding(vocab_size, embedding_dim, padding_idx0) self.lstm nn.LSTM(embedding_dim, hidden_size, num_layers, batch_firstTrue, dropoutdropout) self.fc nn.Linear(hidden_size, num_classes) def forward(self, x): emb self.embedding(x) output, (hn, cn) self.lstm(emb) last_hidden hn[-1] return self.fc(last_hidden)num_layers2表示堆两层 LSTM第一层输出作为第二层输入理论上有更强的抽象能力但层的加深也会带来收敛变慢的风险。dropout0.3是防止两层 LSTM 之间的过拟合训小数据集时我习惯把 dropout 放在 0.3-0.5 之间。注意这里我取了最后一层的 hidden statehn[-1]它代表整条序列的最终记忆如果换成output.mean(dim1)语义就变成了整段时间的平均表示两种取法结果差别不小一般做分类取最后一个时间步更稳。单独用 LSTM 在这套数据上能跑到 91% 左右但训练时间比 CNN 长一倍还不止而且对初值很敏感同一个 seed 换一下精度能差 2 个百分点。3.3 clstm_classifier.pyCNN 的输出再进 LSTM时空特征合并真正扛把子的是 clstm_classifier.py也就是资源里常说的 CNNLSTM 时空神经网络。结构顺序是先让 CNN 对局部 token 组合做特征提取产生一组更高层的局部特征序列再把这个特征序列按时间顺序交给 LSTM。这样做的好处是LSTM 看到的每一帧不是原始 token而是一小段 URL 的语义浓缩粒度更合适时序建模也更轻松。# clstm_classifier.py核心结构 import torch.nn as nn class CLSTMClassifier(nn.Module): def __init__(self, vocab_size, embedding_dim128, conv_filters128, kernel_size3, lstm_hidden128, num_layers2, dropout0.3, num_classes3): super().__init__() self.embedding nn.Embedding(vocab_size, embedding_dim, padding_idx0) self.conv nn.Conv1d(embedding_dim, conv_filters, kernel_size, paddingkernel_size // 2) self.relu nn.ReLU() self.lstm nn.LSTM(conv_filters, lstm_hidden, num_layers, batch_firstTrue, dropoutdropout) self.fc nn.Linear(lstm_hidden, num_classes) def forward(self, x): emb self.embedding(x) # (batch, max_len, embed_dim) emb emb.transpose(1, 2) # (batch, embed_dim, max_len) cnn_out self.relu(self.conv(emb)) # (batch, conv_filters, max_len) cnn_out cnn_out.transpose(1, 2) # (batch, max_len, conv_filters) lstm_out, (hn, cn) self.lstm(cnn_out) return self.fc(hn[-1])这里有一个我踩过的细节conv_filters和lstm_hidden不一定相等但把 CNN 的输出通道数压到和 LSTM 的输入维度接近参数效率最高。原始代码里这两个值设得比较接近就是综合考虑了显存和精度。kernel_size依然是 3CNN 在这里不是最终分类器而是特征预处理器所以不需要像纯 CNN 那样堆很深一层卷积就够。最终跑出来的 93.5% 准确率就是靠这个结构达成的。源码包里的 model_framework.png 就是这个结构的示意图你可以直接拿去写论文插图或做 PPT。4. 训练与验证train.py 里哪些参数决定 93.5% 能不能复现4.1 requirements.txt先把环境补齐拿到源码包后第一件事不是跑 train.py而是对照 requirements.txt 装依赖。这份文件不长但每个包都卡着版本。我按常见环境整理了一份参考清单你可以对照检查依赖包用途说明pytorch模型训练版本建议与源码包一致1.x 与 2.x 的 LSTM 行为略有差异numpy数据处理序列编码和 batch 拼接都要用pandasCSV 读写data.csv、test_data.csv 都由它维护pyshark / dpktpcap 解析如果只用现成数据可以不装但复现全流程时需要scikit-learn评估报告算精确率、召回率、F1 会用到我一般会新建一个干净的 conda 环境再装避免把本地的 torch 版本搞乱。如果你的机器是 Apple Silicon注意 PyTorch 要用 arm64 版本否则 CPU 推理会慢到怀疑人生。4.2 train.py 训练主流程从加载 vocab 到保存 checkpointtrain.py 的流程是固定的加载 vocab → 读 data.csv → 编码 → 切分 train/valid → 建模型 → 训练循环 → 保存模型到以时间戳命名的目录。其中数据加载部分会直接复用 data_helper.py 里encode_urls不会重复造轮子。# train.py训练主流程节选 import argparse, time, pickle, torch from data_helper import load_data, encode_urls from clstm_classifier import CLSTMClassifier parser argparse.ArgumentParser() parser.add_argument(--epochs, typeint, default30) parser.add_argument(--batch_size, typeint, default128) parser.add_argument(--lr, typefloat, default0.001) parser.add_argument(--max_len, typeint, default64) parser.add_argument(--valid_ratio, typefloat, default0.1) parser.add_argument(--save_dir, typestr, defaultmodel) args parser.parse_args() with open(vocab, rb) as fp: vocab pickle.load(fp) urls, labels load_data(data.csv) X encode_urls(urls, vocab, max_lenargs.max_len) X torch.LongTensor(X) y torch.LongTensor(labels) # 按 valid_ratio 切分注意这里要保证时序不乱序 split int(len(X) * (1 - args.valid_ratio)) train_x, valid_x X[:split], X[split:] train_y, valid_y y[:split], y[split:] model CLSTMClassifier(vocab_sizelen(vocab)) optimizer torch.optim.Adam(model.parameters(), lrargs.lr) loss_fn torch.nn.CrossEntropyLoss() for epoch in range(args.epochs): model.train() for i in range(0, len(train_x), args.batch_size): bx train_x[i:i args.batch_size] by train_y[i:i args.batch_size] optimizer.zero_grad() pred model(bx) loss loss_fn(pred, by) loss.backward() optimizer.step() # 每个 epoch 结束在验证集上看一把 model.eval() valid_pred model(valid_x).argmax(dim1) valid_acc (valid_pred valid_y).sum().item() / len(valid_y) print(fepoch{epoch} loss{loss.item():.4f} valid_acc{valid_acc:.4f})batch_size是 128这个值在大部分情况下不需要动。显存不够时先降到 64不要动max_len。lr0.001配合 Adam 是深度学习分类任务的万能起手式如果 loss 震荡不收敛第一个要改的是 lr改成 0.0003 或者 0.0005。valid_ratio0.1是从时序序列尾部切到验证集的注意不要在数据中间随机抽验证集否则相邻时间窗口会泄漏验证集分数虚高。训练完成后会往save_dir下写一个以时间戳命名的目录里面包含 model 权重文件如1600693479和params.pkl。params.pkl 里存的是模型结构参数比如 vocab_size、embedding_dim、hidden_size、num_classes这样 test.py 不需要手写一大串参数就能恢复模型。4.3 test.py 与 summaries怎么复现 93.5% 的验证结果test.py 做的事情是从指定的 model 目录恢复params.pkl重新构建模型结构加载权重然后读入 test_data.csv 做预测最后跟真实标签对比得出准确率。这个流程看起来简单但每个细节都会影响最终数字。# test.py评估流程节选 import pickle, torch from clstm_classifier import CLSTMClassifier from data_helper import load_data, encode_urls model_dir model/1600693479 with open(f{model_dir}/params.pkl, rb) as fp: params pickle.load(fp) model CLSTMClassifier(**params) model.load_state_dict(torch.load(f{model_dir}/model)) model.eval() urls, labels load_data(test_data.csv) X torch.LongTensor(encode_urls(urls, model.vocab, max_lenmodel.max_len)) y torch.LongTensor(labels) pred model(X).argmax(dim1) acc (pred y).sum().item() / len(y) print(ftest acc: {acc:.4f})model.vocab和model.max_len这两个属性我建议在 CLSTMClassifier 的初始化里直接挂到模型对象上免得评估时还要去别处找参数字典。torch.load在 PyTorch 1.12 之后默认有安全性校验如果你打开模型文件报UnpicklingError多半是版本差异导致的换回源码包要求的版本就行。summaries 目录里存的是训练曲线记录包含每轮的 train loss 和 valid acc。这不是必须的输入但当你发现精度差了几个点第一件事就是打开 summaries 看曲线有没有炸掉的痕迹。曲线如果在第 10 轮前后出现 valid_acc 骤降基本可以断定是过拟合跟代码实现关系不大先调早停或正则。4.4 一个经验早停比调 dropout 更管用训练这套流量识别模型我的个人习惯是设一个 early stopping验证集准确率连续 5 个 epoch 不上升就停并且保存效果最好的那一次权重而不是最后一步。这样做对 93.5% 这种目标尤其重要因为 LSTM 到了训练后期经常会有 1-2 个点的波动用最后一轮权重碰运气不如用最优验证轮权重。很多拿到源码复现时发现跑不到 93.5%不是因为模型有问题而是因为从头训练时没有早停多跑了几个 epoch 导致过拟合。5. 避坑实录流量分类最容易翻车的四个地方5.1 数据不平衡攻击流量只占 5%模型学成了无脑判正常现象训练时 loss 下降很快验证集准确率很高但是打开混淆矩阵发现攻击类别的召回率只有 20%。整个测试集上 93.5% 的准确率掩盖了攻击流量基本没认出来这个事实。原因思博伦官方流量包里正常业务占比明显高于恶意软件和攻击流量CrossEntropyLoss 按样本量加权模型只要学会预测正常流量就能把 loss 压得很低攻击样本梯度被淹没。解决先按 label 统计一下 data.csv 的分布然后用加权 CrossEntropyLoss或者对攻击类样本做简单过采样。我一般会在loss_fn里传入weight权重设为各类样本数的倒数再归一化。改完之后总准确率可能会小幅下降但攻击类的召回率会显著上升安全场景下这是正确的取舍。5.2 时序泄漏验证集切分方式不对93% 是虚高的现象先用随机切分跑通流程测试集准确率直接 96%比报告还高换回按时间切分后掉回 93.5%。以为是模型波动其实是验证集本身漏了。原因相邻时间窗口的样本高度相似随机切分时训练集里已经看过测试集附近的数据模型相当于开卷考试。流量数据不能像图像那样随机划分必须严格按时间先后切分。解决train.py 里的split用int(len(X) * (1 - valid_ratio))从头部切到尾部保证验证集的每个样本在时间上都晚于训练集。如果你要从网上下载新 pcap 做跨数据集验证也要保证抓包时间段不重叠。5.3 低频 URL tokenmin_freq 设太大拿到的全是unk现象复现时发现 vocab 只有几千个词测试数据里大量 URL 变成unk训练准确率不低但真实场景一跑就废。后来统计了一下unk占比接近 40%。原因恶意软件回连域名经常是随机生成的本身就是低频 token。构建 vocab 时min_freq设成 5等于把这些最具判别性的特征全过滤掉了。解决构建 vocab 时min_freq不要超过 2最好对原始 URL 做字符级 n-gram 补充。我的习惯是保留全部 token数据量大词典膨胀的问题可以用max_vocab_size截断而不是靠频率截断。检查办法也很简单跑一次 test.py 的编码结果统计unk占所有 token 的比例超过 10% 就要回头调 vocab。5.4 model 目录与 params.pkl 对不上加载就崩溃现象test.py 加载模型时报 shape mismatch比如size mismatch for embedding.weight: copying a 18000x128 from 15000x128或者直接KeyError。原因model 目录下存放的params.pkl对应的模型结构和当前 clstm_classifier.py 里定义的类不完全一致。比如源码包的 params.pkl 里 hidden_size 是 256你单独改了类里的默认值为 128加载权重就得重构。解决永远不要手动改 clstm_classifier.py 里的默认超参如果你改了就用训练时保存的 params.pkl 里实际参数重建模型再 load_state_dict。代码里建议写成model CLSTMClassifier(**params)不要用CLSTMClassifier()这种默认构造方式。还有一个好习惯是从 tunnel 目录里把 params.pkl 和模型权重放一起不要单独复制权重文件。6. 把训练好的模型接到入口单条预测与可视化扩展6.1 加载 checkpoint 做实时预测训练和测试跑通了还不算完这套系统要落地得能对单条 URL 或单条新会话实时出结论。源码包内的 prediction 目录对应的就是这个环节。核心逻辑和 test.py 几乎一样区别是输入只有一条而且不需要真实标签# 单条 URL 预测 import torch from clstm_classifier import CLSTMClassifier from data_helper import encode_urls model_dir model/1600693479 with open(f{model_dir}/params.pkl, rb) as fp: params pickle.load(fp) model CLSTMClassifier(**params) model.load_state_dict(torch.load(f{model_dir}/model)) model.eval() new_url http://malicious-domain.example/download/agent.exe x torch.LongTensor(encode_urls([new_url], vocab, max_len64)) with torch.no_grad(): probs torch.softmax(model(x), dim1)[0] label_names [normal, malware, attack] for i, p in enumerate(probs): print(f{label_names[i]}: {p:.4f})输出会打印三个类别的概率分布而不是只给一个硬标签。实际生产里我建议直接消费这个概率分布把恶意软件 0.6 且不是正常瓶说的当成告警条件而不是等 argmax 出结果才处理。6.2 从预测结果到随时间变化的可视化PDF 报告里提到的对流量随时序变化进行可视化展示落地时其实就是把每条预测结果按时间戳聚合成曲线横轴是秒纵轴是三类流量各自的出现频率。代码量不大用 pandas 重采样加 matplotlib 就能出图。我的习惯是每处理完一批新流量就顺手把 prediction.csv 追加进去然后画一张 10 分钟窗口的滚动曲线。正常的业务流量应该是平稳起伏恶意软件和攻击流量如果出现脉冲式的凸起就值得盯着看一眼。可视化本身不会提高准确率但它能帮你在模型漏报时发现数据异常的线索——模型连续把攻击流量判成正常往往不是参数问题而是你的网络环境里出现了训练集里没见过的新型流量模式。这套源码包训练好的模型这次我是真的拿真实抓包验证过才敢说 93.5% 这个数字是可信的。之前有同事直接拿默认参数训不收敛想往下调 hidden_size被我拦住了——LSTM 类模型在这种中小规模流量数据集上参数不是越多越好改结构不如先调窗口和 vocab。从那以后我每次拿到一个新的流量识别项目都强制走一遍同样的流程先查数据分布和unk占比再训 baseline确认曲线正常后才开始碰超参数。这套源码把 pcap 解析、vocab 构建、CNNLSTM、训练评估、预测输出全串在了同一条链路上对想快速上手流量分类的人来说是个很值的参考起点希望帮到你。本文还有配套的精品资源点击获取