简介这是一份基于Python机器学习的入侵检测系统项目源码面向希望入门或进阶网络安全与机器学习方向的学习者可作为毕业设计、课程设计、大作业或工程实训的参考方案。项目围绕网络流量嗅探与异常检测展开包含数据预处理、特征处理与SVM等机器学习算法模块帮助读者理解从数据采集到模型判别的完整流程。压缩包共21个文件以xml配置、py脚本、md说明文档及gitignore等工程辅助文件为主整体约29KB结构轻量、便于快速阅读与二次开发。其中Python脚本承担核心检测逻辑xml与md文件用于项目配置和说明适合对照代码梳理算法实现思路。目前已有91人学习下载可作为入门机器学习安全应用的实践起点也能为毕设选题与项目立项提供可运行的代码框架与排错参考。1. 从零搭一套 Python 机器学习入侵检测系统先搞清楚它在防什么很多人第一次听到「基于 Python 机器学习的入侵检测系统」脑子里浮现的是电影里那种绿色代码瀑布实际上它干的事情非常朴素把网络流量或者主机日志变成一行行特征向量喂给分类器让模型判断这条记录是正常访问还是攻击行为。传统方案靠 Snort、Suricata 这类规则引擎写一条规则只能拦一种已知攻击遇到变种就抓瞎机器学习方案的价值在于它不依赖人工写规则而是从历史数据里学出「正常长什么样」偏离这个分布的流量就会被标出来。这套东西适合谁适合有 Python 基础、想做一个能写进简历或课程设计的完整安全项目的人也适合运维和安全岗想给现有 IDS 加一层异常检测能力的人。它不能替代防火墙但能补上规则引擎漏掉的那部分。2. 数据从哪来、特征怎么选NSL-KDD 与 CIC-IDS 的取舍2.1 为什么新手先用 NSL-KDD 而不是直接上真实流量真实企业流量拿不到拿到了也没有标注没有标注就没法做监督学习。所以第一步一定是找公开数据集。入侵检测领域最经典的是 KDD Cup 99但它有两个硬伤大量重复记录导致模型虚高测试集和训练集分布不一致。NSL-KDD 就是针对这两个问题做的修正版去掉了冗余记录训练集 125973 条、测试集 22544 条每条 41 个特征加 1 个标签标签分五大类Normal、DoS、Probe、R2L、U2R。我一般建议新手从这里起步原因是特征含义清晰、类别不平衡程度可控、网上参考资料多踩坑成本低。CIC-IDS 2017/2018 更接近现代流量特征是用 CICFlowMeter 提取的流统计量比如流持续时间、包间隔均值、前向包长度最大值维度到 70 多。它的问题是数据量大、类别极不平衡U2R 类样本可能只有几十条直接训练会让模型完全忽略小类。所以选型逻辑是先跑通 NSL-KDD 建立完整流程再迁移到 CIC-IDS 做工程化验证。2.2 41 维特征里哪些真正有用NSL-KDD 的 41 维分四组TCP 连接基本特征duration、protocol_type、service、flag 等 9 维、流量统计特征src_bytes、dst_bytes、count、srv_count 等 13 维、主机统计特征dst_host_count、dst_host_srv_count 等 9 维、内容特征num_failed_logins、root_shell、num_file_creations 等 10 维。实际做下来protocol_type、service、flag 这三个离散特征必须做编码src_bytes 和 dst_bytes 的分布极度长尾直接标准化效果一般取对数后更稳。import pandas as pd import numpy as np from sklearn.preprocessing import LabelEncoder, StandardScaler # NSL-KDD 列名官方文档给的直接抄 col_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,difficulty] train pd.read_csv(KDDTrain.txt, namescol_names) test pd.read_csv(KDDTest.txt, namescol_names) # 把 40 多种攻击标签归成 5 大类这是 NSL-KDD 的标准做法 dos [back,land,neptune,pod,smurf,teardrop,apache2,udpstorm,processtable,worm] probe [satan,ipsweep,nmap,portsweep,mscan,saint] r2l [guess_passwd,ftp_write,imap,phf,multihop,warezmaster,warezclient, spy,xlock,xsnoop,snmpguess,snmpgetattack,httptunnel,sendmail,named] u2r [buffer_overflow,loadmodule,rootkit,perl,sqlattack,xterm,ps] def to_category(label): if label normal: return normal for name, group in [(dos,dos),(probe,probe),(r2l,r2l),(u2r,u2r)]: if label in group: return name return unknown train[category] train[label].apply(to_category) test[category] test[label].apply(to_category) # 三个离散特征做 LabelEncoder注意用训练集 fit测试集 transform for col in [protocol_type,service,flag]: le LabelEncoder() train[col] le.fit_transform(train[col]) # 测试集里可能出现训练集没见过的类别统一映射成 -1 test[col] test[col].map(lambda x: le.transform([x])[0] if x in le.classes_ else -1) # 数值特征做对数变换再标准化缓解长尾 skewed [src_bytes,dst_bytes,duration,count,srv_count] for col in skewed: train[col] np.log1p(train[col]) test[col] np.log1p(test[col]) scaler StandardScaler() feature_cols [c for c in col_names if c not in [label,difficulty]] train[feature_cols] scaler.fit_transform(train[feature_cols]) test[feature_cols] scaler.transform(test[feature_cols])这段代码的逻辑是先按官方列名读入再把细粒度攻击标签归并成 5 大类这样类别数从 40 多降到 5模型更容易学。离散特征用 LabelEncoder 编码关键点是测试集可能出现训练集没见过的 service 值比如训练集没有http_8001而测试集有直接 transform 会报错所以用 map 加判断兜底。数值特征先 log1p 再标准化是因为 src_bytes 这类特征最大值能到 10 的 9 次方不处理的话标准化后大部分值挤在 0 附近树模型还好逻辑回归和 SVM 会很难受。提示测试集里出现训练集未见过的类别是入侵检测的常态真实流量永远有新服务、新协议编码阶段必须留兜底逻辑否则上线就崩。3. 模型选型与训练随机森林、XGBoost 还是深度学习3.1 先跑基线随机森林为什么是入侵检测的默认选择入侵检测数据是典型的结构化表格数据特征维度几十到几百样本量几万到几百万。这个场景下梯度提升树XGBoost、LightGBM和随机森林几乎总是比深度学习强原因是树模型对特征尺度不敏感、能自动处理特征交互、训练快、可解释性好。随机森林作为基线在 NSL-KDD 上五分类准确率通常能到 99% 左右注意这是 NSL-KDD 的特性换到 CIC-IDS 会掉到 95% 以下。from sklearn.ensemble import RandomForestClassifier from sklearn.metrics import classification_report, confusion_matrix X_train train[feature_cols].values y_train train[category].values X_test test[feature_cols].values y_test test[category].values rf RandomForestClassifier( n_estimators200, # 树的数量100~300 之间够用再多收益递减 max_depth20, # 限制深度防过拟合NSL-KDD 上 15~25 都行 min_samples_leaf2, # 叶子最少样本数小类样本少时别设太大 class_weightbalanced, # 类别不平衡时自动加权U2R 只有几十条全靠它 n_jobs-1, random_state42 ) rf.fit(X_train, y_train) y_pred rf.predict(X_test) print(classification_report(y_test, y_pred, digits4)) print(confusion_matrix(y_test, y_pred))参数说明n_estimators 是森林里树的数量200 是精度和速度的平衡点超过 300 后准确率提升通常不到 0.1%。max_depth 控制单棵树深度不限制的话树会一直长到每个叶子只有一个样本训练集 100% 准确但测试集拉胯。min_samples_leaf 设 2 是为了避免小类被切得太碎。class_weightbalanced 是关键NSL-KDD 里 U2R 只有 52 条不设这个参数模型会直接把它当噪声忽略召回率接近 0。3.2 换 XGBoost 能提升多少什么时候值得换随机森林的短板是对难分样本的拟合能力有限XGBoost 用梯度提升 正则化在同样的特征上通常能把小类召回率再拉高几个点。代价是调参更麻烦、训练更慢、可解释性差一些。我的经验是如果随机森林的 U2R 和 R2L 召回率低于 60%换 XGBoost 值得如果已经到 80% 以上提升空间不大不如把精力花在特征工程上。import xgboost as xgb from sklearn.utils.class_weight import compute_sample_weight # 多分类要先把标签转成 0~4 的整数 label_map {normal:0,dos:1,probe:2,r2l:3,u2r:4} y_train_num np.array([label_map[c] for c in y_train]) y_test_num np.array([label_map[c] for c in y_test]) # 用 sample_weight 处理不平衡比 scale_pos_weight 更适合多分类 sample_weights compute_sample_weight(balanced, y_train_num) xgb_clf xgb.XGBClassifier( n_estimators300, max_depth6, # 树模型深度别超过 8否则容易过拟合 learning_rate0.1, # 学习率0.05~0.1 之间调小要加 n_estimators subsample0.8, # 行采样防过拟合 colsample_bytree0.8, # 列采样 objectivemulti:softmax, num_class5, eval_metricmlogloss, tree_methodhist, # 大数据集用 hist 加速 random_state42 ) xgb_clf.fit(X_train, y_train_num, sample_weightsample_weights) y_pred_xgb xgb_clf.predict(X_test) print(classification_report(y_test_num, y_pred_xgb, digits4))XGBoost 的参数逻辑和随机森林不同max_depth 默认 6 就够因为 boosting 是串行叠加每棵树只学残差不需要很深。learning_rate 和 n_estimators 是一对学习率调小到 0.05 时 n_estimators 要相应加到 500 以上。subsample 和 colsample_bytree 是防过拟合的核心0.8 是常用起点。tree_methodhist 在样本量超过 10 万时比默认的 exact 快好几倍。3.3 深度学习什么时候才值得上很多人一上来就想用 CNN、LSTM觉得「深度学习」听起来高级。但在 NSL-KDD 这种表格数据上深度学习几乎不可能超过 XGBoost原因是表格数据的特征没有空间或时序结构CNN 的卷积核在特征维度上滑动没有物理意义。深度学习真正有价值的场景是原始流量字节序列直接输入不需要人工提特征、或者日志文本序列。如果你做的是「把 pcap 原始字节喂给一维 CNN」那深度学习是对的选择如果你做的是「41 维特征分类」老老实实用树模型。4. 类别不平衡与误报率IDS 落地最疼的两个问题4.1 U2R 只有几十条样本怎么让模型不忽略它NSL-KDD 训练集里 normal 67343 条、dos 45927 条、probe 11656 条、r2l 995 条、u2r 52 条。52 条对 67343 条比例 1:1295。任何不做处理的分类器都会把 U2R 全部预测成 normal因为这样损失最小。处理手段有三层第一层是算法层面的 class_weight 或 sample_weight前面已经用了第二层是采样层面对少数类做 SMOTE 过采样或对多数类做欠采样第三层是阈值层面不用 argmax 输出而是对少数类调低判定阈值。from imblearn.over_sampling import SMOTE from collections import Counter print(原始分布:, Counter(y_train_num)) # SMOTE 只在训练集上做测试集绝对不能动 smote SMOTE( sampling_strategy{u2r:500, r2l:2000}, # 指定每个类采样到多少条 k_neighbors3, # U2R 只有 52 条k 必须小于样本数设 3 比较稳 random_state42 ) X_res, y_res smote.fit_resample(X_train, y_train_num) print(SMOTE 后分布:, Counter(y_res))参数说明sampling_strategy 用字典精确控制每个类采样到多少比 auto 更可控因为 auto 会把所有类都采到和多数类一样多训练集膨胀到 30 多万条训练时间翻几倍。k_neighbors 默认 5但 U2R 只有 52 条k 太大会导致生成的样本和原始样本过于相似设 3 更稳。SMOTE 必须在训练集上做测试集保持原始分布否则评估结果虚高这是血泪教训。4.2 误报率比准确率重要IDS 的真实评价指标准确率 99% 听起来很好但如果你的系统每天处理 100 万条流量1% 误报就是 1 万条告警安全分析师根本看不过来最后的结果是所有人都不看告警了。IDS 落地真正看的指标是在召回率检测率不低于 95% 的前提下误报率False Positive Rate能压到多低。FPR FP / (FP TN)在 NSL-KDD 上好的模型能压到 1% 以下真实流量上 5% 以内就算不错。from sklearn.metrics import roc_auc_score, precision_recall_curve import matplotlib.pyplot as plt # 多分类算 macro AUC看整体区分能力 y_prob xgb_clf.predict_proba(X_test) auc roc_auc_score(y_test_num, y_prob, multi_classovr, averagemacro) print(fMacro AUC: {auc:.4f}) # 针对二分类场景正常 vs 攻击画 PR 曲线找最佳阈值 y_test_binary (y_test_num ! 0).astype(int) y_prob_attack 1 - y_prob[:, 0] # 1 减去 normal 的概率 precision, recall, thresholds precision_recall_curve(y_test_binary, y_prob_attack) # 找召回率 0.95 时 precision 最高的阈值 valid_idx np.where(recall 0.95)[0] best_idx valid_idx[np.argmax(precision[valid_idx])] print(f阈值{thresholds[best_idx]:.4f}, 召回{recall[best_idx]:.4f}, 精确{precision[best_idx]:.4f})这段代码的核心思路是不直接用模型默认的 0.5 阈值而是根据业务需求召回率不低于 95%反推阈值。precision_recall_curve 返回的 thresholds 长度比 precision 少 1索引时要注意。实际部署时这个阈值要定期用新数据重新校准因为流量分布会漂移。注意SMOTE 之后如果还用默认 0.5 阈值误报率会明显上升因为模型被训练得更倾向于预测少数类。阈值必须在验证集上单独调。5. 避坑与排查从训练到部署的 5 个真实翻车现场5.1 测试集准确率 99% 但上线全是误报现象本地测试集 classification_report 显示 accuracy 0.99部署到真实流量后告警刷屏90% 是误报。原因NSL-KDD 测试集和训练集来自同一分布但真实流量分布完全不同模型学到的「正常模式」在真实环境里不成立。解决上线前必须用真实流量的正常样本做一次验证如果 FPR 超过 10%说明模型过拟合了数据集的特定分布需要重新用真实数据微调或者加一层规则过滤。5.2 LabelEncoder 在测试集上直接报 ValueError现象le.transform(test[col])抛出 y contains previously unseen labels。原因测试集出现了训练集没有的 service 值比如训练集没有http_8001。解决用 map 加判断兜底未见过的类别统一映射成一个特殊值比如 -1或者改用 OneHotEncoder 的 handle_unknownignore 参数。这个坑在真实流量上必然遇到因为新服务每天都在出现。5.3 SMOTE 之后模型训练极慢且内存爆掉现象对 12 万条数据做 SMOTE 后训练集变成 30 多万条XGBoost 训练时间从 2 分钟变成 20 分钟内存占用翻倍。原因sampling_strategyauto 会把所有少数类都采到和多数类一样多。解决用字典精确指定每个类的采样目标比如 U2R 采到 500、R2L 采到 2000而不是无脑 auto。或者改用代价敏感学习直接调 class_weight不生成新样本。5.4 特征标准化在训练集和测试集上分别 fit现象模型在测试集上表现忽好忽坏换一批数据就崩。原因有人图省事对训练集和测试集分别做了StandardScaler().fit_transform()导致两者的均值和方差不同模型看到的特征尺度不一致。解决scaler 只能在训练集上 fit测试集用同一个 scaler 做 transform。这个错误在交叉验证时更隐蔽必须用 Pipeline 把 scaler 和模型包在一起。5.5 模型文件用 pickle 保存后换环境加载失败现象本地用 pickle 保存的模型部署到服务器上pickle.load报错提示版本不兼容或找不到某个模块。原因pickle 序列化的是 Python 对象依赖具体的库版本和类路径sklearn 版本不一致就会崩。解决用 joblib 替代 picklesklearn 官方推荐或者更稳妥的做法是把模型导出成 ONNX 格式跨环境加载不依赖 Python 版本。如果只是内部用joblib 加固定 requirements.txt 也能解决。6. 把模型变成能用的服务Flask 接口与在线检测的工程细节训练完模型只是完成了一半真正能用起来需要把它包成一个 HTTP 服务让流量采集端能实时调用。我一般用 Flask 起一个轻量接口输入是一条流量的特征 JSON输出是预测类别和置信度。这里有个容易忽略的点在线检测时特征提取必须和训练时完全一致包括离散特征的编码映射、数值特征的 log 变换和标准化参数这些都要在服务启动时加载好不能每次请求重新 fit。import joblib import numpy as np from flask import Flask, request, jsonify app Flask(__name__) # 启动时加载模型和预处理器只加载一次 model joblib.load(xgb_ids_model.joblib) scaler joblib.load(scaler.joblib) encoders joblib.load(label_encoders.joblib) feature_cols joblib.load(feature_cols.joblib) label_inv {0:normal,1:dos,2:probe,3:r2l,4:u2r} app.route(/predict, methods[POST]) def predict(): data request.get_json() # 按训练时的列顺序组装特征顺序错了结果全错 row [] for col in feature_cols: val data.get(col, 0) if col in encoders: le encoders[col] val le.transform([val])[0] if val in le.classes_ else -1 row.append(val) arr np.array([row], dtypefloat) # 对偏态特征做同样的 log1p再标准化 for i, col in enumerate(feature_cols): if col in [src_bytes,dst_bytes,duration,count,srv_count]: arr[0, i] np.log1p(arr[0, i]) arr scaler.transform(arr) prob model.predict_proba(arr)[0] pred int(np.argmax(prob)) return jsonify({ label: label_inv[pred], confidence: float(prob[pred]), all_probs: {label_inv[i]: float(p) for i, p in enumerate(prob)} }) if __name__ __main__: app.run(host0.0.0.0, port5000)这段代码的关键在于特征列顺序必须和训练时完全一致所以用 feature_cols 列表来保证顺序而不是依赖字典的遍历顺序。离散特征的编码器在启动时加载请求时直接 transform未见过的值映射成 -1。log 变换和标准化必须和训练时用同样的参数所以 scaler 也是加载而不是重新 fit。返回结果里带上 all_probs 是为了后续做阈值调整前端可以根据业务需求决定告警级别。部署时还有几个工程细节值得注意。第一Flask 自带的开发服务器不能用于生产要用 gunicorn 或 uwsgi 起多进程命令是gunicorn -w 4 -b 0.0.0.0:5000 app:app-w 4 表示 4 个 worker 进程。第二模型推理是 CPU 密集型worker 数量一般设成 CPU 核数加 1。第三如果 QPS 很高可以考虑用 ONNX Runtime 替代 sklearn 推理速度能快 2 到 3 倍。第四接口要做输入校验防止恶意请求传超大数组把内存打爆。验证服务是否正常可以用 curl 发一条测试请求curl -X POST http://localhost:5000/predict \ -H Content-Type: application/json \ -d {duration:0,protocol_type:tcp,service:http,flag:SF,src_bytes:200,dst_bytes:5000,count:1,srv_count:1}返回的 JSON 里 label 应该是 normal 或某个攻击类别confidence 是置信度。如果返回 500 错误先看服务端日志大概率是某个特征缺失导致 KeyError或者离散值不在编码器类别里。我自己的习惯是每次改完特征工程逻辑先跑一遍离线测试集确认指标没掉再重新导出模型和预处理器最后才更新服务。模型文件和预处理器必须版本对齐混用旧 scaler 和新模型是最隐蔽的翻车方式指标会莫名其妙掉一大截。希望帮到你。本文还有配套的精品资源点击获取