用CNN做网络入侵检测：从流量转图像到PyTorch实战

简介一套基于Python与CNN的网络入侵检测项目源码及文档说明面向需要完成机器学习/深度学习期末大作业的高校学生也适合网络安全方向初学者参考实践。项目涵盖数据预处理、模型构建、训练评估与预测全流程代码经过本地编译可运行难度适中并已通过助教老师审定可直接作为课程设计、毕业设计或竞赛练手的基础方案。资源包共33个文件包含12个数据文件如NSL-KDD数据集相关CSV、7个XML工程配置、4个Python脚本如Train.py、Predict.py、PreHandle.py、1个模型权重文件best_model.pth以及文档说明、训练曲线图等压缩包整体约21.58MB目录结构清晰便于按模块阅读与复现。目前已有355人学习下载。读者可以获得完整的入侵检测源码、数据集转换与预处理思路、模型训练与预测的实现细节以及准确率/精确度可视化图表和README说明能够帮助理解CNN在网络安全领域的典型应用并快速改造适配自己的实验场景。1. 从“把流量当成图像”讲起为什么 CNN 能做网络入侵检测很多人第一次看到“Python 实现 CNN 网络入侵检测”这个题目时第一反应是网络流量不是一串 IP、端口、时间戳吗图像卷积怎么会用得上反过来如果你在期末大作业里只做一棵决策树又会发现 95% 准确率经不起一句追问“U2R 攻击的召回率是多少”这个题目真正的隐含要求是把原始网络流量整理成二维矩阵用 CNN 自动学习局部组合特征而不是手工定义规则去匹配攻击特征。它适用的是有标注流量数据NSL-KDD、UNSW-NB15且需要做多分类的场合常见落地是离线训练、在线对会话打分。适合想做入侵检测课程设计并愿意把“可以跑”升级成“讲得清楚”的 IT 从业者和学生阅读。2. 数据预处理从原始 pcap 到 CNN 能吃的“流量图像”CNN 不直接吃 IP 地址和 TCP 端口它吃的是排列规整的二维矩阵。所以这个项目的分水岭不在模型而在预处理你能不能把一条会话变成一张图并且让这张图在不同攻击类型之间呈现可学习的差异。2.1 两种主流做法结构化特征表格 vs 原始报文转图像现在做 CNN 网络入侵检测预处理大致分两派。结构化特征方案直接复用 NSL-KDD 这类数据集的 41 维特征协议类型、服务类型、flag、连接时长、源字节数、目的字节数、相同主机连接数等。把离散字段做 one-hot连续字段归一化reshape 成 1×8×8 或 1×11×11 的灰度图。这个方案的优点是特征已经由专家提取过训练稳定调参路径清晰缺点是“图像”只是数字矩阵的排列视觉语义有限。原始报文转图像方案是从 pcap 里按五元组切出会话每个包抽取长度、TTL、窗口、到达间隔、TCP 标志位做成 30×14 或 32×32 的矩阵。这个方案更接近题目字面意思也让 CNN 的卷积平移不变性真正派上用场。缺点是从抓包到对齐会话、补齐序列每一步都在制造坑。我一般会先跑通结构化特征版本拿一个基础分再把原始 pcap 转图像作为加分项。两条路线都值得写进文档因为它们对应不同的数据集和不同的“为什么用 CNN”的解释路径。2.2 结构化特征方案NSL-KDD 的字段清洗、类别编码与归一化NSL-KDD 的service字段有 70 多个取值直接 one-hot 会多出 70 列但其中大量低频服务其实只出现在极少数样本里。先按出现次数合并低频项再编码维度会干净很多。import pandas as pd from sklearn.preprocessing import MinMaxScaler RARE_SERVICE_THRESHOLD 100 def map_attack_family(label: str) - str: # 把 41 种攻击名归并成 4 个大类便于课程设计里做 5 分类 attack_map { dos: [back, land, neptune, pod, smurf, teardrop], probe: [ipsweep, nmap, portsweep, satan], r2l: [ftp_write, guess_passwd, imap, multihop, phf, spy, warezclient, warezmaster], u2r: [buffer_overflow, loadmodule, perl, rootkit], } for family, members in attack_map.items(): if label in members: return family return normal def prepare_dataframe(path: str) - pd.DataFrame: df pd.read_csv(path) df[label] df[label].str.strip().str.lower() df[attack] df[label].apply(map_attack_family) service_count df[service].value_counts() rare_services service_count[service_count RARE_SERVICE_THRESHOLD].index df.loc[df[service].isin(rare_services), service] other categorical [protocol_type, service, flag] df pd.concat( [df, pd.get_dummies(df[categorical], prefixcategorical)], axis1 ) df df.drop(columnscategorical) return df train prepare_dataframe(NSL_KDD_Train.csv) test prepare_dataframe(NSL_KDD_Test.csv) feature_cols [c for c in train.columns if c not in {label, attack}] scaler MinMaxScaler() scaler.fit(train[feature_cols]) # 只 fit 训练集防止数据泄漏 train[feature_cols] scaler.transform(train[feature_cols]) test[feature_cols] scaler.transform(test[feature_cols])代码逻辑是先把原始攻击名映射成 5 个大类再把低频service合并到other最后用pd.get_dummies做 one-hot。MinMaxScaler会把连续特征压到 0 到 1避免协议类型编码值支配梯度更新。提示MinMaxScaler只能用训练集拟合验证集和测试集直接transform。这是预处理里最常见的泄漏点也是答辩时老师喜欢问的地方。2.3 把流量会话转成图像的具体代码pcap 切会话、抽取字段、生成灰度图如果不想依赖预处理好的数据集可以自己从 pcap 构造“流量图像”。核心思路是同一个五元组来回方向的包归到同一条会话截取前 30 个包每个包抽取 14 个数值字段最终拼成 30×14 的矩阵再 pad 成 32×32。from collections import defaultdict from scapy.all import rdpcap, IP, TCP, UDP import numpy as np MAX_PACKETS 30 FEATURE_DIM 14 def flow_key(pkt): if IP not in pkt: return None src, dst pkt[IP].src, pkt[IP].dst if TCP in pkt: sport, dport, proto pkt[TCP].sport, pkt[TCP].dport, tcp elif UDP in pkt: sport, dport, proto pkt[UDP].sport, pkt[UDP].dport, udp else: sport, dport, proto 0, 0, other # 用排序后的二元组做 key这样客户端和服务端方向会落到同一条会话 key tuple(sorted([(src, sport), (dst, dport)])) (proto,) return key def packet_vec(pkt, prev_ts): vec np.zeros(FEATURE_DIM, dtypenp.float32) vec[0] min(9999.0, float(len(pkt))) # 包长度 if IP in pkt: vec[1] pkt[IP].ttl # 生存时间 if TCP in pkt: vec[2] pkt[TCP].window # 接收窗口 flag_str str(pkt[TCP].flags) for i, flag in enumerate(FSRPAUEC): vec[4 i] 1.0 if flag in flag_str else 0.0 vec[3] 0.0 if prev_ts 0.0 else min(9999.0, pkt.time - prev_ts) return vec def pcap_to_session_matrix(input_file): packets rdpcap(input_file) flows defaultdict(list) prev_ts defaultdict(float) for pkt in packets: key flow_key(pkt) if key is None: continue vec packet_vec(pkt, prev_ts.get(key, 0.0)) prev_ts[key] pkt.time if len(flows[key]) MAX_PACKETS: flows[key].append(vec) images [] for _, vecs in flows.items(): mat np.zeros((MAX_PACKETS, FEATURE_DIM), dtypenp.float32) mat[:len(vecs)] np.stack(vecs) row_min mat.min(axis1, keepdimsTrue) row_max mat.max(axis1, keepdimsTrue) mat (mat - row_min) / (row_max - row_min 1e-8) img mat.reshape(1, 20, 21) img np.pad(img, ((0, 0), (6, 6), (5, 6)), modeconstant) images.append(img) return imagesMAX_PACKETS30控制了会话长度长会话会被截断短会话补零14 个字段里前 4 个是连续数值后 8 个是 TCP 标志位最后两个位留作方向或扩展字段。你会注意到 30×14 被 reshape 成 20×21因为 30×1442020×21420再 pad 成 32×32后就能直接进入最常见的 CNN 输入尺寸。这里真正决定效果的不是卷积层而是“会话如何切、字段如何选”。把MAX_PACKETS从 30 改成 50矩阵会变大、计算变慢但短会话攻击反而容易被稀释对期末大作业来说30 包起步是合理经验值。2.4 类别不平衡与数据划分label 映射和分层抽样NSL-KDD 的多分类任务里Normal、DoS、Probe 样本量很大R2L、U2R 很少。直接把数据集扫进模型CNN 会学成一个“永远说 normal”的分类器指标照样能到 85% 以上。大类对应的原始攻击名常见数量级Normalnormal数万条DoSback、neptune、pod、smurf、teardrop、land数万条Probeipsweep、nmap、portsweep、satan数千条R2Lguess_passwd、imap、warezclient、warezmaster 等数百到数千条U2Rbuffer_overflow、loadmodule、perl、rootkit几十到几百条针对这个分布训练集内部也要按类别比例切出验证集from sklearn.model_selection import train_test_split train_df, val_df train_test_split( all_df, test_size0.2, stratifyall_df[attack], # 按 attack 列分层保证每类比例一致 random_state42 ) class_counts train_df[attack].value_counts() class_weight class_counts.max() / class_countsstratify会按attack列的比例抽样避免某一类只在训练集或只在验证集里出现。后面的class_weight是给少数类更高的损失权重传给CrossEntropyLoss时能明显拉高 R2L 和 U2R 的召回。3. 搭建 CNN 分类模型网络结构、损失函数与训练参数数据变成 32×32 的灰度图后CNN 的任务就是从这些矩阵里找出攻击特有的局部模式。这个阶段最容易出问题的不是网络不够深而是参数没有对齐输入尺寸。3.1 对小尺寸“流量图像”友好的卷积核与池化设计流量图像不是自然图像它没有明显的边缘和纹理所以不要照搬 ImageNet 上的大卷积核。常见做法是用 3×3 小卷积核堆两层到三层每层之间加 BatchNorm 和 ReLU再用 MaxPool 降低分辨率。输入只有 32×32三层池化后特征图变成 4×4这个尺寸足够把高层语义交给全连接层。选择 3×3 而不是 5×5是因为小卷积核参数量更少对 32×32 这种小图来说收敛更快也不容易过早把空间信息压没。padding1保证卷积前后尺寸不变MaxPool 的 stride2 负责真正的降采样。3.2 完整 PyTorch 训练代码模型定义、训练循环、早停下面是一套可以直接套用的 PyTorch 代码。模型输入是 1×32×32 的张量输出是 5 个类别的 logits。import torch from torch import nn from torch.utils.data import DataLoader, TensorDataset from sklearn.metrics import f1_score class FlowCNN(nn.Module): def __init__(self, num_classes: int 5): super().__init__() self.features nn.Sequential( nn.Conv2d(1, 32, kernel_size3, padding1), nn.BatchNorm2d(32), nn.ReLU(inplaceTrue), nn.MaxPool2d(2), nn.Conv2d(32, 64, kernel_size3, padding1), nn.BatchNorm2d(64), nn.ReLU(inplaceTrue), nn.MaxPool2d(2), nn.Conv2d(64, 128, kernel_size3, padding1), nn.BatchNorm2d(128), nn.ReLU(inplaceTrue), nn.MaxPool2d(2) ) self.classifier nn.Sequential( nn.Flatten(), nn.Dropout(0.5), nn.Linear(128 * 4 * 4, 256), nn.ReLU(inplaceTrue), nn.Dropout(0.3), nn.Linear(256, num_classes) ) def forward(self, x): return self.classifier(self.features(x)) def evaluate(model, loader): model.eval() preds, labels [], [] with torch.no_grad(): for x, y in loader: logits model(x.to(device)) preds.extend(logits.argmax(dim1).cpu().tolist()) labels.extend(y.cpu().tolist()) return f1_score(labels, preds, averagemacro) model FlowCNN(num_classes5).to(device) class_weight torch.tensor([1.0, 1.0, 1.3, 5.0, 12.0], dtypetorch.float32).to(device) criterion nn.CrossEntropyLoss(weightclass_weight) optimizer torch.optim.AdamW(model.parameters(), lr1e-3, weight_decay5e-4) scheduler torch.optim.lr_scheduler.CosineAnnealingLR(optimizer, T_max40) train_loader DataLoader(train_ds, batch_size64, shuffleTrue, num_workers2, pin_memoryTrue) val_loader DataLoader(val_ds, batch_size128, shuffleFalse) best_f1, patience 0.0, 0 for epoch in range(60): model.train() train_loss 0.0 for x_batch, y_batch in train_loader: x_batch, y_batch x_batch.to(device), y_batch.to(device) optimizer.zero_grad() logits model(x_batch) loss criterion(logits, y_batch) loss.backward() optimizer.step() train_loss loss.item() * x_batch.size(0) train_loss / len(train_loader.dataset) val_f1 evaluate(model, val_loader) print(fepoch{epoch:02d} loss{train_loss:.4f} val_f1{val_f1:.4f}) if val_f1 best_f1: best_f1 val_f1 torch.save(model.state_dict(), flow_cnn_best.pt) patience 0 else: patience 1 if patience 8: break scheduler.step()这里模型剪枝到了三层卷积每层只加一个 BatchNorm没有魔改结构。训练循环里每轮结束后计算验证集 macro F1F1 不涨就计数连续 8 轮不涨就早停保存的是验证集表现最好的权重而不是最后一轮的权重。CosineAnnealingLR配合T_max40让学习率在 60 轮里平滑衰减避免后期在局部震荡。class_weight是按类别频率反比设置的示例值。如果你用的数据集分布不同应该用 2.4 节算出的class_counts.max() / class_counts动态生成而不是写死。3.3 训练参数怎么设epoch、batch、学习率与数据增强陷阱参数常见取值为什么这么定输入尺寸1×32×32兼容经典 CNN 结构计算量小batch_size32 或 64流量图像不占显存64 足够稳定初始学习率1e-3AdamW 下收敛较快配合调度器衰减权重衰减5e-4抑制全连接层过拟合Dropout0.5 / 0.3特征层和分类层各放一个patience8验证集 F1 连续 8 轮不升就停数据增强基本不用随机翻转、平移会破坏包顺序语义最后一行值得多说一句流量图像和 CIFAR-10 不一样水平翻转等于把同一条会话的方向反转语义完全变了。常见的RandomHorizontalFlip在这里属于错误的数据增强不要因为准确率掉了才回头查。3.4 用 TensorBoard 看收敛和过拟合训练时除了打印 F1还可以把 train loss 和 val loss 一起写进 TensorBoard观察两边的间隔。from torch.utils.tensorboard import SummaryWriter writer SummaryWriter(runs/flow_cnn) for epoch in range(num_epochs): # train_loss 和 val_loss 在上面训练循环里已经算好 writer.add_scalars(loss, { train: train_loss, val: val_loss, }, epoch) writer.add_scalar(val/f1_macro, val_f1, epoch) writer.close()启动命令是tensorboard --logdir runs --port 6006。如果 train loss 一直降、val loss 在第 15 轮反弹而 val F1 没有跟着涨多半是全连接层过拟合优先调大 Dropout 或减小256这个隐藏层宽度。4. 评估与调优从准确率到误报率期末评分看什么期末评分最看重的不一定是“准确率有多高”而是“你能不能解释每个指标代表什么”。CNN 网络入侵检测里的准确率很容易虚高因为 Normal 和 DoS 占了大多数。4.1 多分类评估指标的取舍在 NSL-KDD 五分类里准确率只能说明总体判断对的比重。R2L 和 U2R 样本很少哪怕模型把它们全分错准确率也只掉一两个百分点。所以要同时看每个类别的 Precision、Recall、F1尤其是少数类的 Recall。入侵检测里还有一对指标常常被写进文档检测率 DRDetection Rate和误报率 FPR。DR 就是攻击类的 RecallFPR 是正常流量被误判成攻击的比例。好不好看要看在低误报率下能不能保住高检测率。4.2 用 sklearn 输出混淆矩阵、精确率、召回率、F1 并画图模型训练完后直接在测试集上做一次完整评估from sklearn.metrics import classification_report, confusion_matrix, ConfusionMatrixDisplay import numpy as np model.load_state_dict(torch.load(flow_cnn_best.pt)) model.eval() y_true, y_prob [], [] with torch.no_grad(): for x, y in test_loader: out model(x.to(device)) y_prob.append(torch.softmax(out, dim1).cpu().numpy()) y_true.extend(y.cpu().tolist()) y_true np.array(y_true) y_prob np.concatenate(y_prob, axis0) y_pred np.argmax(y_prob, axis1) target_names [Normal, DoS, Probe, R2L, U2R] print(classification_report(y_true, y_pred, target_namestarget_names, digits4)) cm confusion_matrix(y_true, y_pred) disp ConfusionMatrixDisplay(cm, display_labelstarget_names) disp.plot(cmapBlues)classification_report会输出每个类别的精确率、召回率和 F1digits4是为了应付答辩时“准确率怎么还是四舍五入”的问题。ConfusionMatrixDisplay能把混淆矩阵画成图写进文档时比只贴数字直观得多。4.3 高误报场景下的阈值调整与代价敏感训练如果测试结果是攻击类误报太多可以先看正常类被误分到哪个攻击类。多数情况下CNN 会把正常流量预测成 Probe因为两者的时间窗口特征很接近。这时候可以调整判定阈值而不是重新训练。normal_id 0 attack_score 1 - y_prob[:, normal_id] # 越接近 1 越像攻击 for threshold in [0.5, 0.7, 0.9]: pred_attack (attack_score threshold).astype(int) y_true_binary (y_true ! normal_id).astype(int) tp ((pred_attack 1) (y_true_binary 1)).sum() fp ((pred_attack 1) (y_true_binary 0)).sum() fn ((pred_attack 0) (y_true_binary 1)).sum() print(fthreshold{threshold} FPR{fp/(fp (y_true_binary0).sum()):.4f} fDR{tp/(tpfn):.4f})把阈值从 0.5 提到 0.9会让模型更保守只有足够确信才判为攻击。这样误报率会降但少量攻击也会漏掉。更好的做法是在训练时给少数类更高的class_weight让模型对 U2R 和 R2L 更敏感然后再结合阈值调整。4.4 常见过拟合与数据泄漏点排查训练结束前按这个清单过一遍归一化统计量有没有泄漏。MinMaxScaler只能 fit 训练集测试集和验证集一律 transform。同一条会话的包有没有同时出现在训练集和验证集。pcap 转图像时先按flow_key分组再按会话 ID 划分不能按包划分。切分前有没有分层。train_test_split不加stratifyU2R 这种小类很容易从验证集消失。验证集 F1 是否和训练集 F1 差距过大。差距超过 0.1 就说明过拟合优先加 Dropout而不是换更深的网络。少数类指标是否全部为 0。如果 U2R 的 recall 是 0先看是不是类别权重没传给损失函数或者数据量太少导致梯度被多数类淹没。5. 把模型用于新流量实时检测的验证方法与落地形态模型在测试集上拿到 90% 多只能说明它在同一分布的数据上有效。真正要验证的是给一条没见过的 pcap它还能不能给出合理的结论。5.1 用测试集以外的 pcap 验证泛化能力把pcap_to_session_matrix和训练好的模型拼起来跑一条全新流量from scapy.all import wrpcap, IP, TCP, Raw pkt IP(src10.0.0.8, dst192.168.1.20) / \ TCP(sport12345, dport80, flagsS) / \ Raw(bGET / HTTP/1.1\r\nHost: example.com\r\n\r\n) wrpcap(one_request.pcap, [pkt]) mat pcap_to_session_matrix(one_request.pcap)[0] img torch.from_numpy(mat).unsqueeze(0).to(device) with torch.no_grad(): proba torch.softmax(model(img), dim1) print(dict(zip(target_names, proba.cpu().numpy()[0].tolist())))小 pcap 只有一两个包补零后很难被判定为攻击这符合直觉。真正有意义的验证方式是抓一小段混合流量里面手工插入一条已知攻击会话再观察模型是否输出对应类别。5.2 用 Grad-CAM 确认 CNN 到底在看哪些特征答辩时最怕被问“你的 CNN 看到了什么”。可以用 Grad-CAM 把最后一层卷积的激活热力图叠加到输入图像上直接证明模型在关注矩阵的哪几行。from torch.nn import functional as F def grad_cam(model, x, target_layer_idx8): model.eval() activations {} def forward_hook(module, input, output): activations[value] output.detach() def backward_hook(module, grad_input, grad_output): activations[grad] grad_output[0].detach() layer model.features[target_layer_idx] layer.register_forward_hook(forward_hook) layer.register_full_backward_hook(backward_hook) x x.clone().requires_grad_(True) out model(x) pred out.argmax(dim1) out[0, pred].backward() weights activations[grad].mean(dim(2, 3), keepdimTrue) cam (weights * activations[value]).sum(dim1, keepdimTrue) cam F.relu(cam).squeeze().cpu().numpy() return camtarget_layer_idx8对应self.features第三层卷积。热力图里亮的位置如果集中在包长度和窗口这几列说明模型主要在学流量规模特征如果散布在 TCP 标志位列说明它更依赖连接状态。这张图比任何口头解释都有说服力。5.3 封装成 REST 服务的快速做法课程设计如果要演示“实时检测”不用做复杂界面一个 FastAPI 接口就够了。from fastapi import FastAPI, UploadFile, File app FastAPI() app.post(/predict) async def predict_pcap(file: UploadFile File(...)): with open(upload.pcap, wb) as f: f.write(await file.read()) mat pcap_to_session_matrix(upload.pcap) if not mat: return {detail: pcap is empty} img torch.from_numpy(mat[0]).unsqueeze(0).to(device) with torch.no_grad(): proba torch.softmax(model(img), dim1) pred proba.argmax(dim1).item() return {label: target_names[pred], confidence: proba[0, pred].item()}接口不做分片上传也不处理超大 pcap但演示“上传抓包文件、返回攻击类别和置信度”已经足够。生产环境还要加流量分割、会话窗口滑动和结果聚合那是另一个话题。5.4 几个会影响检测效果的工程细节别把模型文件单独丢在根目录建议用models/、utils/、api/三层结构让答辩老师一眼能看出预处理、模型、接口的边界。pcap 转图像时如果文件很大rdpcap会把整个文件读进内存改用PcapReader流式读取能省不少内存。类别名列表要固定顺序测试脚本里也必须用同一份target_names否则预测结果和可视化对不上。上线前还要确认训练数据的类别分布和实际网络流量是否一致。如果一个内网里 99% 都是 Web 访问训练集却有一半 DoS真实误报率会和测试集指标差很多。这时候最实用的做法是重新切时间窗口按“最近一周”的流量重训一次而不是抱着 NSL-KDD 模型一直用下去。把这条 Grad-CAM 热力图和同一时段的告警事件放在一张截图里才是这份大作业最值得写进文档的部分。本文还有配套的精品资源点击获取