战役归因证据分析

战役归因证据分析 是做什么的？

概述

战役归因分析涉及系统评估证据，以确定哪个威胁行为者或团体对网络行动负责。本技能涵盖使用钻石模型和ACH（竞争假设分析法）收集和加权归因指标、分析基础设施重叠、TTP一致性、恶意软件代码相似性、操作时间模式以及语言痕迹，从而构建具有置信度权重的归因评估。

何时使用

• 当调查安全事件需要分析战役归因证据时
• 当为该领域构建检测规则或威胁狩猎查询时
• 当SOC分析师需要此分析类型的结构化流程时
• 当验证相关攻击技术安全监控覆盖率时

先决条件

• Python 3.9+，并安装 attackcti、stix2、networkx 库
• 威胁情报平台（MISP、OpenCTI）的访问权限
• 理解入侵分析钻石模型
• 熟悉MITRE ATT&CK威胁团体资料
• 掌握恶意软件分析和基础设施追踪技术

关键概念

归因证据类别

1. 基础设施重叠：共享的C2服务器、域名、IP范围、托管提供商
2. TTP一致性：战役间匹配的ATT&CK技术和子技术
3. 恶意软件代码相似性：共享代码库、编译器、PDB路径、加密例程
4. 操作模式：时间（工作时间、时区）、目标选择模式、操作节奏
5. 语言痕迹：特定语言的嵌入字符串、变量名、错误消息
6. 受害目标特征：目标行业、地理位置和组织概况的一致性

置信度等级

• 高置信度：多个独立证据类别汇聚于同一行为者
• 中等置信度：若干证据类别匹配，但存在一些歧义
• 低置信度：证据有限，可能存在虚假标志或共享工具

竞争假设分析法（ACH）

一种结构化分析方法，对多个竞争性假设进行证据评估。每条证据相对于每个假设被评分为一致、不一致或中立。得分最低的不一致证据的假设被视为更优。

工作流程

步骤1：收集归因证据

from stix2 import MemoryStore, Filter
from collections import defaultdict

class AttributionAnalyzer:
    def __init__(self):
        self.evidence = []
        self.hypotheses = {}

    def add_evidence(self, category, description, value, confidence):
        self.evidence.append({
            "category": category,
            "description": description,
            "value": value,
            "confidence": confidence,
            "timestamp": None,
        })

    def add_hypothesis(self, actor_name, actor_id=""):
        self.hypotheses[actor_name] = {
            "actor_id": actor_id,
            "consistent_evidence": [],
            "inconsistent_evidence": [],
            "neutral_evidence": [],
            "score": 0,
        }

    def evaluate_evidence(self, evidence_idx, actor_name, assessment):
        """将证据与假设进行比对：一致/不一致/中立。"""
        if assessment == "consistent":
            self.hypotheses[actor_name]["consistent_evidence"].append(evidence_idx)
            self.hypotheses[actor_name]["score"] += self.evidence[evidence_idx]["confidence"]
        elif assessment == "inconsistent":
            self.hypotheses[actor_name]["inconsistent_evidence"].append(evidence_idx)
            self.hypotheses[actor_name]["score"] -= self.evidence[evidence_idx]["confidence"] * 2
        else:
            self.hypotheses[actor_name]["neutral_evidence"].append(evidence_idx)

    def rank_hypotheses(self):
        """按归因分数对假设进行排序。"""
        ranked = sorted(
            self.hypotheses.items(),
            key=lambda x: x[1]["score"],
            reverse=True,
        )
        return [
            {
                "actor": name,
                "score": data["score"],
                "consistent": len(data["consistent_evidence"]),
                "inconsistent": len(data["inconsistent_evidence"]),
                "confidence": self._score_to_confidence(data["score"]),
            }
            for name, data in ranked
        ]

    def _score_to_confidence(self, score):
        if score >= 80:
            return "HIGH"
        elif score >= 40:
            return "MODERATE"
        else:
            return "LOW"

步骤2：基础设施重叠分析

def analyze_infrastructure_overlap(campaign_a_infra, campaign_b_infra):
    """比较两个战役的基础设施以进行归因。"""
    overlap = {
        "shared_ips": set(campaign_a_infra.get("ips", [])).intersection(
            campaign_b_infra.get("ips", [])
        ),
        "shared_domains": set(campaign_a_infra.get("domains", [])).intersection(
            campaign_b_infra.get("domains", [])
        ),
        "shared_asns": set(campaign_a_infra.get("asns", [])).intersection(
            campaign_b_infra.get("asns", [])
        ),
        "shared_registrars": set(campaign_a_infra.get("registrars", [])).intersection(
            campaign_b_infra.get("registrars", [])
        ),
    }

    overlap_score = 0
    if overlap["shared_ips"]:
        overlap_score += 30
    if overlap["shared_domains"]:
        overlap_score += 25
    if overlap["shared_asns"]:
        overlap_score += 15
    if overlap["shared_registrars"]:
        overlap_score += 10

    return {
        "overlap": {k: list(v) for k, v in overlap.items()},
        "overlap_score": overlap_score,
        "assessment": "STRONG" if overlap_score >= 40 else "MODERATE" if overlap_score >= 20 else "WEAK",
    }

步骤3：跨战役TTP比较

from attackcti import attack_client

def compare_campaign_ttps(campaign_techniques, known_actor_techniques):
    """将战役TTP与已知威胁行为者资料进行比较。"""
    campaign_set = set(campaign_techniques)
    actor_set = set(known_actor_techniques)

    common = campaign_set.intersection(actor_set)
    unique_campaign = campaign_set - actor_set
    unique_actor = actor_set - campaign_set

    jaccard = len(common) / len(campaign_set.union(actor_set)) if campaign_set.union(actor_set) else 0

    return {
        "common_techniques": sorted(common),
        "common_count": len(common),
        "unique_to_campaign": sorted(unique_campaign),
        "unique_to_actor": sorted(unique_actor),
        "jaccard_similarity": round(jaccard, 3),
        "overlap_percentage": round(len(common) / len(campaign_set) * 100, 1) if campaign_set else 0,
    }

步骤4：生成归因报告

def generate_attribution_report(analyzer):
    """生成结构化的归因评估报告。"""
    rankings = analyzer.rank_hypotheses()

    report = {
        "assessment_date": "2026-02-23",
        "total_evidence_items": len(analyzer.evidence),
        "hypotheses_evaluated": len(analyzer.hypotheses),
        "rankings": rankings,
        "primary_attribution": rankings[0] if rankings else None,
        "evidence_summary": [
            {
                "index": i,
                "category": e["category"],
                "description": e["description"],
                "confidence": e["confidence"],
            }
            for i, e in enumerate(analyzer.evidence)
        ],
    }

    return report

验证标准

• 证据收集覆盖全部六个归因类别
• ACH矩阵正确评估证据与竞争假设的关系
• 基础设施重叠分析识别共享指标
• TTP比较使用ATT&CK技术ID以确保精确性
• 归因置信度等级得到合理证明
• 报告包含备选假设和虚假标志考量

参考资料

• 入侵分析钻石模型
• MITRE ATT&CK团体
• 竞争假设分析法
• 威胁归因框架