killvxk 实现菱形模型分析
概述
菱形入侵分析模型通过审查四个核心特征（对手、能力、基础设施、受害者）提供结构化框架，用于分析网络入侵事件。本技能涵盖以编程方式实现菱形模型，对入侵事件进行分类和关联，构建链接相关事件的活动线程，创建活动攻击图，并从入侵数据中生成可枢纽的情报。
前置条件
- Python 3.9+ 及
networkx、stix2、graphviz库 - 理解菱形模型的核心特征和元特征
- 可访问威胁情报数据（MISP/OpenCTI 事件）
- 熟悉 MITRE ATT&CK 用于能力映射
核心概念
菱形模型核心特征
- 对手（Adversary）：实施入侵的威胁行为者或操作者
- 能力（Capability）：使用的工具、技术和恶意软件（映射到 ATT&CK）
- 基础设施（Infrastructure）：C2 服务器、域名、电子邮件地址、托管提供商
- 受害者（Victim）：目标组织、系统、人员或数据资产
元特征
- 时间戳（Timestamp）：事件发生时间
- 阶段（Phase）：杀伤链阶段（侦察、投递、利用等）
- 结果（Result）：成功、失败或未知
- 方向（Direction）：对手到基础设施、基础设施到受害者等
- 方法论（Methodology）：社会工程学、技术利用、内部威胁
- 资源（Resources）：所需财务、人力、技术资源
活动线程与活动组
- 活动线程（Activity Thread）：来自单一对手操作的菱形事件序列
- 活动组（Activity Group）：归因于同一对手的线程集群
实践步骤
步骤 1：定义菱形事件数据结构
from dataclasses import dataclass, field
from datetime import datetime
from typing import Optional
import json
import uuid
@dataclass
class DiamondEvent:
adversary: str = ""
capability: str = ""
infrastructure: str = ""
victim: str = ""
timestamp: str = ""
phase: str = ""
result: str = ""
direction: str = ""
methodology: str = ""
confidence: int = 0
notes: str = ""
event_id: str = field(default_factory=lambda: str(uuid.uuid4())[:8])
mitre_techniques: list = field(default_factory=list)
iocs: list = field(default_factory=list)
def to_dict(self):
return {
"event_id": self.event_id,
"adversary": self.adversary,
"capability": self.capability,
"infrastructure": self.infrastructure,
"victim": self.victim,
"timestamp": self.timestamp,
"phase": self.phase,
"result": self.result,
"direction": self.direction,
"methodology": self.methodology,
"confidence": self.confidence,
"mitre_techniques": self.mitre_techniques,
"iocs": self.iocs,
"notes": self.notes,
}
步骤 2：从事件构建活动线程
import networkx as nx
class DiamondAnalysis:
def __init__(self):
self.events = []
self.graph = nx.DiGraph()
def add_event(self, event: DiamondEvent):
self.events.append(event)
self.graph.add_node(event.event_id, **event.to_dict())
def build_activity_thread(self):
"""按时间顺序将事件链接为活动线程。"""
sorted_events = sorted(self.events, key=lambda e: e.timestamp)
for i in range(len(sorted_events) - 1):
self.graph.add_edge(
sorted_events[i].event_id,
sorted_events[i + 1].event_id,
relationship="followed_by",
)
def find_pivots(self):
"""查找共享基础设施或能力的事件枢纽点。"""
pivots = {"infrastructure": {}, "capability": {}, "adversary": {}}
for event in self.events:
if event.infrastructure:
pivots["infrastructure"].setdefault(event.infrastructure, []).append(event.event_id)
if event.capability:
pivots["capability"].setdefault(event.capability, []).append(event.event_id)
if event.adversary:
pivots["adversary"].setdefault(event.adversary, []).append(event.event_id)
return {
k: {pk: pv for pk, pv in v.items() if len(pv) > 1}
for k, v in pivots.items()
}
def generate_report(self):
return {
"total_events": len(self.events),
"unique_adversaries": len(set(e.adversary for e in self.events if e.adversary)),
"unique_victims": len(set(e.victim for e in self.events if e.victim)),
"unique_infrastructure": len(set(e.infrastructure for e in self.events if e.infrastructure)),
"pivots": self.find_pivots(),
"events": [e.to_dict() for e in self.events],
}
验证标准
- 菱形事件捕获所有四个核心特征及元特征
- 活动线程按时间顺序链接相关事件
- 枢纽分析识别跨事件的共享基础设施和能力
- 图形可视化正确呈现活动攻击图
- 事件映射到 MITRE ATT&CK 技术用于能力分类