# analyzing-threat-landscape-with-misp

Query a MISP (Malware Information Sharing Platform) instance via PyMISP to compute event statistics, IOC type breakdowns, threat actor galaxy clusters, and tag trends, and generate threat landscape reports with temporal trends. Use when asked to analyze threat intelligence data, summarize top threat actors or malware families, or produce a CTI landscape report from MISP events.

- **Kind:** skill
- **Source:** https://github.com/mukul975/Anthropic-Cybersecurity-Skills
- **Page:** https://forefy.com/skills/e21cd49e-5572-40a6-a37b-aacf3529c0d8
- **API (JSON + files):** https://forefy.com/api/asr/e21cd49e-5572-40a6-a37b-aacf3529c0d8

---

## LICENSE

```

```

## SKILL.md

---
name: analyzing-threat-landscape-with-misp
description: Query a MISP (Malware Information Sharing Platform) instance via PyMISP
  to compute event statistics, IOC type breakdowns, threat actor galaxy clusters,
  and tag trends, and generate threat landscape reports with temporal trends. Use
  when asked to analyze threat intelligence data, summarize top threat actors or
  malware families, or produce a CTI landscape report from MISP events.
domain: cybersecurity
subdomain: threat-intelligence
tags:
- threat-intelligence
- misp
- threat-landscape
- ioc-analysis
- cti
- threat-sharing
version: '1.0'
author: mahipal
license: Apache-2.0
d3fend_techniques:
- File Metadata Consistency Validation
- Application Protocol Command Analysis
- Identifier Analysis
- Content Format Conversion
- Message Analysis
nist_csf:
- ID.RA-01
- ID.RA-05
- DE.CM-01
- DE.AE-02
mitre_attack:
- T1566
- T1071.001
- T1568
- T1583.001
- T1102
---


# Analyzing Threat Landscape with MISP


## When to Use

- When investigating security incidents that require analyzing threat landscape with misp
- When building detection rules or threat hunting queries for this domain
- When SOC analysts need structured procedures for this analysis type
- When validating security monitoring coverage for related attack techniques

## Prerequisites

- Familiarity with threat intelligence concepts and tools
- Access to a test or lab environment for safe execution
- Python 3.8+ with required dependencies installed
- Appropriate authorization for any testing activities

## Instructions

1. Install dependencies: `pip install pymisp`
2. Configure MISP URL and API key.
3. Run the agent to generate threat landscape analysis:
   - Pull event statistics by threat level and date range
   - Analyze attribute type distributions (IP, domain, hash, URL)
   - Identify top MITRE ATT&CK techniques from event tags
   - Track threat actor activity via galaxy clusters
   - Generate temporal trend analysis of IOC submissions

```bash
python scripts/agent.py --misp-url https://misp.local --api-key YOUR_KEY --days 90 --output landscape_report.json
```

## Examples

### Threat Landscape Summary
```
Period: Last 90 days
Events analyzed: 1,247
Top threat level: High (43%)
Top attribute type: ip-dst (31%), domain (22%), sha256 (18%)
Top MITRE technique: T1566 Phishing (89 events)
Top threat actor: APT28 (34 events)
```

## references

```

```

## references/api-reference.md

# API Reference: MISP Threat Landscape Analysis

## PyMISP Connection
```python
from pymisp import PyMISP
misp = PyMISP(url, api_key, ssl=True)
```

## Event Search
```python
events = misp.search(date_from="2025-01-01", pythonify=True)
```
| Parameter | Description |
|-----------|-------------|
| `date_from` | Start date (YYYY-MM-DD) |
| `date_to` | End date |
| `tags` | Filter by tags |
| `threat_level_id` | 1=High, 2=Medium, 3=Low, 4=Undefined |
| `published` | True/False |
| `pythonify` | Return MISPEvent objects |

## Event Object Fields
| Field | Description |
|-------|-------------|
| `id` | Event ID |
| `date` | Event date |
| `threat_level_id` | 1-4 severity level |
| `analysis` | 0=Initial, 1=Ongoing, 2=Completed |
| `info` | Event description |
| `Attribute` | List of IOC attributes |
| `Tag` | List of tags |
| `Orgc` | Contributing organization |

## Attribute Types
| Type | Example |
|------|---------|
| `ip-dst` | Destination IP address |
| `ip-src` | Source IP address |
| `domain` | Domain name |
| `hostname` | FQDN |
| `url` | Full URL |
| `md5` / `sha1` / `sha256` | File hashes |
| `email-src` | Sender email |
| `filename` | Malicious filename |
| `mutex` | Mutex name |
| `regkey` | Registry key |

## Galaxy Tag Prefixes
| Prefix | Content |
|--------|---------|
| `misp-galaxy:mitre-attack-pattern=` | MITRE ATT&CK techniques |
| `misp-galaxy:threat-actor=` | Threat actor groups |
| `misp-galaxy:malpedia=` | Malware families |
| `misp-galaxy:sector=` | Target sectors |
| `misp-galaxy:country=` | Target countries |

## Statistics API
```python
misp.get_community_id()
misp.user_statistics()
misp.attributes_statistics(context="type")
misp.attributes_statistics(context="category")
misp.tags_statistics()
```

## scripts

```

```

## scripts/agent.py

```python
#!/usr/bin/env python3
"""MISP Threat Landscape Analysis Agent - Generates threat landscape reports from MISP event data."""

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

from pymisp import PyMISP

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

THREAT_LEVELS = {1: "High", 2: "Medium", 3: "Low", 4: "Undefined"}
ANALYSIS_LEVELS = {0: "Initial", 1: "Ongoing", 2: "Completed"}

MITRE_TAG_PREFIX = "misp-galaxy:mitre-attack-pattern="
THREAT_ACTOR_PREFIX = "misp-galaxy:threat-actor="
MALWARE_PREFIX = "misp-galaxy:malpedia="


def connect_misp(url, api_key, ssl=True):
    """Connect to MISP instance."""
    misp = PyMISP(url, api_key, ssl=ssl)
    logger.info("Connected to MISP: %s", url)
    return misp


def fetch_events(misp, days=90):
    """Fetch events from the last N days."""
    date_from = (datetime.utcnow() - timedelta(days=days)).strftime("%Y-%m-%d")
    events = misp.search(date_from=date_from, pythonify=True)
    logger.info("Fetched %d events from last %d days", len(events), days)
    return events


def analyze_threat_levels(events):
    """Break down events by threat level."""
    counter = Counter()
    for event in events:
        level = getattr(event, "threat_level_id", 4)
        counter[THREAT_LEVELS.get(int(level), "Undefined")] += 1
    total = sum(counter.values()) or 1
    return {level: {"count": count, "percent": round(count / total * 100, 1)} for level, count in counter.most_common()}


def analyze_attribute_types(events):
    """Analyze distribution of attribute types across events."""
    counter = Counter()
    for event in events:
        for attr in getattr(event, "Attribute", []):
            counter[attr.type] += 1
    total = sum(counter.values()) or 1
    return {
        atype: {"count": count, "percent": round(count / total * 100, 1)}
        for atype, count in counter.most_common(20)
    }


def extract_tags(events):
    """Extract and categorize tags from events."""
    mitre_techniques = Counter()
    threat_actors = Counter()
    malware_families = Counter()
    all_tags = Counter()

    for event in events:
        for tag in getattr(event, "Tag", []):
            tag_name = tag.name
            all_tags[tag_name] += 1
            if tag_name.startswith(MITRE_TAG_PREFIX):
                technique = tag_name[len(MITRE_TAG_PREFIX):].strip('"').strip("'")
                mitre_techniques[technique] += 1
            elif tag_name.startswith(THREAT_ACTOR_PREFIX):
                actor = tag_name[len(THREAT_ACTOR_PREFIX):].strip('"').strip("'")
                threat_actors[actor] += 1
            elif tag_name.startswith(MALWARE_PREFIX):
                malware = tag_name[len(MALWARE_PREFIX):].strip('"').strip("'")
                malware_families[malware] += 1

    return {
        "mitre_techniques": dict(mitre_techniques.most_common(20)),
        "threat_actors": dict(threat_actors.most_common(20)),
        "malware_families": dict(malware_families.most_common(20)),
        "top_tags": dict(all_tags.most_common(30)),
    }


def analyze_temporal_trends(events, days=90):
    """Analyze event creation trends over time (weekly buckets)."""
    buckets = defaultdict(int)
    for event in events:
        event_date = getattr(event, "date", None)
        if event_date:
            if isinstance(event_date, str):
                event_date = datetime.strptime(event_date, "%Y-%m-%d")
            week_start = event_date - timedelta(days=event_date.weekday())
            buckets[week_start.strftime("%Y-%m-%d")] += 1
    return dict(sorted(buckets.items()))


def analyze_organizations(events):
    """Analyze contributing organizations."""
    org_counter = Counter()
    for event in events:
        org = getattr(event, "Orgc", None)
        if org:
            org_name = getattr(org, "name", "Unknown")
            org_counter[org_name] += 1
    return dict(org_counter.most_common(20))


def compute_ioc_stats(events):
    """Compute IOC statistics: total count, unique values, categories."""
    ioc_values = set()
    category_counter = Counter()
    for event in events:
        for attr in getattr(event, "Attribute", []):
            ioc_values.add(attr.value)
            category_counter[attr.category] += 1
    return {
        "total_attributes": sum(category_counter.values()),
        "unique_values": len(ioc_values),
        "categories": dict(category_counter.most_common(15)),
    }


def generate_report(events, threat_levels, attr_types, tags, trends, orgs, ioc_stats, days):
    """Generate threat landscape report."""
    report = {
        "timestamp": datetime.utcnow().isoformat(),
        "period_days": days,
        "total_events": len(events),
        "threat_level_distribution": threat_levels,
        "attribute_type_distribution": attr_types,
        "ioc_statistics": ioc_stats,
        "mitre_attack_techniques": tags["mitre_techniques"],
        "top_threat_actors": tags["threat_actors"],
        "top_malware_families": tags["malware_families"],
        "temporal_trends": trends,
        "contributing_organizations": orgs,
    }
    high_pct = threat_levels.get("High", {}).get("percent", 0)
    top_technique = next(iter(tags["mitre_techniques"]), "N/A")
    top_actor = next(iter(tags["threat_actors"]), "N/A")
    print(f"THREAT LANDSCAPE: {len(events)} events, {high_pct}% high severity, top technique: {top_technique}, top actor: {top_actor}")
    return report


def main():
    parser = argparse.ArgumentParser(description="MISP Threat Landscape Analysis Agent")
    parser.add_argument("--misp-url", required=True, help="MISP instance URL")
    parser.add_argument("--api-key", required=True, help="MISP API key")
    parser.add_argument("--days", type=int, default=90, help="Analysis period in days")
    parser.add_argument("--no-ssl", action="store_true", help="Disable SSL verification")
    parser.add_argument("--output", default="landscape_report.json")
    args = parser.parse_args()

    misp = connect_misp(args.misp_url, args.api_key, ssl=not args.no_ssl)
    events = fetch_events(misp, args.days)

    threat_levels = analyze_threat_levels(events)
    attr_types = analyze_attribute_types(events)
    tags = extract_tags(events)
    trends = analyze_temporal_trends(events, args.days)
    orgs = analyze_organizations(events)
    ioc_stats = compute_ioc_stats(events)

    report = generate_report(events, threat_levels, attr_types, tags, trends, orgs, ioc_stats, args.days)
    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()
```