CNN+LSTM流量分析识别：pcap切流、特征化与部署避坑指南

简介：基于CNN与LSTM的流量分析识别系统设计与实现资料包，面向深度学习、人工智能方向的学生、研究者和网络安全分析人员，用于解决网络流量的实时识别与分类问题。方案采用CNN提取空间特征、LSTM提取时序特征，将思博伦官方pcap包解析为URL序列进行训练，在官方测试流量上达到93.5%的准确率，能有效区分正常业务流量、恶意软件流量和网络攻击流量，并支持随时序变化的可视化展示。资源共24个文件，主要包含6个Python源码脚本（数据预处理、CNN分类器、LSTM分类器、CNN+LSTM联合分类器、训练与测试入口）、TensorFlow模型权重及词表参数、3个CSV数据文件、完整PDF设计报告和使用说明，覆盖从数据处理到模型训练、测试与部署的全流程，目录结构清晰。压缩包仅23.58MB，轻量易用。已有573人学习，适合作为课程设计、毕业设计或流量安全项目的完整参考。

1. 把流量当“时序画像”来读：CNN+LSTM这套方案解决什么问题

被报告里的准确率带偏，是刚接触流量分析系统的人最容易踩的坑：公开数据集上模型刷到99%，拿到自己的环境一测只剩六成。基于CNN和LSTM的流量分析识别系统，思路和深度包检测完全不一样，它不靠特征字符串匹配，而是把每条网络会话当成一段有先后顺序的“一维信号”。CNN卷积神经网络负责抓字节流、包长序列里的局部模式，LSTM神经网络负责记住这些模式在时间上的先后关系，最终输出这条流属于正常业务还是扫描、爆破、远控等风险类别。可解决的问题覆盖加密流量分类、恶意会话识别、数据包取证筛查和CTF流量分析里的攻击定位。适合正在搭安全运营工具、做课程设计或毕业设计的从业者；拿到源码和训练好的模型后，最该花时间的不是把脚本跑通，而是把数据流水线和训练参数吃透。

2. 系统架构与数据准备：先让pcap变成模型能吃的“一维序列”

2.1 整体链路：采集、切流、特征化、推理、处置

用CNN和LSTM做流量识别，模型只是中间一环。一个能真正落到环境的系统，至少要包含整条链路：网络侧抓包，服务器上部署tshark或tcpdump定时采集；pcap到达解析层，按五元组和超时时间切分成会话；每个会话经特征化模块转成长度固定的一维张量；随后进入训练好的分类模型，推理结果连同原始流元数据一起交给上层处置逻辑，记录日志、弹告警或联动防火墙阻断。

市面上的开源实现，切流和特征化通常用Scapy或CICFlowMeter完成，模型训练用Keras/TensorFlow或PyTorch。标题里“源码+模型”这类压缩包，一般已经把这几块打包好了，但源码普遍有个毛病：切流逻辑和模型训练耦合在一起，换个pcap就报错，或者特征维度写死。动手前先把目录拆开，抓包脚本、pcap解析器、特征化、模型定义、训练入口、推理入口六个模块各自独立运行，后续调试成本会低很多。

这个阶段最常见的误区是一上来就调模型，把pcap往训练脚本里塞。实际上流量识别的准确率上限，有一大半在切流和特征化这里。会话切错、方向时序弄反、截断长度不合理，网络结构再强也补不回来。我一般会先把特征化模块单独验证：给一个已知标注的小pcap，打印出每条会话的元数据、字节数和时间戳，和Wireshark的“显示TCP流”功能做对比，一致再继续往后走。

2.2 pcap到会话切分：四元组方向归并是第一个关键步骤

切流的目标，是把双向的原始数据包组装成一段有顺序的字节流。一个TCP会话，客户端发一段请求、服务端回一段响应，模型要同时看到双方数据才能判断完整行为。所以切分逻辑不能只看原始四元组，而要把源、目的互换的包归到同一个会话里。

下面这段基于Scapy的脚本，是能直接跑通的切流实现，我写课程设计和内部工具时都习惯用它做起点：

from scapy.all import rdpcap, TCP, UDP, IP from collections import OrderedDict def make_key(pkt): """生成会话key，方向互换后仍映射到同一个会话""" ip = pkt.getlayer(IP) if not ip: return None if pkt.haslayer(TCP): proto = "tcp" a, b = pkt[TCP].sport, pkt[TCP].dport elif pkt.haslayer(UDP): proto = "udp" a, b = pkt[UDP].sport, pkt[UDP].dport else: return None # 统一方向：让端口号较小的一侧放在前面 if a < b: return (ip.src, a, ip.dst, b, proto) else: return (ip.dst, b, ip.src, a, proto) def split_sessions(pcap_path, idle_timeout=60, max_bytes=8192): pkts = rdpcap(pcap_path) sessions = OrderedDict() for pkt in pkts: key = make_key(pkt) if key is None: continue # 超过空闲时间阈值，视为一条新会话，避免长连接把序列撑爆 if key not in sessions: sessions[key] = {"start": float(pkt.time), "payload": b""} elif float(pkt.time) - sessions[key]["start"] > idle_timeout: sessions[key] = {"start": float(pkt.time), "payload": b""} payload = bytes(pkt.getlayer(TCP).payload) if pkt.haslayer(TCP) else bytes(pkt.getlayer(UDP).payload) sessions[key]["payload"] += payload # 乐观截断，防止下载大文件时把一整条几MB的流全读进内存 if len(sessions[key]["payload"]) >= max_bytes: sessions[key]["payload"] = sessions[key]["payload"][:max_bytes] return sessions if __name__ == "__main__": for key, sess in split_sessions("capture.pcap", idle_timeout=60, max_bytes=8192).items(): print(key, len(sess["payload"]), sess["start"])

make_key函数里的细节值得注意：我没有按原始四元组做key，而是比较端口大小、统一方向，客户端发给服务端和服务端回给客户端的包就归进了同一条会话。idle_timeout参数解决长连接的切分问题，两条相邻报文间隔超过阈值就另起新会话，避免一个小时的空闲占掉序列的主要位置。max_bytes=8192是防止内存被打满的保险丝，因为模型训练时序列长度只取1024左右，后面还会二次截断，这里保住前8KB已经够用。

参数怎么调：普通Web场景idle_timeout取60秒，DNS这类短连接取15秒，数据库长连接取90秒。max_bytes不建议小于4096，否则像HTTP这类前段有大协议头的流量会把关键信息截掉。Scapy跑几百MB的大pcap会明显变慢，可以先tshark按IP段或端口粗过滤一道，再用Scapy做精细切流，批处理效率能快好几倍。

2.3 特征化：字节序列、包长序列与统计特征的组合

切出来的payload不能直接丢给模型，长度不对齐是一回事，更重要的是模型需要的信息不一定都在payload里。我常用的是三通道设计：原始字节流、包长时序、包间隔时序，最后在分类层融合。第一个通道，字节序列，把payload原样落到0到255的整数张量，适合模型学习字节级的局部模式。第二个通道，包长序列，把会话里每个报文的payload长度按时间顺序排开，一维卷积在上面滑动，能学到“小块传输、停顿、大批量发包”这样的节奏特征。第三个通道是包间隔，单位毫秒，反映交互的缓急程度，爆破和扫描的请求间隔通常非常均匀。

拼接成张量的过程可以抽象成一个函数，下面是可复现的核心逻辑：

import numpy as np def flow_to_tensor(session, seq_len=1024, max_pkts=128): # 假设session是解析器输出的字典，包含payload、pkt_lens、inter_arrival等字段 payload = session["payload"] inter = session.get("inter_arrival", []) pkt_lens = session.get("pkt_lens", []) # 通道一：字节序列，定长截断 + 尾部补零 bytes_seq = np.frombuffer(payload[:seq_len], dtype=np.uint8).astype(np.float32) / 255.0 if len(bytes_seq) < seq_len: bytes_seq = np.pad(bytes_seq, (0, seq_len - len(bytes_seq)), "constant") # 通道二：包长序列，按1500字节的MTU做归一化 pkt_lens = pkt_lens[:max_pkts] if len(pkt_lens) < max_pkts: pkt_lens += [0] * (max_pkts - len(pkt_lens)) pkt_len_seq = np.asarray(pkt_lens, dtype=np.float32) / 1500.0 # 通道三：包间隔序列，log1p压缩长尾，避免个别大间隔值主导梯度 ivals = np.asarray(inter[:max_pkts], dtype=np.float32) if len(ivals) < max_pkts: ivals = np.pad(ivals, (0, max_pkts - len(ivals)), "constant") ivals = np.log1p(np.maximum(ivals, 0)) return {"bytes": bytes_seq, "pkt_lens": pkt_len_seq, "intervals": ivals}

seq_len=1024是多数场景的均衡选择，HTTP请求响应够用；如果主要分析SSH、SMTP这类长文本协议，可以提到3072，但训练显存会明显上涨。max_pkts=128的意思是只看会话前128个包，超过部分不管，这是刻意为之：恶意行为大多发生在会话前段，比如连接建立后的握手特征、前几批载荷内容，全塞进来反而让模型学到大量噪声尾巴。

归一化上，字节值除以255、包长除以1500，让三个通道量纲一致，梯度更新更顺。有人习惯用StandardScaler做Z-Score，在流量特征上我不推荐，因为补零产生的稀疏位置很多，中心化会带来不必要的均值偏移。

2.4 数据集与标注：公开数据集的坑和CTF pcap怎么用

训练数据是这个系统最贵的一环。公开数据集里CICIDS2017和UNSW-NB15用得最多，前者覆盖大量应用层会话，后者偏攻击检测。用它们要有个心理准备：样本按天采集，网上很多源码直接train_test_split随机切分，同一IP、同一时段的会话同时进训练集和测试集，模型等于看了答案。规范做法是按时间切分，前几天的数据训练、后几天的数据测试，这样评估出来的指标才可信。更细的说法，验证集的最小时间戳必须大于训练集的最大时间戳，否则就是代码里漏了排序或重新shuffle。

CTF流量包，比如取证题里附带的pcap，包含刻意藏起来的Web攻击或远控通信记录，样本量小、类别少，适合做系统功能验证，不适合训练。扩充样本的正路是在可控环境搭几个服务，用脚本模拟登录、上传、下载、扫描、爆破并抓包，把抓包时间、源IP、目的IP记成清单，再按四元组自动打标签。要注意多标签冲突：同一个会话既有正常登录又有爆破尝试，标签只能取最高风险级别，否则模型学到自相矛盾的映射。

标签体系建议控制在五类以内：正常、扫描、爆破、远控、数据外传。类别再细，样本量跟不上，模型反而学不好。另外公开数据集里正常流量往往占95%以上，类别极度不平衡的问题几乎一定出现，具体处理在第4章展开。

3. 模型设计与训练：Conv1D堆LSTM，网络参数怎么定、训练怎么续

3.1 选型理由：为什么不是纯CNN、纯RNN，也不是Transformer

单独用CNN卷积神经网络做流量分类，常见做法是把会话转成类似图像的特征图，用二维卷积去扫。亲自跑过就会发现，问题在于会话是强时间有序的，拉成二维图会破坏先后关系；而图像任务里旋转、平移不变性的归纳偏置放在流量上是没意义的，反而增加参数。改用一维卷积沿序列方向滑，只关注局部窗口内的字节模式，用不同尺寸的卷积核分别捕捉短签名和长签名，这才是CNN在流量分析里的正确打开方式。

单独用RNN/LSTM也有硬伤。LSTM确实能记住长距离依赖，但逐时间步串行计算，一条上KB的序列又是时间又是显存；而且LSTM对局部精确匹配不敏感，恶意载荷特征出现在第100字节还是第500字节，它的记忆效果不如卷积核直接扫出来可靠。

把两者接起来是常见做法：先用几层Conv1D把输入序列降维、提取局部模式，再把卷积输出按时间步展开给LSTM，让LSTM学习“先握手、传一段数据、突然开始大批量发包”这类行为级时序。对比Transformer和CNN、RNN的区别会发现，Transformer默认全局自注意力，数据量不够时很容易过拟合，而且推理资源要求高；CNN+LSTM的组合在几万条样本规模下训练稳定、CPU推理也扛得住。如果样本到百万级、要做大规模分布式训练，再换Transformer不迟。

还有一个常被忽略的基线：先用XGBoost二分类模型在统计特征上跑一版，很多场景它已经能到0.9以上的AUC，深度学习模型应该在这个分数之上再谈提升，否则先回头检查特征和标签。

3.2 网络结构：一维卷积块加双向LSTM，参数表与Keras实现

我常用的结构分三段：特征提取、时序建模、分类输出。特征提取段是两个Conv1D块，每块包含一维卷积、批归一化、ReLU激活和最大池化。时序段用一层双向LSTM，输出维度128。分类段取LSTM最后一个时间步的输出，接Dropout和Dense层。多分类输出用softmax，二分类用sigmoid。

下面这张参数表是按单字节序列通道设计的，多通道扩展见代码后的说明：

对应的Keras模型定义：

import tensorflow as tf from tensorflow.keras import layers def build_cnn_lstm(input_len=1024, num_classes=5, lstm_units=128): # 输入形状：单个序列，多通道时可把最后一维改成通道数 seq_input = layers.Input(shape=(input_len, 1), name="seq") x = layers.Conv1D(filters=64, kernel_size=5, padding="same")(seq_input) x = layers.BatchNormalization()(x) x = layers.ReLU()(x) x = layers.MaxPooling1D(pool_size=2)(x) x = layers.Conv1D(filters=128, kernel_size=3, padding="same")(x) x = layers.BatchNormalization()(x) x = layers.ReLU()(x) x = layers.MaxPooling1D(pool_size=2)(x) x = layers.Bidirectional(layers.LSTM(lstm_units, dropout=0.3))(x) x = layers.Dropout(0.5)(x) x = layers.Dense(64, activation="relu")(x) output = layers.Dense(num_classes, activation="softmax" if num_classes > 2 else "sigmoid")(x) return tf.keras.Model(inputs=seq_input, outputs=output)

代码的逻辑是：一维卷积沿着序列提取局部特征，BatchNorm缓解梯度消失，ReLU让训练过程不容易炸。池化层把序列长度降为原来的四分之一，LSTM的时间步长也就少了四分之三，训练速度收益非常明显。双向LSTM正向反向各跑一遍再拼接，能同时看到某段局部模式之前和之后发生了什么，对识别会话中段才出现的恶意载荷有实际帮助；代价是参数翻倍、CPU推理变慢，只做实时单包拦截的话可以换回单向LSTM。

参数调整经验：lstm_units=128是均衡点，样本量小降到64，数据量大可以加到256。Dropout两层都要留，LSTM内部的dropout=0.3和分类头的0.5各司其职，省掉任何一个都容易过拟合。第2.3节的三通道特征要并进来的话，最简单的方式是把三个序列拼成(1024, 3)的三通道张量，第一层Conv1D的输入通道从1改成3，其余不用动。

3.3 训练脚本：早停、断点续训、类别权重一次性配齐

训练部分最怕两件事：训练到一半进程被杀，以及过拟合了还在硬跑。标准做法是把ModelCheckpoint、EarlyStopping、class_weight三件事一起挂上：

from tensorflow.keras.callbacks import ModelCheckpoint, EarlyStopping def build_callbacks(ckpt_path="best_model.h5"): ckpt = ModelCheckpoint( ckpt_path, monitor="val_loss", save_best_only=True, # 验证集损失下降才覆盖旧权重 verbose=1 ) early = EarlyStopping( monitor="val_loss", patience=5, # 连续5轮没有改善就停 restore_best_weights=True # 停止时把权重回滚到最优轮次 ) return [ckpt, early] # 类别权重示例：正常类50000条、恶意类3000条时，把恶意类权重提高 class_weight = {0: 1.0, 1: 3.0} model.fit( train_dataset, validation_data=val_dataset, epochs=30, batch_size=64, callbacks=build_callbacks(), class_weight=class_weight, )

ModelCheckpoint保存的是验证集上表现最好的权重，不是最后一轮，避免训练末期过拟合导致模型退化。EarlyStopping的patience=5要注意，如果学习率设得大导致损失曲线震荡，模型可能提前停，这时把patience放宽到10再观察。restore_best_weights=True建议始终打开，否则早停触发时模型停在最后一轮而不是最优轮次。class_weight是最省事的类别不平衡处理手段，不用改数据结构，只在损失层面对少数类加惩罚。

学习率调度也别忽略。流量数据经常第一轮就冲到一个看似不错的点，之后损失几乎不动。给出两条可直接用的经验：Adam优化器初始学习率1e-3，训练10轮后还没收敛就降到1e-4；换AdamW的话把weight_decay设成1e-4，对稀疏字节序列的泛化有帮助。这套调法在LSTM时间序列预测任务的python实现里也通用，拿过去直接套就可以。

3.4 训练与测试划分纪律：按时间切，不按样本ID切

训练集和验证集的切分方式，直接决定最后拿到的准确率是实打实还是自欺欺人。流量数据最大的特点是会话之间高度相关，同一台主机在同一个时间段内产生的会话，源IP、目的端口、TLS指纹都相近。如果像图像分类那样random_split，同源会话会同时出现在训练集和测试集，模型记住IP和端口就够了，根本不用学协议行为。

正确划分是按时间排序后切：前80%的会话训练、中间10%验证、最后10%测试。三个时间粒度要一致：训练数据采集时间段、模型上线时间段、回放验证时间段。如果训练数据采自3月，模型在5月流量上测出来差，那不一定是模型坏了，而是数据分布漂移。后续维护要滚动重训，比如每月把最近三个月流量重跑一遍流水线。

还有一个容易翻车的小细节：切分后检查一次时间戳，确认验证集最小时间大于训练集最大时间，出现交叉就说明代码里漏了排序或又重新shuffle了。

4. 避坑：训练与部署里最常见的5个翻车点

4.1 按IP随机切分导致测试集失真

现象：训练F1高达0.98，模型部署到自己的镜像流量一测只有0.6。

原因：数据集按样本ID随机切分，同一IP、同一端口的会话同时出现在训练集和测试集，模型直接记住IP和标签的对应关系。这就是典型的信息泄露。

解决：改成按时间段切分，并加一道硬校验：遍历测试集，确认每个四元组key不出现在训练集，出现就打印出来人工排查。即使切分正确，新环境流量类别占比大概率不一样，这也是分数掉一半的原因，光看准确率没用，要看各类别召回率。

4.2 类别不平衡，模型全部预测成多数类

现象：训练过程一切正常，推理时所有流量都被判正常，恶意样本检出率几乎为零。

原因：正常会话占比超过95%，模型发现猜“正常”就能拿到95%准确率，于是躺平了。

解决：先用class_weight给少数类加权重，再看混淆矩阵里每个类别的召回率。恶意类召回率低于0.8就说明模型还在划水。还不行就把损失函数换成Focal Loss，它对难分类的少数类更友好。要注意别对序列数据用SMOTE做上采样，合成出来的流量样本和真实网络行为差异很大，只会制造虚假的模式。

4.3 序列长度不合理，训练时显存爆炸

现象：LSTM训练到一半被OOM杀死，或者单个epoch耗时超过十几分钟。

原因：直接把整条会话塞给LSTM，有的流几万字节，时间步长达几万步，显存和算力都扛不住。

解决：在数据生成器里统一截断到1024或2048，超过部分直接丢弃；batch_size从64降到32或16。还炸的话，先加一层GlobalAveragePooling1D压缩序列，再接一个小LSTM。训练前手动打印一个batch的张量形状，第二维必须是预期的seq_len，这一步千万别skip。

4.4 训练中断后白跑，没有断点续训

现象：训练到一半局里断电或开发机重启，打开日志发现前十几个小时白费。

原因：只设了epoch数，没配checkpoint，进程结束什么都没留下。

解决：把ModelCheckpoint和EarlyStopping写进所有训练脚本，checkpoint文件名带上时间和epoch，比如cnn_lstm_epoch12_los0.31.h5。下次启动先load_weights再fit。配合TensorBoard记录每轮损失曲线，翻车时一眼能看出是收敛震荡还是数据问题。

4.5 模型导出格式不对，推理性能上不去

现象：用Python开着模型做在线推理，单次预测几十毫秒，流量一大处理不过来。

原因：把Keras的h5直接丢给生产环境，每次预测都重新加载图形库，纯Python推理路径太慢，预处理补零也没有纳入部署。

解决：用ONNX Runtime或TF-TRT导出。LSTM模型在ONNX Runtime的CPU上通常能快几倍，导出时把序列长度固定成训练值，避免动态shape带来的额外开销。实时性要求高的场景做两级级联：先用一维CNN粗筛，只有疑似可疑的会话才送进LSTM做精细分类，单核CPU也能扛住上千路并发。部署后留一条旁路，模型输入输出同时写日志，方便对比灰度效果。

提示：把训练数据、切流脚本、checkpoint文件全部纳入版本管理。三个月后你会回来查当时某个参数到底怎么改的，没有历史记录就只能从零试。

5. 上线前最后一步：用时间回放验证模型老化

模型部署上线只是开始，真正值钱的是持续验证机制。流量分析系统有一个和图像模型完全不同的特性：流量每天都在变，新协议版本、新攻击手法都会让旧模型的分布假设失效。我习惯在系统里加一个回放模块：把昨天的pcap按小时切块，每个小时跑一遍当前模型，把每小时预测类别分布和实际抽样结果做滑动对比。

对比指标里最关键的是类别分布漂移程度。正常流量的预测分布应该跟随业务曲线波动，白天Web请求多，凌晨时段全是备份任务；如果某天突然多出一批预测为远控的样本，不管真假都要触发人工审计。新模型上线前，先跑一周影子回放，新旧模型并行处理同一份流量，记录各自F1和延迟，汇总成对比表：

回放脚本的骨架不复杂：读pcap、按时间窗切分、逐条预测、汇总分布，但一定要和训练流水线共用同一个特征化函数，否则回放结果没有参考价值。我现在的习惯是，模型文件里直接写进特征化参数、训练数据时间范围、模型版本号，换环境时从不裸拷h5。这套流程踩过太多次第4章那种坑，特征泄露、类别不平衡、断点续训，多数流量分析项目翻车都不是结构不行，而是数据流水线没守住底线。希望这些能帮你在自己的环境里少走弯路，让模型出的每一个数字都经得起回放检验。

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