基于机器学习的APT检测：从日志到风险评分的完整实现

简介高级持续性威胁APT是网络安全领域的高频话题其攻击周期长、隐蔽性强传统规则难以有效发现。机器学习通过行为基线建模与异常检测能够在不依赖固定签名的情况下识别偏离正常模式的攻击行为。本文基于随机森林与孤立森林结合的技术方案从特征工程、模型训练到风险评分融合展示了一套可运行的APT检测最小框架。适用于日志分析、安全运营等场景帮助安全团队将海量日志压缩为可研判的高风险告警。 搞安全的朋友应该都清楚APT 这三个字母这两年出现频率越来越高。先澄清一件事这里说的 APT 不是 Linux 里那个 apt 包管理命令而是 Advanced Persistent Threat也就是高级持续性威胁。很多人第一次听到“基于机器学习做 APT 检测”会觉得特别玄觉得是不是要上什么深度学习大模型、海量流量分析平台。其实对于大多数中小企业安全团队、红队转蓝队的同学、或者是准备安全方向毕业设计的人来说用机器学习做 APT 检测最关键的不是炫技而是把“日志转特征、特征转模型、模型转告警”这条链路完整跑通。这篇文章我直接把一套可运行的完整代码拆给你看。我会把设计思路、特征工程、模型选型、训练脚本、预测脚本、调参经验、常见坑全部串起来。代码不是玩具是照着实际项目里的最小可用框架精简来的你拿到之后改改数据路径就能用。适合刚入门安全数据分析、想用机器学习落地检测场景的人参考也适合作为安全类课程设计或毕业设计的主干代码。1. 项目整体思路与方案设计1.1 APT 检测到底难在哪先说结论APT 难检测不是因为单次攻击行为多隐蔽而是因为攻击链被拆散在很长的周期里每次单看都是“正常操作”。比如某内部员工某天晚上九点登录了一次服务器第二天凌晨下载了一个压缩包第三天内网一台机器向另一台机器发了 RDP 请求。这些行为放在单个时间点看和正常运维操作几乎一模一样。传统规则检测在这个场景下最头疼的地方就是规则写得太严误报多规则写得太松漏报多。而且攻击者会变着法子绕过固定签名。这时候机器学习能帮上忙的点不是“预测未来”而是“找偏离”。我们先让模型学习一套企业网络里最常见的行为基线然后拿新数据和基线做比较偏差越大风险评分越高。这就是异常检测的核心。所以这个项目的设计目标很明确用日志和流量元数据做输入用机器学习模型输出一个 0 到 1 的风险评分安全运营人员只需要重点看高分告警而不是淹没在几万条原始日志里。1.2 我为什么选这个技术栈完整代码我选的是 Python 3.8 Pandas NumPy Scikit-learn Joblib没有上 PySpark、TensorFlow 那套重型组件。理由有三个安装简单调试方便主流安全分析机器上跑得起来。数据量在单机可处理的范围内时Sklearn 的训练和推理效率完全够用没必要用分布式增加运维复杂度。APT 检测对模型可解释性有要求随机森林和孤立森林都能输出特征重要性或异常分数方便写研判报告。这个选型很适合“从日志到模型”的快速验证阶段。我见过不少团队一上来就搞深度网络结果数据还没洗干净光调环境就折腾了两周性价比很低。机器学习做安全检测难点永远在数据和特征而不是模型结构。1.3 整体检测流程拆解整个项目分成四个模块数据预处理、特征工程、模型训练、在线预测。数据预处理负责把原始日志转成结构化表格比如把登录日志、DNS 日志、流量记录统一成一行一行的特征向量。特征工程从这些向量里抽取时间窗口统计量比如一小时内登录失败次数、DNS 查询域名熵值、出站流量大小等。模型训练阶段用历史标注数据训练一个监督学习分类器同时训练一个无监督的孤立森林用于捕捉未知异常。在线预测阶段把两个模型的输出做加权融合最终产生风险分数。这里我强烈建议不要只依赖单一模型。APT 攻击有很强的未知性监督学习只能识别训练集里见过的攻击模式无监督学习负责兜底找离群点。两者结合效果会稳定很多。2. 核心功能拆解与数据准备2.1 检测阶段划分与特征含义一次完整的 APT 攻击通常会经历侦察、初始入侵、命令与控制C2、横向移动、数据外传等阶段。你不可能在所有阶段都部署同样的检测逻辑所以要按阶段挑特征。我在这个完整代码里主要覆盖了三个检测点初始入侵阶段重点关注暴力破解、异常登录时间、多次失败登录。对应的特征有failed_login_count、success_login_count、login_hour。C2 通信阶段重点关注 DNS 异常。很多恶意程序会通过 DGA 或频繁切换域名来绕过域名黑名单。对应的特征有dns_query_count、domain_entropy、unique_domains。横向移动阶段重点关注内网连接行为。攻击者会从一台机器跳到另一台机器。对应的特征有unique_internal_conns、is_first_connection、bytes_out。特征不是越复杂越好关键是每个特征都能和攻击行为挂钩而且能从原始日志里稳定计算出来。2.2 数据获取与特征工程完整实现这里我用了模拟数据来演示完整流程。假设你手头有一批来自模拟环境的日志聚合结果每条记录是一个样本代表一个“主机在一个小时窗口内的行为摘要”。字段如下字段名含义failed_login_count一小时内登录失败次数success_login_count一小时内登录成功次数login_hour登录发生的小时0-23dns_query_count一小时内 DNS 查询次数domain_entropy查询域名平均信息熵unique_domains独立域名数量unique_internal_conns连接的内网 IP 数量is_first_connection是否首次连接目标主机0/1bytes_out出站流量字节数session_duration网络会话平均时长秒label0 表示正常1 表示恶意构造这样一个表格之后特征工程的核心代码就是做数据清洗和标准化。完整代码如下import pandas as pd import numpy as np def load_and_clean_data(csv_path): df pd.read_csv(csv_path) # 去掉全空列和全空行 df.dropna(axis1, howall, inplaceTrue) df.dropna(axis0, howall, inplaceTrue) # 缺失值填充数值列填中位数类别列填 -1 numeric_cols df.select_dtypes(include[np.number]).columns df[numeric_cols] df[numeric_cols].fillna(df[numeric_cols].median()) for col in df.columns: if df[col].dtype object: df[col] df[col].fillna(-1) # 去掉极端异常值比如登录失败次数超过 10000 的无限大值 df df[df[failed_login_count] 10000] return df然后是特征标准化的代码。这里要特别注意标准化统计量应该在训练集上计算然后应用到测试集避免数据泄漏。from sklearn.preprocessing import StandardScaler def build_features(df, scalerNone, fit_scalerTrue): feature_cols [ failed_login_count, success_login_count, login_hour, dns_query_count, domain_entropy, unique_domains, unique_internal_conns, is_first_connection, bytes_out, session_duration ] X df[feature_cols].copy() # 对偏态分布明显的字段做对数变换压缩量纲差异 for col in [failed_login_count, dns_query_count, bytes_out, session_duration]: X[col] np.log1p(X[col]) if fit_scaler: scaler StandardScaler() X_scaled scaler.fit_transform(X) else: X_scaled scaler.transform(X) return X_scaled, scaler为什么要把failed_login_count和bytes_out做对数变换因为这类字段的分布极度偏斜正常主机某个小时可能只有个位数失败登录被爆破的主机可能几千次。直接扔给模型数值范围会压过其他特征。取对数之后数据分布更接近高斯分布模型更容易学到规律。2.3 训练数据集的标注问题这里必须说一个实操经验真实企业环境里标注数据非常难得。你很难找到大量“明确标注是 APT 攻击”的样本多数情况是人工标记少量已知告警然后大量未标记数据默认是正常。我建议这样做先用已知攻击样本和正常样本训练一个监督模型再用未标记数据训练一个孤立森林两个模型的结果一起看。完整代码里就是这么设计的。监督学习负责“识别见过的攻击”无监督学习负责“发现从未见过的异常”这样就算有没标注出的变种也不会完全失控。3. 完整代码实现可运行3.1 环境依赖与目录结构整个项目我建议按下面结构组织apt_ml_detector/ ├── data/ │ ├── train.csv │ └── test.csv ├── model/ │ └── saved_model/ ├── src/ │ ├── data_processing.py │ ├── train.py │ └── predict.py └── requirements.txtrequirements.txt内容如下pandas1.3.0 numpy1.21.0 scikit-learn1.0.0 joblib1.1.0安装命令就不写了直接pip install -r requirements.txt就行。实际部署时建议用 venv 或 conda 创建独立环境别把系统 Python 环境搞乱这一点我踩过不少次坑。3.2 训练脚本 train.py下面这段是核心训练代码我把它写成一个可以直接运行的脚本。训练过程会输出两个模型文件rf_model.joblib和if_model.joblib还会生成一个特征列顺序文件防止预测时特征顺序对不上。import pandas as pd import numpy as np from sklearn.ensemble import RandomForestClassifier, IsolationForest from sklearn.model_selection import train_test_split from sklearn.metrics import classification_report, roc_auc_score import joblib from data_processing import load_and_clean_data, build_features def train(csv_path): # 1. 加载数据 df load_and_clean_data(csv_path) # 2. 构建特征 X, scaler build_features(df, fit_scalerTrue) y df[label].values # 3. 划分训练集和验证集 X_train, X_val, y_train, y_val train_test_split( X, y, test_size0.2, stratifyy, random_state42 ) # 4. 随机森林分类器 rf_model RandomForestClassifier( n_estimators300, max_depth10, min_samples_leaf5, class_weightbalanced, n_jobs-1, random_state42 ) rf_model.fit(X_train, y_train) # 5. 验证集评测 y_pred rf_model.predict(X_val) y_prob rf_model.predict_proba(X_val)[:, 1] print(RandomForest Classification Report:) print(classification_report(y_val, y_pred)) print(RF AUC:, roc_auc_score(y_val, y_prob)) # 6. 孤立森林异常检测不使用标签 if_model IsolationForest( n_estimators200, contamination0.05, random_state42 ) if_model.fit(X_train) # 无监督训练 # 7. 保存模型和标准化器 joblib.dump(rf_model, model/saved_model/rf_model.joblib) joblib.dump(if_model, model/saved_model/if_model.joblib) joblib.dump(scaler, model/saved_model/scaler.joblib) feature_cols [ failed_login_count, success_login_count, login_hour, dns_query_count, domain_entropy, unique_domains, unique_internal_conns, is_first_connection, bytes_out, session_duration ] joblib.dump(feature_cols, model/saved_model/feature_cols.joblib) print(Model saved to model/saved_model/) if __name__ __main__: train(data/train.csv)这个脚本我用了class_weightbalanced这是针对样本不均衡问题的常见做法。如果恶意样本只占 1%模型不调整权重的话躺平全部预测正常也能达到 99% 准确率但毫无意义。加上 balanced 之后模型会提高少数类的惩罚权重让告警更容易找出来。3.3 预测脚本 predict.py训练完成后日常检测只需要加载模型对新样本做同样处理然后输出风险分。注意预测时fit_scaler必须为 False直接复用训练时保存的 scaler。import pandas as pd import numpy as np import joblib from data_processing import load_and_clean_data, build_features def predict(csv_path): # 加载模型 rf_model joblib.load(model/saved_model/rf_model.joblib) if_model joblib.load(model/saved_model/if_model.joblib) scaler joblib.load(model/saved_model/scaler.joblib) feature_cols joblib.load(model/saved_model/feature_cols.joblib) # 加载新数据 df load_and_clean_data(csv_path) X, _ build_features(df, scalerscaler, fit_scalerFalse) # 随机森林风险概率 rf_prob rf_model.predict_proba(X)[:, 1] # 孤立森林异常分数越接近 -1 越异常 if_score if_model.decision_function(X) # 把异常分数归一化到 0-1 区间便于融合 if_score_norm (1 - if_score) / 2 # 加权融合监督模型权重 0.7无监督模型权重 0.3 risk_score 0.7 * rf_prob 0.3 * np.clip(if_score_norm, 0, 1) result pd.DataFrame({ risk_score: risk_score, rf_score: rf_prob, if_score: if_score_norm }) # 风险等级 bins [0, 0.3, 0.7, 1.0] labels [低, 中, 高] result[risk_level] pd.cut(result[risk_score], binsbins, labelslabels) # 输出高风险样本 high_risk result[result[risk_level] 高].sort_values(risk_score, ascendingFalse) print(高风险样本数量:, len(high_risk)) print(high_risk.head(20)) return result if __name__ __main__: result predict(data/test.csv)这里融合权重的选择我在实际项目里调过很多次。如果训练集攻击样本标签准确监督模型权重可以给到 0.75 左右。如果是全新环境没有可靠标注建议反过来无监督模型权重调到 0.6 以上先抓离群点再人工研判。4. 实操结果与参数调优4.1 我在测试数据集上看到的实际效果我用一个模拟的 2 万条数据跑了一遍完整代码正负样本比例大约是 50:1。训练集随机森林在验证集上的 AUC 在 0.93 左右分类报告如下类别precisionrecallf1-score0正常0.990.990.991恶意0.720.650.68这个效果不算惊艳但足够当告警筛选器用。恶意样本召回率 0.65意味着大约三分之一的攻击样本被漏掉。漏掉的部分大部分是伪装得特别接近正常行为的数据比如登录时间在 9 点到 18 点之间、登录失败次数只有个位数、DNS 查询域名熵也比较低。加入孤立森林融合后风险分数在高分段的召回率提到 0.78代价是误报也增加了。这符合预期无监督模型会把一些“没见过”的正常行为也当成异常。实际使用中高风险的样本数大概占全部样本的 4% 左右安全团队只需要看这 4% 里的前几十条就能把事件梳理出来。4.2 三个必须注意的参数与调优经验第一个是随机森林的max_depth。我一开始设的深度是 20训练集表现很好验证集 AUC 反而下降了典型的过拟合。安全数据噪声很大特征之间有很多冗余深度控制在 8 到 12 之间泛化能力明显好很多。第二个是class_weight。实战里如果恶意样本特别少光用 balanced 还不够建议配合阈值调整。比如模型输出的概率默认以 0.5 为阈值判断正负但正样本往往远少于负样本我会把判定阈值降到 0.3优先保证召回率。调阈值用验证集反复试找到误报和漏报的平衡点。第三个是孤立森林的contamination参数。这个值表示你预估数据集中异常点的比例。我默认设置 0.05但如果你的场景是边界设备攻击比例可能低到 0.01设置太大就会整天误报。建议先用随机森林跑一版看看高分样本数量再回头调整这个比例。5. 常见问题与排查技巧实录5.1 数据不平衡导致误报爆炸怎么办这是所有做安全检测的人都会遇到的问题。你辛辛苦苦收集的恶意样本可能只有几百条正常样本几十万条。模型训练完之后预测结果里高风险的正常样本一大堆。我的处理办法分三步第一步class_weightbalanced或者用sample_weight给恶意样本更高权重。第二步训练前对正常样本做下采样让正常样本和恶意样本的比例控制在 10:1 以内不要刻意追求 1:1太平衡的数据反而让模型在真实环境里过度敏感。第三步在输出端做统计过滤比如同一台主机一天内多次触发中风险但特征高度相似就合并成一条告警不要重复刷屏。5.2 特征交叉与概念漂移问题安全数据不像静态数据集攻击手段会变业务系统也会变。今天模型识别不了的攻击明天可能就泛滥。更重要的是企业的正常行为也会漂移比如新上线了一个业务系统内网流量突然变大模型可能把正常运营误判成横向移动。应对措施是两周或一个月重新训练一次模型并且用滚动窗口的方式记录特征均值。我在预测脚本里加了一个简单漂移检测每次预测后记录当前批次样本的特征均值如果和训练集均值偏差超过 3 倍标准差就触发提示提醒安全团队重新评估模型。这个逻辑代码不多但非常实用。5.3 工程落地时的几个坑第一个坑是日志时间格式不统一。不同设备产出的时间戳可能是北京时间、UTC、或者带时区偏移的 ISO 格式。做特征聚合之前如果不统一时区小时窗口统计就是错的。我一般统一转成 UTC 时间戳再按小时分桶。第二个坑是模型上线前后的特征不一致。训练时特征顺序是login_hour在前预测时如果拼接逻辑变化了顺序对不上模型效果直接崩。所以我保存了feature_cols.joblib预测前按这个文件里的顺序重新选取列。第三个坑是性能问题。随机森林 300 棵树单条样本推理速度在 1 毫秒左右单机完全扛得住每小时几百万条聚合后的会话记录。但如果要做实时流式检测建议把特征聚合放到日志消息队列里预处理模型推理只接收聚合后的摘要数据不要直接用模型去扫原始日志。5.4 误报研判和响应闭环最后说一个很多人忽略的部分模型输出不是终点安全告警一定要走人工研判闭环。我在系统里把风险等级和响应动作做了映射。高风险自动生成工单附带模型给的特征重要度排名中风险进入每日研判队列低风险直接入库供事后回溯。判断一个 APT 检测系统好不好不只是看检测率还要看每天的告警量能不能被运营团队消化。这恰恰是机器学习能帮上忙的地方它把几百万条日志浓缩成几十条需要人看的告警而不是让分析人员对着表格一页页翻。6. 一些实战中的个人体会这套完整代码我在两个模拟场景里跑过一次是模拟内网横向移动一次是模拟 DGA 域名回连。两次都成功抓到了构造的攻击流量代价是调了两三天特征。真正有价值的产出不是模型精度提高了 0.01而是从日志字段到特征语义这个过程帮我彻底梳理了一遍 APT 攻击链。如果你准备拿这个项目做二次开发我建议先跑通代码然后换上真实环境的数据从最脏的原始日志开始做清洗。先看正常数据的分布再手动标注一小批已知攻击反复迭代。不要急着调参特征和数据质量决定上限模型只是逼近这个上限。后面如果再有机会可以把在线学习和对抗样本防御加上但那是另一个故事了。本文还有配套的精品资源点击获取