automating-ioc-enrichment

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 Alert → Extract IOCs → Classify Type → Route to enrichment functions
  IP Address → AbuseIPDB + Shodan + VirusTotal IP + MISP
  Domain → VirusTotal Domain + PassiveTotal + Shodan + MISP
  URL → URLScan.io + VirusTotal URL + Google Safe Browse
  File Hash → VirusTotal Files + MalwareBazaar + MISP
→ Aggregate results → Calculate confidence score → Update alert → Notify analyst

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

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

RATE_LIMIT_DELAY = 0.25  # 4 requests/second for VT free tier

@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")

    # VirusTotal IP lookup
    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"]

    # Calculate composite confidence score
    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 の抽出：IP、ドメイン、URL、ハッシュの正規表現パターンを使用する「Extract Indicators」タスク
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 Enterprise：100 万回の検索あたり $X）

主要な概念

用語	定義
SOAR	Security Orchestration, Automation, and Response — セキュリティワークフローを自動化し、さまざまなツールを統合するプラットフォーム
エンリッチメントプレイブック	生のセキュリティイベントにコンテキスト情報を追加する自動ワークフローの一連の処理
レート制限	API プロバイダーが定めるリクエスト頻度の制限（例：VT の無料プランでは 1 分あたり 4 リクエスト）。パイプラインはこれらの制限を守る必要があります
複合信頼度スコア	重み付けされた計算式を使用して、複数のエンリッチメントソースからのシグナルを集約した単一のスコア
ファンアウトパターン	複数のエンリッチメントクエリを同時に並列実行し、エンリッチメント全体のレイテンシを最小限に抑える方法

ツールとシステム

• Cortex XSOAR（Palo Alto）：VT、MISP、Shodan、AbuseIPDB など、700 以上のマーケットプレイス統合機能を備えたエンタープライズ向け SOAR
• Splunk SOAR（Phantom）：Python ベースのプレイブックを備えた SOAR プラットフォーム。Splunk SIEM とネイティブに統合
• Tines：Webhook 駆動の自動化を備えたノーコード SOAR プラットフォーム。小規模チームにとって費用対効果が高い
• TheHive + Cortex：Cortex アナライザーによるオブザーバブルのエンリッチメントが可能なオープンソース IR／エンリッチメントプラットフォーム

よくある落とし穴

• エンリッチメントの遅延による処理の停滞：エンリッチメントに5分以上かかると、アナリストはエンリッチメントされていないアラートの調査を始めるため、本来の目的が損なわれます。タイムアウトを設定し、部分的な結果を提供してください。
• キャッシュがない：同じ IOC を50回クエリすると、不要な API コストが発生します。デフォルトでエンリッチメント結果を24時間キャッシュしてください。
• API の失敗を黙って無視する：エンリッチメント呼び出しの失敗はログに記録し、フォールバック処理を実行してください。クリーンな IOC に見える空の結果を、黙って返してはいけません。
• エンリッチメントスコアだけでブロックを自動化する：複合スコアには誤検知が含まれるため、共有インフラストラクチャをブロックする判断には人間の確認を必須にしてください。