building-threat-feed-aggregation-with-misp

name	building-threat-feed-aggregation-with-misp
description	部署 MISP（恶意软件信息共享平台）来聚合、关联和分发来自多个来源的威胁情报推送，用于集中式 IOC 管理和自动化 SIEM 集成。
domain	cybersecurity
subdomain	threat-intelligence
tags	["misp","threat-feed","aggregation","indicator","sharing","correlation","siem-integration","threat-intelligence"]
version	1.0
author	mahipal
license	Apache-2.0

使用 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

# 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=[e-mail]
      - 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: 
    
       
       
       
       
    
       

  
     
     
     


  
  
  

步骤 2：通过 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"[+] 已连接到 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 = .misp.feeds()
        enabled_count = 
         feed  feeds:
            feed_data = feed.get(, {})
             feed_data.get()  recommended_feeds:
                  feed_data.get():
                    .misp.enable_feed(feed_data[])
                    .misp.enable_feed_cache(feed_data[])
                    enabled_count += 
                    ()

        ()

     ():
        
        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 = 
        feed.publish = 
        feed.distribution =   

        result = .misp.add_feed(feed)
           result:
            feed_id = result[][]
            ()
             feed_id
        :
            ()
             

     ():
        
        feeds = .misp.feeds()
         feed  feeds:
            feed_data = feed.get(, {})
             feed_data.get():
                .misp.fetch_feed(feed_data[])
                ()
        ()

manager = MISPFeedManager(
    ,
    ,
)
manager.enable_default_feeds()
manager.add_custom_feed(
    name=,
    url=,
    provider=,
    feed_format=,
)
manager.fetch_all_feeds()

步骤 3：搜索和关联指标

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", ),
            : attr.get(, ),
            : attr.get(, ),
        })
         tag  attr.get(, []):
            events[event_id][].add(tag.get(, ))

     {: attributes, : events}


misp = manager.misp
results = search_indicators(misp, value=)
results_by_type = search_indicators(misp, type_attribute=, last_days=)
results_by_tag = search_indicators(misp, tags=[, ])

步骤 4：导出到 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):
        """通过 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", )  t  attr.get(, [])],
                },
                : ,
                : ,
                : ,
            }
            resp = requests.post(
                ,
                headers=headers, json=event, verify=,
            )
             resp.status_code == :
                exported += 

        ()

     ():
        
        ioc_types = ioc_types  [, , , ]
        date_from = (datetime.now() - timedelta(days=days)).strftime()

        all_iocs = []
         ioc_type  ioc_types:
            results = .misp.search(
                , type_attribute=ioc_type,
                date_from=date_from, published=,
                enforceWarninglist=,
            )
             attr  results.get(, []):
                all_iocs.append(attr.get(, ))

        unique_iocs = ((all_iocs))
         (output_file, )  f:
             ioc  unique_iocs:
                f.write()

        ()

exporter = MISPSIEMExporter(misp)
exporter.export_ioc_list(, ioc_types=[], days=)

验收标准

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

参考资料

• MISP 项目
• MISP GitHub 仓库
• MISP 默认推送
• PyMISP 文档
• Kraven Security: 使用 MISP 推送
• Cosive: 什么是 MISP