building-threat-intelligence-platform

脅威インテリジェンスプラットフォームの構築

概要

脅威インテリジェンスプラットフォーム（TIP）の構築では、複数の CTI ツールをデプロイして統合し、脅威インテリジェンスを収集、分析、エンリッチメント、配信するための統合システムを構築します。このスキルでは、オープンソースツール（MISP、OpenCTI、TheHive、Cortex）を使用した TIP アーキテクチャの設計、フィード取り込みパイプラインの設定、エンリッチメントワークフローの確立、STIX/TAXII の相互運用性の実装、CTI 運用のためのアナリストダッシュボードの構築について説明します。

使用する場面

• 環境内で脅威インテリジェンスプラットフォーム機能をデプロイまたは設定する場合
• コンプライアンス要件に沿ったセキュリティ制御を確立する場合
• この分野のセキュリティアーキテクチャを構築または改善する場合
• この実装が必要なセキュリティ評価を実施する場合

前提条件

• プラットフォームコンポーネントのデプロイに使用する Docker と Docker Compose
• pymisp、pycti、thehive4py ライブラリを含む Python 3.9+
• データストレージ用の Elasticsearch/OpenSearch クラスター
• メッセージキュー用の Redis と RabbitMQ
• STIX 2.1 データモデルと TAXII 2.1 トランスポートに関する知識
• エンリッチメントサービス（VirusTotal、Shodan、AbuseIPDB）の API キー

主な概念

TIP アーキテクチャのコンポーネント

1. 収集レイヤー：OSINT、商用、内部の各ソースからフィードを取り込む
2. ストレージレイヤー：STIX 2.1 スキーマを使用したインデックス済み CTI データ用の Elasticsearch/OpenSearch
3. 分析レイヤー：ナレッジグラフ分析用の OpenCTI と、IOC 相関分析用の MISP
4. エンリッチメントレイヤー：IOC のエンリッチメントを自動化する Cortex アナライザー
5. 対応レイヤー：ケース管理とインシデント対応の統合を担う TheHive
6. 共有レイヤー：脅威インテリジェンスを外部に共有する TAXII サーバー

プラットフォームの統合ポイント

• MISP <-> OpenCTI：OpenCTI MISP コネクター経由の双方向同期
• OpenCTI <-> TheHive：信頼度の高いインジケーターからのアラート／ケース作成
• TheHive <-> Cortex：ケースのオブザーバブルの分析とエンリッチメントの自動化
• 全体 <-> SIEM：API または Kafka 経由で IOC を Splunk/Elastic にリアルタイムでプッシュ

ワークフロー

ステップ 1：Docker Compose でプラットフォームをデプロイする

version: '3.8'
services:
  # --- Storage Layer ---
  elasticsearch:
    image: docker.elastic.co/elasticsearch/elasticsearch:8.12.0
    environment:
      - discovery.type=single-node
      - xpack.security.enabled=false
      - "ES_JAVA_OPTS=-Xms2g -Xmx2g"
    ports:
      - "9200:9200"
    volumes:
      - es-data:/usr/share/elasticsearch/data

  redis:
    image: redis:7
    ports:
      - "6379:6379"

  rabbitmq:
    image: rabbitmq:3-management
    ports:
      - "5672:5672"
      - "15672:15672"

  minio:
    image: minio/minio
    command: server /data --console-address ":9001"
    ports:
      - "9000:9000"
      - "9001:9001"

  # --- MISP ---
  misp:
    image: ghcr.io/misp/misp-docker/misp-core:latest
    ports:
      - "8443:443"
    environment:
      - [email protected]
      - MISP_BASEURL=https://localhost:8443
    volumes:
      - misp-data:/var/www/MISP/app/files

  # --- OpenCTI ---
  opencti:
    image: opencti/platform:6.4.4
    environment:
      - APP__PORT=8080
      - [email protected]
      - APP__ADMIN__PASSWORD=TIPAdminPassword
      - APP__ADMIN__TOKEN=tip-opencti-token-uuid
      - ELASTICSEARCH__URL=http://elasticsearch:9200
      - MINIO__ENDPOINT=minio
      - RABBITMQ__HOSTNAME=rabbitmq
      - REDIS__HOSTNAME=redis
    ports:
      - "8080:8080"
    depends_on:
      - elasticsearch
      - redis
      - rabbitmq
      - minio

  # --- TheHive ---
  thehive:
    image: strangebee/thehive:5.3
    environment:
      - TH_CORTEX_URL=http://cortex:9001
    ports:
      - "9000:9000"
    depends_on:
      - elasticsearch

  # --- Cortex ---
  cortex:
    image: thehiveproject/cortex:3.1.8
    ports:
      - "9001:9001"
    depends_on:
      - elasticsearch

volumes:
  es-data:
  misp-data:

ステップ 2：フィード取り込みパイプラインを設定する

from pymisp import PyMISP
from pycti import OpenCTIApiClient
import json

class TIPFeedManager:
    """Manage threat intelligence feed ingestion across platform components."""

    def __init__(self, misp_url, misp_key, opencti_url, opencti_token):
        self.misp = PyMISP(misp_url, misp_key, ssl=False)
        self.opencti = OpenCTIApiClient(opencti_url, opencti_token)

    def configure_osint_feeds(self):
        """Enable default OSINT feeds in MISP."""
        osint_feeds = [
            {"name": "CIRCL OSINT", "id": 1},
            {"name": "Botvrij.eu", "id": 2},
            {"name": "abuse.ch URLhaus", "id": 5},
            {"name": "abuse.ch Feodo Tracker", "id": 6},
        ]
        for feed in osint_feeds:
            try:
                self.misp.enable_feed(feed["id"])
                self.misp.fetch_feed(feed["id"])
                print(f"[+] Enabled feed: {feed['name']}")
            except Exception as e:
                print(f"[-] Failed: {feed['name']}: {e}")

    def configure_opencti_connectors(self):
        """List and verify OpenCTI connector status."""
        connectors = self.opencti.connector.list()
        for conn in connectors:
            print(
                f"  Connector: {conn['name']} - "
                f"Active: {conn['active']} - "
                f"Type: {conn['connector_type']}"
            )

    def sync_misp_to_opencti(self):
        """Verify MISP-OpenCTI sync is operational."""
        # OpenCTI MISP connector handles this automatically
        # Check connector status
        connectors = self.opencti.connector.list()
        misp_connector = [
            c for c in connectors if "misp" in c["name"].lower()
        ]
        if misp_connector:
            print(f"[+] MISP connector active: {misp_connector[0]['active']}")
        else:
            print("[-] MISP connector not found - configure in Docker Compose")

ステップ 3：Cortex でエンリッチメントパイプラインを構築する

import requests

class CortexEnrichment:
    """Integrate Cortex analyzers for automated enrichment."""

    def __init__(self, cortex_url, cortex_key):
        self.url = cortex_url
        self.headers = {"Authorization": f"Bearer {cortex_key}"}

    def list_analyzers(self):
        """List available Cortex analyzers."""
        resp = requests.get(
            f"{self.url}/api/analyzer",
            headers=self.headers,
            timeout=30,
        )
        if resp.status_code == 200:
            analyzers = resp.json()
            for a in analyzers:
                print(f"  {a['name']}: {a.get('description', '')[:60]}")
            return analyzers
        return []

    def analyze_observable(self, observable_type, observable_value, analyzer_id):
        """Submit an observable for analysis."""
        job = {
            "data": observable_value,
            "dataType": observable_type,
            "tlp": 2,
            "message": "TIP automated enrichment",
        }
        resp = requests.post(
            f"{self.url}/api/analyzer/{analyzer_id}/run",
            json=job,
            headers=self.headers,
            timeout=30,
        )
        if resp.status_code == 200:
            return resp.json()
        return None

    def get_job_report(self, job_id):
        """Get the report for a completed analysis job."""
        resp = requests.get(
            f"{self.url}/api/job/{job_id}/report",
            headers=self.headers,
            timeout=60,
        )
        if resp.status_code == 200:
            return resp.json()
        return None

ステップ 4：アナリストダッシュボードのメトリクスを実装する

class TIPMetrics:
    """Collect platform metrics for analyst dashboards."""

    def __init__(self, misp, opencti):
        self.misp = misp
        self.opencti = opencti

    def get_platform_stats(self):
        """Collect statistics across all platform components."""
        stats = {}

        # MISP stats
        misp_stats = self.misp.get_server_statistics()
        stats["misp"] = {
            "total_events": misp_stats.get("event_count", 0),
            "total_attributes": misp_stats.get("attribute_count", 0),
            "active_feeds": len([
                f for f in self.misp.feeds()
                if f.get("Feed", {}).get("enabled")
            ]),
        }

        # OpenCTI stats via GraphQL
        stats["opencti"] = {
            "total_indicators": self.opencti.indicator.list(
                first=0, withPagination=True
            ).get("pagination", {}).get("globalCount", 0),
            "total_reports": self.opencti.report.list(
                first=0, withPagination=True
            ).get("pagination", {}).get("globalCount", 0),
        }

        return stats

検証基準

• すべてのプラットフォームコンポーネント（MISP、OpenCTI、TheHive、Cortex）がデプロイされ、アクセス可能である
• MISP と OpenCTI の双方向同期が稼働している
• 3 つ以上の OSINT フィードからデータを取り込んでいる
• Cortex アナライザーが設定され、エンリッチメント結果を返している
• プラットフォームのメトリクスダッシュボードにリアルタイム統計が表示されている
• 脅威インテリジェンスの共有に使用する STIX/TAXII エクスポートが機能している

参考資料

• OpenCTI ドキュメント
• MISP プロジェクト
• TheHive プロジェクト
• Cortex ドキュメント
• MISP-OpenCTI 統合