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にリンクし、キルチェーンフェーズ(戦術)を含み、Intrusion Set、Malware、Toolオブジェクトに関連付けることができます。
検出ルール生成
抽出された攻撃パターンは、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バンドルとしてエクスポート可能であること