building-attack-pattern-library-from-cti-reports

CTI レポートからの攻撃パターンライブラリの構築

概要

Mandiant、CrowdStrike、Talos、Microsoft などのベンダーが公開するサイバー脅威インテリジェンス（CTI）レポートには、抽出、正規化、カタログ化して構造化された攻撃パターンライブラリにまとめられる、攻撃者の行動に関する詳細な記述が含まれています。このスキルでは、CTI レポートを解析して攻撃者のテクニックを抽出し、行動を MITRE ATT&CK のテクニック ID に対応付け、STIX 2.1 の攻撃パターンオブジェクトを作成し、戦術、テクニック、脅威アクターでインデックス化した検索可能なライブラリを構築し、文書化されたパターンから検知ルールのテンプレートを生成する方法を説明します。

使用する状況

• 環境で CTI レポートからの攻撃パターンライブラリ構築機能をデプロイまたは設定する場合
• コンプライアンス要件に沿ったセキュリティ制御を確立する場合
• この領域のセキュリティアーキテクチャを構築または改善する場合
• この実装を必要とするセキュリティ評価を実施する場合

前提条件

• stix2、mitreattack-python、spacy、requests ライブラリを含む Python 3.9 以降
• CTI レポートのコレクション（PDF、HTML、またはテキスト形式）
• MITRE ATT&CK STIX データ（ローカル、または TAXII 経由）
• ATT&CK のテクニック構造と命名規則に関する理解
• 検知エンジニアリングの概念（Sigma、YARA）に関する知識

主な概念

攻撃パターンの抽出

CTI レポートには、攻撃者の行動が自然言語で記述されています。抽出では、ATT&CK のテクニックに対応する動詞や技術用語の特定、ツール名やマルウェアファミリーの認識、インフラストラクチャの指標の特定、行動の一連の流れと攻撃チェーン（キルチェーンのフェーズ）との対応付けを行います。

STIX 2.1 攻撃パターンオブジェクト

STIX では、攻撃パターンは、脅威アクターが標的の侵害を試みる方法を記述する構造化ドメインオブジェクト（SDO）として定義されています。各パターンは外部参照を通じて ATT&CK にリンクされ、キルチェーンのフェーズ（戦術）が含まれ、Intrusion Set、Malware、Tool オブジェクトと関連付けることができます。

検知ルールの生成

抽出した攻撃パターンは、検知エンジニアリングに次の情報を提供します。Sigma ルール作成に使える具体的な手順の例、相関ルールに使える行動シーケンス、YARA と Snort のルールに使える IOC パターン、テレメトリの不足を補うために必要なデータソースです。

ワークフロー

ステップ 1：CTI レポートを解析して行動を抽出する

import re
import json
from collections import defaultdict

class CTIReportParser:
    """Parse CTI reports to extract adversary behaviors."""

    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):
        """Parse a CTI report and extract behaviors."""
        sentences = re.split(r'[.!?]\s+', text)
        behaviors = []

        for sentence in sentences:
            sentence_lower = sentence.lower()
            # Check for behavior indicators
            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"[+] Extracted {len(behaviors)} behavioral indicators from report")
        return behaviors

    def _extract_tools(self, text):
        """Extract tool/malware names from 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):
        """Match text to ATT&CK technique hints."""
        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):
        """Load all ATT&CK techniques for mapping."""
        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"[+] Loaded {len(self.techniques)} ATT&CK techniques")

    def map_behaviors(self, behaviors):
        """Map extracted behaviors to ATT&CK techniques."""
        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"[+] Mapped {len(mapped)} behaviors to ATT&CK techniques")
        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"):
        """Create STIX Attack Pattern from mapped behavior."""
        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"):
        """Build complete attack pattern library from mappings."""
        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"[+] Library: {len(self.patterns)} attack 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"[+] Library exported to {output_file}")

    def generate_detection_templates(self, mapped_behaviors):
        """Generate Sigma rule templates from attack patterns."""
        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"[+] Generated {len(templates)} detection 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 攻撃パターンオブジェクトを作成している
• 戦術、テクニック、脅威アクターでライブラリを検索できる
• 文書化されたパターンから検知テンプレートを生成している
• 共有用にライブラリを STIX バンドルとしてエクスポートできる

参考資料

• MITRE ATT&CK
• STIX 2.1 Attack Pattern SDO
• CISA：ATT&CK マッピングのベストプラクティス
• attackcti Python Library
• Sigma Rules Project
• MITRE ATT&CK STIX Data