简介基于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_features20, num_classes2): super().__init__() self.conv nn.Sequential( nn.Conv1d(1, 64, kernel_size5, padding2), nn.BatchNorm1d(64), nn.ReLU(), nn.Conv1d(64, 64, kernel_size3, padding1), 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 要求的输入形状。两层卷积用的都是 padding2 和 padding1是为了让卷积不缩短特征长度池化才做降维。如果去掉 padding特征长度每过一层卷积就减一截第二层卷积时输入长度只有 18核大小 5 也能算但边界信息的保留方式就不一样了。3.2 训练流程里的两个关键参数梯度裁剪和类别权重训练循环本身不复杂但有两个参数我会特别在意梯度裁剪和类别不平衡权重。网络入侵检测数据集的标签通常极不平衡正常流量可能占 90% 以上直接训练会让模型学会全预测正常类准确率看着很高召回率是零。解决方法是给损失函数传pos_weight放大少数类的梯度信号。另一个是梯度裁剪防止个别异常样本让 loss 爆炸、参数一步跳到坏区域。def train_one_epoch(model, loader, optimizer, criterion, device, clip_norm1.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_norm1.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, dtypetorch.float32) labels torch.tensor(labels, dtypetorch.long) return torch.utils.data.TensorDataset(feats, labels)这个函数看起来短但它把标准化、类型转换都收在了一个入口里。dtypetorch.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_samplesnum_samples, replacementTrue) return sampler权重公式1 / class_count是最简单的逆频率加权少数类的样本每条权重更大。num_samples我一般设成训练集总样本数这样每个 epoch 差不多把训练集完整过一遍。replacementTrue允许同一批数据被重复抽到在做 oversampling 效果的同时不需要复制数据到内存。这个 sampler 要和shuffleFalse一起用因为 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(dim1).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_predargmax(dim1)把模型输出的两个类别的 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 在训练阶段更新验证阶段用的是累计统计量但如果训练中途出现过极端 batchrunning_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, pathflow_cnn.onnx): model.eval() torch.onnx.export( model, sample_input, path, opset_version13, 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 解析那一步。本文还有配套的精品资源点击获取
