ARTICLE · INTELLIGENCE

战地情报 · 详情页

来自尧图项目组的一线实战观察与深度解析

基于Python的APT攻击检测方法:流量特征提取与XGBoost模型部署实践

基于Python的APT攻击检测方法:流量特征提取与XGBoost模型部署实践 简介一套基于Python溯源图的APT攻击检测方法毕业设计源码包面向计算机相关专业在校学生、教师及企业安全从业者覆盖高级持续性威胁APT检测场景既可用于毕业设计、课程设计等正式项目也适合希望从零开始学习图神经网络安全分析的中初级开发者。压缩包共30个文件其中11个py源码为核心实现RGAT、GRU等图神经网络模型、数据加载与训练脚本另有7个xml工程配置、4个md说明文档等整体仅51KB结构紧凑便于快速定位与部署。目前已有50人下载学习源码经测试可运行导师指导认可答辩评分达95分具有较高的完成度与参考价值。内含针对streamspot与darpa-cadets两种数据集的RGAT、GRU模型变体并附部署文档与使用说明可帮助理解溯源图构建与异常检测流程也可在此基础上替换数据集或调整模型结构实现个性化功能拓展。1. 拿到“APT 攻击检测”毕设题目先想清楚什么不该做我见过不少把这题做成“下载数据集 跑 XGBoost 报告 99% 准确率”的同学结果答辩时被一句“你的数据标签是怎么来的没标签的数据怎么办”问住。基于 Python 的 APT 攻击检测方法源码及部署文档资料听起来像一个现成脚手架但真正能落地的做法是先把“检测对象”想窄APT 攻击链那么长第一步千万别做端到端全流程检测那是给自己挖坑。这篇笔记把我在类似方案里验证过的路线讲清楚——主力放在流量侧特征数据清洗、模型训练、部署文档三件事上让你做出来的东西既有源码又能复现而不是一个解释不了的黑匣子。2. 把检测对象拆到三步进攻链选哪个阶段下手什么数据源能支撑2.1 APT 的典型进攻链初始入侵、内网横移、数据外传APT 攻击和普通扫描爆破最大的区别是“慢”和“多阶段”。我把常见模型下的进攻链简化成三步初始入侵、内网横移、数据外传。初始入侵常见手段是钓鱼邮件附件、漏洞利用、弱口令爆破检测价值最高但噪声也最大一封钓鱼邮件可能只产生几个异常包还没法从流量里稳定提取内网横移阶段主机的交互变多域名请求、SMB 连接、端口扫描交织在一起需要日志侧和流量侧配合才有戏数据外传阶段反而最好检测因为攻击者要把数据传出去必然产生“长时间、大包体、固定目标端口”的流量特征而且这种行为一般发生在凌晨或非工作时间和正常业务流偏离明显。所以我的建议是不要在毕设里把三段全部实现一遍把范围压到“内网横移 数据外传”的流量检测或者干脆只做“数据外传”一类把这一步做到能解释、能复现、能验证。这个选择不是偷懒而是把有限时间花在能出结果的地方答辩时也能讲清楚“为什么我选择这个阶段”。2.2 数据源选型网络流量、主机日志、安全设备告警各是什么代价检测 APT 攻击链上的可疑流量通常有三类数据源可选网络流量数据、主机日志、安全设备告警。三种方案落地的成本和效果差异非常大我整理了一个对比表数据源获取难度可解释性毕设推荐度典型翻车点网络流量 pcap / 流特征中公开数据集多高每条流有明确五元组和统计特征高数据集标签不全背景流量混入攻击流量主机日志进程、网络连接、注册表高需要真实主机环境中强依赖宿主环境低没有真实入侵样本实验无法复现安全设备告警IDS/EDR 告警低但脱敏困难中依赖设备品牌和规则质量低告警本身已过一层规则标签主观性强如果选流量侧特征就用 CICFlowMeter 那套网络流特征每条流量是一个包含五元组信息的数据流再从流里提取包长均值、流持续时间、包到达间隔、端口分布等统计特征。这套特征最大的优势是脱离具体主机环境模型在实验网络里训练完拿到另一个网络环境里做推理特征定义仍然成立只是分布会偏移。这也是为什么网安、AI 方向毕设里流量侧检测最稳妥。主机日志方案不是不能做而是你得真实跑一个诱捕环境或者在合规前提下找历史数据否则甚至连“带标签的入侵日志”都凑不齐。安全设备告警方案同样受限于设备训练出来的模型基本是复读设备的规则没法体现你自己的检测逻辑。2.3 第一版模型怎么定只检测一个“外传阶段”用流特征说话定下用网络流量之后第一版模型我建议不要上任何花哨结构用 XGBoost 或随机森林把 20 到 30 个流特征灌进去做二分类恶意流 / 良性流。恶意流不要求细分攻击类型只要求把攻击链阶段的可疑流量和正常业务流量分开。这个定位有三个好处一是数据准备量小不需要每个攻击类型都凑足样本二是特征可解释树模型可以输出特征重要性答辩时能指着一张图说“这条流持续 2 小时、平均包长 800 字节所以判为外传”三是便于后续迭代先有基线再谈优化。第二点要提醒的是流特征检测和传统的“抓包看 SNI”不一样。流特征不解析应用层内容只看元数据因此对加密流量也有效数据外传阶段的加密隧道同样会表现出“包长、时长、间隔”的异常。这也成了这套方案值得写进报告里的加分点。检测对象一旦收敛后面的事就容易推进了数据、特征、模型、部署四步走每一步都控制在你能解释的范围里。3. 数据集与标签清洗把 CICIDS2017 变成能训练的最小样本3.1 数据集的四个“坑前提醒”CICIDS2017 是流量检测方向用得最多的公开数据集之一里面有正常流量和多种攻击流量。但直接用官网原始 CSV 训练会在四个地方踩坑。第一标签极不均匀。某些攻击类型只持续几十分钟对应到 CSV 里可能只有几百条流而良性流量一天有几百万条不处理类别不均衡模型会直接学成“全预测良性”。第二文件不是完整的“一天一个”。官网把 5 天的 pcap 切片导出成 8 个 CSV比如周四上午是 Web 攻击下午是 Infiltration周五上午是正常流量下午是 DDoS每个文件的列名不完全一致个别文件列名里有空格和引号。第三原始 CSV 体积非常大。合起来几个 GB用 pandas 默认参数读取会撑爆内存而且部分数值列有缺失值和 Infinity。第四时间戳范围只有一周 5 天。训练集和验证集的切分不能随机打乱否则同一条 TCP 流可能同时出现在训练和测试里指标虚高。这一节先把合并和标签统一做掉切分策略放到 3.4。3.2 合并 8 个 CSV列名、标签、内存的三个处理点下面这段是我常用的合并脚本适用于把 CICIDS2017 的多个 CSV 合成一个训练文件。import pandas as pd from pathlib import Path RAW_DIR Path(data/raw) MERGED_PATH Path(data/apt_detect_all.csv) frames [] for csv_path in sorted(RAW_DIR.glob(*.csv)): df pd.read_csv(csv_path, encodinglatin1, low_memoryFalse) # CICIDS2017 部分列名带空格和引号统一去掉 df.columns [c.strip().strip() for c in df.columns] frames.append(df) merged pd.concat(frames, ignore_indexTrue) # 统一标签BENIGN 记为 0其余攻击类型记为 1 # 同时保留原始标签方便后面按攻击类型拆分分析 merged[Label] merged[Label].astype(str).str.strip() merged[Label_original] merged[Label] merged[Label] merged[Label].apply(lambda x: 0 if x BENIGN else 1) merged.to_csv(MERGED_PATH, indexFalse) print(merged[Label].value_counts())逻辑说明先用 glob 按文件名排序读取保证拼接顺序固定避免下次运行时数据顺序变化。读文件时用了 encodinglatin1因为原 CSV 个别字符不是 UTF-8 编码默认读会直接报错。low_memoryFalse 是防止 pandas 在读取时把同一列的不同 block 推断成不同 dtype否则后面算特征会莫名报错。参数说明合并时不筛选列先把全量特征都读进来。class 标签里如果出现空格或不可见字符先 astype(str).str.strip() 再比较避免 BENIGN 被分到恶意类。保留 Label_original 是为了后续分析哪类攻击被漏报最多不加这一列后面想按攻击类型复盘还得重新读原始文件。脚本跑完后打印 value_counts这一步非常关键如果发现恶意类只有几十条说明下载的文件不全或者某天的攻击部分没有包含进来。我在跑周四和周五数据时就遇到过合并后恶意样本只有几百条的情况最后查到是漏了一个文件名不带 WorkingHours 的 CSV白白浪费了两小时。3.3 特征选择20 个特征就够起步剩下的等有需要再加CICFlowMeter 产出的特征 80 个左右但不是全都要。树模型对冗余特征不敏感可特征太多会让部署脚本变得笨重而且在真实环境里你未必能从抓包工具里拿到全部 80 项。我一般先选 20 个左右保证 pcap 重放验证时每个特征都能稳定算出来。特征名说明为什么选它Flow Duration流持续时间微秒外传阶段常伴随长连接正常 web 短请求不会持续几小时Fwd Packet Length Mean正向平均包长大包体外传时该值显著偏高Bwd Packet Length Mean反向平均包长区分交互型流量和单向外传流量Packet Length Variance包长方差隧道流量包长方差比交互式流量小Average Packet Size平均包大小区分文本传输与二进制文件传输Init_Win_bytes_forward正向初始窗口大小不同操作系统协议栈差异常用于区分主机类型min_seg_size_forward正向最小段大小恶意流量常表现为小包探测加后续大包传输以上是前几个典型特征剩余特征从 CICFlowMeter 的时长类、标志位类、包间隔类特征里补到 20 个。选择标准就三条能在 CICFlowMeter 输出里稳定得到、不在多分类场景下失效、和攻击链阶段有业务含义上的关联。把特征表写进毕设文档时记得对每个特征加一句“为什么对 APT 检测有效”这比贴一张 80 行的特征列表更有说服力。3.4 训练集划分按时间切片别拿随机打乱骗自己流量检测的训练集划分和图像分类不一样不能直接 train_test_split(random_state42)。原因是同一个 IP、同一种攻击方式产生的流在时间上高度聚合随机打乱后攻击流量会被均匀分到训练和验证里模型相当于“见过答案再考试”验证集指标虚高。我自己踩过这个坑随机切分版准确率 99.2%改成按天切分后掉到 96.8%后者才是真实水平。正确做法是按 CSV 的来源文件切比如前 4 天的文件合并做训练集周五做验证集。如果周五的攻击类型和前几天完全不重合指标会很难看但这正是真实场景——新攻击类型是没见过的检测模型必须对未知攻击有一定泛化能力。如果你选的标题方向包含“部署后效果”这部分一定写清楚因为验证方式决定了你报告里那个准确率到底是模型的本事还是数据划分的幻觉。4. 模型训练与部署文档从 Notebook 到能交付的源码包4.1 用 XGBoost 训练第一版确认正负样本比例合并好的 CSV 可以直接进入训练环节。训练代码我放在 train_xgb.py下面是核心部分。import pandas as pd import joblib from xgboost import XGBClassifier from sklearn.model_selection import train_test_split df pd.read_csv(data/apt_detect_all.csv) # 去掉非数值列保留标签 features [c for c in df.columns if c not in ( Timestamp, Flow ID, Src IP, Src Port, Dst IP, Dst Port, Protocol, Label, Label_original )] X df[features].fillna(0) y df[Label] X_train, X_test, y_train, y_test train_test_split( X, y, test_size0.2, stratifyy, random_state42 ) # 用正负样本比例给 scale_pos_weight缓解类别不均衡 pos_ratio (y_train 0).sum() / (y_train 1).sum() model XGBClassifier( n_estimators300, max_depth6, learning_rate0.05, scale_pos_weightpos_ratio, eval_metricaucpr, n_jobs-1, random_state42, ) model.fit(X_train, y_train) joblib.dump(model, models/apt_xgb.joblib)逻辑说明features 列表先排除所有非数值列再 fillna(0) 处理缺失值。这里针对 CICIDS2017 已经够用因为原始数据里的缺失值大多是除法产生的XGBoost 能处理 NaN但统一填 0 更容易部署时保持一致。class_weight 不用 sklearn 版本转而用 XGBoost 自带 scale_pos_weight这样训练和推理都留在 xgboost 生态里。参数说明n_estimators300 对 20 个特征来说足够继续加大收益很小还拖慢重放验证。max_depth6 在流特征上是经验值深度太浅欠拟合太深过拟合到具体 IP 和端口。eval_metric 用 aucpr 而不是 accuracy因为恶意流量是少数类accuracy 会被大量良性流量淹没。第一次训练完务必打印测试集 recall 和 precision不要只打印 accuracy如果发现 recall 低于 0.5下一步先调 scale_pos_weight 而不是堆模型。4.2 轻量推理脚本给单个 pcap 流出检测结果训练完要有一个能对单个 pcap 出结果的推理入口不能每次都在 Notebook 里跑。下面是最小推理示例import joblib import pandas as pd model joblib.load(models/apt_xgb.joblib) def predict_flow(feature_row): # feature_row 是一个 dictkey 必须和训练时的特征名一致 df pd.DataFrame([feature_row]) df df[model.get_booster().feature_names] prob model.predict_proba(df)[0][1] return prob # 示例用 CICFlowMeter 对某个 pcap 输出的一行特征 sample_feature { Flow Duration: 360000000, Fwd Packet Length Mean: 780.5, Bwd Packet Length Mean: 12.3, Packet Length Variance: 1024.0, } prob predict_flow(sample_feature) print(恶意概率:, round(prob, 3))逻辑说明推理脚本强调两件事——特征名顺序要对上训练时的顺序否则 xgboost 会报 feature_names 不匹配输入必须是数值型pandas 会自动处理但类型错乱时容易报 dtype 错误。用 model.get_booster().feature_names 取特征列表再让输入的 DataFrame 按这个顺序重排一列是最稳妥的做法。这一版推理还只是“单条流打分”。部署到真实环境时通常需要一个抓包进程持续产流再把每条流的特征送进模型打分超过阈值的流汇总成告警。第一版先把单条流跑通后面在验证章节再补在线抓包部分。书本上的黑匣子表现再好看落到实际流量里还是要靠这个入口做验证。4.3 部署文档要写什么环境、接口、复现步骤、实验结果标题里写了“部署文档资料”这部分是很多人忽略的。我见过源码写得很漂亮、部署文档是空白的毕设评审老师拿到后根本跑不起来。部署文档不用写成长篇大论但必须包含下面四块内容文档章节要写的内容写不清楚会怎样运行环境Python 版本、操作系统、依赖库版本清单requirements.txt换一台机器直接报 ImportError复现失败数据准备数据集来源、目录结构、下载后放哪个路径接手的人不知道 raw 数据放哪脚本跑不了推理接口输入是什么pcap / CSV / 单条特征输出是什么恶意概率 / 告警别人无法把模型接到自己的流量上验证流程用哪个 pcap 重放、预期输出、耗时多少无法确认模型部署后是否正常工作环境部分我一般会额外写一句“部署机用 Python 3.10使用 linux 系统安装 python 的方式尽量装官方版本别用发行版自带旧版本”。读者如果用的是 windows也能通过 vscode 配置 python 或 pycharm 配置 python 环境来跑但真正做重放验证时 Linux 更顺手。部署文档的目标是让零基础接手者只读文档就能把模型从零跑起来多写一句“执行 python train_xgb.py 后会在 models/ 下生成 apt_xgb.joblib”比写一百句理论都有用。4.4 源码目录的交付结构源码笔记放一起README 说清楚入口源码包不能只有训练脚本我交付时一般按下面这个结构组织文件/目录作用README.md入口说明从数据集到模型的完整复现步骤requirements.txt固定依赖版本防止环境漂移build_dataset.py原始 CSV 合并、标签统一、特征筛选train_xgb.py训练主脚本输出 joblib 模型predict_flow.py单条流 / 单个 pcap 的推理入口docs/deploy.md部署文档环境、接口、验证流程docs/eval_report.md实验结果准确率、误报率、按攻击类型拆分的召回率这里“源码笔记”放在一起README 是入口。很多同学喜欢把笔记和代码分成两个压缩包其实没必要评审老师和接手的人只想知道一件事按什么顺序跑完这些脚本。一份 README 写清楚“第一步 build_dataset第二步 train_xgb第三步 predict_flow”整份代码的可用性立刻不一样。这也是“优秀毕业设计”和普通作业的区别——普通作业交一堆散脚本优秀毕设交一套能复现的流水线。5. 避坑记录五个让检测结果失效的常见错误5.1 训练集里没有恶意标签模型悄悄学成了“全部输出 0”现象训练完准确率 99.5%但打印测试集预测分布时发现一条恶意流量都没报出来recall 是 0。原因合并 CSV 时漏掉了含有攻击流量的文件或者标签里有不可见字符导致 BENIGN 之外的标签没有被正确映射为 1训练集里实际只有良性类。解决每次合并完强制打印 value_counts确认正负样本比例在可接受范围如果恶意样本占比低于 0.5%先排查文件是否齐全再检查标签列是否有空格和 BOM。这个坑花不了几分钟排查但一旦忽略后面整个模型都是废的。5.2 准确率 99% 但误报率没人看上线就是事故现象验证集上准确率 99%看起来完美一接入真实抓包流量告警刷屏全是误报。原因CICIDS2017 里的良性流量和真实业务流量分布差异很大模型学到的是“这个数据集的良性流量长什么样”而不是“所有正常流量长什么样”。解决训练时把 eval_metric 换成 aucpr固定阈值前先画 PR 曲线把阈值拉到误报率低于 0.5% 的位置报告里同时给出 recall、precision、F1、误报率四件套别只写 accuracy。误报问题在流量检测里就是信誉问题告警多到没人看模型再好也没用。5.3 换台电脑模型就崩joblib 不是后悔药现象在自己机器上 joblib.load 正常换成服务器或室友电脑后报“model format not supported”或 xgboost 版本不一致的错误。原因xgboost 版本升级后模型文件格式不向下兼容joblib 里还保留了训练时的 Python 环境信息。解决把 requirements.txt 锁死版本另外保留一份 train_xgb.py 训练脚本换机器时直接重新训练而不是靠模型文件打天下。模型文件只是产物训练脚本才是后悔药。5.4 tcpreplay 重放攻击 pcap模型一条都不报现象用 tcpreplay 在实验网络里重放包含攻击流量的 pcap结果模型一条探测流都没报出来而离线测试时召回率是正常的。原因pcap 里的时间戳是原采集时刻的重放时如果原流量是某天下午 3 点捕获的流之间的时间间隔按真实时间流逝但 tcpreplay 默认快速发包导致 Flow Duration、Flow IAT Mean 等时间类特征全部失真特征分布和训练时完全不一样。解决重放时用 --pps 设置每秒包数保证流的持续时间和原 pcap 一致。给 pcap 做时间戳整形让重放出的流量特征和训练数据 match这一步在部署文档的验证流程里必须写清楚。5.5 Windows 日志路径踩坑抓取脚本一换机器就失效现象在 Windows 上调试好的定时抓取脚本放到 Linux 上一跑就报路径错放到另一台 Windows 上也可能因为盘符不同挂掉。原因路径写死了形如 C:/Users/xxx/Desktop/capture.pcap没有用 pathlib 或 os.path.join 拼接。解决所有路径统一用 Path(file).parent 基于脚本所在目录构造数据目录和输出目录在 README 里用相对路径说明。这个问题看似低级但在交付源码时最容易让接手的人卡在第一步比模型参数坑人多了。6. 上线前的验证与进阶用重放和滑动窗口把模型调实6.1 离线验证别用随机切分按时间段留出验证日离线实验的验证集划分直接决定你报告里的数字可不可信。前面提到不要随机打乱具体操作是按原始 CSV 文件的时间顺序前 4 天做训练第五天做验证。这样恶意流量的“浓度”分布最接近真实场景训练里见过的攻击类型可能第五天完全不出现。如果验证集指标掉得厉害说明模型对未知攻击类型没有泛化能力你就知道下一步得补数据而不是调超参。训练脚本里保留 train_test_split 只是为了快速看曲线提交报告时用按天切分的结果。6.2 在线验证最小闭环pyshark 抓包 滑动窗口出特征离线验证通过后可以做一个小型在线验证用 pyshark 抓实验网络的实时流量每次抓 60 秒把 pcap 交给 CICFlowMeter 出特征再喂给模型做批量打分。这里有一个核心技巧60 秒滑动窗口窗口大小对应流持续时间的上限和 python 量化交易策略代码里用滚动窗口算波动率是同一个套路。import pyshark capture pyshark.LiveCapture(interfaceeth0, output_filecapture.pcap) print(开始抓取按 CtrlC 结束) for packet in capture.sniff_continuously(packet_count500): # 简单打印包级信息窗口结束后统一交给 CICFlowMeter 提特征 print(packet.ip.src, packet.ip.dst, packet.length)逻辑说明pyshark 这里只负责抓包落盘特征提取还是交给 CICFlowMeter 离线做因为在线实时提流特征的工作量大且容易丢包。抓包窗口结束后运行 CICFlowMeter 生成新 CSV再用 predict_flow.py 批量打分。如果环境里没有图形界面CICFlowMeter 有命令行版本把输入目录和输出目录配好就能一键出特征。参数说明packet_count500 是个很小的值只在连通性测试时用实际验证我一般抓 10 分钟约 10 万个包然后滑动 5 分钟再抓下一轮。窗口太长会导致短流无法及时出结果太短则长连接特征失真。抓包时用 eth0 这种具体接口避免直接抓 any否则会重复统计包。6.3 给毕设加分的一个习惯人工复核结果回流训练进阶做法是把检测系统的输出变成一个持续迭代的闭环模型每轮打分后把置信度在 0.4 到 0.7 之间的流量交给人工复核确认后把新标注样本追加到训练集再执行训练脚本更新模型。这个方法在大厂安全运营里叫“AI 辅助专家复核”关键点在于人不是去看所有流量只用看模型“拿不准”的那部分。做毕设时把这个流程写进部署文档哪怕只模拟了一轮都能明显提升系统设计的完整性。我自己的习惯是每次调完阈值都重新看一下误报样本长什么样很多特征工程思路就是从误报样本里来的。这比死磕超参数有用得多希望帮到你。本文还有配套的精品资源点击获取
RELATED READING

延伸阅读

更多一线实战笔记与深度复盘,助您持续精进