threat intelligence

Building Attack Pattern Library from CTI Reports

Extract and catalog attack patterns from cyber threat intelligence reports into a structured STIX-based library mapped to MITRE ATT&CK for detection engineering and threat-informed defense.

attack-patterncti-reportsdetection-engineeringextractionmitre-attacknlpstixthreat-intelligence
Install this skill
npx skills add mukul975/Anthropic-Cybersecurity-Skills
Framework mappings

Overview

Cyber threat intelligence (CTI) reports from vendors like Mandiant, CrowdStrike, Talos, and Microsoft contain detailed descriptions of adversary behaviors that can be extracted, normalized, and cataloged into a structured attack pattern library. This skill covers parsing CTI reports to extract adversary techniques, mapping behaviors to MITRE ATT&CK technique IDs, creating STIX 2.1 Attack Pattern objects, building a searchable library indexed by tactic, technique, and threat actor, and generating detection rule templates from documented patterns.

When to Use

  • When deploying or configuring building attack pattern library from cti reports capabilities in your environment
  • When establishing security controls aligned to compliance requirements
  • When building or improving security architecture for this domain
  • When conducting security assessments that require this implementation

Prerequisites

  • Python 3.9+ with stix2, mitreattack-python, spacy, requests libraries
  • Collection of CTI reports (PDF, HTML, or text format)
  • MITRE ATT&CK STIX data (local or via TAXII)
  • Understanding of ATT&CK technique structure and naming conventions
  • Familiarity with detection engineering concepts (Sigma, YARA)

Key Concepts

Attack Pattern Extraction

CTI reports describe adversary behaviors in natural language. Extraction involves identifying action verbs and technical terms that map to ATT&CK techniques, recognizing tool names and malware families, identifying infrastructure indicators, and mapping sequences of behaviors to attack chains (kill chain phases).

STIX 2.1 Attack Pattern Objects

STIX defines Attack Pattern as a Structured Domain Object (SDO) that describes ways threat actors attempt to compromise targets. Each pattern links to ATT&CK via external references, includes kill chain phases (tactics), and can be related to Intrusion Sets, Malware, and Tool objects.

Detection Rule Generation

Extracted attack patterns inform detection engineering by providing: specific procedure examples for Sigma rule creation, behavioral sequences for correlation rules, IOC patterns for YARA and Snort rules, and data source requirements for telemetry gaps.

Workflow

Step 1: Parse CTI Reports and Extract Behaviors

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)

Step 2: Map Behaviors to ATT&CK Techniques

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)

Step 3: Create STIX 2.1 Attack Pattern Library

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)

Validation Criteria

  • CTI report parsed and behavioral indicators extracted
  • Behaviors mapped to ATT&CK techniques with confidence
  • STIX 2.1 Attack Pattern objects created with proper references
  • Library searchable by tactic, technique, and threat actor
  • Detection templates generated from documented patterns
  • Library exportable as STIX bundle for sharing

References

Source materials

References and resources

Everything below is rendered for inspection. Script files are read-only and never run.

References 1

api-reference.md1.4 KB

API Reference: Attack Pattern Library from CTI Reports

Technique Extraction Patterns

Technique Regex Pattern
T1566.001 spearphish.*attach
T1059.001 powershell, invoke-expression
T1053.005 scheduled task, schtasks
T1547.001 registry run key, CurrentVersion\\Run
T1003.001 lsass, credential dump, mimikatz
T1486 ransomware encrypt
T1048 exfiltration, data theft

IOC Extraction Regex

IOC Type Pattern
IPv4 \b(?:\d{1,3}\.){3}\d{1,3}\b
Domain [a-zA-Z0-9-]+\.(?:com|net|org)
MD5 [a-fA-F0-9]{32}
SHA-256 [a-fA-F0-9]{64}
Defanged URL hxxps?://[^\s]+
Explicit technique T\d{4}(?:\.\d{3})?

STIX Attack Pattern

{
  "type": "attack-pattern",
  "name": "Spearphishing Attachment",
  "external_references": [
    {"source_name": "mitre-attack", "external_id": "T1566.001"}
  ],
  "kill_chain_phases": [
    {"phase_name": "initial-access"}
  ]
}

Library Output Structure

Field Description
technique_frequency Count per technique across reports
technique_report_map Which reports mention each technique
total_unique_techniques Distinct techniques found

MITRE ATT&CK STIX Data

https://raw.githubusercontent.com/mitre/cti/master/enterprise-attack/enterprise-attack.json

Scripts 1

agent.py5.1 KB
Display-only source. This catalog never executes bundled scripts.
#!/usr/bin/env python3
"""Attack Pattern Library Builder Agent - Extracts attack patterns from CTI reports and maps to MITRE ATT&CK."""

import json
import re
import logging
import argparse
from datetime import datetime
from collections import Counter, defaultdict


logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
logger = logging.getLogger(__name__)

TECHNIQUE_PATTERNS = {
    "T1566.001": [r"spearphish(?:ing)?\s+attach", r"malicious\s+(?:email\s+)?attachment"],
    "T1566.002": [r"spearphish(?:ing)?\s+link", r"phishing\s+(?:url|link)"],
    "T1059.001": [r"powershell", r"invoke-(?:expression|command|webrequest)"],
    "T1059.003": [r"cmd\.exe", r"command\s+(?:prompt|shell|line)"],
    "T1053.005": [r"scheduled\s+task", r"schtasks"],
    "T1547.001": [r"registry\s+run\s+key", r"autostart", r"CurrentVersion\\\\Run"],
    "T1003.001": [r"lsass", r"credential\s+dump", r"mimikatz"],
    "T1021.001": [r"remote\s+desktop", r"rdp\s+lateral"],
    "T1021.002": [r"smb\s+share", r"admin\s*\$", r"C\s*\$\s+share"],
    "T1071.001": [r"http\s+c2", r"web\s+(?:beacon|c2)", r"https?\s+callback"],
    "T1486": [r"encrypt(?:ion|ed)\s+(?:file|data)", r"ransomware\s+encrypt"],
    "T1048": [r"exfiltrat(?:e|ion)", r"data\s+(?:theft|steal|upload)"],
    "T1105": [r"download(?:ed)?\s+(?:payload|malware|tool)", r"ingress\s+tool\s+transfer"],
    "T1027": [r"obfuscat(?:e|ion|ed)", r"encoded\s+(?:payload|script)"],
    "T1562.001": [r"disable\s+(?:antivirus|defender|security)", r"tamper\s+protection"],
}


def extract_techniques_from_text(text):
    """Extract MITRE ATT&CK techniques from report text."""
    text_lower = text.lower()
    matched = {}
    for tech_id, patterns in TECHNIQUE_PATTERNS.items():
        for pattern in patterns:
            if re.search(pattern, text_lower):
                matched[tech_id] = {"pattern_matched": pattern, "technique_id": tech_id}
                break
    explicit = re.findall(r"T\d{4}(?:\.\d{3})?", text)
    for tid in explicit:
        if tid not in matched:
            matched[tid] = {"pattern_matched": "explicit_reference", "technique_id": tid}
    return matched


def extract_iocs_from_text(text):
    """Extract IOCs from report text."""
    iocs = {
        "ips": list(set(re.findall(r"\b(?:\d{1,3}\.){3}\d{1,3}\b", text))),
        "domains": list(set(re.findall(r"\b(?:[a-zA-Z0-9-]+\.)+(?:com|net|org|io|xyz|top|info|ru|cn)\b", text))),
        "hashes_md5": list(set(re.findall(r"\b[a-fA-F0-9]{32}\b", text))),
        "hashes_sha256": list(set(re.findall(r"\b[a-fA-F0-9]{64}\b", text))),
        "urls": list(set(re.findall(r"hxxps?://[^\s<>\"]+", text))),
    }
    return iocs


def process_report(report_text, report_name=""):
    """Process a single CTI report to extract attack patterns."""
    techniques = extract_techniques_from_text(report_text)
    iocs = extract_iocs_from_text(report_text)
    return {
        "report_name": report_name,
        "techniques_found": len(techniques),
        "technique_ids": list(techniques.keys()),
        "technique_details": techniques,
        "ioc_counts": {k: len(v) for k, v in iocs.items()},
        "iocs": iocs,
    }


def build_pattern_library(processed_reports):
    """Build a consolidated attack pattern library from multiple reports."""
    technique_frequency = Counter()
    technique_reports = defaultdict(list)
    for report in processed_reports:
        for tid in report["technique_ids"]:
            technique_frequency[tid] += 1
            technique_reports[tid].append(report["report_name"])
    library = {
        "technique_frequency": dict(technique_frequency.most_common()),
        "technique_report_map": {t: r for t, r in technique_reports.items()},
        "total_unique_techniques": len(technique_frequency),
        "total_reports_processed": len(processed_reports),
    }
    return library


def generate_report(processed_reports, library):
    """Generate attack pattern library report."""
    report = {
        "timestamp": datetime.utcnow().isoformat(),
        "library": library,
        "report_details": processed_reports,
    }
    print(f"PATTERN LIBRARY: {library['total_unique_techniques']} techniques from {library['total_reports_processed']} reports")
    return report


def main():
    parser = argparse.ArgumentParser(description="Attack Pattern Library Builder Agent")
    parser.add_argument("--report-files", nargs="+", required=True, help="CTI report text files")
    parser.add_argument("--output", default="pattern_library.json")
    args = parser.parse_args()

    processed = []
    for filepath in args.report_files:
        with open(filepath, "r", encoding="utf-8", errors="ignore") as f:
            text = f.read()
        result = process_report(text, filepath)
        processed.append(result)
        logger.info("Processed %s: %d techniques", filepath, result["techniques_found"])

    library = build_pattern_library(processed)
    report = generate_report(processed, library)
    with open(args.output, "w") as f:
        json.dump(report, f, indent=2)
    logger.info("Report saved to %s", args.output)


if __name__ == "__main__":
    main()
Keep exploring