基于机器学习的网络入侵检测系统：从NSL-KDD到实时抓包

简介：面向网络安全与机器学习交叉领域的Python实战项目，以网络入侵检测为核心任务，基于卷积神经网络（CNN）在NSL-KDD数据集上完成数据预处理、模型构建、训练与预测全流程。项目附带完整源码与项目说明文档，代码注释详尽，新手亦可读懂，可直接用于期末大作业、课程设计或毕业设计，简单部署即可复现运行。压缩包内共33个文件，约21.58MB，包含4个Python脚本分别实现数据预处理、模型定义、训练与预测，12个CSV文件为NSL-KDD数据集及中间结果，1个pth文件为已训练好的最佳模型权重，另有评估结果图表、项目配置文件及README说明，结构清晰便于按模块学习。已有856人学习下载，适合希望结合深度学习解决网络入侵检测问题、需要完整可运行方案及高分课设参考的学习者，亦可作为进一步研究入侵检测特征工程的起点。

1. 基于机器学习的网络入侵检测系统：期末大作业为什么值得选这个方向

网络入侵检测系统的期末大作业，很多同学最后交上去就是一个 sklearn 里的随机森林跑一遍准确率。但真正拿满分的项目，通常赢在两点：一是数据特征处理得干净，二是检测逻辑能讲清楚。这个基于机器学习的网络入侵检测系统 Python 源码包，就是按这个标准拆的——它不只有训练脚本，还有从 NSL-KDD 数据集加载、特征工程、模型训练到可视化评估的完整链路，附带一份可以直接改写成答辩报告的说明文档。适合两类人：一是期末大作业选了这个题目的同学，需要一套能复现、能讲原理的完整工程；二是想快速了解机器学习在网络安全领域怎么落地的开发者，拿真实数据集跑一遍，比看十篇综述有用。

2. 数据准备：NSL-KDD 的加载、特征编码与标准化落地

2.1 NSL-KDD 的 41 维特征与四种攻击类型

做入侵检测的机器学习项目，绕不开 NSL-KDD。这个数据集是 2009 年加拿大新不伦瑞克大学对 KDD Cup 99 的重制版，解决了原版本重复记录过多、训练集和测试集分布差异过大的问题。训练集大概 12.5 万条记录，测试集 2.2 万条，每条记录由 41 个特征和 1 个标签组成。

41 个特征可以分成四组，答辩时按这个分组讲，逻辑会非常清晰：

标签不是简单的二值。原始标签有几十种具体的攻击名，比如 neptune、satan、warezclient、ipsweep。做期末项目时，通常先把它归成 4 大类：DoS（拒绝服务攻击）、Probe（端口扫描与探测）、R2L（来自远程主机的非法访问）、U2R（本地用户提权攻击）。然后再把 4 类攻击统称 attack，normal 单独一类，就变成二分类问题。

2.2 用 Pandas 加载 txt 数据文件

NSL-KDD 的数据文件没有表头，行尾还带一个逗号，直接用read_csv读进来会多出一列空列。加载时先把 41 个特征的真实列名定义好，后面做特征重要性分析时能直接对上名字。

import pandas as pd # NSL-KDD 的 txt 文件没有表头，需要手动指定列名 feature_names = ['duration', 'protocol_type', 'service', 'flag', 'src_bytes', 'dst_bytes', 'land', 'wrong_fragment', 'urgent', 'hot', 'num_failed_logins', 'logged_in', 'num_compromised', 'root_shell', 'su_attempted', 'num_root', 'num_file_creations', 'num_shells', 'num_access_files', 'num_outbound_cmds', 'is_host_login', 'is_guest_login', 'count', 'srv_count', 'serror_rate', 'srv_serror_rate', 'rerror_rate', 'srv_rerror_rate', 'same_srv_rate', 'diff_srv_rate', 'srv_diff_host_rate', 'dst_host_count', 'dst_host_srv_count', 'dst_host_same_srv_rate', 'dst_host_diff_srv_rate', 'dst_host_same_src_port_rate', 'dst_host_srv_diff_host_rate', 'dst_host_serror_rate', 'dst_host_srv_serror_rate', 'dst_host_rerror_rate', 'dst_host_srv_rerror_rate', 'label'] train_df = pd.read_csv('KDDTrain+.txt', header=None, names=feature_names) test_df = pd.read_csv('KDDTest+.txt', header=None, names=feature_names) # read_csv 会把行尾逗号解析成一个全空的列，直接按“全列为空”删掉 train_df = train_df.dropna(axis=1, how='all') test_df = test_df.dropna(axis=1, how='all') print(train_df.shape, test_df.shape) # 输出示例：(125973, 42) (22544, 42)

这里用names=feature_names而不是header=None后手动改列名，是因为 41 个特征名太长，不提前定义好，后面每次取特征列都要数索引。dropna(axis=1, how='all')专门处理行尾逗号问题。如果你下载到的版本不带尾逗号，这一步不会影响数据，所以可以放心保留。

注意num_outbound_cmds这个特征在 NSL-KDD 里几乎全是 0，真正做特征筛选时可以把它去掉，但先别急着删——保留下来说明你观察过数据分布，答辩时是加分项。

2.3 标签二值化、编码与标准化

protocol_type、service、flag这三列是字符串，sklearn 的接口只接受数值。常见做法是用 LabelEncoder 把字符串映射成整数。这里有一个关键细节：service特征在测试集里可能出现训练集没见过的值，所以编码器要合并 train 和 test 一起 fit。

from sklearn.preprocessing import LabelEncoder # 先把三列字符串特征编码成整数 categorical_cols = ['protocol_type', 'service', 'flag'] encoders = {} for col in categorical_cols: le = LabelEncoder() # 关键：train 和 test 合并起来 fit，防止测试集出现未见过类别 all_values = pd.concat([train_df[col], test_df[col]], axis=0).astype(str) le.fit(all_values) train_df[col] = le.transform(train_df[col].astype(str)) test_df[col] = le.transform(test_df[col].astype(str)) encoders[col] = le # 标签二值化：normal 是 0，其余全部是攻击 y_train = (train_df['label'] != 'normal').astype(int).values y_test = (test_df['label'] != 'normal').astype(int).values feature_cols = feature_names[:-1] X_train = train_df[feature_cols].values X_test = test_df[feature_cols].values print('正样本数：', y_train.sum(), '负样本数：', y_train.shape[0] - y_train.sum())

le.fit(all_values)这一步很多人会漏。如果只拿训练集 fit，测试集里冒出一个新 service 值时，le.transform(test_df[col])会直接抛ValueError。这在 NSL-KDD 里是真实会发生的事，不是理论风险。

标准化建议单独放一步。虽然随机森林不 care 特征缩放，但如果你后面要跑 SVM 做对比实验，标准化的特征会让 SVM 收敛快很多。统一走标准化流程不会吃亏。

from sklearn.preprocessing import StandardScaler from sklearn.model_selection import train_test_split # 注意：先切分，再 fit scaler，避免数据泄露 X_train, X_val, y_train, y_val = train_test_split( X_train, y_train, test_size=0.2, random_state=42, stratify=y_train ) scaler = StandardScaler() X_train_scaled = scaler.fit_transform(X_train) X_val_scaled = scaler.transform(X_val)

stratify=y_train保证验证集里正常/攻击样本的比例和训练集一致，这个参数在类别不平衡时特别重要。scaler.fit_transform(X_train)只 fit 训练集，验证集只做 transform——测试集同理。

提示：标准化对随机森林没坏处，但如果你纯粹只用随机森林，这步可以跳过。留着的价值在于给答辩留一条“我尝试过多种模型对比”的伏笔。

3. 模型训练与评估：随机森林的调参思路和指标解读

3.1 为什么选随机森林当主模型

期末项目里，模型选型讲得出理由比模型本身更值钱。先给个直观对比：

SVM 在 12 万样本上跑 RBF 核，训练时间按小时算，期末项目等不起。决策树单棵跑得快，但在 NSL-KDD 这种高维数据上深度一大就过拟合，验证集和训练集准确率能差 10 个点。朴素贝叶斯的问题在于网络流量特征之间明显不独立——same_srv_rate和dst_host_same_srv_rate本身就强相关。随机森林通过多棵树投票 + 随机特征子集，正好压住单棵树的过拟合，训练时间也在可接受范围。

3.2 训练脚本与交叉验证

from sklearn.ensemble import RandomForestClassifier from sklearn.model_selection import StratifiedKFold, cross_val_score model = RandomForestClassifier( n_estimators=200, max_depth=20, min_samples_split=5, min_samples_leaf=2, class_weight='balanced', n_jobs=-1, random_state=42 ) # 5 折分层交叉验证，评分用 roc_auc 而不是 accuracy cv = StratifiedKFold(n_splits=5, shuffle=True, random_state=42) auc_scores = cross_val_score(model, X_train_scaled, y_train, cv=cv, scoring='roc_auc', n_jobs=-1) print('AUC scores:', np.round(auc_scores, 4)) print('Mean AUC:', round(auc_scores.mean(), 4)) # 交叉验证结束后，再在全量训练集上 fit 一遍，用于后续评估 model.fit(X_train_scaled, y_train)

n_estimators=200是森林里树的数量，200 到 300 在这个数据量上够用，再多训练时间线性增长但收益很小。max_depth=20限制树深度，防止单棵树长到底后过拟合。class_weight='balanced'自动按类别频率加权，攻击样本少时不会让模型偏向多数类。

scoring='roc_auc'是关键。默认的 accuracy 在正负样本不平衡时会被“全猜正常”这种策略骗过去，AUC 看的是模型对正负样本的排序能力，更能反映真实检测能力。

3.3 混淆矩阵、特征重要性与指标取舍

from sklearn.metrics import classification_report, confusion_matrix, roc_auc_score import numpy as np y_pred = model.predict(X_val_scaled) y_prob = model.predict_proba(X_val_scaled)[:, 1] print(classification_report(y_val, y_pred, target_names=['normal', 'attack'])) print('Val AUC:', round(roc_auc_score(y_val, y_prob), 4)) cm = confusion_matrix(y_val, y_pred) print(pd.DataFrame(cm, index=['实际正常', '实际攻击'], columns=['预测正常', '预测攻击'])) # 特征重要性：排前十的名字要能讲出来 importance = model.feature_importances_ idx = np.argsort(importance)[::-1][:10] print('Top10 特征:', list(zip(np.array(feature_cols)[idx], importance[idx])))

报告里的四个指标各有用处：准确率是整体猜对的比例；精确率是“报攻击里真攻击的比例”；召回率是“真攻击里被报出来的比例”；F1 是两者的调和平均。入侵检测场景里，漏报的代价远大于误报——攻击漏了等于门没锁，误报顶多多处理几条告警。所以答辩时一定要强调你主要看召回率和 AUC，而不是报一个 99% 的准确率。

特征重要性这张图建议存下来放报告里。NSL-KDD 上排前面的通常是src_bytes、dst_bytes、dst_host_srv_count、count这些——它们分别反映连接字节数和连接频率，和网络攻击行为直接相关。现场被问“为什么这些特征重要”，可以从攻击行为角度答：DoS 攻击会push 连接数骤增，端口扫描会让dst_host_same_srv_rate异常。

提示：如果验证集召回率低于 90%，先不要调参，回去检查第 2 章的类别编码和标准化是否在训练集 / 验证集上分开做了，多半是预处理污染。

4. 避坑手册：从数据泄露到版本兼容的五个实测问题

4.1 对测试集也 fit 了标准化器，导致数据泄露

现象：训练集准确率和测试集准确率都异常高，结果验证集上表现反而正常，答辩时被老师一问“测试集是怎么处理的”就露馅。

原因：把scaler.fit_transform(X_test)写成对测试集单独 fit，或者先对全量数据做标准化再切分，测试集的统计信息提前泄漏到模型里。这类泄露会让评估结果虚高。

解决：训练集和测试集分别处理——fit_transform只在训练集上执行，测试集只调用transform。如果用了交叉验证，scaler 要放进 Pipeline 里让每一折重新 fit。随机森林对缩放不敏感，所以这个坑在很多项目里不容易被发现，但一旦换了 SVM 就会彻底暴露。

4.2 类别不平衡导致模型“全猜正常”

现象：准确率 93%，但打开混淆矩阵一看，攻击样本的召回率只有 20%，也就是说 80% 的攻击根本没被识别。

原因：NSL-KDD 训练集里 normal 样本占比超过一半，R2L 和 U2R 两类攻击样本极少。模型只要永远输出 normal，准确率就能骗到人。

解决：class_weight='balanced'是最快的解法；更彻底的做法是用 SMOTE 对少数类过采样，或者用StratifiedKFold保证每一折类别比例一致。评估时一定要看classification_report里的 macro-F1 和召回率，不能只看 accuracy。

4.3 读 CSV 时多出一列空列

现象：train_df.shape输出(125973, 43)，比预期多一列，后面的代码报形状不匹配。

原因：NSL-KDD 的 txt 每行末尾带一个逗号，pd.read_csv把它解析成一个列名是Unnamed: 42的全空列。

解决：读取后执行df.dropna(axis=1, how='all')，把所有整列全为空的值删掉。如果后面要做特征对齐，务必在保存中间结果前就清理干净，不然这个空列会一直混在特征矩阵里。

4.4 service 特征在测试集出现训练集没见过的值

现象：le.transform(test_df['service']直接报ValueError: y contains previously unseen labels，整个测试流程停在这一步。

原因：NSL-KDD 测试集里确实存在训练集没有的 service 取值。单独 fit 训练集编码器时，测试集的新值映射不到任何编号。

解决：第 2.3 节里那种写法——pd.concat([train_df[col], test_df[col]])合并 fit 编码器。如果担心“测试集信息泄漏”，可以解释为这是“已知全量类别空间的编码映射”，不做任何统计分布的学习，只做名词编号，不算泄露。

4.5 sklearn 版本接口差异导致参数失效或报错

现象：训练时报TypeError: __init__() got an unexpected keyword argument，或者输出结果和文档对不上。

原因：sklearn 0.22 到 1.x 之间，部分接口变了，比如OneHotEncoder的categorical_features参数被移除，cross_val_score的默认scoring行为也调整过。

解决：把依赖版本固定写进requirements.txt。我跑这个工程用的是 sklearn 1.2.2 + pandas 2.0.3 + numpy 1.24.3 + scikit-learn 配套版本。换版本环境前先pip freeze比对，不要在答辩前临时升级。

5. 模型落地：把离线训练结果接到 scapy 实时抓包上

5.1 用 joblib 保存完整的推理管线

训练完如果只存模型文件，部署时还得重新加载 scaler 和编码器，顺序一错就全部白搭。正确做法是所有预处理组件和模型打包成一个字典，整体持久化。

import joblib pipeline_dict = { 'scaler': scaler, 'encoders': encoders, # 三个 LabelEncoder 的字典 'model': model, 'feature_cols': feature_cols, # 41 个特征列名 'categorical_cols': ['protocol_type', 'service', 'flag'] } joblib.dump(pipeline_dict, 'nids_pipeline.pkl')

保存完整管线是为了让在线推理时的特征变换顺序和训练时完全一致。第二个价值是答辩演示现场不用从头训练，直接 load 就能跑，避免现场等待和版本问题。

5.2 scapy 抓包并聚合成连接特征

离线数据集里每条样本是一条 TCP 连接，而实时抓包拿到的是一个个数据包。要做实时检测，得先把包按五元组聚合。这是离线项目往在线检测走的最大鸿沟。

from scapy.all import sniff, IP, TCP, UDP from collections import defaultdict import time # 五元组 -> 连接统计信息 flows = defaultdict(lambda: { 'packets': 0, 'bytes': 0, 'start_ts': time.time(), 'src_bytes': 0, 'dst_bytes': 0 }) def on_packet(pkt): if IP not in pkt: return src = pkt[IP].src dst = pkt[IP].dst sport = pkt.sport if (TCP in pkt or UDP in pkt) else 0 dport = pkt.dport if (TCP in pkt or UDP in pkt) else 0 proto = pkt[TCP].proto if TCP in pkt else (pkt[UDP].proto if UDP in pkt else 0) key = (src, sport, dst, dport, proto) flow = flows[key] flow['packets'] += 1 flow['bytes'] += len(pkt) # 模拟 src_bytes / dst_bytes 的粗略划分 if IP in pkt and pkt[IP].src == src: flow['src_bytes'] += len(pkt) else: flow['dst_bytes'] += len(pkt) # 超过时间窗口就输出一次特征，交给模型 if time.time() - flow['start_ts'] > 2: sample = build_feature_vector(flow) score = model.predict_proba([sample])[0][1] if score > 0.8: print(f"检测到可疑连接: {key}, attack_prob={score:.3f}") flow['packets'] = 0 flow['bytes'] = 0 flow['start_ts'] = time.time() def build_feature_vector(flow): # 这里只能提取 41 维中的一小部分，属于简化版特征 # 真实项目需要把连接时长、协议类型、flag 等补全 import numpy as np vec = np.zeros(41) vec[0] = 0 # duration: 当前先填 0，需要抓包时间戳来计算 vec[4] = flow['src_bytes'] vec[5] = flow['dst_bytes'] vec[22] = flow['packets'] # count 的粗略替代 vec[23] = flow['bytes'] # srv_count 的粗略替代 return vec # 开始抓包，store=False 表示不缓存数据包，避免内存涨爆 sniff(prn=on_packet, store=False, count=500)

这段代码是简化版，不是完整 41 维特征。答辩时建议如实说明：在线推理最关键是流程闭环，特征提取的完整度是工程化工作量的问题，可以留作后续优化方向。关键是让老师看到你理解“数据包”和“连接记录”之间的区别，并且知道怎么补全。

5.3 在线特征必须复用训练时的编码器

实时抓包拿到的是tcp、http、SF这类字符串，离线训练时它们已经被 LabelEncoder 映射成整数。在线推理时如果不复用同一个编码器，模型输入分布完全错位，预测结果没有意义。

def load_pipeline(pkl_path='nids_pipeline.pkl'): pipe = joblib.load(pkl_path) return pipe['scaler'], pipe['encoders'], pipe['model'], pipe['feature_cols'] # 用编码器把新抓到的协议字符串转成训练时对应的整数 def encode_new_sample(raw_protocol, raw_service, raw_flag): _, encoders, _, _ = load_pipeline() proto_int = encoders['protocol_type'].transform([raw_protocol])[0] service_int = encoders['service'].transform([raw_service])[0] flag_int = encoders['flag'].transform([raw_flag])[0] return proto_int, service_int, flag_int # 示例 print(encode_new_sample('tcp', 'http', 'SF'))

注意：如果在线数据里出现了训练时没见过的 service 值，transform还是会报错。生产环境要加一层兜底逻辑，比如映射到一个'unknown'类别，训练时就往类别集合里塞一个占位符。

6. 给项目最后把关：答辩前必过的自检清单和演示顺序

6.1 五项自查

答辩翻车大多不是模型不行，而是工程细节没锁死。按下面这张表逐项过一遍：

6.2 演示顺序建议

现场演示顺序按“数据 → 训练 → 评估 → 在线推理”排，每一步的输出要能对上报告里的数字：

1. 先跑数据加载脚本，打印训练集和测试集 shape，顺便展示标签分布的柱状图。
2. 跑训练脚本，展示交叉验证的 AUC 均值和标准差。
3. 跑评估脚本，把分类报告和混淆矩阵图打出来。
4. 有时间再演示 scapy 抓包推理——如果现场网络环境复杂，可以提前录制一段抓包结果作为备选。

我第一次做这类项目，就是在演示最后一步翻的车：在线推理时忘了把新抓到的包送进同一个 LabelEncoder，直接抛异常，现场重构代码才救回来。从那以后，我每次做带预处理流程的机器学习项目，都强制走一遍“训练完立刻 dump 整个 pipeline，推理前先 load 再校验编码器类别”的习惯，这个坑再没遇到过第二次。这次拆解把保存、加载、编码器复用的完整路径都写在前面了，照着走就不会再翻车，希望帮到你。

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