$npx -y skills add killvxk/cybersecurity-skills-zh --skill analyzing-campaign-attribution-evidence攻击活动溯源归因分析涉及系统性地评估证据,以确定哪个威胁行为者或组织对某次网络行动负责。本技能涵盖使用 Diamond Model 和 ACH(竞争假设分析)收集并加权溯源归因指标、分析基础设施重叠、TTP 一致性、恶意软件代码相似性、操作时序模式和语言痕迹,以构建置信度加权的溯源归因评估。
| 1 | # 分析攻击活动溯源归因证据 |
| 2 | |
| 3 | ## 概述 |
| 4 | |
| 5 | 攻击活动溯源归因(Attribution)分析涉及系统性地评估证据,以确定哪个威胁行为者(Threat Actor)或组织对某次网络行动负责。本技能涵盖使用 Diamond Model 和 ACH(竞争假设分析)收集并加权溯源指标、分析基础设施重叠、TTP 一致性、恶意软件代码相似性、操作时序模式和语言痕迹,以构建置信度加权的归因评估报告。 |
| 6 | |
| 7 | ## 前置条件 |
| 8 | |
| 9 | - Python 3.9+,安装 `attackcti`、`stix2`、`networkx` 库 |
| 10 | - 访问威胁情报平台(MISP、OpenCTI) |
| 11 | - 了解 Diamond Model 入侵分析框架 |
| 12 | - 熟悉 MITRE ATT&CK 威胁组织画像 |
| 13 | - 掌握恶意软件分析和基础设施追踪技术 |
| 14 | |
| 15 | ## 核心概念 |
| 16 | |
| 17 | ### 溯源证据类别 |
| 18 | 1. **基础设施重叠**:共享 C2 服务器、域名、IP 范围、托管服务商 |
| 19 | 2. **TTP 一致性**:跨攻击活动中匹配的 ATT&CK 技术和子技术 |
| 20 | 3. **恶意软件代码相似性**:共享代码库、编译器、PDB 路径、加密例程 |
| 21 | 4. **操作模式**:时序(工作时间、时区)、目标模式、操作节奏 |
| 22 | 5. **语言痕迹**:特定语言的嵌入字符串、变量名、错误消息 |
| 23 | 6. **受害者学**:目标行业、地理位置和组织画像一致性 |
| 24 | |
| 25 | ### 置信度级别 |
| 26 | - **高置信度**:多个独立证据类别聚焦于同一行为者 |
| 27 | - **中置信度**:若干证据类别匹配,但存在一定模糊性 |
| 28 | - **低置信度**:证据有限,可能存在伪旗或共享工具 |
| 29 | |
| 30 | ### 竞争假设分析(ACH) |
| 31 | 一种结构化分析方法,针对多个竞争假设评估证据。每条证据针对每个假设被评分为一致、不一致或中性。不一致证据最少的假设为优先假设。 |
| 32 | |
| 33 | ## 实践步骤 |
| 34 | |
| 35 | ### 步骤 1:收集溯源证据 |
| 36 | |
| 37 | ```python |
| 38 | from stix2 import MemoryStore, Filter |
| 39 | from collections import defaultdict |
| 40 | |
| 41 | class AttributionAnalyzer: |
| 42 | def __init__(self): |
| 43 | self.evidence = [] |
| 44 | self.hypotheses = {} |
| 45 | |
| 46 | def add_evidence(self, category, description, value, confidence): |
| 47 | self.evidence.append({ |
| 48 | "category": category, |
| 49 | "description": description, |
| 50 | "value": value, |
| 51 | "confidence": confidence, |
| 52 | "timestamp": None, |
| 53 | }) |
| 54 | |
| 55 | def add_hypothesis(self, actor_name, actor_id=""): |
| 56 | self.hypotheses[actor_name] = { |
| 57 | "actor_id": actor_id, |
| 58 | "consistent_evidence": [], |
| 59 | "inconsistent_evidence": [], |
| 60 | "neutral_evidence": [], |
| 61 | "score": 0, |
| 62 | } |
| 63 | |
| 64 | def evaluate_evidence(self, evidence_idx, actor_name, assessment): |
| 65 | """评估证据与假设的关系:一致/不一致/中性。""" |
| 66 | if assessment == "consistent": |
| 67 | self.hypotheses[actor_name]["consistent_evidence"].append(evidence_idx) |
| 68 | self.hypotheses[actor_name]["score"] += self.evidence[evidence_idx]["confidence"] |
| 69 | elif assessment == "inconsistent": |
| 70 | self.hypotheses[actor_name]["inconsistent_evidence"].append(evidence_idx) |
| 71 | self.hypotheses[actor_name]["score"] -= self.evidence[evidence_idx]["confidence"] * 2 |
| 72 | else: |
| 73 | self.hypotheses[actor_name]["neutral_evidence"].append(evidence_idx) |
| 74 | |
| 75 | def rank_hypotheses(self): |
| 76 | """按溯源分数对假设进行排序。""" |
| 77 | ranked = sorted( |
| 78 | self.hypotheses.items(), |
| 79 | key=lambda x: x[1]["score"], |
| 80 | reverse=True, |
| 81 | ) |
| 82 | return [ |
| 83 | { |
| 84 | "actor": name, |
| 85 | "score": data["score"], |
| 86 | "consistent": len(data["consistent_evidence"]), |
| 87 | "inconsistent": len(data["inconsistent_evidence"]), |
| 88 | "confidence": self._score_to_confidence(data["score"]), |
| 89 | } |
| 90 | for name, data in ranked |
| 91 | ] |
| 92 | |
| 93 | def _score_to_confidence(self, score): |
| 94 | if score >= 80: |
| 95 | return "HIGH" |
| 96 | elif score >= 40: |
| 97 | return "MODERATE" |
| 98 | else: |
| 99 | return "LOW" |
| 100 | ``` |
| 101 | |
| 102 | ### 步骤 2:基础设施重叠分析 |
| 103 | |
| 104 | ```python |
| 105 | def analyze_infrastructure_overlap(campaign_a_infra, campaign_b_infra): |
| 106 | """比较两个攻击活动的基础设施以进行溯源。""" |
| 107 | overlap = { |
| 108 | "shared_ips": set(campaign_a_infra.get("ips", [])).intersection( |
| 109 | campaign_b_infra.get("ips", []) |
| 110 | ), |
| 111 | "shared_domains": set(campaign_a_infra.get("domains", [])).intersection( |
| 112 | campaign_b_infra.get("domains", []) |
| 113 | ), |
| 114 | "shared_asns": set(campaign_a_infra.get("asns", [])).intersection( |
| 115 | campaign_b_infra.get("asns", []) |
| 116 | ), |
| 117 | "shared_registrars": set(campaign_a_infra.get("registrars", [])).intersection( |
| 118 | campaign_b_infra.get("registrars", []) |
| 119 | ), |
| 120 | } |
| 121 | |
| 122 | overlap_score = 0 |
| 123 | if overlap["shared_ips"]: |
| 124 | overlap_score += 30 |
| 125 | if overlap["shared_domains"]: |
| 126 | overlap_score += 25 |
| 127 | if overlap["shared_asns"]: |
| 128 | overlap_score += 15 |
| 129 | if overlap["shared_registrars"]: |
| 130 | overlap_score += 10 |
| 131 | |
| 132 | return { |
| 133 | "overlap": {k: list(v) for k, v in overlap.items()}, |
| 134 | "overlap_score": overlap_score, |
| 135 | "assessment": "STRONG" if overlap_score >= 40 else "MODERATE" if overlap_score >= 20 else "WEAK", |
| 136 | } |
| 137 | ``` |
| 138 | |
| 139 | ### 步骤 3:跨攻击活动的 TTP 对比 |
| 140 | |
| 141 | ```python |
| 142 | from attackcti import attack_client |
| 143 | |
| 144 | def compare_campaign_ttps(campaign_techniques, known_actor_techniques): |
| 145 | """将攻击活动 TTP 与已知威胁行为者画像进行对比。""" |
| 146 | campaign_set = set(campaign_techniques) |
| 147 | actor_set = set(known_actor_techniques) |
| 148 | |
| 149 | common = campaign_set.intersection(actor_set) |
| 150 | unique_campaign = campaign_set - actor_set |
| 151 | unique_actor = actor_set - campaign_set |
| 152 | |
| 153 | jaccard = len(common) / le |