Security Skills Atlas
threat intelligence

Implementing Diamond Model Analysis

The Diamond Model of Intrusion Analysis provides a structured framework for analyzing cyber intrusions by examining four core features - Adversary, Capability, Infrastructure, and Victim. This skill covers implementing the Diamond Model programmatically to classify and correlate intrusion events, build activity threads, and generate pivot-ready intelligence.

ctidiamond-modelintrusion-analysisiocmitre-attackstixthreat-intelligence
Install this skill
npx skills add mukul975/Anthropic-Cybersecurity-Skills
Framework mappings

MITRE ATT&CK

T0816 on the official MITRE ATT&CK siteT1589 on the official MITRE ATT&CK siteT1591 on the official MITRE ATT&CK siteT1592 on the official MITRE ATT&CK siteT1593 on the official MITRE ATT&CK site

NIST CSF 2.0

DE.AE-02 on the official NIST CSF 2.0 siteDE.CM-01 on the official NIST CSF 2.0 siteID.RA-01 on the official NIST CSF 2.0 siteID.RA-05 on the official NIST CSF 2.0 site

Overview

The Diamond Model of Intrusion Analysis provides a structured framework for analyzing cyber intrusions by examining four core features: Adversary, Capability, Infrastructure, and Victim. This skill covers implementing the Diamond Model programmatically to classify and correlate intrusion events, build activity threads linking related events, create activity-attack graphs, and generate pivot-ready intelligence from intrusion data.

When to Use

  • When deploying or configuring implementing diamond model analysis 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 networkx, stix2, graphviz libraries
  • Understanding of the Diamond Model core and meta-features
  • Access to threat intelligence data (MISP/OpenCTI events)
  • Familiarity with MITRE ATT&CK for capability mapping

Key Concepts

Diamond Model Core Features

  • Adversary: The threat actor or operator conducting the intrusion
  • Capability: The tools, techniques, and malware used (maps to ATT&CK)
  • Infrastructure: C2 servers, domains, email addresses, hosting providers
  • Victim: Target organization, system, person, or data asset

Meta-Features

  • Timestamp: When the event occurred
  • Phase: Kill chain stage (recon, delivery, exploitation, etc.)
  • Result: Success, failure, or unknown
  • Direction: Adversary-to-infrastructure, infrastructure-to-victim, etc.
  • Methodology: Social engineering, technical exploit, insider threat
  • Resources: Financial, human, technical resources required

Activity Threads and Groups

  • Activity Thread: Sequence of Diamond events from a single adversary operation
  • Activity Group: Cluster of threads attributed to the same adversary

Workflow

Step 1: Define Diamond Event Data Structure

from dataclasses import dataclass, field
from datetime import datetime
from typing import Optional
import json
import uuid
 
@dataclass
class DiamondEvent:
    adversary: str = ""
    capability: str = ""
    infrastructure: str = ""
    victim: str = ""
    timestamp: str = ""
    phase: str = ""
    result: str = ""
    direction: str = ""
    methodology: str = ""
    confidence: int = 0
    notes: str = ""
    event_id: str = field(default_factory=lambda: str(uuid.uuid4())[:8])
    mitre_techniques: list = field(default_factory=list)
    iocs: list = field(default_factory=list)
 
    def to_dict(self):
        return {
            "event_id": self.event_id,
            "adversary": self.adversary,
            "capability": self.capability,
            "infrastructure": self.infrastructure,
            "victim": self.victim,
            "timestamp": self.timestamp,
            "phase": self.phase,
            "result": self.result,
            "direction": self.direction,
            "methodology": self.methodology,
            "confidence": self.confidence,
            "mitre_techniques": self.mitre_techniques,
            "iocs": self.iocs,
            "notes": self.notes,
        }

Step 2: Build Activity Thread from Events

import networkx as nx
 
class DiamondAnalysis:
    def __init__(self):
        self.events = []
        self.graph = nx.DiGraph()
 
    def add_event(self, event: DiamondEvent):
        self.events.append(event)
        self.graph.add_node(event.event_id, **event.to_dict())
 
    def build_activity_thread(self):
        """Link events chronologically into activity threads."""
        sorted_events = sorted(self.events, key=lambda e: e.timestamp)
        for i in range(len(sorted_events) - 1):
            self.graph.add_edge(
                sorted_events[i].event_id,
                sorted_events[i + 1].event_id,
                relationship="followed_by",
            )
 
    def find_pivots(self):
        """Find pivot points where events share infrastructure or capabilities."""
        pivots = {"infrastructure": {}, "capability": {}, "adversary": {}}
 
        for event in self.events:
            if event.infrastructure:
                pivots["infrastructure"].setdefault(event.infrastructure, []).append(event.event_id)
            if event.capability:
                pivots["capability"].setdefault(event.capability, []).append(event.event_id)
            if event.adversary:
                pivots["adversary"].setdefault(event.adversary, []).append(event.event_id)
 
        return {
            k: {pk: pv for pk, pv in v.items() if len(pv) > 1}
            for k, v in pivots.items()
        }
 
    def generate_report(self):
        return {
            "total_events": len(self.events),
            "unique_adversaries": len(set(e.adversary for e in self.events if e.adversary)),
            "unique_victims": len(set(e.victim for e in self.events if e.victim)),
            "unique_infrastructure": len(set(e.infrastructure for e in self.events if e.infrastructure)),
            "pivots": self.find_pivots(),
            "events": [e.to_dict() for e in self.events],
        }

Validation Criteria

  • Diamond events capture all four core features with meta-features
  • Activity threads link related events chronologically
  • Pivot analysis identifies shared infrastructure and capabilities across events
  • Graph visualization renders the activity-attack graph correctly
  • Events map to MITRE ATT&CK techniques for capability classification

References

Source materials

References and resources

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

References 3

api-reference.md1.7 KB

API Reference: Diamond Model Intrusion Analysis Agent

Dependencies

Library Version Purpose
(stdlib only) Python 3.8+ Dataclass-based Diamond Model event modeling

CLI Usage

python scripts/agent.py --data /intel/events.json --output-dir /reports/

Functions

DiamondEvent (dataclass)

Four vertices: adversary, capability, infrastructure, victim. Plus: phase, result, confidence, notes.

create_event(adversary, capability, infrastructure, victim, **kwargs) -> DiamondEvent

Factory for creating Diamond Model events with auto-generated ID and timestamp.

load_events(data_path) -> list

Loads events from JSON file with {"events": [...]} structure.

pivot_on_vertex(events, vertex, value) -> list

Analytic pivot: returns all events sharing a specific vertex value.

build_activity_thread(events, adversary) -> dict

Groups events by adversary chronologically. Lists capabilities, infrastructure, victims.

cluster_by_infrastructure(events) -> dict

Groups event IDs by shared infrastructure for campaign identification.

compute_vertex_statistics(events) -> dict

Counts unique values per vertex and confidence distribution.

Input Format

{
  "events": [{
    "adversary": "APT29",
    "capability": "Cobalt Strike",
    "infrastructure": "185.220.101.42",
    "victim": "finance-server-01",
    "phase": "Lateral Movement",
    "confidence": "high"
  }]
}

Output Schema

{
  "statistics": {"total_events": 15, "unique_adversaries": 2},
  "activity_threads": [{"adversary": "APT29", "event_count": 8}],
  "infrastructure_clusters": {"185.220.101.42": ["evt1", "evt5"]}
}
standards.md1.1 KB

Standards and Frameworks Reference

Applicable Standards

  • STIX 2.1: Structured Threat Information eXpression for CTI data representation
  • TAXII 2.1: Transport protocol for sharing CTI over HTTPS
  • MITRE ATT&CK: Adversary tactics, techniques, and procedures taxonomy
  • Diamond Model: Intrusion analysis framework (Adversary, Capability, Infrastructure, Victim)
  • Traffic Light Protocol (TLP): Information sharing classification (CLEAR, GREEN, AMBER, RED)

MITRE ATT&CK Relevance

  • Technique mapping for threat actor behavior classification
  • Data sources for detection capability assessment
  • Mitigation strategies linked to specific techniques

Industry Frameworks

  • NIST Cybersecurity Framework (CSF) 2.0 - Identify function
  • ISO 27001:2022 - A.5.7 Threat Intelligence
  • FIRST Standards - TLP, CSIRT, vulnerability coordination

References

workflows.md1.4 KB

Diamond Model Analysis Workflows

Workflow 1: Collection and Analysis

[Intelligence Sources] --> [Data Collection] --> [Analysis] --> [Reporting]
        |                        |                   |               |
        v                        v                   v               v
  OSINT/HUMINT/SIGINT    Normalize/Enrich    Assess/Correlate  Disseminate

Steps:

  1. Planning: Define intelligence requirements and collection priorities
  2. Collection: Gather data from relevant sources
  3. Processing: Normalize data formats and filter noise
  4. Analysis: Apply analytical frameworks and correlate findings
  5. Production: Generate intelligence products and reports
  6. Dissemination: Share with stakeholders via appropriate channels
  7. Feedback: Collect consumer feedback to refine future collection

Workflow 2: Continuous Monitoring

[Watchlist] --> [Automated Monitoring] --> [Change Detection] --> [Alert/Update]

Steps:

  1. Define Watchlist: Identify indicators, actors, and topics to monitor
  2. Configure Monitoring: Set up automated collection from relevant sources
  3. Change Detection: Identify new or changed intelligence
  4. Assessment: Evaluate significance of changes
  5. Alerting: Notify stakeholders of significant intelligence updates
  6. Archive: Store intelligence for historical analysis and trending

Scripts 2

agent.py5.0 KB
Display-only source. This catalog never executes bundled scripts.
#!/usr/bin/env python3
"""Diamond Model intrusion analysis agent for structuring threat intelligence events."""

import argparse
import json
import logging
import os
import uuid
from dataclasses import asdict, dataclass, field
from datetime import datetime
from typing import Dict, List

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


@dataclass
class DiamondEvent:
    """A Diamond Model event with four core vertices."""
    event_id: str = field(default_factory=lambda: str(uuid.uuid4())[:8])
    timestamp: str = ""
    adversary: str = ""
    capability: str = ""
    infrastructure: str = ""
    victim: str = ""
    phase: str = ""
    result: str = ""
    direction: str = ""
    methodology: str = ""
    confidence: str = "medium"
    notes: str = ""


def create_event(adversary: str, capability: str, infrastructure: str,
                 victim: str, **kwargs) -> DiamondEvent:
    """Create a Diamond Model event from the four vertices."""
    return DiamondEvent(
        adversary=adversary, capability=capability,
        infrastructure=infrastructure, victim=victim,
        timestamp=datetime.utcnow().isoformat(), **kwargs)


def load_events(data_path: str) -> List[DiamondEvent]:
    """Load Diamond Model events from JSON file."""
    with open(data_path) as f:
        data = json.load(f)
    events = []
    for item in data.get("events", []):
        events.append(DiamondEvent(**{k: v for k, v in item.items()
                                      if k in DiamondEvent.__dataclass_fields__}))
    return events


def pivot_on_vertex(events: List[DiamondEvent], vertex: str, value: str) -> List[DiamondEvent]:
    """Pivot analysis: find all events sharing a vertex value."""
    return [e for e in events if getattr(e, vertex, "") == value]


def build_activity_thread(events: List[DiamondEvent], adversary: str) -> dict:
    """Build an activity thread for an adversary across events."""
    thread_events = [e for e in events if e.adversary == adversary]
    thread_events.sort(key=lambda e: e.timestamp)
    return {
        "adversary": adversary,
        "event_count": len(thread_events),
        "first_seen": thread_events[0].timestamp if thread_events else "",
        "last_seen": thread_events[-1].timestamp if thread_events else "",
        "capabilities_used": list({e.capability for e in thread_events if e.capability}),
        "infrastructure_used": list({e.infrastructure for e in thread_events if e.infrastructure}),
        "victims_targeted": list({e.victim for e in thread_events if e.victim}),
        "phases": [e.phase for e in thread_events if e.phase],
    }


def cluster_by_infrastructure(events: List[DiamondEvent]) -> Dict[str, List[str]]:
    """Cluster events by shared infrastructure to identify campaigns."""
    clusters = {}
    for e in events:
        if e.infrastructure:
            clusters.setdefault(e.infrastructure, []).append(e.event_id)
    return clusters


def compute_vertex_statistics(events: List[DiamondEvent]) -> dict:
    """Compute statistics across all Diamond Model vertices."""
    return {
        "total_events": len(events),
        "unique_adversaries": len({e.adversary for e in events if e.adversary}),
        "unique_capabilities": len({e.capability for e in events if e.capability}),
        "unique_infrastructure": len({e.infrastructure for e in events if e.infrastructure}),
        "unique_victims": len({e.victim for e in events if e.victim}),
        "confidence_distribution": {
            "high": sum(1 for e in events if e.confidence == "high"),
            "medium": sum(1 for e in events if e.confidence == "medium"),
            "low": sum(1 for e in events if e.confidence == "low"),
        },
    }


def generate_report(data_path: str) -> dict:
    """Generate Diamond Model analysis report."""
    events = load_events(data_path)
    stats = compute_vertex_statistics(events)
    adversaries = {e.adversary for e in events if e.adversary}
    threads = [build_activity_thread(events, adv) for adv in adversaries]
    clusters = cluster_by_infrastructure(events)
    return {
        "analysis_date": datetime.utcnow().isoformat(),
        "statistics": stats,
        "activity_threads": threads,
        "infrastructure_clusters": clusters,
        "events": [asdict(e) for e in events],
    }


def main():
    parser = argparse.ArgumentParser(description="Diamond Model Intrusion Analysis Agent")
    parser.add_argument("--data", required=True, help="Path to events JSON")
    parser.add_argument("--output-dir", default=".")
    parser.add_argument("--output", default="diamond_report.json")
    args = parser.parse_args()

    os.makedirs(args.output_dir, exist_ok=True)
    report = generate_report(args.data)
    out_path = os.path.join(args.output_dir, args.output)
    with open(out_path, "w") as f:
        json.dump(report, f, indent=2)
    logger.info("Report saved to %s", out_path)
    print(json.dumps(report["statistics"], indent=2))


if __name__ == "__main__":
    main()
process.py4.1 KB
Display-only source. This catalog never executes bundled scripts.
#!/usr/bin/env python3
"""
Diamond Model of Intrusion Analysis Implementation

Creates Diamond Model events, builds activity threads, and performs pivot analysis.

Requirements: pip install networkx stix2

Usage:
    python process.py --events events.json --output analysis.json
    python process.py --demo --output demo_analysis.json
"""

import argparse
import json
import uuid
from collections import defaultdict
from dataclasses import dataclass, field
from typing import Optional


@dataclass
class DiamondEvent:
    adversary: str = ""
    capability: str = ""
    infrastructure: str = ""
    victim: str = ""
    timestamp: str = ""
    phase: str = ""
    result: str = "success"
    confidence: int = 0
    mitre_techniques: list = field(default_factory=list)
    iocs: list = field(default_factory=list)
    event_id: str = field(default_factory=lambda: str(uuid.uuid4())[:8])

    def to_dict(self):
        return vars(self)


class DiamondModelAnalyzer:
    def __init__(self):
        self.events = []

    def add_event(self, event: DiamondEvent):
        self.events.append(event)

    def load_events(self, filepath):
        with open(filepath) as f:
            data = json.load(f)
        for e in data:
            self.events.append(DiamondEvent(**e))

    def find_pivots(self):
        pivots = {"infrastructure": defaultdict(list), "capability": defaultdict(list),
                  "adversary": defaultdict(list), "victim": defaultdict(list)}
        for e in self.events:
            for feature in pivots:
                val = getattr(e, feature, "")
                if val:
                    pivots[feature][val].append(e.event_id)
        return {k: {pk: pv for pk, pv in v.items() if len(pv) > 1} for k, v in pivots.items()}

    def build_activity_threads(self):
        threads = defaultdict(list)
        for e in sorted(self.events, key=lambda x: x.timestamp):
            key = e.adversary or "unknown"
            threads[key].append(e.to_dict())
        return dict(threads)

    def generate_report(self):
        return {
            "total_events": len(self.events),
            "unique_adversaries": len(set(e.adversary for e in self.events if e.adversary)),
            "unique_victims": len(set(e.victim for e in self.events if e.victim)),
            "unique_infrastructure": len(set(e.infrastructure for e in self.events if e.infrastructure)),
            "pivots": self.find_pivots(),
            "activity_threads": self.build_activity_threads(),
            "events": [e.to_dict() for e in self.events],
        }


def run_demo():
    analyzer = DiamondModelAnalyzer()
    analyzer.add_event(DiamondEvent(
        adversary="APT29", capability="Cobalt Strike", infrastructure="198.51.100.1",
        victim="Gov Agency A", timestamp="2025-06-01T10:00:00Z", phase="initial-access",
        mitre_techniques=["T1566.001"], confidence=80,
    ))
    analyzer.add_event(DiamondEvent(
        adversary="APT29", capability="Custom Backdoor", infrastructure="198.51.100.1",
        victim="Gov Agency A", timestamp="2025-06-01T12:00:00Z", phase="persistence",
        mitre_techniques=["T1547.001"], confidence=85,
    ))
    analyzer.add_event(DiamondEvent(
        adversary="APT29", capability="Mimikatz", infrastructure="198.51.100.2",
        victim="Gov Agency A", timestamp="2025-06-02T09:00:00Z", phase="credential-access",
        mitre_techniques=["T1003.001"], confidence=90,
    ))
    return analyzer.generate_report()


def main():
    parser = argparse.ArgumentParser(description="Diamond Model Analyzer")
    parser.add_argument("--events", help="Events JSON file")
    parser.add_argument("--demo", action="store_true", help="Run demo analysis")
    parser.add_argument("--output", default="diamond_analysis.json")
    args = parser.parse_args()

    if args.demo:
        report = run_demo()
    elif args.events:
        analyzer = DiamondModelAnalyzer()
        analyzer.load_events(args.events)
        report = analyzer.generate_report()
    else:
        parser.print_help()
        return

    print(json.dumps(report, indent=2))
    with open(args.output, "w") as f:
        json.dump(report, f, indent=2)


if __name__ == "__main__":
    main()

Assets 1

template.mdtext/markdown · 1.0 KB
Keep exploring
Analyzing Campaign Attribution EvidenceCampaign attribution analysis involves systematically evaluating evidence to determine which threat actor or group is responsible for a cyber operation. This skill covers collecting and weighting attribution indicators using the Diamond Model and ACH (Analysis of Competing Hypotheses), analyzing infrastructure overlaps, TTP consistency, malware code similarities, operational timing patterns, and language artifacts to build confidence-weighted attribution assessments.Analyzing Threat Actor TTPs with MITRE ATT&CKMITRE ATT&CK is a globally-accessible knowledge base of adversary tactics, techniques, and procedures (TTPs) based on real-world observations. This skill covers systematically mapping threat actor behavior to the ATT&CK framework, building technique coverage heatmaps using the ATT&CK Navigator, identifying detection gaps, and producing actionable intelligence reports that link observed IOCs to specific adversary techniques across the Enterprise, Mobile, and ICS matrices.Building IOC Enrichment Pipeline with OpenCTIOpenCTI is an open-source platform for managing cyber threat intelligence knowledge, built on STIX 2.1 as its native data model. This skill covers building an automated IOC enrichment pipeline using OpenCTI's connector ecosystem to enrich indicators with context from VirusTotal, Shodan, AbuseIPDB, GreyNoise, and other sources. The pipeline automatically enriches newly ingested indicators, correlates them with known threat actors and campaigns, and scores them for analyst prioritization.Building Threat Intelligence PlatformBuilding a Threat Intelligence Platform (TIP) involves deploying and integrating multiple CTI tools into a unified system for collecting, analyzing, enriching, and disseminating threat intelligence. This skill covers designing TIP architecture using open-source tools (MISP, OpenCTI, TheHive, Cortex), configuring feed ingestion pipelines, establishing enrichment workflows, implementing STIX/TAXII interoperability, and building analyst dashboards for CTI operations.Collecting Threat Intelligence with MISPMISP (Malware Information Sharing Platform) is an open-source threat intelligence platform for gathering, sharing, storing, and correlating Indicators of Compromise (IOCs) of targeted attacks, threat intelligence, financial fraud information, vulnerability information, or counter-terrorism information. This skill covers deploying MISP, configuring threat feeds, using the PyMISP API for programmatic access, and building automated collection pipelines that aggregate IOCs from multiple community and commercial sources.Implementing STIX/TAXII Feed IntegrationSTIX (Structured Threat Information eXpression) and TAXII (Trusted Automated eXchange of Intelligence Information) are OASIS open standards for representing and transporting cyber threat intelligence. This skill covers implementing a STIX/TAXII 2.1 feed consumer and producer using Python, configuring TAXII server discovery, collection management, polling for new intelligence, parsing STIX 2.1 objects, and integrating feeds into SIEM and TIP platforms.