building-threat-feed-aggregation-with-misp

部署 MISP（恶意软件信息共享平台）来聚合、关联和分发来自多个来源的威胁情报推送，用于集中式 IOC 管理和自动化 SIEM 集成。

9 stars

Best use case

building-threat-feed-aggregation-with-misp is best used when you need a repeatable AI agent workflow instead of a one-off prompt.

部署 MISP（恶意软件信息共享平台）来聚合、关联和分发来自多个来源的威胁情报推送，用于集中式 IOC 管理和自动化 SIEM 集成。

Teams using building-threat-feed-aggregation-with-misp should expect a more consistent output, faster repeated execution, less prompt rewriting.

When to use this skill

You want a reusable workflow that can be run more than once with consistent structure.

When not to use this skill

You only need a quick one-off answer and do not need a reusable workflow.
You cannot install or maintain the underlying files, dependencies, or repository context.

Installation

Claude Code / Cursor / Codex

$curl -o ~/.claude/skills/building-threat-feed-aggregation-with-misp/SKILL.md --create-dirs "https://raw.githubusercontent.com/killvxk/cybersecurity-skills-zh/main/skills/building-threat-feed-aggregation-with-misp/SKILL.md"

Manual Installation

Download SKILL.md from GitHub
Place it in .claude/skills/building-threat-feed-aggregation-with-misp/SKILL.md inside your project
Restart your AI agent — it will auto-discover the skill

How building-threat-feed-aggregation-with-misp Compares

Feature / Agent	building-threat-feed-aggregation-with-misp	Standard Approach
Platform Support	Not specified	Limited / Varies
Context Awareness	High	Baseline
Installation Complexity	Unknown	N/A

Frequently Asked Questions

What does this skill do?

部署 MISP（恶意软件信息共享平台）来聚合、关联和分发来自多个来源的威胁情报推送，用于集中式 IOC 管理和自动化 SIEM 集成。

Where can I find the source code?

You can find the source code on GitHub using the link provided at the top of the page.

SKILL.md Source

# 使用 MISP 构建威胁推送聚合

## 概述

MISP 是领先的开源威胁情报平台，用于收集、存储、分发和共享网络安全指标和威胁情报。它将来自 OSINT 来源、商业提供商和共享社区的推送聚合到统一平台，具备自动关联、STIX/TAXII 导出，以及与 SIEM 和安全工具的直接集成。本技能涵盖通过 Docker 部署 MISP、配置来自 abuse.ch、AlienVault OTX 和 CIRCL 等来源的推送、设置自动推送同步，以及与 Splunk、Elasticsearch 和 SOAR 平台集成。

## 前置条件

- 用于部署的 Docker 和 Docker Compose
- Python 3.9+，安装 `pymisp` 库用于 API 交互
- 生产部署需要 8GB+ RAM 的 Linux 服务器
- 了解 IOC 类型和威胁情报生命周期
- 访问外部推送 URL 的网络访问权限

## 核心概念

### MISP 架构

MISP 将威胁情报存储为包含属性（IOC）的事件，按类型和类别组织。事件可以有标签（MITRE ATT&CK、TLP 标记、行业标签）、Galaxy（威胁行为者档案、恶意软件家族、攻击模式）和对象（相关属性的结构化分组）。事件在实例间自动关联。

### 推送类型

MISP 支持三种推送格式：MISP 格式（原生 JSON 事件）、CSV（逗号分隔的 IOC）和自由文本（非结构化文本，自动提取 IOC）。推送可以是远程的（从 URL 获取）或本地的（上传文件）。MISP 内置了 80 多个默认 OSINT 推送，包括 abuse.ch URLhaus、Botvrij、CIRCL OSINT 和恶意软件流量分析。

### 共享与同步

MISP 实例可通过推/拉机制与其他 MISP 实例同步。共享组控制分发范围（仅限本组织、本社区、已连接社区、所有社区）。TAXII 服务器模块支持与 STIX/TAXII 消费者集成。

## 实践步骤

### 步骤 1：使用 Docker 部署 MISP

```yaml
# docker-compose.yml 用于 MISP 部署
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:
```

### 步骤 2：通过 PyMISP API 配置推送

```python
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"[+] 已连接到 MISP: {misp_url}")

    def list_feeds(self):
        """列出所有已配置的推送。"""
        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"[+] 推送: {len(enabled)} 个已启用, {len(disabled)} 个已禁用")
        return feeds

    def enable_default_feeds(self):
        """启用推荐的默认 OSINT 推送。"""
        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"  [+] 已启用: {feed_data['name']}")

        print(f"[+] 共启用 {enabled_count} 个推送")

    def add_custom_feed(self, name, url, provider, feed_format="csv",
                        input_source="network", enabled=True):
        """添加自定义威胁情报推送。"""
        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"  # 所有社区

        result = self.misp.add_feed(feed)
        if "Feed" in result:
            feed_id = result["Feed"]["id"]
            print(f"[+] 已添加推送: {name} (ID: {feed_id})")
            return feed_id
        else:
            print(f"[-] 添加推送失败: {result}")
            return None

    def fetch_all_feeds(self):
        """触发所有已启用推送的获取。"""
        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"  [*] 正在获取: {feed_data['name']}")
        print("[+] 已触发所有已启用推送的获取")

manager = MISPFeedManager(
    "https://misp.organization.com",
    "YOUR_MISP_API_KEY",
)
manager.enable_default_feeds()
manager.add_custom_feed(
    name="Abuse.ch MalwareBazaar 近期",
    url="https://bazaar.abuse.ch/export/csv/recent/",
    provider="abuse.ch",
    feed_format="csv",
)
manager.fetch_all_feeds()
```

### 步骤 3：搜索和关联指标

```python
def search_indicators(misp, value=None, type_attribute=None, tags=None, last_days=30):
    """在 MISP 中搜索并关联指标。"""
    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"[+] 搜索返回 {len(attributes)} 个属性")

    # 按事件分组以获取上下文
    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}

# 搜索特定 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"])
```

### 步骤 4：导出到 SIEM（Splunk / Elasticsearch）

```python
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):
        """通过 HEC 将近期 MISP 指标导出到 Splunk。"""
        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=False,
            )
            if resp.status_code == 200:
                exported += 1

        print(f"[+] 已导出 {exported}/{len(attributes)} 个指标到 Splunk")

    def export_ioc_list(self, output_file, ioc_types=None, days=30):
        """导出防火墙/代理封锁列表的平面 IOC 列表。"""
        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"[+] 已导出 {len(unique_iocs)} 个唯一 IOC 到 {output_file}")

exporter = MISPSIEMExporter(misp)
exporter.export_ioc_list("blocklist_ips.txt", ioc_types=["ip-dst"], days=7)
```

## 验收标准

- MISP 已部署并可通过 Web 界面和 API 访问
- 默认 OSINT 推送已启用并正在获取数据
- 自定义推送已添加并正在摄取指标
- 指标可搜索并跨事件关联
- IOC 已成功导出到 SIEM（Splunk/Elasticsearch）
- 防火墙/代理集成的封锁列表已生成

## 参考资料

- [MISP 项目](https://www.misp-project.org/)
- [MISP GitHub 仓库](https://github.com/MISP/MISP)
- [MISP 默认推送](https://www.misp-project.org/feeds/)
- [PyMISP 文档](https://pymisp.readthedocs.io/)
- [Kraven Security: 使用 MISP 推送](https://kravensecurity.com/threat-intelligence-with-misp-part-4-using-feeds/)
- [Cosive: 什么是 MISP](https://www.cosive.com/blog/what-is-misp-the-ultimate-introduction)

Related Skills

tracking-threat-actor-infrastructure

from killvxk/cybersecurity-skills-zh

威胁行为者基础设施追踪涉及使用被动 DNS、证书透明度日志、Shodan/Censys 扫描、WHOIS 分析和网络指纹技术，对对手控制的 C2 服务器、钓鱼域名和暂存服务器等资产进行监控、映射和持续追踪

profiling-threat-actor-groups

from killvxk/cybersecurity-skills-zh

通过聚合 TTP 文档、历史活动数据、工具指纹和来自多个情报源的归因指标，为 APT 组织、犯罪组织和黑客活动组织开发全面的威胁行为者画像。适用于就行业特定威胁向管理层汇报、更新威胁模型假设，或针对特定对手优先部署防御控制措施。当涉及 MITRE ATT&CK 组织、Mandiant APT 画像、CrowdStrike 对手命名或行业特定威胁简报时激活。

processing-stix-taxii-feeds

from killvxk/cybersecurity-skills-zh

处理通过 TAXII 2.1 服务器分发的 STIX 2.1 威胁情报包，将对象规范化为平台原生模式并路由到相应消费系统。适用于接入新的 TAXII 集合端点、自动化与 ISAC 的双向情报共享，或为格式错误的 STIX 包构建管道验证。当涉及 OASIS STIX、TAXII 服务器配置、MISP TAXII 或 Cortex XSOAR 情报源集成时激活。

performing-threat-modeling-with-owasp-threat-dragon

from killvxk/cybersecurity-skills-zh

使用 OWASP Threat Dragon 创建数据流图，运用 STRIDE 和 LINDDUN 方法论识别威胁，并生成威胁模型报告用于安全设计审查。

performing-threat-landscape-assessment-for-sector

from killvxk/cybersecurity-skills-zh

通过分析威胁行为者定向攻击模式、常见攻击向量和行业特定漏洞，开展行业特定威胁态势评估，为组织风险管理提供决策依据

performing-threat-intelligence-sharing-with-misp

from killvxk/cybersecurity-skills-zh

使用 PyMISP 在 MISP 平台上创建、丰富和共享威胁情报事件，包括 IOC 管理、情报源集成、STIX 导出及社区共享工作流

performing-threat-hunting-with-yara-rules

from killvxk/cybersecurity-skills-zh

使用 YARA 模式匹配规则在文件系统和内存转储中狩猎恶意软件、可疑文件和入侵指标。涵盖规则编写、yara-python 扫描以及与威胁情报源的集成。

performing-threat-hunting-with-elastic-siem

from killvxk/cybersecurity-skills-zh

使用 KQL/EQL 查询、检测规则和 Timeline 调查在 Elastic Security SIEM 中执行主动威胁狩猎，识别绕过自动检测的威胁。适用于 SOC 团队针对特定 ATT&CK 技术进行狩猎、调查异常行为，或使用 Elasticsearch 和 Kibana Security 验证检测覆盖缺口。

performing-threat-emulation-with-atomic-red-team

from killvxk/cybersecurity-skills-zh

使用 atomic-operator Python 框架执行 Atomic Red Team 测试，进行 MITRE ATT&CK 技术验证。从 YAML 原子测试加载测试定义、运行攻击模拟并验证检测覆盖率。适用于测试 SIEM 检测规则、验证 EDR 覆盖率或开展紫队演练。

performing-insider-threat-investigation

from killvxk/cybersecurity-skills-zh

调查内部威胁事件，涉及滥用授权访问权限窃取数据、破坏系统或违反安全策略的员工、承包商或受信任合作伙伴。结合数字取证、用户行为分析以及 HR/法务协调，构建基于证据的案例。适用于内部威胁调查、员工数据盗窃、权限滥用、用户行为异常或内部威胁检测等请求场景。

performing-dark-web-monitoring-for-threats

from killvxk/cybersecurity-skills-zh

暗网威胁监控涉及系统性扫描 Tor 隐藏服务、地下论坛、粘贴站点和暗网市场，以识别针对组织的威胁，包括泄露凭据、数据泄露、威胁行为者讨论、漏洞利用工具和预谋攻击。

investigating-insider-threat-indicators

from killvxk/cybersecurity-skills-zh

调查内部威胁（Insider Threat）指标，包括数据渗漏（Data Exfiltration）尝试、未授权访问模式、策略违规和离职前行为，结合 SIEM 分析、DLP 告警和 HR 数据关联进行调查。适用于 SOC 团队收到 HR 内部威胁转介、检测到员工异常数据移动，或需要为潜在内部威胁建立调查时间线时。