pip install requests stix2 feedparser

abuse.ch Feeds:
├── MalwareBazaar   — malware samples (hash, signature, tags)
├── ThreatFox       — IOCs (domains, IPs, URLs) with malware family
├── URLhaus         — malicious URLs (distribution sites, C2)
├── Feodo Tracker   — botnet C2 servers (Dridex, Emotet, TrickBot)
└── SSL Blacklist   — malicious SSL certificates

AlienVault OTX:
├── Pulses          — community-contributed IOC collections
└── Subscriptions   — curated feeds by sector/region

MISP Default Feeds:
├── CIRCL OSINT     — curated open-source intelligence
├── Botvrij.eu      — botnet and malware IOCs
└── abuse.ch feeds  — integrated via MISP modules

{
  "value": "evil.com",
  "type": "domain",
  "source": "threatfox",
  "confidence": 75,
  "first_seen": "2026-01-15T00:00:00Z",
  "last_seen": "2026-03-01T00:00:00Z",
  "tags": ["cobalt-strike", "c2"],
  "malware_family": "CobaltStrike",
  "tlp": "white",
  "expiry": "2026-04-01T00:00:00Z"
}

import requests
from datetime import datetime

def fetch_urlhaus() -> list[dict]:
    """Fetch recent malicious URLs from URLhaus."""
    resp = requests.get(
        "https://urlhaus-api.abuse.ch/v1/urls/recent/", timeout=30,
    )
    if resp.status_code == 200:
        return [
            {"value": u["url"], "type": "url", "source": "urlhaus",
             "tags": u.get("tags", []), "threat": u.get("threat", ""),
             "date_added": u.get("date_added", "")}
            for u in resp.json().get("urls", [])[:100]
        ]
    return []

def fetch_feodotracker() -> list[dict]:
    """Fetch botnet C2 IPs from Feodo Tracker."""
    resp = requests.get(
        "https://feodotracker.abuse.ch/downloads/ipblocklist_recommended.json",
        timeout=30,
    )
    if resp.status_code == 200:
        return [
            {"value": entry["ip_address"], "type": "ip", "source": "feodotracker",
             "port": entry.get("port"), "malware": entry.get("malware", ""),
             "first_seen": entry.get("first_seen", "")}
            for entry in resp.json()[:100]
        ]
    return []

def fetch_otx_pulse(pulse_id: str, api_key: str) -> list[dict]:
    """Fetch indicators from an AlienVault OTX pulse."""
    resp = requests.get(
        f"https://otx.alienvault.com/api/v1/pulses/{pulse_id}/indicators",
        headers={"X-OTX-API-KEY": api_key}, timeout=30,
    )
    if resp.status_code == 200:
        return [
            {"value": i["indicator"], "type": i["type"], "source": "otx",
             "title": i.get("title", ""), "description": i.get("description", "")}
            for i in resp.json().get("results", [])
        ]
    return []

from collections import defaultdict

def normalize_indicators(raw_feeds: list[list[dict]]) -> list[dict]:
    """Normalize and deduplicate indicators across feeds."""
    seen: dict[str, dict] = {}
    for feed in raw_feeds:
        for indicator in feed:
            key = f"{indicator['type']}:{indicator['value'].lower()}"
            if key in seen:
                seen[key]["sources"].append(indicator["source"])
                seen[key]["confidence"] = min(seen[key]["confidence"] + 10, 100)
            else:
                seen[key] = {
                    "value": indicator["value"].lower(),
                    "type": indicator["type"],
                    "sources": [indicator["source"]],
                    "confidence": 50,
                    "tags": indicator.get("tags", []),
                    "malware": indicator.get("malware", indicator.get("threat", "")),
                    "first_seen": indicator.get("first_seen", indicator.get("date_added", "")),
                }
    return list(seen.values())

def check_feed_health(feed_name: str, fetch_fn: callable) -> dict:
    """Monitor feed availability and freshness."""
    try:
        start = datetime.now()
        results = fetch_fn()
        elapsed = (datetime.now() - start).total_seconds()
        return {
            "feed": feed_name,
            "status": "healthy",
            "indicator_count": len(results),
            "response_time_s": elapsed,
            "timestamp": datetime.now().isoformat(),
        }
    except Exception as e:
        return {
            "feed": feed_name,
            "status": "error",
            "error": str(e),
            "timestamp": datetime.now().isoformat(),
        }

def push_to_misp(indicators: list[dict], misp_url: str, misp_key: str, event_id: int) -> dict:
    """Push normalized indicators to MISP event."""
    type_map = {"ip": "ip-dst", "domain": "domain", "url": "url", "sha256": "sha256"}
    pushed = 0
    for ind in indicators:
        misp_type = type_map.get(ind["type"], "text")
        payload = {
            "value": ind["value"], "type": misp_type,
            "to_ids": ind["confidence"] >= 70,
            "comment": f"Sources: {', '.join(ind['sources'])}. Confidence: {ind['confidence']}",
        }
        resp = requests.post(
            f"{misp_url}/attributes/add/{event_id}",
            headers={"Authorization": misp_key, "Content-Type": "application/json"},
            json=payload,
        )
        if resp.status_code == 200:
            pushed += 1
    return {"pushed": pushed, "total": len(indicators), "event_id": event_id}