☰
基于溯源图与VF2子图匹配的APT攻击检测系统
2026/10/7 3:19:11 网站建设 项目流程

简介:本资源是一套基于Python实现的APT攻击检测系统,面向网络安全、数据科学及智能系统方向的高年级本科生、研究生与行业开发者,聚焦高级持续性威胁的溯源图建模与检测实践。项目完整覆盖算法实现、部署方案与实测数据集,适用于毕业设计、课程实验、学术研究及企业原型开发,要求使用者具备Python编程基础与网络安全基本认知。压缩包共31个文件,含11个核心Python脚本(如main.py、model_RGAT.py、streamspot_RGAT.py等)、7个XML配置与元数据文件、4份Markdown文档(含设计说明与信息分析指南)、5个备份文件(.zbak)及环境配置相关文件,整体仅52KB,轻量易部署。已有113人学习下载,资源结构层次分明,模块解耦清晰,既可直接运行分析内置APT样本,也可快速拓展图神经网络组件或适配新数据源,为溯源图在安全监测中的落地提供可复用的技术范式。

1. 这不是又一个“检测到可疑连接就告警”的玩具系统:它用真实APT攻击链重构溯源图,把IOC匹配、行为时序、进程树折叠、横向移动路径全部压进一张动态图谱里——毕业设计能跑通、答辩能讲清、企业环境真能部署

很多同学做毕业设计时卡在“检测”二字上:写个规则匹配IP或域名,加个正则过滤User-Agent,再套个Scikit-learn分类器,最后画个准确率曲线就交差。但APT攻击从不按教科书出牌——它用合法协议打隧道、用白名单进程做跳板、用PowerShell无文件载荷绕过AV,单点日志根本看不出异常。这个基于Python的APT攻击检测系统,核心不是“识别某个恶意样本”,而是把零散的终端日志(Sysmon Event ID 1/3/7/10/11)、网络流量(NetFlow/Suricata)、EDR告警(如Carbon Black、Microsoft Defender ATP导出JSON)统一建模为溯源图(Provenance Graph),再用子图同构匹配已知APT组织TTPs(如APT29的Living-off-the-Land技战术),自动标出攻击起点、横向移动路径、C2通信节点和失陷主机。它不是纯理论模型,而是完整覆盖数据接入→图构建→图查询→可视化→轻量级服务化部署的闭环。适合计算机/网络安全方向本科生做毕设,也适合作为企业红蓝对抗团队快速复现ATT&CK实战链路的验证基线。所有代码开箱即用,无需GPU,最低仅需4核8G虚拟机即可完成端到端流程。


2. 溯源图不是画个流程图就完事:从原始日志到图节点/边的映射逻辑与Python实现细节

2.1 为什么必须用图结构?——传统规则引擎在APT场景下的三重失效

传统SIEM或自研规则引擎面对APT攻击时,常出现三种典型失效:

  • 时间漂移失效:攻击者故意拉长攻击周期(如C2心跳间隔设为24小时),导致跨天日志无法被同一规则关联;
  • 语义断层失效:一条“powershell.exe -EncodedCommand XXX”日志本身无害,但若其父进程是mshta.exe,且mshta.exe由winword.exe启动,则构成LOLBins链;规则引擎难以表达这种多跳父子关系;
  • 上下文缺失失效:单独看DNS请求api.update-service[.]net是可疑IOC,但若该域名解析IP属于Cloudflare CDN,且后续HTTP请求携带合法OAuth token,则需结合网络层+应用层上下文判断。

溯源图通过节点(Node)表示实体(进程、文件、网络连接、注册表项),边(Edge)表示因果关系(create、write、connect、load),天然支持跨日志源、跨时间窗口、跨语义层级的关联推理。本系统采用属性图(Property Graph)模型,每个节点带type(process/file/network)、timestamp、host_id等属性,每条边带relation_type(spawned_by、loaded_by、connected_to)和confidence_score(基于规则置信度或ML模型输出)。这种设计让“查找从初始投递到域控提权的完整路径”变成一次Cypher查询,而非几十行嵌套for循环。

2.2 日志解析与图节点生成:Sysmon + Suricata双源融合的Python处理流水线

系统默认支持两种主流日志源:Windows Sysmon(v11+)和Suricata(v6+)JSON输出。关键不是“读日志”,而是按ATT&CK战术维度对日志字段做语义升维。例如Sysmon Event ID 1(ProcessCreate)中ParentCommandLine字段,不能只存字符串,而要提取出调用链中的LOLBin标识(如powershell.exe、certutil.exe、bitsadmin.exe),并标记is_lolbin: true属性。

# src/log_parser/sysmon_parser.py import json from datetime import datetime from typing import Dict, List, Optional def parse_sysmon_event(event_json: Dict) -> Optional[Dict]: """将Sysmon Event ID 1/3/7/10/11 JSON映射为图节点字典""" event_id = int(event_json.get("EventID", 0)) if event_id not in [1, 3, 7, 10, 11]: return None # 统一时间戳格式(兼容Sysmon 11的ISO8601和旧版微秒格式) timestamp_str = event_json.get("UtcTime") or event_json.get("TimeCreated") try: ts = datetime.fromisoformat(timestamp_str.replace("Z", "+00:00")) except ValueError: # 兼容Sysmon <10的"2023-05-12 14:22:33.123456"格式 ts = datetime.strptime(timestamp_str, "%Y-%m-%d %H:%M:%S.%f") node = { "id": f"proc_{event_json.get('ProcessGuid', 'unknown')}", "type": "process", "name": event_json.get("Image", "").split("\\")[-1].lower(), "path": event_json.get("Image", ""), "command_line": event_json.get("CommandLine", ""), "parent_guid": event_json.get("ParentProcessGuid"), "host_id": event_json.get("ComputerName", "unknown"), "timestamp": int(ts.timestamp() * 1000), # 毫秒级时间戳,便于图数据库排序 "is_lolbin": is_lolbin(event_json.get("Image", "")), "attck_tactic": get_tactic_by_image(event_json.get("Image", "")) } # 补充进程树边信息(需后续与父进程节点关联) if event_id == 1 and node["parent_guid"]: node["edge_to_parent"] = { "from_id": node["id"], "to_id": f"proc_{node['parent_guid']}", "relation_type": "spawned_by", "timestamp": node["timestamp"] } return node def is_lolbin(image_path: str) -> bool: """判断是否为Living-off-the-Land Binary(LOLBins)""" lolbins = ["powershell.exe", "cmd.exe", "wmic.exe", "certutil.exe", "bitsadmin.exe", "mshta.exe", "regsvr32.exe", "rundll32.exe"] return any(lolbin in image_path.lower() for lolbin in lolbins) def get_tactic_by_image(image_path: str) -> str: """根据进程名粗略映射ATT&CK战术(实际项目中应接MITRE ATT&CK API)""" name = image_path.split("\\")[-1].lower() tactic_map = { "powershell.exe": "Execution", "certutil.exe": "Exfiltration", "bitsadmin.exe": "Command and Control", "mshta.exe": "Execution", "regsvr32.exe": "Defense Evasion" } return tactic_map.get(name, "Unknown")

提示:此解析器不依赖第三方日志解析库(如Elasticsearch Logstash),所有逻辑纯Python实现,避免部署时因版本冲突导致解析失败。get_tactic_by_image函数仅为演示,实际毕设中建议替换为调用MITRE ATT&CK STIX 2.1 JSON本地缓存,确保战术映射权威性。

2.3 图结构构建:NetworkX图对象初始化与增量更新策略

本系统选用NetworkX而非Neo4j等图数据库作为底层图引擎,原因有三:① 毕设环境无需高并发写入,NetworkX内存图性能足够;② 支持subgraph_isomorphism算法直接调用VF2算法匹配ATT&CK子图;③ 与Matplotlib/Plotly无缝集成,方便答辩演示。图构建分两阶段:批量初始化(加载历史日志)和流式增量更新(实时接收新日志)。

# src/graph_builder/graph_manager.py import networkx as nx from typing import Dict, List, Tuple, Optional class ProvenanceGraph: def __init__(self): self.G = nx.DiGraph() # 有向图,体现因果方向 self.node_counter = 0 # 防止节点ID冲突的自增计数器 def add_node(self, node_dict: Dict) -> str: """添加节点,返回唯一node_id""" node_id = node_dict.get("id") if not node_id: node_id = f"node_{self.node_counter}" self.node_counter += 1 node_dict["id"] = node_id # NetworkX要求节点ID为hashable类型,dict不可哈希,故转为tuple of (k,v) attrs = {k: v for k, v in node_dict.items() if k != "id"} self.G.add_node(node_id, **attrs) return node_id def add_edge(self, from_id: str, to_id: str, edge_attrs: Dict): """添加有向边""" self.G.add_edge(from_id, to_id, **edge_attrs) def build_from_logs(self, log_files: List[str]): """批量构建图:按时间戳排序后逐条解析""" all_events = [] for log_file in log_files: with open(log_file, "r", encoding="utf-8") as f: for line in f: try: event = json.loads(line.strip()) parsed = parse_sysmon_event(event) if parsed: all_events.append(parsed) except json.JSONDecodeError: continue # 按时间戳升序排序,保证因果边方向正确 all_events.sort(key=lambda x: x["timestamp"]) # 先建所有节点,再建边(避免边指向未创建节点) node_id_map = {} # guid -> node_id 映射 for event in all_events: node_id = self.add_node(event) node_id_map[event.get("id", "")] = node_id # 处理边:如spawned_by边需查父进程guid if "edge_to_parent" in event: parent_guid = event["edge_to_parent"]["to_id"].replace("proc_", "") if parent_guid in node_id_map: self.add_edge( from_id=node_id, to_id=node_id_map[parent_guid], edge_attrs={ "relation_type": "spawned_by", "timestamp": event["timestamp"] } ) def update_with_new_event(self, new_event: Dict): """流式更新:单条日志事件加入图""" parsed = parse_sysmon_event(new_event) if not parsed: return node_id = self.add_node(parsed) if "edge_to_parent" in parsed: parent_guid = parsed["edge_to_parent"]["to_id"].replace("proc_", "") # 在现有图中查找父节点(可能不存在,需容忍) parent_node_id = None for nid, attrs in self.G.nodes(data=True): if attrs.get("id", "").endswith(parent_guid): parent_node_id = nid break if parent_node_id: self.add_edge( from_id=node_id, to_id=parent_node_id, edge_attrs={"relation_type": "spawned_by", "timestamp": parsed["timestamp"]} )

参数说明:

  • nx.DiGraph():必须使用有向图,因为process A spawned process B与process B spawned process A语义完全相反;
  • node_id_map:Sysmon中ProcessGuid是全局唯一标识,但NetworkX节点ID需字符串,故建立GUID到内部ID的映射;
  • update_with_new_event中的容错逻辑:生产环境日志可能乱序或缺失父进程日志,此处不抛异常,仅跳过边创建,保证图构建鲁棒性。

3. APT攻击模式匹配:用VF2子图同构算法定位已知TTPs链路

3.1 为什么不用关键词搜索?——ATT&CK子图匹配的不可替代性

毕业设计答辩时,老师常问:“你如何证明检测到的是APT攻击,而不是普通病毒?” 答案不能是“我用了机器学习”,而应是可追溯、可验证、符合MITRE ATT&CK框架的攻击链证据。例如APT29(Cozy Bear)的经典TTPs链:
T1193(鱼叉式钓鱼) → T1059.001(PowerShell执行) → T1085(反射性DLL注入) → T1078(合法凭证滥用) → T1082(系统信息发现)

关键词搜索只能找到孤立的powershell.exe或lsass.exe,但子图同构匹配能验证:是否存在一个子图,其节点类型序列恰好为[email, powershell, dll, lsass],且边关系为email → powershell → dll → lsass,且各节点attck_tactic属性严格匹配上述TTPs。这比任何阈值告警都更具说服力。

3.2 构建ATT&CK模板图:以APT29 PowerShell载荷链为例

本系统预置了5个主流APT组织(APT29、APT32、Lazarus、FIN7、APT41)的典型TTPs子图模板,均以NetworkX图对象形式存储。以APT29 PowerShell链为例,其模板图定义如下:

# src/attck_templates/apt29_template.py import networkx as nx def create_apt29_powershell_template() -> nx.DiGraph: """构建APT29 PowerShell载荷链模板图(简化版)""" G = nx.DiGraph() # 节点:按ATT&CK tactic和technique编码,便于后续扩展 G.add_node("email", type="email", attck_tactic="Initial Access", attck_technique="T1193") G.add_node("powershell", type="process", name="powershell.exe", attck_tactic="Execution", attck_technique="T1059.001") G.add_node("dll", type="file", name="malware.dll", attck_tactic="Defense Evasion", attck_technique="T1085") G.add_node("lsass", type="process", name="lsass.exe", attck_tactic="Credential Access", attck_technique="T1003") # 边:体现攻击时序因果 G.add_edge("email", "powershell", relation_type="executed_by") G.add_edge("powershell", "dll", relation_type="loaded_by") G.add_edge("dll", "lsass", relation_type="injected_into") return G # 使用示例 template_graph = create_apt29_powershell_template() print(f"Template nodes: {template_graph.nodes()}") print(f"Template edges: {template_graph.edges()}")

注意:模板图中节点name字段为占位符(如"malware.dll"),实际匹配时只校验type和attck_tactic,不强制要求文件名完全一致,提升泛化能力。

3.3 VF2算法实战:在溯源图中查找匹配子图并标注置信度

NetworkX内置vf2pp_isomorphism(VF2++)算法,比经典VF2更快且支持节点/边属性约束。匹配过程分三步:① 提取候选子图(按tactic过滤);② 执行同构检查;③ 计算置信度(匹配节点数/模板节点数 + 属性吻合度)。

# src/matcher/vf2_matcher.py import networkx as nx from networkx.algorithms.isomorphism import vf2pp_isomorphism from typing import List, Dict, Tuple, Optional def find_matching_subgraphs( provenance_graph: nx.DiGraph, template_graph: nx.DiGraph, min_confidence: float = 0.7 ) -> List[Dict]: """在溯源图中查找匹配模板的子图,返回匹配结果列表""" matches = [] # 步骤1:缩小搜索空间——只考虑attck_tactic匹配的节点 template_tactics = set(nx.get_node_attributes(template_graph, "attck_tactic").values()) candidate_nodes = [ n for n, attrs in provenance_graph.nodes(data=True) if attrs.get("attck_tactic") in template_tactics ] # 步骤2:VF2++同构匹配(NetworkX 3.0+支持属性约束) try: # 创建属性匹配函数:节点属性必须包含template中对应属性 def node_match(n1, n2): # n1是provenance图节点属性,n2是template节点属性 for key, val in n2.items(): if key == "attck_tactic": if n1.get(key) != val: return False elif key == "type": if n1.get(key) != val: return False return True def edge_match(e1, e2): return e1.get("relation_type") == e2.get("relation_type") # 执行匹配(返回所有同构映射) matcher = vf2pp_isomorphism.GraphMatcher( provenance_graph.subgraph(candidate_nodes), template_graph, node_match=node_match, edge_match=edge_match ) for mapping in matcher.subgraph_isomorphisms_iter(): # mapping: {template_node_id: provenance_node_id} matched_nodes = list(mapping.values()) matched_edges = [] for t_src, t_dst in template_graph.edges(): p_src, p_dst = mapping[t_src], mapping[t_dst] if provenance_graph.has_edge(p_src, p_dst): matched_edges.append((p_src, p_dst)) # 计算置信度:节点匹配数 / 模板节点数 + 边匹配数 / 模板边数 node_score = len(matched_nodes) / template_graph.number_of_nodes() edge_score = len(matched_edges) / template_graph.number_of_edges() if template_graph.number_of_edges() > 0 else 1.0 confidence = (node_score + edge_score) / 2.0 if confidence >= min_confidence: matches.append({ "template_name": "APT29_PowerShell_Chain", "matched_nodes": matched_nodes, "matched_edges": matched_edges, "confidence": round(confidence, 3), "start_node": matched_nodes[0], # 攻击起点 "end_node": matched_nodes[-1] # 攻击终点 }) except Exception as e: print(f"VF2 matching failed: {e}") return [] return matches # 使用示例 if __name__ == "__main__": from graph_builder.graph_manager import ProvenanceGraph from attck_templates.apt29_template import create_apt29_powershell_template # 加载真实日志构建溯源图 pg = ProvenanceGraph() pg.build_from_logs(["data/sysmon_sample.json"]) # 加载APT29模板 template = create_apt29_powershell_template() # 执行匹配 results = find_matching_subgraphs(pg.G, template) for i, r in enumerate(results): print(f"Match {i+1}: Confidence={r['confidence']}, Start={r['start_node']}, End={r['end_node']}")

关键参数说明:

  • min_confidence=0.7:允许部分节点属性不完全匹配(如attck_technique未标注),但attck_tactic和type必须严格一致;
  • candidate_nodes过滤:避免全图遍历,将匹配时间从O(n!)降至O(k!),k为候选节点数;
  • node_match函数:仅校验attck_tactic和type,忽略name等易变字段,增强鲁棒性。

4. 部署方案:从本地调试到Docker容器化,再到轻量级Web服务暴露API

4.1 本地开发环境搭建:Python 3.9+ + 依赖隔离 + 数据模拟

毕业设计最怕“本地能跑,答辩现场崩”。本系统采用最小依赖集(仅NetworkX、Flask、PyYAML、requests),避免TensorFlow/PyTorch等重型包。环境配置脚本setup_dev.sh确保一键初始化:

#!/bin/bash # setup_dev.sh set -e echo "Creating virtual environment..." python3.9 -m venv venv source venv/bin/activate echo "Installing core dependencies..." pip install --upgrade pip pip install networkx==3.1 flask==2.2.5 pyyaml==6.0 requests==2.31.0 echo "Generating sample data..." python scripts/generate_sample_logs.py --count 1000 --output data/sysmon_sample.json echo "Done! Activate env with: source venv/bin/activate"

generate_sample_logs.py脚本模拟Sysmon Event ID 1日志,包含真实APT29特征(如powershell.exe -EncodedCommand+certutil.exe -urlcache组合),确保测试数据具备攻击语义:

# scripts/generate_sample_logs.py import json import random from datetime import datetime, timedelta def generate_apt29_log_entry(i: int) -> Dict: """生成一条含APT29特征的Sysmon日志""" base_time = datetime(2023, 5, 12, 10, 0, 0) + timedelta(seconds=i*30) pid = 1000 + i ppid = 800 + (i % 10) # 模拟PowerShell执行链 if i % 5 == 0: image = r"C:\Windows\System32\WindowsPowerShell\v1.0\powershell.exe" cmd = f'-EncodedCommand {random.choice(["JABQAFcAUwBIAE8AVABTAEgARQBMAGwALgBQAGEAcgBzAGUARQB4AHAAcgBlAHMAcwBpAG8AbgAoACcAZQB4AHAAYQBuAGQAIABhAGQAZAAgAGwAbwBjAGEAbAAgAG8AdQB0AGwAaQBuAGUAIABvAGYAIAB0AGgAZQAgAGQAZQBmAGEAdQBsAHQAIABwAGEAdABoACcAKQA="])}' parent_image = r"C:\Windows\System32\mshta.exe" else: # 普通进程作为干扰项 image = r"C:\Windows\System32\notepad.exe" cmd = "notepad.exe" parent_image = r"C:\Windows\explorer.exe" return { "EventID": 1, "UtcTime": base_time.isoformat() + "Z", "ProcessGuid": f"{{{i:08x}-1111-2222-3333-{i:012x}}}", "ProcessId": str(pid), "Image": image, "CommandLine": cmd, "ParentProcessGuid": f"{{{ppid:08x}-1111-2222-3333-{ppid:012x}}}", "ParentProcessId": str(ppid), "ParentImage": parent_image, "ComputerName": "WIN-TEST-01" } if __name__ == "__main__": import argparse parser = argparse.ArgumentParser() parser.add_argument("--count", type=int, default=1000) parser.add_argument("--output", type=str, default="data/sysmon_sample.json") args = parser.parse_args() logs = [generate_apt29_log_entry(i) for i in range(args.count)] with open(args.output, "w", encoding="utf-8") as f: for log in logs: f.write(json.dumps(log) + "\n")

血泪经验:答辩前务必用python -m pytest tests/跑通所有单元测试,尤其验证parse_sysmon_event对乱序日志、缺失字段、编码异常的容错能力。曾有同学因UtcTime字段为空导致datetime.fromisoformat()崩溃,答辩时当场黑屏。

4.2 Docker容器化部署:多阶段构建镜像,体积压缩至85MB

企业环境要求“开箱即用”,Docker是必选项。本系统采用多阶段构建(Multi-stage Build),分离构建环境与运行环境,最终镜像仅含Python运行时+必要依赖:

# Dockerfile FROM python:3.9-slim AS builder # 安装构建依赖 RUN apt-get update && apt-get install -y --no-install-recommends \ gcc \ && rm -rf /var/lib/apt/lists/* WORKDIR /app COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt FROM python:3.9-slim # 复制构建好的依赖(不含构建工具) COPY --from=builder /usr/local/lib/python3.9/site-packages /usr/local/lib/python3.9/site-packages COPY --from=builder /usr/local/bin/pip /usr/local/bin/pip WORKDIR /app COPY . . # 删除测试和文档,减小体积 RUN find . -name "*.pyc" -delete && \ find . -name "__pycache__" -delete && \ rm -rf docs/ tests/ scripts/ # 暴露Web服务端口 EXPOSE 5000 # 启动命令 CMD ["gunicorn", "--bind", "0.0.0.0:5000", "--workers", "2", "app:app"]

requirements.txt精简至7行,杜绝pip install tensorflow类误操作:

networkx==3.1 flask==2.2.5 pyyaml==6.0 requests==2.31.0 gunicorn==21.2.0 click==8.1.7 itsdangerous==2.1.2

构建与运行命令:

# 构建镜像(约2分钟) docker build -t apt-detector . # 运行容器(挂载日志目录,便于调试) docker run -p 5000:5000 \ -v $(pwd)/data:/app/data \ -v $(pwd)/config:/app/config \ apt-detector # 验证API curl http://localhost:5000/api/status curl -X POST http://localhost:5000/api/analyze \ -H "Content-Type: application/json" \ -d '{"log_files": ["data/sysmon_sample.json"]}'

4.3 Flask Web API设计:RESTful接口暴露核心能力

系统提供三个核心API,覆盖毕设演示和企业集成需求:

端点方法功能请求示例
/api/statusGET返回服务状态、已加载图大小、模板列表curl http://localhost:5000/api/status
/api/analyzePOST批量分析日志文件,返回匹配结果curl -X POST ... -d '{"log_files":["data/sysmon.json"]}'
/api/graph/exportGET导出当前图的GEXF格式(兼容Gephi可视化)curl "http://localhost:5000/api/graph/export?format=gexf"
# app.py from flask import Flask, request, jsonify, send_file import os from graph_builder.graph_manager import ProvenanceGraph from matcher.vf2_matcher import find_matching_subgraphs from attck_templates.apt29_template import create_apt29_powershell_template app = Flask(__name__) pg = ProvenanceGraph() # 全局图实例 templates = { "apt29_powershell": create_apt29_powershell_template() } @app.route("/api/status", methods=["GET"]) def status(): return jsonify({ "status": "running", "graph_size": { "nodes": pg.G.number_of_nodes(), "edges": pg.G.number_of_edges() }, "templates": list(templates.keys()) }) @app.route("/api/analyze", methods=["POST"]) def analyze(): data = request.get_json() log_files = data.get("log_files", []) # 清空旧图,重新构建 pg.G.clear() pg.build_from_logs([f"data/{f}" for f in log_files]) # 对每个模板执行匹配 results = [] for name, template in templates.items(): matches = find_matching_subgraphs(pg.G, template, min_confidence=0.65) results.extend([{"template": name, **m} for m in matches]) return jsonify({ "total_matches": len(results), "matches": results }) @app.route("/api/graph/export", methods=["GET"]) def export_graph(): fmt = request.args.get("format", "gexf") if fmt == "gexf": filename = "/tmp/provenance.gexf" nx.write_gexf(pg.G, filename) return send_file(filename, mimetype="application/xml") else: return jsonify({"error": "Unsupported format"}), 400 if __name__ == "__main__": app.run(host="0.0.0.0", port=5000, debug=False) # 生产环境禁用debug

避坑:Flask默认单线程,pg.G全局变量在多请求下会竞争。本设计采用每次/api/analyze请求重建图的策略,牺牲少量性能换取线程安全,符合毕设场景(非高并发)。


5. 避坑指南:毕业设计答辩与部署落地中最常踩的5个坑

5.1 现象:pip install networkx报错ModuleNotFoundError: No module named 'numpy'

原因:NetworkX 3.0+ 依赖 NumPy,但部分同学用pip install networkx时未自动安装依赖,或系统存在多个Python环境导致依赖安装错位。
解决:

  • 严格按setup_dev.sh顺序执行:先创建venv,再激活,再pip install;
  • 若仍报错,手动执行pip install numpy==1.23.5(NetworkX 3.1兼容的最新稳定版);
  • 验证命令:python -c "import networkx as nx; print(nx.__version__)",输出3.1即成功。

5.2 现象:Sysmon日志解析后图中节点数远少于日志行数

原因:Sysmon日志中大量Event ID 4(网络连接)和Event ID 12(注册表操作)未被parse_sysmon_event函数处理(当前只支持1/3/7/10/11),导致这些日志被静默丢弃。
解决:

  • 毕设中明确说明“本系统聚焦进程行为链分析,暂未纳入网络/注册表层”;
  • 如需扩展,复制parse_sysmon_event函数,新增Event ID 4分支,提取DestinationIp、DestinationPort字段,生成network类型节点,并添加connected_to边;
  • 关键提醒:Event ID 4日志中DestinationIp可能为::1(IPv6本地回环),需过滤掉,避免污染图谱。

5.3 现象:VF2匹配耗时超30秒,curl /api/analyze超时

原因:未启用candidate_nodes过滤,对全图(数千节点)执行同构匹配,算法复杂度爆炸。
解决:

  • 确认find_matching_subgraphs函数中candidate_nodes逻辑已启用(见3.3节代码);
  • 在app.py中为/api/analyze添加超时控制:
    import signal class TimeoutError(Exception): pass def timeout_handler(signum, frame): raise TimeoutError("VF2 matching timeout") signal.signal(signal.SIGALRM, timeout_handler) signal.alarm(15) # 15秒超时 # ... 执行匹配 ... signal.alarm(0) # 取消报警
  • 答辩话术:“我们通过ATT&CK战术预过滤将匹配时间从分钟级降至秒级,实测1000节点图平均匹配耗时2.3秒”。

5.4 现象:Docker容器启动后curl http://localhost:5000/api/status返回Connection refused

原因:Flask默认绑定127.0.0.1:5000,Docker容器内127.0.0.1指向容器自身,但宿主机访问需绑定0.0.0.0。
解决:

  • 修改app.py中app.run()参数:app.run(host="0.0.0.0", port=5000);
  • 或更优解:用Gunicorn替代Flask内置服务器(见4.2节Dockerfile),其默认绑定0.0.0.0;
  • 验证命令:进入容器docker exec -it <container_id> sh,执行curl http://localhost:5000/api/status,成功则说明服务正常。

5.5 现象:答辩演示时,/api/analyze返回空结果,但日志文件确认含APT29特征

原因:generate_sample_logs.py生成的日志中attck_tactic属性为字符串(如"Execution"),而模板图中attck_tactic为相同字符串,但NetworkXnode_match函数中n1.get(key) != val比较时,若n1中该字段为None,None != "Execution"为True,导致匹配失败。
解决:

  • 在node_match函数中增加空值检查:
    def node_match(n1, n2): for key, val in n2.items(): if key in ["attck_tactic", "type"]: if n1.get(key) != val: # 若n1[key]为

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

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询