從 CTI 報告建立攻擊樣式資料庫 是做什麼的？
概覽
來自 Mandiant、CrowdStrike、Talos 和 Microsoft 等供應商的網路威脅情資 (CTI) 報告中，包含詳細的對手行為描述，這些描述可以被提取、標準化，並編目成結構化的攻擊樣式資料庫。此技能涵蓋：解析 CTI 報告以提取對手技術、將行為對應到 MITRE ATT&CK 技術 ID、建立 STIX 2.1 Attack Pattern 物件、建立按戰術、技術和威脅行為者索引的可搜尋資料庫，以及根據已記錄的模式產生偵測規則範本。
使用時機
- 在您的環境中部署或設定建置 CTI 報告攻擊樣式資料庫功能時
- 建立符合合規要求的安全控制措施時
- 建置或改善此領域的安全架構時
- 進行需要此實作的安全評估時
先決條件
- Python 3.9+ 及
stix2、mitreattack-python、spacy、requests函式庫 - CTI 報告集合（PDF、HTML 或文字格式）
- MITRE ATT&CK STIX 資料（本地或透過 TAXII）
- 了解 ATT&CK 技術結構與命名慣例
- 熟悉偵測工程概念（Sigma、YARA）
關鍵概念
攻擊樣式提取
CTI 報告以自然語言描述對手行為。提取過程包括：識別能對應到 ATT&CK 技術的動作動詞和技術術語、辨識工具名稱與惡意軟體家族、找出基礎設施指標，以及將行為序列對應到攻擊鏈（殺傷鏈階段）。
STIX 2.1 Attack Pattern 物件
STIX 將 Attack Pattern 定義為一種結構化領域物件 (SDO)，用於描述威脅行為者試圖危害目標的方式。每個樣式透過外部參考連結到 ATT&CK，包含殺傷鏈階段（戰術），且可與入侵集合、惡意軟體和工具等物件建立關聯。
偵測規則產生
提取的攻擊樣式可為偵測工程提供：用於 Sigma 規則建立的具體程序範例、用於關聯規則的行為序列、用於 YARA 和 Snort 規則的 IOC 模式，以及用於找出遙測缺口所需的資料來源需求。
工作流程
步驟 1：解析 CTI 報告並提取行為
import re
import json
from collections import defaultdict
class CTIReportParser:
"""解析 CTI 報告以提取對手行為。"""
BEHAVIOR_INDICATORS = [
"used", "executed", "deployed", "leveraged", "exploited",
"established", "created", "modified", "downloaded", "uploaded",
"exfiltrated", "injected", "enumerated", "spawned", "dropped",
"persisted", "escalated", "moved laterally", "collected",
"encrypted", "compressed", "encoded", "obfuscated",
]
TOOL_PATTERNS = [
r'\b(Cobalt Strike|Mimikatz|PsExec|BloodHound|Rubeus|Impacket)\b',
r'\b(PowerShell|cmd\.exe|WMI|WMIC|certutil|bitsadmin)\b',
r'\b(Metasploit|Empire|Covenant|Sliver|Brute Ratel)\b',
r'\b(Lazagne|SharpHound|ADFind|Sharphound|Invoke-Obfuscation)\b',
]
TECHNIQUE_KEYWORDS = {
"spearphishing": "T1566",
"phishing attachment": "T1566.001",
"phishing link": "T1566.002",
"powershell": "T1059.001",
"command line": "T1059.003",
"scheduled task": "T1053.005",
"registry run key": "T1547.001",
"process injection": "T1055",
"dll side-loading": "T1574.002",
"credential dumping": "T1003",
"lsass": "T1003.001",
"kerberoasting": "T1558.003",
"pass the hash": "T1550.002",
"remote desktop": "T1021.001",
"smb": "T1021.002",
"winrm": "T1021.006",
"data staging": "T1074",
"exfiltration over c2": "T1041",
"dns tunneling": "T1071.004",
"web shell": "T1505.003",
}
def parse_report(self, text, report_metadata=None):
"""解析 CTI 報告並提取行為。"""
sentences = re.split(r'[.!?]\s+', text)
behaviors = []
for sentence in sentences:
sentence_lower = sentence.lower()
# 檢查行為指示詞
for indicator in self.BEHAVIOR_INDICATORS:
if indicator in sentence_lower:
behavior = {
"sentence": sentence.strip(),
"action": indicator,
"tools": self._extract_tools(sentence),
"technique_hints": self._match_techniques(sentence_lower),
}
if behavior["technique_hints"]:
behaviors.append(behavior)
break
print(f"[+] 從報告中提取了 {len(behaviors)} 個行為指示詞")
return behaviors
def _extract_tools(self, text):
"""從文字中提取工具/惡意軟體名稱。"""
tools = set()
for pattern in self.TOOL_PATTERNS:
matches = re.findall(pattern, text, re.IGNORECASE)
tools.update(matches)
return list(tools)
def _match_techniques(self, text):
"""將文字比對到 ATT&CK 技術提示。"""
matches = []
for keyword, tech_id in self.TECHNIQUE_KEYWORDS.items():
if keyword in text:
matches.append({"keyword": keyword, "technique_id": tech_id})
return matches
parser = CTIReportParser()
sample_report = """
The threat actor used spearphishing attachments with macro-enabled documents to
gain initial access. Once inside, they executed PowerShell scripts to download
additional tooling. The actor leveraged Mimikatz to dump credentials from LSASS
memory. They then used pass the hash techniques for lateral movement via SMB
to multiple systems. Data was staged in a compressed archive and exfiltrated
over the existing C2 channel. The actor established persistence through
scheduled tasks and registry run keys.
"""
behaviors = parser.parse_report(sample_report)
步驟 2：將行為對應到 ATT&CK 技術
from attackcti import attack_client
class ATTACKMapper:
def __init__(self):
self.lift = attack_client()
self.techniques = {}
self._load_techniques()
def _load_techniques(self):
"""載入所有 ATT&CK 技術以進行對應。"""
all_techs = self.lift.get_enterprise_techniques()
for tech in all_techs:
tech_id = ""
for ref in tech.get("external_references", []):
if ref.get("source_name") == "mitre-attack":
tech_id = ref.get("external_id", "")
break
if tech_id:
self.techniques[tech_id] = {
"name": tech.get("name", ""),
"description": tech.get("description", "")[:500],
"tactics": [p.get("phase_name") for p in tech.get("kill_chain_phases", [])],
"platforms": tech.get("x_mitre_platforms", []),
"data_sources": tech.get("x_mitre_data_sources", []),
}
print(f"[+] 已載入 {len(self.techniques)} 個 ATT&CK 技術")
def map_behaviors(self, behaviors):
"""將提取的行為對應到 ATT&CK 技術。"""
mapped = []
for behavior in behaviors:
for hint in behavior.get("technique_hints", []):
tech_id = hint["technique_id"]
if tech_id in self.techniques:
tech_info = self.techniques[tech_id]
mapped.append({
"technique_id": tech_id,
"technique_name": tech_info["name"],
"tactics": tech_info["tactics"],
"source_sentence": behavior["sentence"],
"tools_observed": behavior["tools"],
"keyword_matched": hint["keyword"],
"data_sources": tech_info["data_sources"],
})
print(f"[+] 已將 {len(mapped)} 個行為對應到 ATT&CK 技術")
return mapped
mapper = ATTACKMapper()
mapped_behaviors = mapper.map_behaviors(behaviors)
步驟 3：建立 STIX 2.1 攻擊樣式資料庫
from stix2 import AttackPattern, Relationship, Bundle, TLP_GREEN
from datetime import datetime
class AttackPatternLibrary:
def __init__(self):
self.patterns = []
self.relationships = []
def add_pattern_from_mapping(self, mapping, report_source="CTI Report"):
"""根據對應的行為建立 STIX Attack Pattern。"""
pattern = AttackPattern(
name=mapping["technique_name"],
description=f"Observed: {mapping['source_sentence']}\n\n"
f"Tools: {', '.join(mapping['tools_observed']) or 'None identified'}\n"
f"Source: {report_source}",
external_references=[{
"source_name": "mitre-attack",
"external_id": mapping["technique_id"],
"url": f"https://attack.mitre.org/techniques/{mapping['technique_id'].replace('.', '/')}/",
}],
kill_chain_phases=[{
"kill_chain_name": "mitre-attack",
"phase_name": tactic,
} for tactic in mapping["tactics"]],
object_marking_refs=[TLP_GREEN],
)
self.patterns.append(pattern)
return pattern
def build_library(self, mapped_behaviors, report_source="CTI Report"):
"""根據對應結果建立完整的攻擊樣式資料庫。"""
seen_techniques = set()
for mapping in mapped_behaviors:
tech_id = mapping["technique_id"]
if tech_id not in seen_techniques:
self.add_pattern_from_mapping(mapping, report_source)
seen_techniques.add(tech_id)
bundle = Bundle(objects=self.patterns + self.relationships)
print(f"[+] 資料庫：{len(self.patterns)} 個攻擊樣式")
return bundle
def export_library(self, output_file="attack_pattern_library.json"):
bundle = Bundle(objects=self.patterns + self.relationships)
with open(output_file, "w") as f:
f.write(bundle.serialize(pretty=True))
print(f"[+] 資料庫已匯出至 {output_file}")
def generate_detection_templates(self, mapped_behaviors):
"""根據攻擊樣式產生 Sigma 規則範本。"""
templates = []
for mapping in mapped_behaviors:
template = {
"title": f"Detection: {mapping['technique_name']} ({mapping['technique_id']})",
"status": "experimental",
"description": f"Detects {mapping['technique_name']} based on CTI report observation",
"references": [
f"https://attack.mitre.org/techniques/{mapping['technique_id'].replace('.', '/')}/",
],
"tags": [
f"attack.{mapping['tactics'][0]}" if mapping['tactics'] else "attack.unknown",
f"attack.{mapping['technique_id'].lower()}",
],
"data_sources": mapping.get("data_sources", []),
"observed_tools": mapping.get("tools_observed", []),
"source_context": mapping["source_sentence"],
}
templates.append(template)
with open("detection_templates.json", "w") as f:
json.dump(templates, f, indent=2)
print(f"[+] 已產生 {len(templates)} 個偵測範本")
return templates
library = AttackPatternLibrary()
bundle = library.build_library(mapped_behaviors, "Sample CTI Report")
library.export_library()
templates = library.generate_detection_templates(mapped_behaviors)
驗證標準
- CTI 報告已解析且行為指示詞已提取
- 行為已對應到 ATT&CK 技術並具有一定信心水準
- STIX 2.1 Attack Pattern 物件已建立且包含正確的參考
- 資料庫可按戰術、技術和威脅行為者進行搜尋
- 偵測範本已根據記錄的模式產生
- 資料庫可作為 STIX bundle 匯出以供分享