soc operations

Implementing Threat Modeling with MITRE ATT&CK

Implements threat modeling using the MITRE ATT&CK framework to map adversary TTPs against organizational assets, assess detection coverage gaps, and prioritize defensive investments. Use when SOC teams need to align detection engineering with threat landscape, conduct threat assessments for new environments, or justify security tool procurement.

attack-navigatordetection-coveragemitre-attackrisk-assessmentsocthreat-modelingttp
Install this skill
npx skills add mukul975/Anthropic-Cybersecurity-Skills
Framework mappings

When to Use

Use this skill when:

  • SOC teams need to assess detection coverage against relevant threat actors and their TTPs
  • Security leadership requires threat-informed defense prioritization
  • New environments (cloud migration, OT integration) need detection strategy planning
  • Purple team exercises require structured adversary emulation based on threat models
  • Annual risk assessments need ATT&CK-based threat landscape analysis

Do not use as a one-time exercise — threat models must be continuously updated as adversary TTPs evolve and organizational attack surface changes.

Prerequisites

  • MITRE ATT&CK framework knowledge (Enterprise, ICS, Mobile, or Cloud matrices)
  • ATT&CK Navigator tool (web or local) for layer visualization
  • Current detection rule inventory mapped to ATT&CK technique IDs
  • Threat intelligence on adversary groups targeting your sector
  • Organizational asset inventory with criticality classifications

Workflow

Step 1: Identify Relevant Threat Actors

Research adversary groups targeting your sector using MITRE ATT&CK Groups:

import requests
import json
 
# Download ATT&CK STIX data
response = requests.get(
    "https://raw.githubusercontent.com/mitre/cti/master/enterprise-attack/enterprise-attack.json"
)
attack_data = response.json()
 
# Extract groups and their techniques
groups = {}
for obj in attack_data["objects"]:
    if obj["type"] == "intrusion-set":
        group_name = obj["name"]
        aliases = obj.get("aliases", [])
        description = obj.get("description", "")
        groups[group_name] = {
            "aliases": aliases,
            "description": description[:200],
            "techniques": []
        }
 
# Map techniques to groups via relationships
relationships = [obj for obj in attack_data["objects"] if obj["type"] == "relationship"]
techniques = {obj["id"]: obj for obj in attack_data["objects"]
              if obj["type"] == "attack-pattern"}
 
for rel in relationships:
    if rel["relationship_type"] == "uses":
        source = rel["source_ref"]
        target = rel["target_ref"]
        for group_name, group_data in groups.items():
            if source == group_data.get("id") and target in techniques:
                tech = techniques[target]
                ext_refs = tech.get("external_references", [])
                for ref in ext_refs:
                    if ref.get("source_name") == "mitre-attack":
                        group_data["techniques"].append(ref["external_id"])
 
# Example: Financial sector threat actors
financial_actors = ["FIN7", "FIN8", "Carbanak", "APT38", "Lazarus Group"]
for actor in financial_actors:
    if actor in groups:
        print(f"{actor}: {len(groups[actor]['techniques'])} techniques")
        print(f"  Top techniques: {groups[actor]['techniques'][:10]}")

Step 2: Build Threat Actor TTP Profile

Create ATT&CK Navigator layers for priority threat actors:

import json
 
def create_attack_layer(actor_name, techniques, color="#ff6666"):
    """Generate ATT&CK Navigator JSON layer for a threat actor"""
    layer = {
        "name": f"{actor_name} TTP Profile",
        "versions": {
            "attack": "15",
            "navigator": "5.0",
            "layer": "4.5"
        },
        "domain": "enterprise-attack",
        "description": f"Techniques associated with {actor_name}",
        "techniques": [
            {
                "techniqueID": tech_id,
                "tactic": "",
                "color": color,
                "comment": f"Used by {actor_name}",
                "enabled": True,
                "score": 1
            }
            for tech_id in techniques
        ],
        "gradient": {
            "colors": ["#ffffff", color],
            "minValue": 0,
            "maxValue": 1
        }
    }
    return layer
 
# Create layers for top threat actors
fin7_techniques = ["T1566.001", "T1059.001", "T1053.005", "T1547.001",
                    "T1078", "T1021.001", "T1003", "T1071.001", "T1041"]
layer = create_attack_layer("FIN7", fin7_techniques, "#ff6666")
 
with open("fin7_layer.json", "w") as f:
    json.dump(layer, f, indent=2)

Step 3: Map Current Detection Coverage

Export current detection rules mapped to ATT&CK:

--- Extract ATT&CK technique mappings from Splunk ES correlation searches
| rest /services/saved/searches
  splunk_server=local
| where match(title, "^(COR|ESCU|RBA):")
| eval techniques = if(isnotnull(action.correlationsearch.annotations),
                       spath(action.correlationsearch.annotations, "mitre_attack"),
                       "unmapped")
| stats count by techniques
| mvexpand techniques
| stats count by techniques
| rename techniques AS technique_id, count AS rule_count

Create detection coverage layer:

def create_coverage_layer(detection_rules):
    """Generate coverage layer from detection rule inventory"""
    technique_counts = {}
    for rule in detection_rules:
        for tech in rule.get("techniques", []):
            technique_counts[tech] = technique_counts.get(tech, 0) + 1
 
    layer = {
        "name": "SOC Detection Coverage",
        "versions": {"attack": "15", "navigator": "5.0", "layer": "4.5"},
        "domain": "enterprise-attack",
        "techniques": [
            {
                "techniqueID": tech_id,
                "color": "#31a354" if count >= 2 else "#a1d99b" if count == 1 else "",
                "score": count,
                "comment": f"{count} detection rule(s)"
            }
            for tech_id, count in technique_counts.items()
        ],
        "gradient": {
            "colors": ["#ffffff", "#a1d99b", "#31a354"],
            "minValue": 0,
            "maxValue": 3
        }
    }
    return layer

Step 4: Perform Gap Analysis

Overlay threat actor TTPs against detection coverage:

def gap_analysis(threat_techniques, covered_techniques):
    """Identify detection gaps for specific threat actor"""
    gaps = set(threat_techniques) - set(covered_techniques)
    covered = set(threat_techniques) & set(covered_techniques)
 
    print(f"Threat Actor Techniques: {len(threat_techniques)}")
    print(f"Detected: {len(covered)} ({len(covered)/len(threat_techniques)*100:.0f}%)")
    print(f"Gaps: {len(gaps)} ({len(gaps)/len(threat_techniques)*100:.0f}%)")
 
    # Prioritize gaps by kill chain phase
    priority_order = {
        "TA0001": 1, "TA0002": 2, "TA0003": 3, "TA0004": 4,
        "TA0005": 5, "TA0006": 6, "TA0007": 7, "TA0008": 8,
        "TA0009": 9, "TA0010": 10, "TA0011": 11, "TA0040": 12
    }
 
    gap_details = []
    for tech_id in gaps:
        gap_details.append({
            "technique": tech_id,
            "priority": "HIGH" if tech_id.split(".")[0] in ["T1003", "T1021", "T1059"] else "MEDIUM",
            "recommendation": f"Build detection for {tech_id}"
        })
 
    return {
        "total_actor_techniques": len(threat_techniques),
        "covered": len(covered),
        "gaps": len(gaps),
        "coverage_pct": round(len(covered)/len(threat_techniques)*100, 1),
        "gap_details": sorted(gap_details, key=lambda x: x["priority"])
    }
 
# Run analysis
result = gap_analysis(fin7_techniques, current_coverage)

Step 5: Create Prioritized Remediation Plan

Build a detection engineering roadmap:

threat_model_remediation_plan:
  assessed_date: 2024-03-15
  primary_threats:
    - FIN7 (Financial sector)
    - APT38 (DPRK financial)
    - Lazarus Group (Destructive)
 
  current_coverage: 64%
  target_coverage: 80%
 
  priority_1_gaps: # 30-day target
    - technique: T1021.002
      name: SMB/Windows Admin Shares
      data_source: Windows Security Event 5140
      effort: Low
      detection_approach: Monitor admin share access from non-admin workstations
 
    - technique: T1003.006
      name: DCSync
      data_source: Windows Security Event 4662
      effort: Medium
      detection_approach: Detect DS-Replication-Get-Changes from non-DC sources
 
  priority_2_gaps: # 60-day target
    - technique: T1055
      name: Process Injection
      data_source: Sysmon EventCode 8, 10
      effort: High
      detection_approach: Monitor cross-process memory access patterns
 
    - technique: T1071.001
      name: Web Protocols (C2)
      data_source: Proxy/Firewall logs
      effort: Medium
      detection_approach: Detect beaconing patterns in HTTP/S traffic
 
  priority_3_gaps: # 90-day target
    - technique: T1070.004
      name: File Deletion
      data_source: Sysmon EventCode 23
      effort: Low
      detection_approach: Monitor mass file deletion in sensitive directories

Step 6: Validate with Adversary Emulation

Test coverage using MITRE Caldera or Atomic Red Team:

# Using Atomic Red Team to validate coverage for FIN7 techniques
# T1566.001 — Spearphishing Attachment
Invoke-AtomicTest T1566.001
 
# T1059.001 — PowerShell
Invoke-AtomicTest T1059.001 -TestNumbers 1,2,3
 
# T1053.005 — Scheduled Task
Invoke-AtomicTest T1053.005
 
# T1547.001 — Registry Run Keys
Invoke-AtomicTest T1547.001
 
# T1003 — Credential Dumping
Invoke-AtomicTest T1003 -TestNumbers 1,2
 
# Verify detections
# Check SIEM for corresponding alerts within 15 minutes

Document emulation results to validate threat model accuracy.

Key Concepts

Term Definition
MITRE ATT&CK Knowledge base of adversary tactics, techniques, and procedures based on real-world observations
TTP Tactics, Techniques, and Procedures — the behavioral patterns of adversary groups
ATT&CK Navigator Web tool for visualizing ATT&CK matrices as layered heatmaps showing coverage or threat profiles
Gap Analysis Process of comparing threat actor TTPs against detection coverage to identify blind spots
Threat-Informed Defense Security strategy prioritizing defenses based on actual adversary behaviors rather than theoretical risks
Adversary Emulation Controlled simulation of threat actor TTPs to validate detection and response capabilities

Tools & Systems

  • MITRE ATT&CK Navigator: Web-based visualization tool for creating and overlaying ATT&CK technique layers
  • MITRE Caldera: Automated adversary emulation platform for testing detection coverage at scale
  • Atomic Red Team: Open-source library of ATT&CK technique tests for security control validation
  • CTID ATT&CK Workbench: MITRE tool for customizing ATT&CK knowledge base with organizational context
  • Tidal Cyber: Commercial platform for threat-informed defense planning using ATT&CK framework

Common Scenarios

  • Annual Threat Assessment: Map top 5 threat actors to ATT&CK, overlay against detection, produce gap analysis
  • Cloud Migration Planning: Model cloud-specific threats (T1078.004, T1537) and plan detection coverage
  • M&A Security Assessment: Threat model the acquired company's environment against relevant threat actors
  • Budget Justification: Use gap analysis to demonstrate detection blind spots requiring tool investment
  • Purple Team Planning: Select adversary emulation scenarios based on highest-priority gaps from threat model

Output Format

THREAT MODEL ASSESSMENT — Financial Services Division
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
Date:             2024-03-15
Threat Actors:    FIN7, APT38, Lazarus Group
Techniques Total: 87 unique techniques across all actors
 
DETECTION COVERAGE:
  Covered:     56/87 (64%)
  Gaps:        31/87 (36%)
 
  Tactic Coverage Breakdown:
    Initial Access:      78%  ████████░░
    Execution:           82%  █████████░
    Persistence:         71%  ████████░░
    Priv Escalation:     65%  ███████░░░
    Defense Evasion:     52%  ██████░░░░  <-- Priority gap
    Credential Access:   58%  ██████░░░░  <-- Priority gap
    Discovery:           45%  █████░░░░░
    Lateral Movement:    61%  ███████░░░
    Collection:          50%  ██████░░░░
    Exfiltration:        55%  ██████░░░░
    C2:                  67%  ███████░░░
 
TOP PRIORITY GAPS (30-day remediation):
  1. T1055 Process Injection — used by all 3 actors, 0 detections
  2. T1003.006 DCSync — used by FIN7 and Lazarus, 0 detections
  3. T1070.004 File Deletion — evidence destruction, 0 detections
 
INVESTMENT RECOMMENDATION:
  Closing top 10 gaps requires: 2 detection engineer FTEs, 60 days
  Expected coverage improvement: 64% -> 76%
Source materials

References and resources

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

References 1

api-reference.md2.2 KB

API Reference: Implementing Threat Modeling with MITRE ATT&CK

Libraries

attackcti (MITRE ATT&CK CTI)

  • Install: pip install attackcti
  • Docs: https://attackcti.readthedocs.io/
  • attack_client() -- Initialize ATT&CK client
  • get_groups() -- All threat actor groups
  • get_techniques() -- All techniques (Enterprise, Mobile, ICS)
  • get_techniques_used_by_group(group) -- Techniques per group
  • get_mitigations() -- Defensive mitigations
  • get_software() -- Malware and tools catalog

mitreattack-python

  • Install: pip install mitreattack-python
  • Docs: https://mitreattack-python.readthedocs.io/
  • MitreAttackData(stix_filepath) -- Load STIX bundle
  • get_groups_using_technique(technique_stix_id) -- Groups per technique
  • get_datacomponents_detecting_technique() -- Detection data sources

ATT&CK Navigator Layer Format

Field Description
name Layer display name
domain enterprise-attack, mobile-attack, ics-attack
techniques[] List of technique annotations
techniques[].techniqueID ATT&CK ID (e.g., T1059)
techniques[].score Numeric score for heat map
techniques[].color Hex color override
gradient Color scale definition

Threat Modeling Workflow

  1. Identify industry-relevant threat actors
  2. Map actor TTPs to ATT&CK techniques
  3. Assess current detection coverage
  4. Identify coverage gaps
  5. Prioritize defensive investments
  6. Export Navigator layer for visualization

Industry Threat Actor Mapping

  • Financial: APT38, FIN7, Carbanak, Lazarus
  • Healthcare: APT41, FIN12, Wizard Spider
  • Government: APT28, APT29, Turla, Sandworm
  • Technology: APT41, APT10, Hafnium
  • Energy: Sandworm, Dragonfly, APT33

Priority Scoring

  • CRITICAL: Technique used by 3+ relevant threat actors
  • HIGH: Technique used by 2 relevant threat actors
  • MEDIUM: Technique used by 1 relevant threat actor

External References

Scripts 1

agent.py8.1 KB
Display-only source. This catalog never executes bundled scripts.
#!/usr/bin/env python3
"""Threat modeling agent using MITRE ATT&CK framework with attackcti."""

import json
import sys
import argparse
from datetime import datetime
from collections import Counter

try:
    from attackcti import attack_client
except ImportError:
    print("Install attackcti: pip install attackcti")
    sys.exit(1)


INDUSTRY_THREAT_ACTORS = {
    "financial": ["APT38", "FIN7", "Carbanak", "Lazarus Group", "FIN8"],
    "healthcare": ["APT41", "FIN12", "Wizard Spider"],
    "government": ["APT28", "APT29", "Turla", "Sandworm Team", "Mustang Panda"],
    "technology": ["APT41", "APT10", "Hafnium", "Nobelium"],
    "energy": ["Sandworm Team", "Dragonfly", "Berserk Bear", "APT33"],
    "defense": ["APT28", "APT29", "Turla", "Lazarus Group", "Kimsuky"],
    "retail": ["FIN6", "FIN7", "FIN8", "Magecart"],
}


def get_group_techniques(group_name):
    """Get all ATT&CK techniques used by a specific threat group."""
    client = attack_client()
    groups = client.get_groups()
    target = None
    for g in groups:
        aliases = [a.lower() for a in g.get("aliases", [])]
        if group_name.lower() in g["name"].lower() or group_name.lower() in aliases:
            target = g
            break
    if not target:
        return None
    techniques = client.get_techniques_used_by_group(target)
    return [{"id": t["external_references"][0]["external_id"],
             "name": t["name"],
             "tactics": [p["phase_name"] for p in t.get("kill_chain_phases", [])]}
            for t in techniques]


def build_threat_profile(industry):
    """Build a threat profile for an industry based on relevant threat actors."""
    actors = INDUSTRY_THREAT_ACTORS.get(industry.lower(), [])
    if not actors:
        print(f"[!] Industry '{industry}' not found. Available: {list(INDUSTRY_THREAT_ACTORS.keys())}")
        return None

    profile = {"industry": industry, "threat_actors": [], "all_techniques": [],
               "tactic_coverage": Counter()}

    for actor_name in actors:
        techniques = get_group_techniques(actor_name)
        if techniques:
            profile["threat_actors"].append({
                "name": actor_name,
                "technique_count": len(techniques),
                "techniques": techniques,
            })
            for t in techniques:
                profile["all_techniques"].append(t["id"])
                for tac in t["tactics"]:
                    profile["tactic_coverage"][tac] += 1

    profile["unique_techniques"] = list(set(profile["all_techniques"]))
    profile["tactic_coverage"] = dict(profile["tactic_coverage"])
    return profile


def assess_detection_coverage(profile, existing_detections=None):
    """Assess detection coverage gaps against threat profile."""
    if existing_detections is None:
        existing_detections = []
    unique_techniques = set(profile.get("unique_techniques", []))
    covered = set(existing_detections)
    gaps = unique_techniques - covered
    coverage_pct = round(len(covered.intersection(unique_techniques)) /
                         max(len(unique_techniques), 1) * 100, 1)
    return {
        "total_threat_techniques": len(unique_techniques),
        "detected": len(covered.intersection(unique_techniques)),
        "gaps": sorted(gaps),
        "coverage_pct": coverage_pct,
        "priority_gaps": sorted(gaps)[:10],
    }


def generate_navigator_layer(profile, layer_name="Threat Model"):
    """Generate ATT&CK Navigator layer JSON for visualization."""
    technique_counts = Counter(profile.get("all_techniques", []))
    techniques = []
    for tech_id, count in technique_counts.items():
        color_map = {1: "#fcf3cf", 2: "#f9e79f", 3: "#f4d03f"}
        techniques.append({
            "techniqueID": tech_id,
            "score": count,
            "color": color_map.get(min(count, 3), "#f4d03f"),
            "comment": f"Used by {count} threat actor(s)",
            "enabled": True,
        })
    layer = {
        "name": layer_name,
        "versions": {"attack": "14", "navigator": "4.9.1", "layer": "4.5"},
        "domain": "enterprise-attack",
        "description": f"Threat model for {profile.get('industry', 'unknown')} industry",
        "techniques": techniques,
        "gradient": {"colors": ["#ffffff", "#f4d03f", "#e74c3c"], "minValue": 0, "maxValue": 3},
        "legendItems": [
            {"label": "1 actor", "color": "#fcf3cf"},
            {"label": "2 actors", "color": "#f9e79f"},
            {"label": "3+ actors", "color": "#f4d03f"},
        ],
    }
    return layer


def prioritize_defenses(profile):
    """Prioritize defensive investments based on threat model."""
    technique_counts = Counter(profile.get("all_techniques", []))
    top_techniques = technique_counts.most_common(15)

    client = attack_client()
    all_techniques = {t["external_references"][0]["external_id"]: t
                      for t in client.get_techniques()
                      if t.get("external_references")}

    priorities = []
    for tech_id, count in top_techniques:
        tech_data = all_techniques.get(tech_id, {})
        priorities.append({
            "technique": tech_id,
            "name": tech_data.get("name", "Unknown"),
            "actor_count": count,
            "tactics": [p["phase_name"] for p in tech_data.get("kill_chain_phases", [])],
            "priority": "CRITICAL" if count >= 3 else "HIGH" if count >= 2 else "MEDIUM",
        })
    return priorities


def run_threat_model(industry, existing_detections=None):
    """Run full threat modeling exercise for an industry."""
    print(f"\n{'='*60}")
    print(f"  MITRE ATT&CK THREAT MODEL")
    print(f"  Industry: {industry}")
    print(f"  Generated: {datetime.utcnow().strftime('%Y-%m-%d %H:%M:%S')} UTC")
    print(f"{'='*60}\n")

    profile = build_threat_profile(industry)
    if not profile:
        return None

    print(f"--- THREAT ACTORS ({len(profile['threat_actors'])}) ---")
    for actor in profile["threat_actors"]:
        print(f"  {actor['name']}: {actor['technique_count']} techniques")

    print(f"\n--- TECHNIQUE SUMMARY ---")
    print(f"  Total technique usage: {len(profile['all_techniques'])}")
    print(f"  Unique techniques:     {len(profile['unique_techniques'])}")

    print(f"\n--- TACTIC DISTRIBUTION ---")
    for tac, count in sorted(profile["tactic_coverage"].items(), key=lambda x: -x[1]):
        bar = "#" * min(count, 30)
        print(f"  {tac:<30} {bar} ({count})")

    coverage = assess_detection_coverage(profile, existing_detections or [])
    print(f"\n--- DETECTION COVERAGE ---")
    print(f"  Coverage: {coverage['coverage_pct']}%")
    print(f"  Gaps: {len(coverage['gaps'])} techniques undetected")
    if coverage["priority_gaps"]:
        print(f"  Priority gaps: {', '.join(coverage['priority_gaps'][:5])}")

    priorities = prioritize_defenses(profile)
    print(f"\n--- DEFENSE PRIORITIES ---")
    for p in priorities[:10]:
        print(f"  [{p['priority']}] {p['technique']} {p['name']} (used by {p['actor_count']} actors)")

    print(f"\n{'='*60}\n")
    return {"profile": profile, "coverage": coverage, "priorities": priorities}


def main():
    parser = argparse.ArgumentParser(description="Threat Modeling with MITRE ATT&CK Agent")
    parser.add_argument("--industry", required=True,
                        choices=list(INDUSTRY_THREAT_ACTORS.keys()),
                        help="Industry for threat profile")
    parser.add_argument("--detections", nargs="*", help="List of detected technique IDs")
    parser.add_argument("--navigator", help="Export ATT&CK Navigator layer to JSON file")
    parser.add_argument("--output", help="Save full report to JSON")
    args = parser.parse_args()

    result = run_threat_model(args.industry, args.detections)
    if result and args.navigator:
        layer = generate_navigator_layer(result["profile"], f"{args.industry} Threat Model")
        with open(args.navigator, "w") as f:
            json.dump(layer, f, indent=2)
        print(f"[+] Navigator layer saved to {args.navigator}")
    if result and args.output:
        with open(args.output, "w") as f:
            json.dump(result, f, indent=2, default=str)
        print(f"[+] Report saved to {args.output}")


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