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

从 CTI 报告构建攻击模式库

概述

来自 Mandiant、CrowdStrike、Talos 和 Microsoft 等厂商的网络威胁情报（CTI）报告包含对对手行为的详细描述，这些描述可以被提取、规范化并编目到结构化的攻击模式库中。本技能涵盖解析 CTI 报告以提取对手技术、将行为映射到 MITRE ATT&CK 技术 ID、创建 STIX 2.1 攻击模式对象、构建可按战术、技术和威胁行为者索引的可搜索库，以及从已记录的模式生成检测规则模板。

使用场景

• 在环境中部署或配置从 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 攻击模式对象

STIX 将攻击模式定义为结构化域对象（SDO），用于描述威胁行为者试图入侵目标的方式。每个模式通过外部引用链接到 ATT&CK，包含杀伤链阶段（战术），并可与入侵集合、恶意软件和工具对象关联。

检测规则生成

提取的攻击模式通过以下方式为检测工程提供信息：用于创建 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 bundle 以便共享

参考资料

• MITRE ATT&CK
• STIX 2.1 Attack Pattern SDO
• CISA: Best Practices for ATT&CK Mapping
• attackcti Python Library
• Sigma Rules Project
• MITRE ATT&CK STIX Data