☰
CNN网络入侵检测实战:pcap解析、特征工程与一维卷积模型
2026/10/10 21:39:52 网站建设 项目流程

简介:基于Python与卷积神经网络(CNN)的网络入侵检测项目,面向计算机、数学、电子信息等专业学生,适合作为课程设计、期末大作业或毕业设计的参考实现。项目以NSL-KDD这一经典入侵检测数据集为对象,涵盖数据读取、归一化、CNN模型构建、训练与预测完整流程,并附有项目说明文档,帮助读者理解从特征处理到模型评估的每个环节。压缩包共33个文件,包括4个Python核心脚本、12个CSV数据文件、1个PyTorch模型权重、若干XML工程配置,以及训练过程的准确率、精确率曲线和数据可视化图片,整体大小约21.58MB,目录结构清晰,可直接解压运行,也便于按模块检索与复用。另含README说明与工程配置文件,可辅助复现实验、调整网络参数、替换数据集或扩展算法。当前已有631人学习下载,适合希望系统掌握深度学习在网络安全领域应用、并愿意钻研代码进行二次开发的读者。

1. 网络入侵检测算法源码包:先别看模型,先问数据从哪来

拿到「基于 Python + CNN 的网络入侵检测算法源码+项目说明.zip」这类项目,很多人第一反应是打开模型文件看网络结构,但我建议你先压住这个冲动。网络入侵检测(Network Intrusion Detection)这个方向,真正的门槛不在 CNN 本身,而在于三件事:有没有带标签的流量数据、特征怎么从 pcap 里提出来、训练集和测试集有没有泄漏。卷积神经网络在这里做的,是把每个网络会话压缩成的特征向量分类成正常或攻击,但特征向量做得脏,后面调什么参数都是白费力气。

这个源码包适合两类人:一类是想复现「流量分类」实验的在校学生,另一类是公司里要自己做 IDS 原型验证的工程师。你需要准备的东西不复杂:一台带 Python 3.8 以上环境的机器、一份或多份 pcap 抓包文件、PyTorch 和 scapy 两个核心依赖。下面我按自己做过类似项目的路径,从数据解析一路拆到模型验证,尽量让每个步骤你都能照着跑通。

2. 从 pcap 到特征矩阵:解析、五元组聚合与标签构造

2.1 选解析库:scapy 上手快,但要按迭代器读文件

常见做法是用 scapy 做 pcap 解析,因为它能直接读 pcap 和 pcapng 两种格式,也能拿到 IP 层和 TCP/UDP 层字段。但它有个坑:rdpcap()会把整个文件一次性读进内存,一个几百 MB 的 pcap 就能吃掉几个 GB 内存,跑训练前机器先卡死。我一般用PcapReader()做流式迭代,逐包处理,内存占用稳定在几百 MB 以内。

from scapy.all import PcapReader, IP, TCP, UDP def iter_pcap(path: str, max_packets: int = 200_000): count = 0 with PcapReader(path) as reader: for pkt in reader: if count >= max_packets: break if IP not in pkt: continue ip = pkt[IP] proto = None sport = dport = 0 if TCP in pkt: proto = "tcp" sport, dport = pkt[TCP].sport, pkt[TCP].dport elif UDP in pkt: proto = "udp" sport, dport = pkt[UDP].sport, pkt[UDP].dport else: continue yield { "ts": float(pkt.time), "src_ip": ip.src, "dst_ip": ip.dst, "sport": sport, "dport": dport, "proto": proto, "len": len(pkt), "flags": pkt[TCP].flags if TCP in pkt else "", } count += 1

这段代码里有几个值得注意的设计:max_packets参数用来限制解析规模,调试时先用小值跑通全流程;每条流记录保留五元组(源 IP、目的 IP、源端口、目的端口、协议),这是后面聚合会话的依据;包长len(pkt)是抓包文件里的实际长度,不是 IP 头里的长度字段,两者相差 14 字节以太网头,统计特征时保持口径一致即可。

2.2 按五元组聚合成会话:双向流量合并是第一个大坑

单包特征没法直接做入侵检测,攻击行为体现在一连串包的统计规律上,比如端口扫描有大量短连接、DDoS 有异常高的包速率。所以要把同一个五元组的双向包合并成一个「流会话」。这里有个常见误操作:只按(src_ip, src_port, dst_ip, dst_port)聚合,结果同一 TCP 连接的请求和响应被拆成两条流,特征被切碎。正确做法是先把四元组排序成无序对,再和协议一起作为 key。

from collections import defaultdict import statistics def build_flows(packets): flows = defaultdict(list) for pkt in packets: src = (pkt["src_ip"], pkt["sport"]) dst = (pkt["dst_ip"], pkt["dport"]) key_src, key_dst = sorted([src, dst]) flows[(key_src[0], key_src[1], key_dst[0], key_dst[1], pkt["proto"])].append(pkt) return flows

聚合之后每个 key 对应一个包列表,但要小心 UDP 的聚合口径。UDP 没有连接概念,实践中常用「超时窗口」切分会话,比如相邻两个 UDP 包间隔超过 60 秒就视为新会话。scapy 抓包里经常出现 DNS 查询这种短 UDP 流,如果不做窗口切割,一个 DNS 服务器的所有流量都会聚成一条巨型流,特征直接失真。

特征提取阶段,我从每个会话里抽出 20 个统计量,包括包长均值、包长标准差、最大包长、SYN 包数、RST 包数、FIN 包数、平均到达间隔、到达间隔标准差、上下游包数比、上下游字节比等。这部分是特征工程的核心,CNN 后面能学到什么,完全取决于这里。

2.3 打标签:靠 IP 黑白名单和端口规则,不靠人工审计

带标签的训练数据是这个项目里最难搞的一环。公开数据集如 NSL-KDD、UNSW-NB15 可以直接用,但它们和真实抓包的分布差距大。源码包里的常见做法是:自己抓一份正常业务流量,再注入攻击工具生成的流量,然后脚本自动打标签。我给标签脚本的规则是:攻击机 IP 匹配到黑名单 IP 段或目的端口命中高危端口列表,该会话标为 1,否则标为 0。

ATTACK_IPS = {"10.0.0.5", "192.168.1.100"} HIGH_RISK_PORTS = {22, 23, 3389, 6379, 9200, 4444, 5555} def label_flow(key, packets): src_ip = key[0] dst_port = key[3] is_attack_ip = src_ip in ATTACK_IPS or key[2] in ATTACK_IPS is_risk_port = dst_port in HIGH_RISK_PORTS return 1 if (is_attack_ip or is_risk_port) else 0

这种打标方式有明显局限:只依赖 IP 和端口,攻击行为如果走的是合法端口,比如 SQL 注入走 80 端口,就会被漏标。更严谨的方案是先跑 Suricata 规则做初步标注,再用人工抽样校正。但对一个源码包项目,脚本化打标能让你把全流程跑通,后续替换成更精细的标注器不影响 CNN 训练部分的代码结构。

3. 一维 CNN 网络结构设计:为什么不用 2D-CNN 处理流量特征

3.1 流量特征是一维向量,Conv1d 比 Conv2d 更合适

很多人一听说 CNN 就默认要构造二维矩阵输入,把特征 reshape 成类似灰度图的形状喂给Conv2d。这个做法能做,但没必要。我们的每个会话特征是一维的 20 维向量,Conv1d直接沿着特征维度做卷积,参数量小、训练快、可解释性也更好。只有当你把多个会话按时间顺序排成序列,想捕捉会话之间的时序关系时,才需要把输入组织成二维结构,但那就是 LSTM 或 Transformer 的领域了。

我设计的网络只有两层卷积加两层全连接,结构很朴素。第一层 Conv1d 把 20 维特征映射到 64 个通道,核大小为 5;第二层 Conv1d 保持通道数不变,核大小为 3,然后接全局平均池化,把序列维度压成 1;最后接两个全连接层,输出二分类 logits。中间加 BatchNorm1d 和 Dropout,防止训练波动和过拟合。

import torch import torch.nn as nn class FlowCNN(nn.Module): def __init__(self, in_features=20, num_classes=2): super().__init__() self.conv = nn.Sequential( nn.Conv1d(1, 64, kernel_size=5, padding=2), nn.BatchNorm1d(64), nn.ReLU(), nn.Conv1d(64, 64, kernel_size=3, padding=1), nn.BatchNorm1d(64), nn.ReLU(), ) self.pool = nn.AdaptiveAvgPool1d(1) self.classifier = nn.Sequential( nn.Dropout(0.3), nn.Linear(64, 32), nn.ReLU(), nn.Linear(32, num_classes), ) def forward(self, x): x = x.unsqueeze(1) x = self.conv(x) x = self.pool(x).squeeze(-1) x = self.classifier(x) return x

这里in_features必须和特征工程输出的维度一致,改特征时要记得同步改这里。unsqueeze(1)是为了把(batch, features)变成(batch, 1, features),满足 Conv1d 要求的输入形状。两层卷积用的都是 padding=2 和 padding=1,是为了让卷积不缩短特征长度,池化才做降维。如果去掉 padding,特征长度每过一层卷积就减一截,第二层卷积时输入长度只有 18,核大小 5 也能算,但边界信息的保留方式就不一样了。

3.2 训练流程里的两个关键参数:梯度裁剪和类别权重

训练循环本身不复杂,但有两个参数我会特别在意:梯度裁剪和类别不平衡权重。网络入侵检测数据集的标签通常极不平衡,正常流量可能占 90% 以上,直接训练会让模型学会全预测正常类,准确率看着很高,召回率是零。解决方法是给损失函数传pos_weight,放大少数类的梯度信号。另一个是梯度裁剪,防止个别异常样本让 loss 爆炸、参数一步跳到坏区域。

def train_one_epoch(model, loader, optimizer, criterion, device, clip_norm=1.0): model.train() total_loss = 0.0 for x, y in loader: x, y = x.to(device), y.to(device) optimizer.zero_grad() logits = model(x) loss = criterion(logits, y) loss.backward() torch.nn.utils.clip_grad_norm_(model.parameters(), clip_norm) optimizer.step() total_loss += loss.item() * x.size(0) return total_loss / len(loader.dataset)

pos_weight的计算方法在下面的章节会写,这里先讲梯度裁剪。clip_norm=1.0的意思是所有参数的梯度 L2 范数超过 1.0 时就等比缩放到 1.0,可以想象成给梯度加了一个限幅器。流量特征里如果出现极端的包长标准差,比如某个会话有 1500 字节的大包和 40 字节的小包混在一起,计算出的梯度可能很大,裁剪能避免这种离群点主导训练方向。

3.3 验证集与早停:CNN 在入侵检测里最容易过拟合

训练集 loss 不断下降,验证集 loss 却反弹,这是流量分类里最常见的翻车场景,因为流量特征不像图像有丰富的语义信息,20 维统计特征对模型来说太简单,容量稍大就背样本。我一般用验证集 loss 做早停,连续 5 个 epoch 不下降就保存当前最优权重。这段逻辑要放在训练循环外面包一层,不要在循环里顺手写。

best_loss = float("inf") patience = 5 bad_epochs = 0 for epoch in range(epochs): train_loss = train_one_epoch(...) val_loss = evaluate(model, val_loader, criterion, device) if val_loss < best_loss: best_loss = val_loss torch.save(model.state_dict(), "best_model.pt") bad_epochs = 0 else: bad_epochs += 1 if bad_epochs >= patience: break

早停的 patience 值我习惯设为 5 到 8。太小的话,验证 loss 因为 batch 采样波动下降一点就触发保存,但你不知道后面还能不能继续下降;太大的话,训练时间拉长,而且后期过拟合的风险累积。另外注意保存的是state_dict()而不是整个模型对象,加载时用FlowCNN(...)建好结构再load_state_dict。

4. 数据集构造与训练评估:平衡采样、DataLoader 与混淆矩阵

4.1 把特征和标签组装成 PyTorch Dataset

特征矩阵和标签准备好之后,需要包一层Dataset才能喂给 DataLoader。这一步有个小坑:不要在__getitem__里做标准化,训练集和验证集的均值和标准差必须一致。我都是先在全部训练集上算好均值和标准差,把标准化参数固化下来,再传给 Dataset。测试集不能重新计算均值和标准差,否则分布偏移,训练时的早停判断就失真了。

def build_dataset(features, labels, mean, std): feats = (features - mean) / std feats = torch.tensor(feats, dtype=torch.float32) labels = torch.tensor(labels, dtype=torch.long) return torch.utils.data.TensorDataset(feats, labels)

这个函数看起来短,但它把标准化、类型转换都收在了一个入口里。dtype=torch.float32是 PyTorch 默认的,不用 float64 是因为显卡加速对 float32 支持最好,float64 在 GPU 上慢很多。labels用 long 类型是因为分类任务的交叉熵损失函数期望目标值是 long 类型,传 float 会报类型错误。

4.2 平衡采样:用 WeightedRandomSampler 替代随机欠采样

类别不平衡的处理除了改损失函数权重,还可以在采样层面做。两种做法我都在项目里试过,随机欠采样会把多数类样本丢掉一部分,训练数据量变小,模型方差变大;WeightedRandomSampler 则是每个 batch 按权重抽取样本,少数类被抽中的概率高,但每个 epoch 都能看到全部样本。后者在流量场景更稳。

from torch.utils.data import WeightedRandomSampler def make_sampler(labels, num_samples): class_counts = torch.bincount(labels) weights = 1.0 / class_counts[labels].double() sampler = WeightedRandomSampler(weights, num_samples=num_samples, replacement=True) return sampler

权重公式1 / class_count是最简单的逆频率加权,少数类的样本每条权重更大。num_samples我一般设成训练集总样本数,这样每个 epoch 差不多把训练集完整过一遍。replacement=True允许同一批数据被重复抽到,在做 oversampling 效果的同时不需要复制数据到内存。这个 sampler 要和shuffle=False一起用,因为 sampler 本身已经控制了抽样顺序。

4.3 评估指标:别只看 accuracy,看混淆矩阵和 ROC-AUC

入侵检测的评估指标如果只看 accuracy,模型几乎一定「表演性」地好——全部预测正常类也能拿到 90% 以上的准确率。所以我固定输出四样东西:混淆矩阵、精确率、召回率、F1。其中召回率对入侵检测最敏感,漏报一个攻击连接比错报一个正常连接代价更大。

from sklearn.metrics import confusion_matrix, classification_report def evaluate_model(model, loader, device): model.eval() y_true, y_pred = [], [] with torch.no_grad(): for x, y in loader: logits = model(x.to(device)) pred = logits.argmax(dim=1).cpu() y_true.extend(y.tolist()) y_pred.extend(pred.tolist()) print(confusion_matrix(y_true, y_pred)) print(classification_report(y_true, y_pred, target_names=["normal", "attack"])) return y_true, y_pred

argmax(dim=1)把模型输出的两个类别的 logits 转成预测类别。torch.no_grad()告诉 PyTorch 这段前向计算不需要记录梯度,能省不少内存。分类报告里的 precision 和 recall 分别对应误报率和漏报率,如果你要写成论文,F1 是审稿人最关心的数字。我还会额外算 ROC-AUC,因为这个指标不依赖阈值选择,模型好坏看得更客观。

5. 避坑指南:解析、训练、评估三阶段的实战踩坑记录

5.1 训练集和验证集来自同一段 pcap,时间重叠导致数据泄漏

现象:验证集 F1 很高,但放到新抓的流量上一测,马上掉到不可用。原因:流量包是按时间顺序抓的,直接按 8:2 随机切分,同一条 TCP 连接的包被分到训练集和验证集两侧,模型在验证集上「见过」连接的片段。解决:按时间戳排序后,把完整会话分配到某一侧,或直接按时间切分,前 80% 时间段做训练,后 20% 做验证。按会话 ID 做 GroupShuffleSplit 是最稳妥的,但前提是会话 ID 的特征构造阶段已经稳定。

5.2 scapy 解析时 pcap 的时间戳精度丢失

现象:同一个会话的包被拆成多条流,到达间隔特征全是 0。原因:某些 pcap 文件的时间戳是 microseconds 精度,scapy 的pkt.time返回 float,但在 Windows 上精度被截断成毫秒。解决:解析时直接用原始浮点值,不要在解析中途做四舍五入;如果发现时间戳一致,用pkt[Raw].load的长度变化来补充切分会话的依据。这个问题只在 Windows 环境复现,Linux 和 macOS 上比较少遇到,但一旦发生,特征工程整层要重做。

5.3 Dataset 里样本顺序和特征顺序错位,标签张量对不上

现象:训练 loss 正常下降,但验证集 loss 永远在 0.7 左右不降。原因:分开保存特征矩阵和标签数组,中间做过下采样或清洗后忘记同步索引,导致特征和标签错位。解决:特征和标签打包成一个 numpy 结构或字典保存,清洗时同步操作,或每次清洗后用np.random.RandomState(seed).permutation同时打乱两个数组。我把特征和标签写进同一个 npz 文件,读回来直接当训练集,能减少很多低级错误。

5.4 BatchNorm 在验证阶段仍使用训练阶段的动量平均,导致验证 loss 异常

现象:训练 loss 正常下降,验证 loss 在某个 epoch 突然变成 NaN。原因:BatchNorm 的 running_mean 和 running_var 在训练阶段更新,验证阶段用的是累计统计量,但如果训练中途出现过极端 batch,running_var 可能变成负数或无限大。解决:在验证前调用model.eval(),训练前调用model.train(),这是 pytorch 的基本约定。另一个补救办法是把 BatchNorm 换成 LayerNorm,对 batch 大小不敏感,但会牺牲一点训练速度。

5.5 代码里用 list 当 defaultdict 的 key,导致会话聚合失效

现象:流数据量统计数量级不对,正常流量和攻击流量特征一模一样。原因:字典 key 用了可变类型 list,而用四元组或字符串做 key,值不会被正确哈希。解决:构建 key 时用 tuple 包裹所有字段,且保证字段顺序一致。排序后的五元组必须用(src_ip, src_port, dst_ip, dst_port, proto)这个 tuple,少一个字段都不行。这是最不显眼但最致命的一个坑,会静默污染整个下游数据。

6. 进阶验证:构造最小样本集、导出 ONNX 与运行时推断

源码包到了训练完成这一步,严格说还没到「能交付」的状态。我最后会做三件事:第一,构造一个几十条记录的最小样本集,跑通从 pcap 到预测的完整链路,验证代码部署后不会在某个环境变量上翻车;第二,把模型导出成 ONNX,脱离 PyTorch 也能跑;第三,写一个推理用的 Python 脚本,接收 pcap 文件路径输出每条会话的分类结果。

def export_onnx(model, sample_input, path="flow_cnn.onnx"): model.eval() torch.onnx.export( model, sample_input, path, opset_version=13, input_names=["features"], output_names=["logits"], dynamic_axes={"features": {0: "batch"}}, )

sample_input的形状是(1, 20),对应一条会话的特征。dynamic_axes让 batch 维度可变,这样部署时可以一次传多条会话。ONNX 导出成功不代表推理结果和 PyTorch 一致,我通常会在导出前后对同一个样本做推理,比较输出 logits 的最大误差,误差超过 1e-4 就说明哪里没对齐。ONNX Runtime 打开模型后,输入是一个(batch, 20)的 float32 数组,输出直接是 logits,不需要再经过 argmax 以外的任何处理。

回看这个项目,最有价值的不是模型层,而是数据链路层。网络入侵检测的 CNN 结构现在已经被做得很成熟,任何一个拿到了带标签流量数据集的人,用默认的两层卷积都能训出可用的分类器,真正决定项目成败的是解析的准确性、打标的口径、泄漏的规避和评估指标的选择。我自己的教训是死了很多次都在重复这几个老问题,现在我把检查确认输入输出长度的断言放在第一个步骤里,防止一上来就失控。希望这份项目说明能帮你在本地跑通第一个可用的模型,而不是卡在 pcap 解析那一步。

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

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询