简介本资源是一套基于Python机器学习实现的高精度网络入侵检测系统源码面向计算机、自动化等专业的本科生及初阶从业者适用于毕业设计、课程大作业与安全方向实践项目。系统采用CNN等主流模型在KDD99数据集上实测准确率达99.5%代码经严格调试并获毕业答辩98分高分评价具备完整训练、测试与日志分析流程。压缩包共16个文件含4个核心Python脚本如main.py、cnn_main.py、2个KDD99数据压缩包.gz、4个IDE配置XML文件、3个TensorFlow事件文件.zjx-24000635及README说明文档等结构清晰便于理解模型构建、数据预处理与结果可视化全流程。资源大小为17.52MB目录组织规范含train/test子目录与日志模块已支持开箱运行。目前已有825人学习下载适合夯实机器学习工程能力、复现经典IDS方案或在此基础上拓展异常流量识别功能。1. 这不是调个 sklearn 就能跑通的“99.5%”——一个真实可用的 Python 网络入侵检测系统必须直面数据失衡、特征漂移与工业级流量解析你下载了那个标着“正确率99.5%”的.zip包解压后发现train.py里只有一行model.fit(X_train, y_train)测试集是用make_classification生成的模拟数据——这根本不是网络入侵检测这是机器学习课设演示。真正的网络入侵检测NIDS系统面对的是 NetFlow、PCAP、Suricata 日志或 Zeek原 Bro提取的 conn.log、http.log、dns.log 等多源异构日志流它要区分 SYN Flood 和正常突发访问要识别 DNS Tunneling 而非简单过滤非常规端口它的“99.5%”必须是在 CIC-IDS2017 或 UNSW-NB15 这类含真实攻击流量、时间序列强相关、类别极度倾斜如 DoS 攻击占 78%而 Web Attack 仅占 0.3%的数据集上经 StratifiedKFold 交叉验证、在测试集严格留出攻击样本后测得的 F1-score而非 accuracy。本文不讲理论推导只带你用 Python 复现一套可部署、可监控、可解释的 NIDS 流水线从原始 PCAP 解析 → 特征工程 → 处理类别不平衡 → 训练 XGBoost SHAP 解释器 → 部署为轻量 API。适合有 Python 基础、接触过 Scikit-learn 但没处理过真实网络日志的工程师也包含对模型泛化能力敏感的进阶参数调优逻辑。2. 用 Scapy Pandas 构建可复现的流量特征提取流水线从 raw PCAP 到结构化 DataFrame网络入侵检测的第一道关卡从来不是模型而是特征是否真正反映网络行为本质。直接用scapy.rdpcap()读取 PCAP 后逐包解析效率低且易丢关键上下文如 TCP 会话状态、HTTP 请求/响应配对。我们必须构建一个分层解析流水线先按五元组src_ip, src_port, dst_ip, dst_port, proto聚合会话再在会话粒度上计算统计特征。这不是“写个 for 循环”而是用 Pandas 的groupby().agg()结合自定义函数实现向量化计算。2.1 安装依赖与验证 PCAP 解析环境确保你的环境中已安装scapy,pandas,numpy,tqdm。注意Scapy 在 Linux 下需 root 权限抓包但解析本地 PCAP 不需要。验证安装pip install scapy pandas numpy tqdm scikit-learn xgboost shap imbalanced-learn # 验证 Scapy 是否能正确解析常见协议 python -c from scapy.all import *; pkt IP()/TCP(dport80)/Raw(bGET / HTTP/1.1); print(pkt[TCP].dport)提示若遇到ImportError: No module named scapy.all请确认未将脚本命名为scapy.py—— 这是最常见的命名冲突错误会导致 Python 优先导入当前目录下的同名文件而非库。2.2 会话级特征提取为什么不能只用单包特征单包特征如包长、TTL、TCP 标志位极易被混淆例如正常 HTTPS 握手与恶意 TLS 指纹探测在单包层面高度相似。而会话级特征session-level features捕捉通信模式一个 SSH 会话的平均包间隔、重传率、数据载荷熵值比单个 SYN 包更能暴露暴力破解行为。我们定义一个会话为同一五元组方向上的连续 TCP/UDP 流忽略反向 ACK并提取以下 23 维特征特征名计算逻辑业务含义durationmax(timestamp) - min(timestamp)会话持续时间秒orig_bytessum(ip_len for pkt in orig_pkts)源端发送总字节数resp_bytessum(ip_len for pkt in resp_pkts)目的端发送总字节数orig_pktslen(orig_pkts)源端发送包数resp_pktslen(resp_pkts)目的端发送包数orig_ip_bytes_meanmean([pkt[IP].len for pkt in orig_pkts])源端平均 IP 包长resp_ip_bytes_stdstd([pkt[IP].len for pkt in resp_pkts])目的端 IP 包长标准差tcp_flags_syn_ratiocount(SYN)/len(orig_pkts)源端 SYN 包占比探测特征entropy_payloadshannon_entropy(concat(payloads))所有载荷拼接后的香农熵检测加密隧道2.2.1 实现extract_session_features函数import numpy as np import pandas as pd from scapy.all import * from collections import defaultdict, Counter from tqdm import tqdm def shannon_entropy(data): 计算字节序列的香农熵 if not data: return 0.0 counter Counter(data) length len(data) entropy -sum((count / length) * np.log2(count / length) for count in counter.values()) return entropy def extract_session_features(pcap_path, max_packets10000): 从 PCAP 提取会话级特征 DataFrame :param pcap_path: PCAP 文件路径 :param max_packets: 最大解析包数防大文件阻塞 :return: pd.DataFrame, columns[src_ip,src_port,dst_ip,dst_port,proto,label, ...features] sessions defaultdict(list) # key: (src,sp,dst,dp,proto), value: list of packets packets rdpcap(pcap_path, countmax_packets) for pkt in tqdm(packets, descParsing packets): if IP not in pkt or (TCP not in pkt and UDP not in pkt): continue ip pkt[IP] proto tcp if TCP in pkt else udp sport pkt[TCP].sport if TCP in pkt else pkt[UDP].sport dport pkt[TCP].dport if TCP in pkt else pkt[UDP].dport # 会话键源→目的方向固定顺序避免双向重复 key (ip.src, sport, ip.dst, dport, proto) sessions[key].append(pkt) feature_list [] for (src, sp, dst, dp, proto), pkts in sessions.items(): if len(pkts) 3: # 过滤过短会话如单个 ICMP continue timestamps [pkt.time for pkt in pkts] orig_pkts [p for p in pkts if IP in p and p[IP].src src] resp_pkts [p for p in pkts if IP in p and p[IP].src dst] # 提取基础统计 duration max(timestamps) - min(timestamps) orig_bytes sum(p[IP].len for p in orig_pkts) if orig_pkts else 0 resp_bytes sum(p[IP].len for p in resp_pkts) if resp_pkts else 0 orig_pkts_cnt len(orig_pkts) resp_pkts_cnt len(resp_pkts) # 计算 IP 包长统计 orig_ip_lens [p[IP].len for p in orig_pkts] if orig_pkts else [0] resp_ip_lens [p[IP].len for p in resp_pkts] if resp_pkts else [0] # TCP 标志位统计仅 TCP 会话 syn_ratio 0.0 if proto tcp and orig_pkts: syn_count sum(1 for p in orig_pkts if TCP in p and p[TCP].flags 0x02) # SYN flag syn_ratio syn_count / len(orig_pkts) # 载荷熵取前 10 个包防内存爆炸 payloads b for p in pkts[:10]: if Raw in p: payloads bytes(p[Raw]) entropy shannon_entropy(payloads) # 构建一行特征 row { src_ip: src, src_port: sp, dst_ip: dst, dst_port: dp, proto: proto, duration: duration, orig_bytes: orig_bytes, resp_bytes: resp_bytes, orig_pkts: orig_pkts_cnt, resp_pkts: resp_pkts_cnt, orig_ip_bytes_mean: np.mean(orig_ip_lens), orig_ip_bytes_std: np.std(orig_ip_lens), resp_ip_bytes_mean: np.mean(resp_ip_lens), resp_ip_bytes_std: np.std(resp_ip_lens), tcp_flags_syn_ratio: syn_ratio, entropy_payload: entropy, label: normal # 占位符后续替换为真实标签 } feature_list.append(row) return pd.DataFrame(feature_list) # 示例解析一个小型 PCAP # df_features extract_session_features(sample.pcap) # print(df_features.shape) # 输出 (N, 24)这段代码的关键在于所有计算均基于会话session而非单包packet且使用tqdm提供进度反馈避免在大型 PCAP 上长时间无响应。shannon_entropy函数直接作用于原始字节流能有效区分随机加密流量高熵与明文 HTTP低熵。注意label字段暂置为normal因为真实标签需从外部标注文件如 CIC-IDS2017 的 Labels.csv关联注入这是下一步的重点。3. 处理真实世界的数据失衡与标签噪声用 SMOTEENN Stratified Sampling 构建鲁棒训练集当你把 CIC-IDS2017 的Monday-WorkingHours.pcap解析成 DataFrame 后执行df[label].value_counts()大概率会看到这样的分布BENIGN 2,100,000,DDoS 180,000,PortScan 12,000,WebAttack 320。这就是典型的长尾分布少数几类攻击样本极少而模型在训练时会天然偏向多数类。此时若直接用accuracy 99.5%实际意味着模型把所有样本都预测为BENIGN—— 这完全无效。我们必须用组合策略先用分层抽样Stratified Sampling保证各类别在训练/验证集中比例一致再用 SMOTEENN 对少数类过采样多数类欠采样。3.1 加载并清洗 CIC-IDS2017 标签数据CIC-IDS2017 的标签并非嵌入 PCAP而是存于独立 CSV 文件如Labels-Monday-WorkingHours.csv其列名为Destination Port,Protocol,Timestamp,Flow Duration,Total Fwd Packets,Label。我们需要将其与上一步提取的df_features关联。关联键不是 IP 地址易变而是时间窗口 协议 端口组合def merge_labels_with_features(features_df, labels_csv, time_window_sec5): 将 CIC-IDS2017 标签 CSV 与特征 DataFrame 按时间窗口和端口匹配 :param features_df: extract_session_features() 输出的 DataFrame :param labels_csv: CIC-IDS2017 的 Labels-*.csv 路径 :param time_window_sec: 时间匹配窗口秒 :return: 带 label 列的 features_df labels pd.read_csv(labels_csv) # 清洗标签去除空格、统一大小写 labels[Label] labels[Label].str.strip().str.upper() # 将标签中的 BENIGN 映射为 normal其他攻击映射为 attack labels[label] labels[Label].apply(lambda x: normal if x BENIGN else attack) # 构建时间索引取每个会话的起始时间min timestamp features_df[session_start] features_df[duration].index.map( lambda i: features_df.iloc[i][duration] - features_df.iloc[i][duration] ) # 此处需实际计算简化示意 # 实际中需为 features_df 添加 start_time 列在 extract_session_features 中记录 min(pkt.time) # 关键按 (src_port, dst_port, proto) 和时间窗口 join # 此处省略具体 join 逻辑因原始标签无 src_ip实践中常用 Zeek log 替代 # 真实项目推荐直接用 Zeek 的 conn.log known-compromised.log 关联更可靠 return features_df # 实践建议跳过脆弱的 PCAP-CSV 时间对齐改用 Zeek 日志 # zeek -r sample.pcap cat conn.log | bro-cut id.orig_h id.orig_p id.resp_h id.resp_p proto service duration orig_bytes resp_bytes orig_pkts resp_pkts conn_features.csv注意PCAP 与 CSV 标签的时间对齐是 CIC-IDS2017 数据集公认的难点误差常达秒级。生产环境强烈推荐放弃 PCAP 解析直接使用 ZeekBro生成的结构化日志。Zeek 的conn.log已包含所有会话级统计且known-compromised.log提供精确攻击标记无需手动匹配。3.2 应用 SMOTEENN 处理类别不平衡imblearn.combine.SMOTEEEN是SMOTE合成少数类样本与EditedNearestNeighbours删除多数类边界噪声点的组合比单独使用 SMOTE 更鲁棒能有效抑制过拟合from imblearn.combine import SMOTEEEN from sklearn.model_selection import train_test_split from sklearn.preprocessing import StandardScaler # 假设 df_labeled 是已合并标签的 DataFrame X df_labeled.drop([src_ip,dst_ip,proto,label], axis1) y df_labeled[label] # 分层划分确保 train/test 中 normal/attack 比例一致 X_train, X_test, y_train, y_test train_test_split( X, y, test_size0.2, stratifyy, random_state42 ) # 标准化XGBoost 对特征尺度不敏感但标准化有助于 SHAP 解释一致性 scaler StandardScaler() X_train_scaled scaler.fit_transform(X_train) X_test_scaled scaler.transform(X_test) # 应用 SMOTEENN smoteenn SMOTEEEN(random_state42, sampling_strategyauto) # auto 表示平衡所有类 X_train_res, y_train_res smoteenn.fit_resample(X_train_scaled, y_train) print(fResampled training set shape: {X_train_res.shape}) print(fResampled label distribution:\n{pd.Series(y_train_res).value_counts()}) # 输出示例 # Resampled training set shape: (124560, 20) # Resampled label distribution: # attack 62280 # normal 62280 # dtype: int64此步骤后训练集y_train_res中attack与normal样本数严格相等。sampling_strategyauto是关键参数它让 SMOTEENN 自动判断哪些类是少数类attack并进行过采样同时对多数类normal执行 ENN 清洗。random_state42保证结果可复现。注意不要对测试集X_test_scaled做任何重采样否则评估将严重失真。4. 训练高精度 XGBoost 模型并用 SHAP 解释决策逻辑不只是“99.5%”更要“为什么是攻击”准确率 99.5% 的模型若无法解释就是生产环境的定时炸弹。运维人员需要知道“这个告警为什么触发是因为entropy_payload过高还是tcp_flags_syn_ratio异常” XGBoost 本身是黑盒但shap库能提供局部可解释性。我们不追求全局特征重要性图而是为每一个预测样本生成力导向图force plot直观展示各特征对最终输出的贡献值。4.1 XGBoost 参数调优聚焦 recallattack 而非 accuracy入侵检测的核心指标是Recall查全率即“真实攻击中有多少被检出”。宁可误报False Positive不可漏报False Negative。因此我们的目标函数应设为focal_loss或直接优化recall。XGBoost 提供eval_metricrec:binary二分类 recallimport xgboost as xgb from sklearn.metrics import classification_report, confusion_matrix # 构建 DMatrixXGBoost 高效输入格式 dtrain xgb.DMatrix(X_train_res, labely_train_res, enable_categoricalFalse) dtest xgb.DMatrix(X_test_scaled, labely_test, enable_categoricalFalse) # 关键参数说明 # - objectivebinary:logistic: 二分类任务 # - eval_metricrec:binary: 优化召回率非 accuracy # - scale_pos_weight: 根据正负样本比设置此处为 1.0因已平衡 # - max_depth6: 防止过拟合深度8 在网络数据上易过拟合 # - subsample0.8, colsample_bytree0.8: 引入随机性提升泛化 params { objective: binary:logistic, eval_metric: rec:binary, # 核心优化 recall max_depth: 6, learning_rate: 0.1, subsample: 0.8, colsample_bytree: 0.8, scale_pos_weight: 1.0, seed: 42 } # 训练模型监控验证集 recall model xgb.train( params, dtrain, num_boost_round200, evals[(dtrain, train), (dtest, test)], early_stopping_rounds30, verbose_eval10 ) # 预测概率 y_pred_proba model.predict(dtest) y_pred (y_pred_proba 0.5).astype(int) print(classification_report(y_test, y_pred, target_names[normal, attack])) # 输出应显示 attack 类的 recall ≥ 0.98运行后classification_report将显示attack类的recall查全率。若低于 0.95需调整scale_pos_weight增大该值会提升对正样本的关注或降低learning_rate并增加num_boost_round。4.2 用 SHAP 生成可操作的告警解释SHAP 的核心是计算每个特征对单个预测的边际贡献。我们为测试集中一个真实攻击样本生成解释import shap # 创建 explainer使用训练数据子集加速 explainer shap.TreeExplainer(model) # 为单个样本索引 0计算 SHAP 值 sample_idx 0 shap_values explainer.shap_values(X_test_scaled[sample_idx:sample_idx1]) # 生成 force plot横向条形图显示各特征如何将 base_value 推向 output_value shap.initjs() shap.force_plot( explainer.expected_value, shap_values[0], X_test_scaled[sample_idx], feature_namesX.columns.tolist(), matplotlibTrue, showFalse ).savefig(attack_explanation.png, bbox_inchestight)生成的attack_explanation.png将清晰显示例如entropy_payload 7.2远高于正常值 3.5贡献了 0.42 的 logits而duration 0.02极短会话贡献了 0.18二者共同将模型输出从基线值-1.2推至2.1最终判定为attack。这就是运维人员真正需要的“为什么”——他们可据此快速确认是否为 DNS Tunneling高熵或 SYN Flood短时高 SYN 比。5. 部署为轻量 API 并集成实时告警用 Flask Redis 实现每秒百次检测模型训练完成只是开始。生产环境要求1API 响应延迟 200ms2支持并发请求3告警能推送至企业微信/钉钉。我们用Flask搭建 REST API用Redis缓存高频攻击特征如某 IP 的 5 分钟内 SYN ratio避免重复计算。5.1 构建 Flask API接收 JSON 特征返回预测与解释from flask import Flask, request, jsonify import joblib import redis import json app Flask(__name__) # 加载训练好的模型与 scaler model joblib.load(xgb_model.pkl) scaler joblib.load(scaler.pkl) # 初始化 Redis 连接用于缓存 r redis.Redis(hostlocalhost, port6379, db0, decode_responsesTrue) app.route(/predict, methods[POST]) def predict(): try: data request.get_json() # data 格式: {src_ip:192.168.1.100,src_port:54321,dst_ip:10.0.0.1,dst_port:80,proto:tcp,...} # 提取特征向量需与训练时顺序严格一致 feature_vector [ data[duration], data[orig_bytes], data[resp_bytes], data[orig_pkts], data[resp_pkts], data[orig_ip_bytes_mean], data[orig_ip_bytes_std], data[resp_ip_bytes_mean], data[resp_ip_bytes_std], data[tcp_flags_syn_ratio], data[entropy_payload] ] # 标准化 scaled_vec scaler.transform([feature_vector]) # 预测 pred_proba model.predict(xgb.DMatrix(scaled_vec))[0] is_attack bool(pred_proba 0.5) # 生成 SHAP 解释仅对攻击样本节省资源 explanation {} if is_attack: # 此处应调用 SHAP 计算为简化返回 top3 特征贡献 # 实际中可预计算 SHAP 值表或用 KernelExplainer explanation { top_features: [ {feature: entropy_payload, value: data[entropy_payload], contribution: 0.42}, {feature: tcp_flags_syn_ratio, value: data[tcp_flags_syn_ratio], contribution: 0.18}, {feature: duration, value: data[duration], contribution: -0.15} ] } return jsonify({ prediction: attack if is_attack else normal, confidence: float(pred_proba), explanation: explanation, timestamp: int(time.time()) }) except Exception as e: return jsonify({error: str(e)}), 400 if __name__ __main__: app.run(host0.0.0.0, port5000, debugFalse) # 生产环境禁用 debug启动服务python app.py然后用 curl 测试curl -X POST http://localhost:5000/predict \ -H Content-Type: application/json \ -d {duration:0.02,orig_bytes:120,resp_bytes:0,orig_pkts:3,resp_pkts:0,orig_ip_bytes_mean:40,orig_ip_bytes_std:5,resp_ip_bytes_mean:0,resp_ip_bytes_std:0,tcp_flags_syn_ratio:1.0,entropy_payload:7.2}5.2 Redis 缓存攻击指纹实现 IP 级实时风控为防止同一攻击 IP 在短时间内反复触发告警我们在 Redis 中维护一个attack_ip_historySorted Set以时间戳为 scoredef record_attack_ip(ip, timestamp, confidence): 记录攻击 IP 到 Redis保留最近 5 分钟 key fattack_ip_history:{ip} r.zadd(key, {str(confidence): timestamp}) r.expire(key, 300) # 5 分钟过期 def is_ip_suspicious(ip, window_sec300, min_confidence0.9): 检查 IP 在窗口期内是否高频攻击 key fattack_ip_history:{ip} # 获取过去 window_sec 秒内的记录 now time.time() recent r.zrangebyscore(key, now - window_sec, now, withscoresTrue) if len(recent) 5 and all(conf min_confidence for conf, _ in recent): return True return False # 在 predict() 函数中当 is_attackTrue 时调用 # record_attack_ip(data[src_ip], time.time(), pred_proba) # if is_ip_suspicious(data[src_ip]): # send_alert_to_dingtalk(data[src_ip], explanation)此机制让系统具备基础的“攻击链”识别能力单次高置信攻击触发告警若同一 IP 在 5 分钟内出现 5 次以上高置信攻击则升级为“可疑僵尸网络”自动推送至安全运营中心SOC。提示生产部署必须添加 gunicorn 或 uWSGI 作为 WSGI 服务器并配置 Nginx 反向代理与负载均衡。单进程 Flask 仅适用于验证无法支撑高并发。6. 验证模型泛化能力用 CIC-IDS2017 的 Friday-WorkingHours 测试集做跨天验证模型在 Monday 数据上达到 99.5% 并不意味它能在 Friday 有效。网络流量具有强时间依赖性工作日的 Web 流量模式与周末不同防火墙策略更新也会导致特征漂移。真正的鲁棒性必须通过跨天、跨周、跨数据集的验证来证明。CIC-IDS2017 提供了完整的周五攻击数据Friday-WorkingHours.pcap这是最严格的测试场景。6.1 执行跨天验证的最小命令# 步骤1用 extract_session_features 解析 Friday-WorkingHours.pcap python -c from features import extract_session_features df_fri extract_session_features(Friday-WorkingHours.pcap, max_packets50000) df_fri.to_csv(friday_features.csv, indexFalse) # 步骤2加载训练好的模型对 Friday 特征进行预测 python -c import pandas as pd import joblib import xgboost as xgb df pd.read_csv(friday_features.csv) scaler joblib.load(scaler.pkl) model joblib.load(xgb_model.pkl) X df.drop([src_ip,dst_ip,proto,label], axis1) X_scaled scaler.transform(X) preds model.predict(xgb.DMatrix(X_scaled)) print(Friday attack recall:, (preds 1).sum() / len(preds)) 若输出Friday attack recall: 0.92说明模型存在明显漂移——它在 Monday 学到的模式无法泛化到 Friday。此时必须引入在线学习Online Learning用xgboost.train()的xgb_model.boost()方法在 Friday 数据上增量训练而非从头训练。或者更优方案是加入概念漂移检测Concept Drift Detection如alibi-detect库的KSDrift当检测到 Friday 特征分布显著偏移时自动触发模型重训流程。6.2 三个必调参数决定你的模型能否走出实验室参数位置默认值调优建议影响eval_metricXGBoosttrain()logloss必须改为rec:binary直接决定模型优化目标影响 recall 与 precision 的权衡scale_pos_weightXGBoostparams1.0设为len(negative)/len(positive)未平衡时强制模型关注少数类防止全预测为 normalmax_depthXGBoostparams6网络数据建议 4~6深度 8 易过拟合流量噪声4 则无法捕获复杂攻击模式这三个参数是区分“课设模型”与“生产模型”的分水岭。它们不写在任何论文里却真实存在于每一次线上告警的准确率曲线中。本文还有配套的精品资源点击获取