Install this skill

npx skills add mukul975/Anthropic-Cybersecurity-Skills

Framework mappings

MITRE ATT&CK

NIST CSF 2.0

Overview

MISP is the leading open-source threat intelligence platform for collecting, storing, distributing, and sharing cybersecurity indicators and threat intelligence. It aggregates feeds from OSINT sources, commercial providers, and sharing communities into a unified platform with automatic correlation, STIX/TAXII export, and direct integration with SIEMs and security tools. This skill covers deploying MISP via Docker, configuring feeds from sources like abuse.ch, AlienVault OTX, and CIRCL, setting up automated feed synchronization, and integrating with Splunk, Elasticsearch, and SOAR platforms.

When to Use

When deploying or configuring building threat feed aggregation with misp 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

Docker and Docker Compose for deployment
Python 3.9+ with pymisp library for API interaction
Linux server with 8GB+ RAM for production deployment
Understanding of IOC types and threat intelligence lifecycle
Network access to external feed URLs

Key Concepts

MISP Architecture

MISP stores threat intelligence as Events containing Attributes (IOCs) organized by type and category. Events can have Tags (MITRE ATT&CK, TLP marking, sector tags), Galaxies (threat actor profiles, malware families, attack patterns), and Objects (structured groupings of related attributes). Events are correlated automatically across the instance.

Feed Types

MISP supports three feed formats: MISP format (native JSON events), CSV (comma-separated IOCs), and freetext (unstructured text with automatic IOC extraction). Feeds can be remote (fetched from URLs) or local (uploaded files). MISP ships with 80+ default OSINT feeds including abuse.ch URLhaus, Botvrij, CIRCL OSINT, and malware traffic analysis.

Sharing and Synchronization

MISP instances can synchronize with other MISP instances via push/pull mechanisms. Sharing groups control distribution (organization only, this community, connected communities, all communities). The TAXII server module enables integration with STIX/TAXII consumers.

Workflow

Step 1: Deploy MISP with Docker

# docker-compose.yml for MISP deployment
version: '3.8'
services:
  misp:
    image: coolacid/misp-docker:core-latest
    container_name: misp
    restart: unless-stopped
    ports:
      - "443:443"
      - "80:80"
    environment:
      - MYSQL_HOST=misp-db
      - MYSQL_DATABASE=misp
      - MYSQL_USER=misp
      - MYSQL_PASSWORD=misp_db_password_change_me
      - MISP_ADMIN_EMAIL=admin@organization.com
      - MISP_ADMIN_PASSPHRASE=admin_password_change_me
      - MISP_BASEURL=https://misp.organization.com
      - POSTFIX_RELAY_HOST=smtp.organization.com
      - TIMEZONE=UTC
    volumes:
      - misp-data:/var/www/MISP/app/files
      - misp-config:/var/www/MISP/app/Config
    depends_on:
      - misp-db
      - misp-redis
 
  misp-db:
    image: mysql:8.0
    container_name: misp-db
    restart: unless-stopped
    environment:
      - MYSQL_DATABASE=misp
      - MYSQL_USER=misp
      - MYSQL_PASSWORD=misp_db_password_change_me
      - MYSQL_ROOT_PASSWORD=root_password_change_me
    volumes:
      - misp-db-data:/var/lib/mysql
 
  misp-redis:
    image: redis:7
    container_name: misp-redis
    restart: unless-stopped
 
volumes:
  misp-data:
  misp-config:
  misp-db-data:

Step 2: Configure Feeds via PyMISP API

from pymisp import PyMISP, MISPFeed
import json
 
class MISPFeedManager:
    def __init__(self, misp_url, misp_key, verify_ssl=False):
        self.misp = PyMISP(misp_url, misp_key, verify_ssl)
        print(f"[+] Connected to MISP: {misp_url}")
 
    def list_feeds(self):
        """List all configured feeds."""
        feeds = self.misp.feeds()
        enabled = [f for f in feeds if f.get("Feed", {}).get("enabled")]
        disabled = [f for f in feeds if not f.get("Feed", {}).get("enabled")]
        print(f"[+] Feeds: {len(enabled)} enabled, {len(disabled)} disabled")
        return feeds
 
    def enable_default_feeds(self):
        """Enable recommended default OSINT feeds."""
        recommended_feeds = [
            "CIRCL OSINT Feed",
            "Botvrij.eu - Indicators of Compromise",
            "abuse.ch URLhaus Host file",
            "abuse.ch Feodo Tracker",
            "abuse.ch SSL Blacklist",
            "malwaredomainlist",
            "CyberCure - IP Feed",
        ]
 
        feeds = self.misp.feeds()
        enabled_count = 0
        for feed in feeds:
            feed_data = feed.get("Feed", {})
            if feed_data.get("name") in recommended_feeds:
                if not feed_data.get("enabled"):
                    self.misp.enable_feed(feed_data["id"])
                    self.misp.enable_feed_cache(feed_data["id"])
                    enabled_count += 1
                    print(f"  [+] Enabled: {feed_data['name']}")
 
        print(f"[+] Enabled {enabled_count} feeds")
 
    def add_custom_feed(self, name, url, provider, feed_format="csv",
                        input_source="network", enabled=True):
        """Add a custom threat intelligence feed."""
        feed = MISPFeed()
        feed.name = name
        feed.provider = provider
        feed.url = url
        feed.source_format = feed_format
        feed.input_source = input_source
        feed.enabled = enabled
        feed.caching_enabled = True
        feed.publish = False
        feed.distribution = "3"  # All communities
 
        result = self.misp.add_feed(feed)
        if "Feed" in result:
            feed_id = result["Feed"]["id"]
            print(f"[+] Added feed: {name} (ID: {feed_id})")
            return feed_id
        else:
            print(f"[-] Error adding feed: {result}")
            return None
 
    def fetch_all_feeds(self):
        """Trigger fetch for all enabled feeds."""
        feeds = self.misp.feeds()
        for feed in feeds:
            feed_data = feed.get("Feed", {})
            if feed_data.get("enabled"):
                self.misp.fetch_feed(feed_data["id"])
                print(f"  [*] Fetching: {feed_data['name']}")
        print("[+] Feed fetch triggered for all enabled feeds")
 
manager = MISPFeedManager(
    "https://misp.organization.com",
    "YOUR_MISP_API_KEY",
)
manager.enable_default_feeds()
manager.add_custom_feed(
    name="Abuse.ch MalwareBazaar Recent",
    url="https://bazaar.abuse.ch/export/csv/recent/",
    provider="abuse.ch",
    feed_format="csv",
)
manager.fetch_all_feeds()

Step 3: Search and Correlate Indicators

def search_indicators(misp, value=None, type_attribute=None, tags=None, last_days=30):
    """Search MISP for indicators with correlation."""
    from datetime import datetime, timedelta
    date_from = (datetime.now() - timedelta(days=last_days)).strftime("%Y-%m-%d")
 
    search_params = {
        "date_from": date_from,
        "published": True,
        "enforceWarninglist": True,
    }
    if value:
        search_params["value"] = value
    if type_attribute:
        search_params["type_attribute"] = type_attribute
    if tags:
        search_params["tags"] = tags
 
    results = misp.search("attributes", **search_params)
    attributes = results.get("Attribute", [])
    print(f"[+] Search returned {len(attributes)} attributes")
 
    # Group by event for context
    events = {}
    for attr in attributes:
        event_id = attr.get("event_id", "")
        if event_id not in events:
            events[event_id] = {"attributes": [], "tags": set()}
        events[event_id]["attributes"].append({
            "type": attr.get("type", ""),
            "value": attr.get("value", ""),
            "category": attr.get("category", ""),
            "timestamp": attr.get("timestamp", ""),
        })
        for tag in attr.get("Tag", []):
            events[event_id]["tags"].add(tag.get("name", ""))
 
    return {"attributes": attributes, "events": events}
 
# Search for specific IOC
misp = manager.misp
results = search_indicators(misp, value="203.0.113.1")
results_by_type = search_indicators(misp, type_attribute="ip-dst", last_days=7)
results_by_tag = search_indicators(misp, tags=["tlp:white", "type:OSINT"])

Step 4: Export to SIEM (Splunk / Elasticsearch)

import requests
from datetime import datetime, timedelta
 
class MISPSIEMExporter:
    def __init__(self, misp_client):
        self.misp = misp_client
 
    def export_to_splunk(self, splunk_url, hec_token, days=7):
        """Export recent MISP indicators to Splunk via HEC."""
        date_from = (datetime.now() - timedelta(days=days)).strftime("%Y-%m-%d")
        results = self.misp.search("attributes", date_from=date_from,
                                    published=True, enforceWarninglist=True)
        attributes = results.get("Attribute", [])
 
        headers = {"Authorization": f"Splunk {hec_token}"}
        exported = 0
        for attr in attributes:
            event = {
                "event": {
                    "ioc_type": attr.get("type", ""),
                    "ioc_value": attr.get("value", ""),
                    "category": attr.get("category", ""),
                    "event_id": attr.get("event_id", ""),
                    "timestamp": attr.get("timestamp", ""),
                    "tags": [t.get("name", "") for t in attr.get("Tag", [])],
                },
                "sourcetype": "misp:attribute",
                "source": "misp",
                "index": "threat_intel",
            }
            resp = requests.post(
                f"{splunk_url}/services/collector/event",
                headers=headers, json=event,
                verify=not os.environ.get("SKIP_TLS_VERIFY", "").lower() == "true",  # Set SKIP_TLS_VERIFY=true for self-signed certs in lab environments
            )
            if resp.status_code == 200:
                exported += 1
 
        print(f"[+] Exported {exported}/{len(attributes)} indicators to Splunk")
 
    def export_ioc_list(self, output_file, ioc_types=None, days=30):
        """Export flat IOC list for firewall/proxy blocklists."""
        ioc_types = ioc_types or ["ip-dst", "domain", "hostname", "url"]
        date_from = (datetime.now() - timedelta(days=days)).strftime("%Y-%m-%d")
 
        all_iocs = []
        for ioc_type in ioc_types:
            results = self.misp.search(
                "attributes", type_attribute=ioc_type,
                date_from=date_from, published=True,
                enforceWarninglist=True,
            )
            for attr in results.get("Attribute", []):
                all_iocs.append(attr.get("value", ""))
 
        unique_iocs = sorted(set(all_iocs))
        with open(output_file, "w") as f:
            for ioc in unique_iocs:
                f.write(f"{ioc}\n")
 
        print(f"[+] Exported {len(unique_iocs)} unique IOCs to {output_file}")
 
exporter = MISPSIEMExporter(misp)
exporter.export_ioc_list("blocklist_ips.txt", ioc_types=["ip-dst"], days=7)

Validation Criteria

MISP deployed and accessible via web interface and API
Default OSINT feeds enabled and fetching data
Custom feeds added and ingesting indicators
Indicators searchable with correlation across events
IOCs exported to SIEM (Splunk/Elasticsearch) successfully
Blocklists generated for firewall/proxy integration

References

Source materials

References and resources

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

References 1

api-reference.md2.1 KB

API Reference: Threat Feed Aggregation with MISP

PyMISP Python Client

Installation

pip install pymisp

Client Initialization

from pymisp import PyMISP
 
misp = PyMISP(
    url="https://misp.example.org",
    key=os.environ.get("MISP_API_KEY", ""),
    ssl=True
)

Feed Management

# List all feeds
feeds = misp.feeds()
 
# Enable a feed
misp.enable_feed(feed_id=1)
 
# Fetch feed data
misp.fetch_feed(feed_id=1)
 
# Cache feed locally
misp.cache_feeds()
 
# Add new feed
feed = misp.add_feed(
    name="Abuse.ch URLhaus",
    provider="abuse.ch",
    url="https://urlhaus.abuse.ch/downloads/csv_recent/",
    input_source="network",
    source_format="csv"
)

Event Operations

# Search events by tag
events = misp.search(tags=["tlp:white", "type:OSINT"])
 
# Get event attributes
event = misp.get_event(event_id=42)
for attr in event.Attribute:
    print(f"{attr.type}: {attr.value}")
 
# Add attribute to event
misp.add_attribute(event_id=42, type="ip-dst", value="198.51.100.1")

STIX/TAXII Export

# STIX export via REST
curl -H "Authorization: $MISP_KEY" \
  "https://misp.example.org/events/restSearch/stix2"
 
# TAXII collection
curl "https://misp.example.org/taxii2/collections"

Common Feed Sources

Feed	URL	Format
Abuse.ch URLhaus	https://urlhaus.abuse.ch/downloads/csv_recent/	CSV
Abuse.ch Feodo	https://feodotracker.abuse.ch/downloads/ipblocklist.csv	CSV
CIRCL OSINT	https://www.circl.lu/doc/misp/feed-osint/	MISP
Botvrij.eu	https://www.botvrij.eu/data/feed-osint/	MISP
PhishTank	https://data.phishtank.com/data/online-valid.json	JSON

Feed Configuration Fields

Field	Description
name	Human-readable feed name
provider	Organization providing the feed
url	Feed URL or local path
input_source	"network" or "local"
source_format	"misp", "csv", "freetext", "stix"
enabled	Boolean to activate feed
distribution	0=Org, 1=Community, 2=Connected, 3=All
delta_merge	Only import new/changed data

Scripts 1

agent.py6.6 KB

Display-only source. This catalog never executes bundled scripts.

#!/usr/bin/env python3
"""Threat Feed Aggregation Agent - Aggregates and correlates threat intelligence feeds using MISP."""

import json
import logging
import os
import argparse
from datetime import datetime
from collections import defaultdict

import requests

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


def misp_request(url, key, endpoint, method="GET", data=None):
    """Make authenticated MISP API request."""
    headers = {"Authorization": key, "Accept": "application/json", "Content-Type": "application/json"}
    full_url = f"{url}/{endpoint}"
    try:
        if method == "GET":
            resp = requests.get(full_url, headers=headers, timeout=30, verify=not os.environ.get("SKIP_TLS_VERIFY", "").lower() == "true")  # Set SKIP_TLS_VERIFY=true for self-signed certs in lab environments
        else:
            resp = requests.post(full_url, headers=headers, json=data or {}, timeout=30, verify=not os.environ.get("SKIP_TLS_VERIFY", "").lower() == "true")  # Set SKIP_TLS_VERIFY=true for self-signed certs in lab environments
        resp.raise_for_status()
        return resp.json()
    except requests.RequestException as e:
        logger.error("MISP request failed: %s", e)
        return {"error": str(e)}


def list_feeds(url, key):
    """List configured MISP feeds."""
    data = misp_request(url, key, "feeds/index", method="POST")
    feeds = data if isinstance(data, list) else data.get("Feed", [])
    result = []
    for feed in feeds:
        f = feed.get("Feed", feed) if isinstance(feed, dict) and "Feed" in feed else feed
        result.append({"id": f.get("id"), "name": f.get("name"), "provider": f.get("provider"),
                       "url": f.get("url"), "enabled": f.get("enabled"), "source_format": f.get("source_format"),
                       "caching_enabled": f.get("caching_enabled")})
    logger.info("Found %d configured feeds", len(result))
    return result


def fetch_feed_data(url, key, feed_id):
    """Fetch and cache data from a specific feed."""
    result = misp_request(url, key, f"feeds/cacheFeeds/{feed_id}", method="POST")
    logger.info("Cached feed %s", feed_id)
    return result


def search_attributes(url, key, attr_type=None, value=None, last_days=30):
    """Search MISP attributes across all events."""
    search_body = {"returnFormat": "json", "limit": 1000, "last": f"{last_days}d"}
    if attr_type:
        search_body["type"] = attr_type
    if value:
        search_body["value"] = value
    data = misp_request(url, key, "attributes/restSearch", method="POST", data=search_body)
    attributes = data.get("response", {}).get("Attribute", [])
    logger.info("Found %d attributes (type=%s, last %dd)", len(attributes), attr_type, last_days)
    return attributes


def aggregate_feed_statistics(url, key, last_days=30):
    """Aggregate statistics across all feeds."""
    events_data = misp_request(url, key, "events/restSearch", method="POST",
                                data={"returnFormat": "json", "limit": 500, "last": f"{last_days}d"})
    events = events_data.get("response", [])
    stats = {"total_events": len(events), "by_threat_level": defaultdict(int),
             "by_org": defaultdict(int), "by_tag": defaultdict(int), "attribute_types": defaultdict(int)}
    threat_levels = {"1": "High", "2": "Medium", "3": "Low", "4": "Undefined"}
    for event_wrap in events:
        event = event_wrap.get("Event", event_wrap)
        tl = threat_levels.get(str(event.get("threat_level_id", 4)), "Undefined")
        stats["by_threat_level"][tl] += 1
        org = event.get("Orgc", {}).get("name", "Unknown")
        stats["by_org"][org] += 1
        for tag in event.get("Tag", []):
            stats["by_tag"][tag.get("name", "")] += 1
        for attr in event.get("Attribute", []):
            stats["attribute_types"][attr.get("type", "unknown")] += 1
    return {k: dict(v) if isinstance(v, defaultdict) else v for k, v in stats.items()}


def correlate_across_feeds(url, key, ioc_value):
    """Correlate an IOC across all feed events."""
    data = misp_request(url, key, "attributes/restSearch", method="POST",
                        data={"returnFormat": "json", "value": ioc_value, "limit": 100})
    attributes = data.get("response", {}).get("Attribute", [])
    correlations = []
    seen_events = set()
    for attr in attributes:
        event_id = attr.get("event_id")
        if event_id not in seen_events:
            seen_events.add(event_id)
            correlations.append({"event_id": event_id, "type": attr.get("type"), "category": attr.get("category"),
                                 "comment": attr.get("comment", "")[:100]})
    logger.info("IOC '%s' found in %d events", ioc_value, len(correlations))
    return correlations


def assess_feed_health(feeds):
    """Assess health and coverage of configured feeds."""
    total = len(feeds)
    enabled = sum(1 for f in feeds if f.get("enabled"))
    cached = sum(1 for f in feeds if f.get("caching_enabled"))
    return {"total_feeds": total, "enabled": enabled, "disabled": total - enabled,
            "caching_enabled": cached, "health_score": round(enabled / total * 100, 1) if total else 0}


def generate_report(feeds, stats, feed_health):
    """Generate threat feed aggregation report."""
    report = {
        "timestamp": datetime.utcnow().isoformat(),
        "feed_inventory": feeds,
        "feed_health": feed_health,
        "aggregated_statistics": stats,
    }
    print(f"FEED REPORT: {feed_health['total_feeds']} feeds, {feed_health['enabled']} enabled, "
          f"{stats.get('total_events', 0)} events")
    return report


def main():
    parser = argparse.ArgumentParser(description="Threat Feed Aggregation with MISP")
    parser.add_argument("--url", required=True, help="MISP instance URL")
    parser.add_argument("--key", required=True, help="MISP API key")
    parser.add_argument("--days", type=int, default=30, help="Look-back period in days")
    parser.add_argument("--correlate", help="IOC value to correlate across feeds")
    parser.add_argument("--output", default="feed_aggregation_report.json")
    args = parser.parse_args()

    feeds = list_feeds(args.url, args.key)
    stats = aggregate_feed_statistics(args.url, args.key, args.days)
    feed_health = assess_feed_health(feeds)
    report = generate_report(feeds, stats, feed_health)
    if args.correlate:
        report["correlation_results"] = correlate_across_feeds(args.url, args.key, args.correlate)
    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