Python实战网络入侵检测系统：从PCAP解析到XGBoost+SHAP部署

简介：本资源是一套基于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(dport=80)/Raw(b'GET / 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 维特征：

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_packets=10000): """ 从 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, count=max_packets) for pkt in tqdm(packets, desc="Parsing 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_sec=5): """ 将 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 解析，直接使用 Zeek（Bro）生成的结构化日志。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'], axis=1) y = df_labeled['label'] # 分层划分：确保 train/test 中 normal/attack 比例一致 X_train, X_test, y_train, y_test = train_test_split( X, y, test_size=0.2, stratify=y, random_state=42 ) # 标准化：XGBoost 对特征尺度不敏感，但标准化有助于 SHAP 解释一致性 scaler = StandardScaler() X_train_scaled = scaler.fit_transform(X_train) X_test_scaled = scaler.transform(X_test) # 应用 SMOTEENN smoteenn = SMOTEEEN(random_state=42, sampling_strategy='auto') # auto 表示平衡所有类 X_train_res, y_train_res = smoteenn.fit_resample(X_train_scaled, y_train) print(f"Resampled training set shape: {X_train_res.shape}") print(f"Resampled 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_strategy='auto'是关键参数，它让 SMOTEENN 自动判断哪些类是少数类（attack）并进行过采样，同时对多数类（normal）执行 ENN 清洗。random_state=42保证结果可复现。注意：不要对测试集X_test_scaled做任何重采样，否则评估将严重失真。

4. 训练高精度 XGBoost 模型并用 SHAP 解释决策逻辑：不只是“99.5%”，更要“为什么是攻击”

准确率 99.5% 的模型若无法解释，就是生产环境的定时炸弹。运维人员需要知道：“这个告警为什么触发？是因为entropy_payload过高，还是tcp_flags_syn_ratio异常？” XGBoost 本身是黑盒，但shap库能提供局部可解释性。我们不追求全局特征重要性图，而是为每一个预测样本生成力导向图（force plot），直观展示各特征对最终输出的贡献值。

4.1 XGBoost 参数调优：聚焦 recall@attack 而非 accuracy

入侵检测的核心指标是Recall（查全率），即“真实攻击中有多少被检出”。宁可误报（False Positive），不可漏报（False Negative）。因此，我们的目标函数应设为focal_loss或直接优化recall。XGBoost 提供eval_metric='rec:binary'（二分类 recall）：

import xgboost as xgb from sklearn.metrics import classification_report, confusion_matrix # 构建 DMatrix（XGBoost 高效输入格式） dtrain = xgb.DMatrix(X_train_res, label=y_train_res, enable_categorical=False) dtest = xgb.DMatrix(X_test_scaled, label=y_test, enable_categorical=False) # 关键参数说明： # - objective='binary:logistic': 二分类任务 # - eval_metric='rec:binary': 优化召回率（非 accuracy！） # - scale_pos_weight: 根据正负样本比设置，此处为 1.0（因已平衡） # - max_depth=6: 防止过拟合（深度>8 在网络数据上易过拟合） # - subsample=0.8, colsample_bytree=0.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_round=200, evals=[(dtrain, 'train'), (dtest, 'test')], early_stopping_rounds=30, verbose_eval=10 ) # 预测概率 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_idx+1]) # 生成 force plot：横向条形图，显示各特征如何将 base_value 推向 output_value shap.initjs() shap.force_plot( explainer.expected_value, shap_values[0], X_test_scaled[sample_idx], feature_names=X.columns.tolist(), matplotlib=True, show=False ).savefig('attack_explanation.png', bbox_inches='tight')

生成的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 实现每秒百次检测

模型训练完成只是开始。生产环境要求：1）API 响应延迟 < 200ms；2）支持并发请求；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(host='localhost', port=6379, db=0, decode_responses=True) @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(host='0.0.0.0', port=5000, debug=False) # 生产环境禁用 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，以时间戳为 score：

def record_attack_ip(ip, timestamp, confidence): """记录攻击 IP 到 Redis，保留最近 5 分钟""" key = f"attack_ip_history:{ip}" r.zadd(key, {str(confidence): timestamp}) r.expire(key, 300) # 5 分钟过期 def is_ip_suspicious(ip, window_sec=300, min_confidence=0.9): """检查 IP 在窗口期内是否高频攻击""" key = f"attack_ip_history:{ip}" # 获取过去 window_sec 秒内的记录 now = time.time() recent = r.zrangebyscore(key, now - window_sec, now, withscores=True) if len(recent) >= 5 and all(conf >= min_confidence for conf, _ in recent): return True return False # 在 predict() 函数中，当 is_attack=True 时调用： # 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_packets=50000) df_fri.to_csv('friday_features.csv', index=False) " # 步骤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'], axis=1) 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 三个必调参数：决定你的模型能否走出实验室

这三个参数，是区分“课设模型”与“生产模型”的分水岭。它们不写在任何论文里，却真实存在于每一次线上告警的准确率曲线中。

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