简介基于机器学习的APT检测完整代码面向网络安全研究人员、蓝队工程师及机器学习初学者解决高级持续性威胁难以通过传统规则发现的问题。项目以随机森林算法为核心融合异常检测思路借助欠采样、过采样及SMOTE合成样本处理类不平衡数据涵盖从数据预处理、特征工程到模型训练评估的完整流程。压缩包共869个文件包含Python源码、pyc编译文件、设计文档、数据集记录及模型权重等其中326个py脚本可直接查看算法实现14个pem证书、12个pkl模型文件用于实际部署另有docx/doc软件文档与png图表辅助理解整体架构资源包约215.7MB。目前已有676人学习适合希望获得一套可运行、可扩展的APT检测基线方案的读者尤其能帮助理解随机森林在异常流量识别中的参数调优与样本平衡策略。1. APT 检测为什么让传统安全设备集体失守机器学习到底在解决什么APT高级持续性威胁检测是安全运营里最磨人的一块攻击者往往已经在内网潜伏了好几个月传统的 EDR 和防火墙规则却一条告警都没出。不是设备刷不动而是 APT 的动作单看每步都合规——用 PowerShell 执行脚本、走 RDP 横向移动、凌晨三点登录域控规则引擎根本不知道这些组合在一起意味着什么。基于机器学习做 APT 检测核心思路是放弃“匹配已知攻击”改成“度量行为偏离正常的程度”让模型从流量、进程、日志里把规则写不出的相关性挖出来。这篇内容会从任务定义、数据准备、特征工程一路讲到能跑的完整代码和上线后的踩坑记录适合理清思路再动手的安全工程师也适合已经被误报淹没、想换检测思路的运营同学。2. 把 APT 检测定义成机器学习任务数据来源、标签策略与公开数据集很多人接到 APT 检测需求第一反应是“先找个模型跑一下”。但模型是最后一步。第一步是把 APT 检测翻译成机器学习能处理的任务定义——你到底是做分类还是做异常检测标签从哪来数据长什么样。这一步定错了后面所有代码都是白写。2.1 二分类还是异常检测先看你的数据里有没有恶意样本APT 检测在建模层面通常有两条路选哪条取决于你手里有哪些数据第一条是有监督二分类把每台主机、每个 IP 的某段时间行为标成“恶意”或“良性”训练一个分类器。这条路的难点在标签——真实 APT 事件一年可能就几次攻击流量在内网总量里占比远低于百分之一安全运营团队手工确认的样本根本不够训练一个像样的分类器。而且攻击者会针对检测改手法昨天的恶意行为特征今天不一定复用。第二条是无监督异常检测只喂正常行为数据让模型学会“正常长什么样”之后偏离基线的地方就是可疑点。这条路的落地难度低很多正常流量永远不缺也不需要事前标注。代价是误报率天然比有监督高因为“没见过”不等于“攻击”。还有一个折中的半监督思路用少量已确认的恶意样本做种子先跑无监督召回一批候选再人工确认、把确认结果回填训练集。真实项目里我一般建议从无监督起步跑一段时间积累告警和确认记录等恶意样本攒到几百条以上再切到有监督为主。如果你看过的机器学习入门教程都在教你“准备标签——切分——训练——评估”那在 APT 场景要先泼一盆冷水标签恰恰是这个领域最稀缺的资源。2.2 数据从哪来公开数据集与自采流量怎么选没有真实内网数据时公开数据集可以用来调通流程、对比算法但它们离真实 APT 场景有距离。我整理过一份对比数据集攻击覆盖面适合用途要留意的坑NSL-KDD经典 DoS / R2L / U2R跑基线、验证代码链路数据太老特征维度少离现代流量很远CICIDS2017 / 2019Web 攻击、暴力破解、DDoS、扫描等算法选型和调参参考攻击样本占比过高直接训练会把误报率带偏UNSW-NB15混合攻击接近真实分布半监督 / 无监督评测标注密度和真实内网差异明显这三类公开数据集的共同问题是攻击流量大多是“单发的独立事件”而 APT 是跨小时、跨天、跨主机的长链条行为。用它们调模型可以但别指望模型在真实环境直接复现同样的效果。真实落地时我一般会走自采流量的路线在核心交换机的出口或分区边界做端口镜像抓一段时间的网络元数据不存 payload 只存会话统计五元组、包长分布、字节数、协议、时间戳用 Zeek 或 Suricata 解析后落成 CSV 或 Parquet。这样既规避了流量隐私问题又保留了做特征工程所需的全部信息。抓多长时间我的实践经验是至少三到四周的持续数据才能建出可用的正常基线少于一周模型会把工作日和周末的差异、早晚高峰的起伏全当成异常误报率直接爆炸。2.3 标签策略时间窗口标签与三个关键坑数据到位后怎么打标签是个隐蔽但致命的问题。给单条 TCP 会话打标签几乎没有意义——APT 的恶意很难体现在单条流上更多体现为“某台主机一段时间内的行为组合”。所以常见的做法是把数据按实体源 IP、主机 ID加时间窗口聚合比如每 5 分钟一条样本代表“这台主机这 5 分钟干了什么”标签打在窗口上而不是单条连接上。这里有三个坑第一个坑是攻击行为的滞后性。攻击者的扫描侦察可能发生在正式利用前的几天甚至几周时间窗开得太短会把早期关联行为漏掉。我见过有人把窗口设成 10 秒横向移动的慢速扫描被切成几十个“正常小片段”特征完全散掉。一般从 5 分钟到 30 分钟之间尝试低慢型 C2 回连甚至要按小时聚。第二个坑是负样本污染。内网蠕虫、误操作、员工下班后挂机的下载任务都会让“正常基线”里混入异常行为。无监督模型会把这类历史污染当成正常模式的一部分之后同类真实攻击来了反而测不出。处理办法是训练前先做一轮简单的离群剔除或者用更新鲜的干净数据做基线。第三个坑是标签穿越。如果某主机事后被确认失陷你往回给它的所有历史窗口都标成恶意模型会学到“这台主机的正常业务流量也可疑”上线后误报集中在业务服务器上。正确做法是只标注确认失陷时间点之后的窗口失陷前的行为是否属于早期渗透证据要单独论证不能一刀切。3. 特征工程与模型选型从流量五元组到进程行为序列数据准备好之后接下来是特征工程。APT 检测的特征分两个域网络侧和主机侧。大量入门项目只做网络侧因为抓包容易、解析方便但真实 APT 有一半以上的动作发生在主机侧——进程执行、文件操作、登录跳板。我的建议是至少把这两个域都建成结构化的表再合并成一份宽表喂给模型。3.1 网络侧特征会话统计、DNS 隧道与加密流的出路网络侧特征可以从三个层次来取会话统计层就是把每条流聚合出包长均值、包长方差、流持续时间、上下行字节比例、TCP 标志位分布。APT 的 C2 通道常见特征是单包极小、间隔规律、长时间低频回连这些在会话统计里会表现为“下行包长方差异常大”或“单位时间连接数极低却持续不断”。DNS 层是 APT 检测的高价值区。DNS 隧道和域名生成算法DGA有个共性域名长度偏长、字符熵值偏高、查询频率有固定节奏。一个正常业务域名平均长度在 15 个字符以内C2 常用的随机字母串域名经常到 25 位以上子域名每段都在 8 位以上的随机字符。把这些熵值、长度、NXD 响应率、查询时间间隔做成特征列能覆盖很大一部分隐蔽信道。加密流层不用强行解密用元数据就够了TLS 证书的剩余有效期C2 常用短命证书、JA3 指纹、SNI 里的随机化子域。特征表大致形如特征列名计算方式检测价值flow_cnt窗口内会话数横向移动、端口扫描bytes_out_std上行包长标准差C2 小包规律通信dns_name_entropy查询域名字符熵DGA、DNS 隧道tls_cert_days证书剩余有效期短命证书 C2hour窗口起始小时凌晨异常活跃3.2 主机侧特征进程链、文件操作与登录行为主机侧特征需要 EDR 或 Sysmon 类数据支撑但信息密度比网络侧更高。核心是三类进程行为父子进程链深度、命令行长度、命令行中的空格数量、进程运行频率。正常的系统管理命令和攻击者的混淆命令在特征上有系统性差异——攻击负载怕被截断命令行往往超长而且空格数异常多。内网大量用 PowerShell 横移的场景里“PowerShell 进程的父进程不是控制台而是 Word/Outlook”就是一条高权重特征。文件行为短时间内的文件重命名频次、对共享目录的访问速度。横向移动的典型动作是在多个共享目录里批量翻文件正常的业务访问在一段时间内目标更集中。登录行为一个账号的登录源 IP 数、登录时间分布、登录目标主机数。账号从 15 个不同 IP 跳登录的熵值远高于正常运维模式。这些主机特征和网络侧特征一样按时间窗口聚合之后和网络特征在 src_ip time_bucket 上做 join就能得到模型真正消费的宽表。3.3 模型选型孤立森林、随机森林、自编码器怎么排优先级模型层面的选择我的经验是按阶段推进不要一开始就上复杂模型模型监督类型优势劣势在 APT 场景的角色孤立森林无监督训练快、不需要标签对弱异常敏感度一般第一版基线首选随机森林 / XGBoost有监督特征重要性可解释依赖历史标签有标签后的主力自编码器无监督能学高维非线性关联训练调参偏玄学网络行为建模的进阶选项LSTM / Transformer有监督抓时间序列关联需要长序列样本样本充足后再说落地路径通常是这样先用孤立森林跑出异常分数分布观测告警量和误报水平同时在运营过程中积累确认数据样本到几百条后切到随机森林或 XGBoost用特征重要性反推哪些检测维度真正有效等时序样本攒够、有足够的跨天攻击案例后再考虑序列模型。所谓自适应入侵检测是一个长期目标工程上基本都要走完这个阶梯想一步到位上深度模型的团队大多在数据量这一关就卡死了。4. 完整代码从数据预处理到 APT 检测模型训练与推理这一章给出完整可跑的代码骨架。为了让你本地能直接复现我用一个模拟数据生成器产生带少量 APT 行为的流量会话 CSV然后走完整的特征工程、训练、推理流程。项目结构是这样apt_detector/ ├── generate_demo_data.py # 生成模拟流量元数据 ├── features.py # 特征工程窗口聚合 ├── train.py # 训练双模型并保存 ├── detect.py # 对特征打分并输出告警 └── evaluate.py # 离线评估这个骨架是真实系统的最小形态生产环境里这几个文件会被包进调度任务或服务接口但核心逻辑不变。4.1 生成模拟流量数据# generate_demo_data.py import pandas as pd import numpy as np np.random.seed(42) n_normal 8000 n_mal 30 # 正常主机工作时间活跃连接目标集中 normal pd.DataFrame({ ts: pd.date_range(2024-06-01 09:00:00, periodsn_normal, freq5s), src_ip: np.random.choice([10.0.1.10, 10.0.1.11, 10.0.1.12], n_normal), dst_ip: np.random.choice([10.0.2.20, 10.0.2.21, 10.0.2.22], n_normal), dst_port: np.random.choice([443, 80, 22], n_normal), bytes_out: np.random.exponential(800, n_normal).astype(int), bytes_in: np.random.exponential(2000, n_normal).astype(int), proto: np.random.choice([tcp, tcp, udp], n_normal), }) # 恶意主机凌晨低频活动目标分散上下行比例异常 mal pd.DataFrame({ ts: pd.date_range(2024-06-03 02:00:00, periodsn_mal, freq45s), src_ip: [10.0.1.13] * n_mal, dst_ip: np.random.choice([10.0.5.10, 10.0.5.11, 10.0.5.12, 10.0.6.1], n_mal), dst_port: np.random.choice([445, 8443], n_mal), bytes_out: np.random.exponential(80, n_mal).astype(int), bytes_in: np.random.exponential(20000, n_mal).astype(int), proto: np.random.choice([tcp, udp], n_mal), }) df pd.concat([normal, mal], ignore_indexTrue).sort_values(ts).reset_index(dropTrue) df[is_malicious] 0 df.loc[mal.index, is_malicious] 1 # concat 后原 index 保留 df.to_csv(flow_logs.csv, indexFalse)这段代码生成 8030 条会话记录8000 条正常流量在白天以 5 秒间隔均匀出现30 条恶意流量在凌晨 2 点以 45 秒间隔低频出现。恶意样本有刻意设计的特征上班时间09:00-17:00和凌晨02:00的时段差异、目标 IP 分散度、上下行包长反比。is_malicious列只在离线评测时使用真实场景下没有这一列它是从事件记录里回填的。4.2 特征工程10 分钟窗口聚合# features.py import pandas as pd import numpy as np def build_features(df, window10T): df df.copy() df[ts] pd.to_datetime(df[ts]) df[time_bucket] df[ts].dt.floor(window) df[hour] df[ts].dt.hour feats df.groupby([time_bucket, src_ip]).agg( flow_cnt(ts, count), bytes_out_mean(bytes_out, mean), bytes_in_mean(bytes_in, mean), bytes_out_std(bytes_out, std), bytes_in_std(bytes_in, std), dst_unique(dst_ip, nunique), port_unique(dst_port, nunique), hour(hour, first), is_malicious(is_malicious, max), ).reset_index() # 单个会话窗口的 std 是 NaN直接填 0 feats feats.fillna(0) return feats if __name__ __main__: df pd.read_csv(flow_logs.csv) feats build_features(df, window10T) feats.to_csv(features.csv, indexFalse) print(feats.head())window10T是 pandas 的 offset alias代表 10 分钟。这意味着每台源 IP 每 10 分钟生成一条特征样本一天 144 条。窗口大小的选择直接决定检测粒度5 分钟适合快速横向移动的抓取30 分钟更贴近慢速 C2 回连的节奏。实战中我会用多个窗口各建一套特征再拼接但初版代码里先固定 10 分钟跑通链路。聚合后的 8 个特征列里dst_unique和port_unique是扫描行为的直接指标bytes_out_std和bytes_in_std是 C2 小包通信的抓手is_malicious取 max 是因为一个窗口内只要出现过恶意流量这个窗口整体就该算恶意。fillna(0)处理的是单条会话窗口里标准差为 NaN 的问题这一步漏掉的后果在避坑章节会专门讲。4.3 训练双模型孤立森林 随机森林# train.py import pandas as pd import joblib from sklearn.ensemble import IsolationForest, RandomForestClassifier df pd.read_csv(features.csv) feature_cols [ flow_cnt, bytes_out_mean, bytes_in_mean, bytes_out_std, bytes_in_std, dst_unique, port_unique, hour ] # 无监督只用正常数据训练孤立森林 normal_feats df[df[is_malicious] 0] X_normal normal_feats[feature_cols].values iso IsolationForest( n_estimators200, contamination0.005, # 预期异常占比宁低勿高 random_state42 ) iso.fit(X_normal) # 有监督全量数据训练随机森林演示用真实场景需单独打标签 from sklearn.model_selection import train_test_split X_all df[feature_cols].values y_all df[is_malicious].values rf RandomForestClassifier( n_estimators300, max_depth10, class_weightbalanced, random_state42 ) rf.fit(X_all, y_all) joblib.dump({iso: iso, rf: rf}, apt_detector.model) print(model saved)孤立森林的contamination参数是模型认为的数据集中异常比例。这里填 0.005意思是预计 0.5% 的窗口行为异常。这个参数宁低勿高——设高了模型强制挑出大量正常窗口当异常告警风暴一来安全团队两小时就要求回滚。随机森林侧class_weightbalanced是应对恶意样本占比过低的手段让模型对少数类给更高惩罚权重。max_depth10限制树的深度防止对少量恶意样本过拟合。真实场景里有监督这一步不会这么简单它消耗的是你两三个月积累的人工确认告警记录。演示数据里直接用is_malicious训练逻辑上等价但不要误以为生产系统可以直接照抄这一步。4.4 实时打分与告警输出# detect.py import pandas as pd import joblib model joblib.load(apt_detector.model) iso model[iso] rf model[rf] def score_row(feature_values): 返回 (孤立森林异常分数, 随机森林恶意概率) iso_score iso.decision_function([feature_values])[0] rf_prob rf.predict_proba([feature_values])[0][1] return iso_score, rf_prob def make_alert(row): vals row[feature_cols].values iso_score, rf_prob score_row(vals) # 双模型联合判定减少单模型误报 if iso_score -0.1 and rf_prob 0.7: return { time_bucket: str(row[time_bucket]), src_ip: row[src_ip], level: high, reason: fiso{iso_score:.3f}, rf{rf_prob:.3f} } if iso_score -0.1 or rf_prob 0.7: return { time_bucket: str(row[time_bucket]), src_ip: row[src_ip], level: medium, reason: single_model } return None # 对全量特征跑一轮告警 if __name__ __main__: df pd.read_csv(features.csv) alerts [] for _, row in df.iterrows(): a make_alert(row) if a: alerts.append(a) print(pd.DataFrame(alerts).to_string())decision_function返回负值表示异常负得越多越异常predict_proba输出的是随机森林判恶意类别的概率。两个模型一个抓“偏离正常”一个抓“像历史恶意”组合起来能压掉相当一部分误报。阈值iso_score -0.1和rf_prob 0.7是按模拟数据调出来的真实环境要重新根据告警量标定——先跑一天看输出多少条告警再决定放松还是收紧。5. APT 检测模型常见踩坑5 个必须提前处理的工程问题这一章全是实战里踩出来的坑每一条都对应过真实的上线事故。按“现象 → 原因 → 解决”的方式写清楚。5.1 随机切分数据集导致时间穿越离线评测虚高现象验证集准确率达到 99%上线当天被真实流量击穿漏报一堆。原因你用了train_test_split(random_state42)做随机切分。APT 行为在时间上是自相关的——攻击者扫描后紧跟着利用、利用后紧跟着横向移动相邻时间窗口的特征高度相似。随机切分会把同一台主机的相邻窗口分别丢进训练集和测试集模型相当于“背答案”。这在时间序列类数据里叫时间穿越。解决按时间顺序切分前 80% 的时间窗口做训练、后 20% 做验证。更严格的做法是“块状评估”连续若干小时的窗口整体作为验证块模拟模型面对一段从未见过的真实时间区间。evaluate.py里要按这个逻辑写不要用 sklearn 默认的随机切分。5.2 只盯准确率模型悄悄学会了“全部判正常”现象模型训练完准确率 99.7%告警数却是 0。安全团队巡检发现内网早已被植入后门。原因APT 恶意样本占比远低于 1%模型只需要把所有样本都判为正常就能拿到极高的准确率。这本质上不是模型的问题是评估指标选错了。解决类不平衡场景下一律用召回率、误报率、PR-AUC 而不是准确率。给运营设一个最低召回目标比如“宁可多告警 10 倍不能漏掉已知攻击类型”再在这个约束下去压误报。PR 曲线的形状比准确率诚实得多。5.3 特征里有 NaN 和无穷值模型分数像抽风现象孤立森林的decision_function输出在 -0.5 到 0.5 之间乱跳同一个特征值重复跑两次结果不一样其实不是模型随机是数据里有非数。原因std这类统计特征在窗口里只有一条会话时是 NaN除法运算中除数为 0 会产生inf。sklearn 对 NaN 的处理是直接忽略该特征或报错对inf则可能输出无意义分数。解决特征工程末尾统一做数据清洗fillna(0)之外还要检查inffeats feats.replace([np.inf, -np.inf], 0).fillna(0)这条建议看起来基础但几乎所有第一次跑流量数据的同事都在这翻过车。用np.isinf全表扫一遍确认没有残留再进模型。5.4 contamination 参数拍脑袋乱填开门就是告警风暴现象模型上线第一天输出 3 万条告警安全运营团队当场要求回滚版本。原因把contamination设成 0.1等于强制模型承认 10% 的行为是异常的。真实内网的异常行为占比通常在 0.1% 到 1% 之间设大了模型会从正常流量里硬挑“异常”。这是参数含义理解错位。解决先不设contamination用iso.fit(X)后取出所有正常数据的分数分布画直方图找拐点。拐点值对应的百分位就是合理的contamination参考值一般落在 0.001 到 0.01 之间。我在实操里还会乘一个 0.5 的折扣系数先低后高告警从少到多逐步放开比一次性拉满稳得多。5.5 模型上线三个月后检测率退化模型成了黑匣子现象新攻击变种漏检旧攻击的误报率反而上升安全团队对模型输出开始不信任。原因业务基线变了。版本迭代、业务迁移、员工规模变化都会让流量分布偏移三个月前学到的“正常”已经不适合今天。攻击者的 C2 手法也在演进老模型没有见过新形态。解决按月重训是底线。更可靠的是监控特征分布漂移——每月对关键特征算一次均值、方差和分位数跟训练时的基线做对比变化超过 20% 就触发重训。随机森林每个树深度有限定期重训之后模型保持可用性没问题。但别指望一个模型吃一年安全对抗本质上是持续更新的游戏这部分没有后悔药只能建立定期重训机制。6. 上线前必须做的验证用一个模拟攻击场景评估模型模型训练完别急着接告警。先做一轮端到端验证确认链路通了、指标达标了再以静默模式观察一段时间。验证手法造一个模拟 DNS 隧道攻击的特征样本走一遍detect.py的打分逻辑确认能产出告警。具体做法是构造低字节数、高频 DNS 查询、高域名字符熵的窗口特征喂给score_row看rf_prob是否超过 0.7。这一步不是验证模型能检测真实攻击而是验证从特征到告警的整条链路没断——很多项目的模型是好的死在特征列名对不上、模型文件版本加载出错这些低级问题上。评估指标上除了 PR-AUC我强烈建议多看一眼“固定误报率下的召回率”。比如设定误报率 0.5%去看模型能召回多少真实攻击。这个数字直接对应安全团队每天处理多少条无效告警比 AUC 更能说明上线后的体验。我在evaluate.py里会用precision_recall_curve画曲线同时打印 top 1% 异常分数里真实恶意窗口的覆盖率——这个指标对运营最有感知。上线节奏分三步走第一周静默模式模型打分只写日志不告警每天人工核对分数榜里的记录第二周开中等级告警看安全团队的反馈速度第三周全量放开。起步阶段宁可漏报也不要海量误报——误报会透支团队信任信任一旦丢了后面再好的模型都拉不回来。选型建议再强调一次如果你的环境只有规则引擎加少量日志先把数据采集补齐再谈模型如果已经有半年以上的流量元数据积累这套完整代码可以直接迁移上线。模型的价值在于从你每天视而不见的正常流量里挑出真正值得人看一眼的东西。我最早那版模型上线第一周就靠一个凌晨的 DNS 异常熵抓到了真实的内网失陷迹象但之后也经历了自己把自己告警淹没的阶段。先把数据质量和评估口径做对再谈算法迭代这是我做过最值的一笔投入。希望帮到你。本文还有配套的精品资源点击获取