building-threat-intelligence-feed-integration

脅威インテリジェンスフィード統合の構築

使用するタイミング

次のような場合にこのスキルを使用します。

• SOC チームが、脅威インテリジェンスフィードを SIEM プラットフォームに自動取り込みする必要がある
• 複数の TI ソースを共通形式（STIX 2.1）に正規化する必要がある
• 検知システムが、ネットワークおよびエンドポイントのテレメトリに対して IOC をリアルタイムで照合する必要がある
• TI フィードの品質評価と重複排除プロセスを確立する必要がある

手動での IOC 検索には使用しないでください。単発のクエリには、専用のエンリッチメントツール（VirusTotal、AbuseIPDB）を使用してください。

前提条件

• フィードを集約するための MISP インスタンスまたは脅威インテリジェンスプラットフォーム（TIP）
• STIX/TAXII クライアントライブラリ（taxii2-client、stix2 Python パッケージ）
• TI フレームワークを設定済みの SIEM プラットフォーム（Splunk ES、Elastic Security、または Sentinel）
• 商用およびオープンソースフィード（AlienVault OTX、Abuse.ch、CISA AIS）の API キー
• フィード処理を自動化するための Python 3.8+

ワークフロー

手順 1：インテリジェンスソースを特定してカタログ化する

利用可能なフィードをタイプ、形式、更新頻度ごとに整理します。

フィードソース	形式	IOC タイプ	更新頻度	費用
AlienVault OTX	STIX/JSON	IP、ドメイン、ハッシュ、URL	リアルタイム	無料
Abuse.ch URLhaus	CSV/JSON	URL、ドメイン	5 分ごと	無料
Abuse.ch MalwareBazaar	JSON API	ファイルハッシュ	リアルタイム	無料
CISA AIS	STIX/TAXII 2.1	すべてのタイプ	毎日	無料（米国政府）
CrowdStrike Intel	STIX/JSON	すべてのタイプ＋攻撃者の TTP	リアルタイム	商用
Mandiant Advantage	STIX 2.1	すべてのタイプ＋レポート	リアルタイム	商用

手順 2：STIX/TAXII フィードを取り込む

TAXII 2.1 サーバーに接続して、インジケーターをダウンロードします。

from taxii2client.v21 import Server, Collection
from stix2 import parse

# Connect to TAXII server (example: CISA AIS)
server = Server(
    "https://taxii.cisa.gov/taxii2/",
    user="your_username",
    password="your_password"
)

# List available collections
for api_root in server.api_roots:
    print(f"API Root: {api_root.title}")
    for collection in api_root.collections:
        print(f"  Collection: {collection.title} (ID: {collection.id})")

# Fetch indicators from a collection
collection = Collection(
    "https://taxii.cisa.gov/taxii2/collections/COLLECTION_ID/",
    user="your_username",
    password="your_password"
)

# Get indicators added in last 24 hours
from datetime import datetime, timedelta
added_after = (datetime.utcnow() - timedelta(days=1)).strftime("%Y-%m-%dT%H:%M:%S.000Z")

response = collection.get_objects(added_after=added_after, type=["indicator"])
for obj in response.get("objects", []):
    indicator = parse(obj)
    print(f"Type: {indicator.type}")
    print(f"Pattern: {indicator.pattern}")
    print(f"Valid Until: {indicator.valid_until}")
    print(f"Confidence: {indicator.confidence}")
    print("---")

手順 3：オープンソースフィードを取り込む

Abuse.ch URLhaus フィード：

import requests
import csv
from io import StringIO

# Download URLhaus recent URLs
response = requests.get("https://urlhaus.abuse.ch/downloads/csv_recent/")
reader = csv.reader(StringIO(response.text), delimiter=',')

indicators = []
for row in reader:
    if row[0].startswith("#"):
        continue
    indicators.append({
        "id": row[0],
        "dateadded": row[1],
        "url": row[2],
        "url_status": row[3],
        "threat": row[5],
        "tags": row[6]
    })

print(f"Ingested {len(indicators)} URLs from URLhaus")

# Filter for active threats only
active = [i for i in indicators if i["url_status"] == "online"]
print(f"Active threats: {len(active)}")

AlienVault OTX Pulse フィード：

from OTXv2 import OTXv2, IndicatorTypes

otx = OTXv2("YOUR_OTX_API_KEY")

# Get subscribed pulses (last 24 hours)
pulses = otx.getall(modified_since="2024-03-14T00:00:00")

for pulse in pulses:
    print(f"Pulse: {pulse['name']}")
    print(f"Tags: {pulse['tags']}")
    for indicator in pulse["indicators"]:
        print(f"  IOC: {indicator['indicator']} ({indicator['type']})")

Abuse.ch Feodo Tracker（C2 IP）：

response = requests.get("https://feodotracker.abuse.ch/downloads/ipblocklist_recommended.json")
c2_data = response.json()

for entry in c2_data:
    print(f"IP: {entry['ip_address']}:{entry['port']}")
    print(f"Malware: {entry['malware']}")
    print(f"First Seen: {entry['first_seen']}")
    print(f"Last Online: {entry['last_online']}")

手順 4：正規化して重複を排除する

標準化のため、すべてのフィードを STIX 2.1 形式に変換します。

from stix2 import Indicator, Bundle
import hashlib

def create_stix_indicator(ioc_value, ioc_type, source, confidence=50):
    """Convert raw IOC to STIX 2.1 indicator"""
    pattern_map = {
        "ipv4": f"[ipv4-addr:value = '{ioc_value}']",
        "domain": f"[domain-name:value = '{ioc_value}']",
        "url": f"[url:value = '{ioc_value}']",
        "sha256": f"[file:hashes.'SHA-256' = '{ioc_value}']",
        "md5": f"[file:hashes.MD5 = '{ioc_value}']",
    }

    return Indicator(
        name=f"{ioc_type}: {ioc_value}",
        pattern=pattern_map[ioc_type],
        pattern_type="stix",
        valid_from="2024-03-15T00:00:00Z",
        confidence=confidence,
        labels=[source],
        custom_properties={"x_source_feed": source}
    )

# Deduplicate across sources
seen_iocs = set()
unique_indicators = []

for ioc in all_collected_iocs:
    ioc_hash = hashlib.sha256(f"{ioc['type']}:{ioc['value']}".encode()).hexdigest()
    if ioc_hash not in seen_iocs:
        seen_iocs.add(ioc_hash)
        unique_indicators.append(
            create_stix_indicator(ioc["value"], ioc["type"], ioc["source"])
        )

bundle = Bundle(objects=unique_indicators)
print(f"Unique indicators: {len(unique_indicators)}")

手順 5：SIEM の脅威インテリジェンスフレームワークに送信する

Splunk ES の脅威インテリジェンスに送信する：

import requests

splunk_url = "https://splunk.company.com:8089"
headers = {"Authorization": f"Bearer {splunk_token}"}

for indicator in unique_indicators:
    # Extract IOC value from STIX pattern
    ioc_value = indicator.pattern.split("'")[1]

    # Upload to Splunk ES threat intel collection
    data = {
        "ip": ioc_value,
        "description": indicator.name,
        "weight": indicator.confidence // 10,
        "threat_key": indicator.id,
        "source_feed": indicator.get("x_source_feed", "unknown")
    }

    requests.post(
        f"{splunk_url}/services/data/threat_intel/item/ip_intel",
        headers=headers, data=data,
        verify=not os.environ.get("SKIP_TLS_VERIFY", "").lower() == "true",  # Set SKIP_TLS_VERIFY=true for self-signed certs in lab environments
    )

一元管理のために MISP に送信する：

from pymisp import PyMISP, MISPEvent, MISPAttribute

misp = PyMISP("https://misp.company.com", "YOUR_MISP_API_KEY")

# Create event for feed batch
event = MISPEvent()
event.info = f"TI Feed Import - {datetime.now().strftime('%Y-%m-%d')}"
event.threat_level_id = 2  # Medium
event.analysis = 2  # Completed

# Add indicators as attributes
for ioc in unique_indicators:
    attr = MISPAttribute()
    attr.type = "ip-dst" if "ipv4" in ioc.pattern else "domain"
    attr.value = ioc.pattern.split("'")[1]
    attr.to_ids = True
    attr.comment = f"Source: {ioc.get('x_source_feed', 'mixed')}"
    event.add_attribute(**attr)

result = misp.add_event(event)
print(f"MISP Event created: {result['Event']['id']}")

手順 6：フィードの稼働状況と品質を監視する

フィードの有効性に関する指標を追跡します。

index=threat_intel sourcetype="threat_intel_manager"
| stats count AS total_iocs,
        dc(threat_key) AS unique_iocs,
        dc(source_feed) AS feed_count
  by source_feed
| join source_feed [
    search index=notable source="Threat Intelligence"
    | stats count AS matches by source_feed
  ]
| eval match_rate = round(matches / unique_iocs * 100, 2)
| sort - match_rate
| table source_feed, unique_iocs, matches, match_rate

主な概念

用語	定義
STIX 2.1	Structured Threat Information Expression — 脅威インテリジェンスオブジェクトの共有に使用する標準化 JSON 形式
TAXII	Trusted Automated eXchange of Indicator Information — REST API を介して STIX データを共有するための転送プロトコル
TIP	Threat Intelligence Platform — 脅威インテリジェンスを集約、スコアリング、配布する一元管理システム
IOC スコアリング	ソースの信頼性と裏付け情報に基づいて、インジケーターに信頼度を割り当てるプロセス
フィードの重複排除	複数のソースにわたる重複 IOC を削除しつつ、複数ソースからの帰属情報を保持すること
IOC の有効期限	古くなったインジケーターを削除するための TTL ポリシー（IP：30 日、ドメイン：90 日、ハッシュ：1 年）

ツールとシステム

• MISP：フィードの集約、相関分析、共有に対応したオープンソースの脅威インテリジェンスプラットフォーム
• AlienVault OTX：コミュニティパルスフィードを提供する無料の脅威インテリジェンス共有プラットフォーム
• Abuse.ch：無料の脅威フィード一式（URLhaus、MalwareBazaar、Feodo Tracker、ThreatFox）
• OpenCTI：STIX 2.1 のネイティブストレージに対応したオープンソースのサイバー脅威インテリジェンスプラットフォーム
• TAXII2 Client：インジケーターを自動取得するため、STIX/TAXII 2.1 サーバーへの接続に使用する Python ライブラリ

一般的なシナリオ

• 新しいフィードの導入：フィードの品質を評価し、フィールドを STIX にマッピングして、自動取り込みパイプラインを設定します
• 複数の SIEM への配信：MISP から正規化された IOC を Splunk、Elastic、Sentinel に同時に送信します
• 誤検知の削減：ソース数と経過時間に基づいて IOC をスコアリングし、古いインジケーターを自動的に期限切れにします
• フィード品質の監査：フィード間の検知一致率を比較し、最も価値の高いソースを特定します
• インシデント IOC の共有：調査で得られた IOC を STIX バンドルとしてパッケージ化し、TAXII 経由で ISAC と共有します

出力形式

THREAT INTEL FEED STATUS — Daily Report
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
Date:         2024-03-15
Total IOCs:   45,892 active indicators

Feed Health:
  Feed                  IOCs    Matches  Match Rate  Status
  Abuse.ch URLhaus      12,340  47       0.38%       HEALTHY
  AlienVault OTX        18,567  23       0.12%       HEALTHY
  Abuse.ch Feodo        1,203   12       1.00%       HEALTHY
  CISA AIS              8,945   8        0.09%       HEALTHY
  CrowdStrike Intel     4,837   31       0.64%       HEALTHY

Actions Today:
  New IOCs ingested:    1,247
  IOCs expired:         892
  Duplicates removed:   156
  SIEM matches:         121 notable events generated
  False positives:      3 (CDN IPs removed from feed)