IOCエンリッチメント自動化

IOCエンリッチメント自動化 は何をしますか？

使用タイミング

以下のような場面でこのスキルを使用します：

• アナリストにルーティングする前に、SIEMアラートに脅威インテリジェンスコンテキストを自動的に付与するSOARプレイブックを構築する場合
• フィッシングメール報告から抽出したIOCを一括エンリッチメントするPythonパイプラインを作成する場合
• VT、Shodan、MISPのデータでアラートコンテキストを事前入力し、アナリストの平均トリアージ時間（MTTT）を短縮する場合

このスキルを、人間のレビューなしで完全に自動化されたブロック判断に使用しないでください — エンリッチメントの自動化は意思決定を支援するものであり、影響の大きいアクションを自律的に実行するものではありません。

前提条件

• SOARプラットフォーム（Cortex XSOAR、Splunk SOAR、Tines、n8n）またはPython 3.9以上の環境
• APIキー：VirusTotal、AbuseIPDB、Shodan、および最低1つのTIP（MISPまたはOpenCTI）
• アラート消費のためのSIEM統合エンドポイント
• APIごとに文書化されたレート制限予算（VT：無料4回/分、エンタープライズ500回/分）

ワークフロー

ステップ1：エンリッチメントパイプラインアーキテクチャの設計

IOCタイプごとにエンリッチメントフローを定義します：

SIEMアラート→IOC抽出→タイプ分類→エンリッチメント関数へルーティング
  IPアドレス→AbuseIPDB + Shodan + VirusTotal IP + MISP
  ドメイン→VirusTotalドメイン + PassiveTotal + Shodan + MISP
  URL→URLScan.io + VirusTotal URL + Google Safe Browse
  ファイルハッシュ→VirusTotalファイル + MalwareBazaar + MISP
→結果集約→信頼度スコア計算→アラート更新→アナリスト通知

ステップ2：Pythonエンリッチメント関数の実装

import requests
import time
from dataclasses import dataclass, field
from typing import Optional

RATE_LIMIT_DELAY = 0.25  # VT無料枠：4リクエスト/秒

@dataclass
class EnrichmentResult:
    ioc_value: str
    ioc_type: str
    vt_malicious: int = 0
    vt_total: int = 0
    abuse_confidence: int = 0
    shodan_ports: list = field(default_factory=list)
    misp_events: list = field(default_factory=list)
    confidence_score: int = 0

def enrich_ip(ip: str, vt_key: str, abuse_key: str, shodan_key: str) -> EnrichmentResult:
    result = EnrichmentResult(ip, "ip")

    # VirusTrustal IPルックアップ
    vt_resp = requests.get(
        f"https://www.virustotal.com/api/v3/ip_addresses/{ip}",
        headers={"x-apikey": vt_key}
    )
    if vt_resp.status_code == 200:
        stats = vt_resp.json()["data"]["attributes"]["last_analysis_stats"]
        result.vt_malicious = stats.get("malicious", 0)
        result.vt_total = sum(stats.values())

    time.sleep(RATE_LIMIT_DELAY)

    # AbuseIPDB
    abuse_resp = requests.get(
        "https://api.abuseipdb.com/api/v2/check",
        headers={"Key": abuse_key, "Accept": "application/json"},
        params={"ipAddress": ip, "maxAgeInDays": 90}
    )
    if abuse_resp.status_code == 200:
        result.abuse_confidence = abuse_resp.json()["data"]["abuseConfidenceScore"]

    # 複合信頼度スコアの計算
    result.confidence_score = min(
        (result.vt_malicious / max(result.vt_total, 1)) * 60 +
        (result.abuse_confidence / 100) * 40, 100
    )

    return result

def enrich_hash(sha256: str, vt_key: str) -> EnrichmentResult:
    result = EnrichmentResult(sha256, "sha256")
    vt_resp = requests.get(
        f"https://www.virustotal.com/api/v3/files/{sha256}",
        headers={"x-apikey": vt_key}
    )
    if vt_resp.status_code == 200:
        stats = vt_resp.json()["data"]["attributes"]["last_analysis_stats"]
        result.vt_malicious = stats.get("malicious", 0)
        result.vt_total = sum(stats.values())
        result.confidence_score = int((result.vt_malicious / max(result.vt_total, 1)) * 100)
    return result

ステップ3：SOARプレイブックの構築（Cortex XSOAR）

Cortex XSOARでエンリッチメントプレイブックを作成します：

1. トリガー：SIEMでアラートが作成されたとき（Webhookまたはポーリング経由）
2. IOC抽出：「Extract Indicators」タスクを使用し、IP、ドメイン、URL、ハッシュの正規表現パターンで抽出
3. 並列エンリッチメント：複数のエンリッチメントタスクに同時にファンアウト
4. VTエンリッチメント：!vt-file-scan または !vt-ip-scan コマンドを呼び出し
5. AbuseIPDBチェック：!abuseipdb-check-ip コマンドを呼び出し
6. MISPルックアップ：!misp-search を呼び出し、相互参照
7. スコア集約：複合スコアを計算するPython変換タスク
8. 条件付きルーティング：スコアが70以上→高優先度キュー、40～69→中優先度、40未満→メモ付きで自動クローズ
9. アラートエンリッチメント：エンリッチメント結果をアラートコンテキストに書き込み、アナリストが表示できるようにする

ステップ4：レート制限と障害への対処

import time
from functools import wraps

def rate_limited(max_per_second):
    min_interval = 1.0 / max_per_second
    def decorator(func):
        last_called = [0.0]
        @wraps(func)
        def wrapper(*args, **kwargs):
            elapsed = time.time() - last_called[0]
            wait = min_interval - elapsed
            if wait > 0:
                time.sleep(wait)
            result = func(*args, **kwargs)
            last_called[0] = time.time()
            return result
        return wrapper
    return decorator

def retry_on_429(max_retries=3):
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            for attempt in range(max_retries):
                response = func(*args, **kwargs)
                if response.status_code == 429:
                    retry_after = int(response.headers.get("Retry-After", 60))
                    time.sleep(retry_after)
                else:
                    return response
        return wrapper
    return decorator

ステップ5：メトリクスとチューニング

パイプラインのパフォーマンスを週単位で追跡します：

• エンリッチメントレイテンシ：目標：アラートトリガーからエンリッチメント出力まで30秒未満
• API成功率：目標：99%超（レート制限や停止イベントを特定）
• 真陽性率：自動化された信頼度スコアに対するアナリストのオーバーライドを追跡
• コスト：予算に対するAPI呼び出し量を追跡（VTエンタープライズ：100万ルックアップあたりXドル）

主要概念

用語	定義
SOAR	Security Orchestration, Automation, and Response — セキュリティワークフローを自動化し、異なるツールを統合するプラットフォーム
エンリッチメントプレイブック	未加工のセキュリティイベントに状況インテリジェンスを追加する自動化ワークフローシーケンス
レート制限	APIプロバイダーによるリクエスト頻度の制限（例：VT無料枠：4リクエスト/分）；パイプラインはこれらの制限を尊重する必要がある
複合信頼度スコア	重み付けされた式を使用して複数のエンリッチメントソースからのシグナルを集約した単一スコア
ファンアウトパターン	複数のエンリッチメントクエリを同時に並列実行し、全体のエンリッチメントレイテンシを最小化するパターン

ツールとシステム

• Cortex XSOAR（Palo Alto）：VT、MISP、Shodan、AbuseIPDBを含む700以上のマーケットプレイス統合を備えたエンタープライズSOAR
• Splunk SOAR（Phantom）：Pythonベースのプレイブックを備えたSOARプラットフォーム；ネイティブSplunk SIEM統合
• Tines：Webhook駆動の自動化を備えたノーコードSOARプラットフォーム；小規模チームにコスト効率が良い
• TheHive + Cortex：オープンソースのIR/エンリッチメントプラットフォーム；Cortexアナライザーを介した観測可能なエンリッチメント

よくある落とし穴

• エンリッチメントレイテンシによるブロッキング：エンリッチメントに5分以上かかると、アナリストはエンリッチメントされていないアラートの処理を開始し、目的が達成されなくなります。タイムアウト制限を設定し、部分的な結果を提供してください。
• キャッシュなし：同じIOCに50回クエリを実行すると、不要なAPIコストが発生します。デフォルトで24時間、エンリッチメント結果をキャッシュしてください。
• API障害を無視する：失敗したエンリッチメント呼び出しはログに記録し、フォールバックロジックをトリガーする必要があります。サイレントに空の結果を生成し、クリーンなIOCとして表示させてはいけません。
• エンリッチメントスコアのみでブロックを自動化する：複合スコアには誤検出が含まれます。共有インフラに対するブロック判断には、人間の確認が必要です。