OpenCTIを使用したIOCエンリッチメントパイプラインの構築 は何をしますか？
概要
OpenCTIは、STIX 2.1をネイティブデータモデルとして採用した、サイバー脅威インテリジェンス知識を管理するためのオープンソースプラットフォームです。このスキルでは、OpenCTIのコネクタエコシステムを使用して、VirusTotal、Shodan、AbuseIPDB、GreyNoiseなどのソースからのコンテキストでインジケータをエンリッチする、自動化されたIOCエンリッチメントパイプラインの構築について説明します。パイプラインは、新しく取り込まれたインジケータを自動的にエンリッチし、既知の脅威アクターやキャンペーンと関連付け、アナリストの優先順位付けのためにスコアリングします。
使用するタイミング
- 環境内でOpenCTIを使用したIOCエンリッチメントパイプラインの構築または設定を展開する場合
- コンプライアンス要件に沿ったセキュリティコントロールを確立する場合
- このドメインのセキュリティアーキテクチャを構築または改善する場合
- この実装を必要とするセキュリティ評価を実施する場合
前提条件
- OpenCTIデプロイメント用のDockerとDocker Compose
pyctiライブラリを使用したPython 3.9+- エンリッチメントサービスのAPIキー：VirusTotal、Shodan、AbuseIPDB、GreyNoise
- STIX 2.1データモデルとリレーションシップの理解
- OpenCTIバックエンド用のElasticSearchまたはOpenSearch
- コネクタメッセージング用のRabbitMQまたはRedis
主要な概念
OpenCTIアーキテクチャ
OpenCTIは、GraphQL APIフロントエンドを使用し、バックエンドにElasticSearch、コネクタ通信にRedis/RabbitMQを採用しています。データはリレーションシップを持つSTIX 2.1オブジェクトとしてネイティブに保存されます。コネクタは、外部インポート（フィード取り込み）、内部インポート（ファイル解析）、内部エンリッチメント（コンテキスト追加）、ストリーム（リアルタイムエクスポート）に分類されます。
エンリッチメントコネクタモデル
内部エンリッチメントコネクタは、新しいオブザーバブルが作成されると自動的にトリガーされるか、アナリストが手動でトリガーします。各コネクタはSTIXオブジェクトを受信し、外部サービスにクエリを実行し、元のオブザーバブルに追加のコンテキスト、ラベル、リレーションシップを追加するSTIX 2.1バンドルを返します。
信頼度スコアリング
OpenCTIはインジケータに0～100の信頼度スケールを使用します。エンリッチメントコネクタは、外部検証（VirusTotalの検出率、Shodanの露出データ、AbuseIPDBのレポート数、GreyNoiseの分類結果）に基づいて信頼度スコアを更新できます。
ワークフロー
ステップ1：Docker Composeを使用したOpenCTIのデプロイ
# docker-compose.yml（主要サービス）
version: '3'
services:
opencti:
image: opencti/platform:6.4.4
environment:
- APP__PORT=8080
- [email protected]
- APP__ADMIN__PASSWORD=ChangeMeNow
- APP__ADMIN__TOKEN=your-admin-token-uuid
- ELASTICSEARCH__URL=http://elasticsearch:9200
- MINIO__ENDPOINT=minio
- RABBITMQ__HOSTNAME=rabbitmq
ports:
- "8080:8080"
depends_on:
- elasticsearch
- minio
- rabbitmq
- redis
connector-virustotal:
image: opencti/connector-virustotal:6.4.4
environment:
- OPENCTI_URL=http://opencti:8080
- OPENCTI_TOKEN=your-admin-token-uuid
- CONNECTOR_ID=connector-virustotal-id
- CONNECTOR_NAME=VirusTotal
- CONNECTOR_SCOPE=StixFile,Artifact,IPv4-Addr,Domain-Name,Url
- CONNECTOR_AUTO=true
- VIRUSTOTAL_TOKEN=your-vt-api-key
- VIRUSTOTAL_MAX_TLP=TLP:AMBER
connector-shodan:
image: opencti/connector-shodan:6.4.4
environment:
- OPENCTI_URL=http://opencti:8080
- OPENCTI_TOKEN=your-admin-token-uuid
- CONNECTOR_ID=connector-shodan-id
- CONNECTOR_NAME=Shodan
- CONNECTOR_SCOPE=IPv4-Addr
- CONNECTOR_AUTO=true
- SHODAN_TOKEN=your-shodan-api-key
- SHODAN_MAX_TLP=TLP:AMBER
connector-abuseipdb:
image: opencti/connector-abuseipdb:6.4.4
environment:
- OPENCTI_URL=http://opencti:8080
- OPENCTI_TOKEN=your-admin-token-uuid
- CONNECTOR_ID=connector-abuseipdb-id
- CONNECTOR_NAME=AbuseIPDB
- CONNECTOR_SCOPE=IPv4-Addr
- CONNECTOR_AUTO=true
- ABUSEIPDB_API_KEY=your-abuseipdb-key
ステップ2：カスタムエンリッチメントコネクタの構築
import os
from pycti import OpenCTIConnectorHelper, get_config_variable
from stix2 import (
Bundle, Indicator, Note, Relationship,
IPv4Address, DomainName
)
import requests
class CustomEnrichmentConnector:
def __init__(self):
config = {
"opencti": {
"url": os.environ.get("OPENCTI_URL"),
"token": os.environ.get("OPENCTI_TOKEN"),
},
"connector": {
"id": os.environ.get("CONNECTOR_ID"),
"name": "CustomEnrichment",
"scope": "IPv4-Addr,Domain-Name,Url",
"auto": True,
"type": "INTERNAL_ENRICHMENT",
},
}
self.helper = OpenCTIConnectorHelper(config)
self.helper.listen(self._process_message)
def _process_message(self, data):
entity_id = data["entity_id"]
stix_object = self.helper.api.stix_cyber_observable.read(id=entity_id)
if not stix_object:
return "オブザーバブルが見つかりません"
observable_type = stix_object["entity_type"]
observable_value = stix_object.get("value", "")
enrichment_results = []
if observable_type == "IPv4-Addr":
enrichment_results = self._enrich_ip(observable_value, entity_id)
elif observable_type == "Domain-Name":
enrichment_results = self._enrich_domain(observable_value, entity_id)
if enrichment_results:
bundle = Bundle(objects=enrichment_results, allow_custom=True)
self.helper.send_stix2_bundle(bundle.serialize())
return "エンリッチメントが完了しました"
def _enrich_ip(self, ip_address, entity_id):
"""GreyNoise、AbuseIPDBのコンテキストでIPアドレスをエンリッチします。"""
objects = []
# GreyNoise Community API
try:
gn_response = requests.get(
f"https://api.greynoise.io/v3/community/{ip_address}",
headers={"key": os.environ.get("GREYNOISE_API_KEY")},
timeout=30,
)
if gn_response.status_code == 200:
gn_data = gn_response.json()
classification = gn_data.get("classification", "unknown")
noise = gn_data.get("noise", False)
riot = gn_data.get("riot", False)
note_content = (
f"## GreyNoise Enrichment\n"
f"- 分類: {classification}\n"
f"- インターネットノイズ: {noise}\n"
f"- RIOT（良性サービス）: {riot}\n"
f"- 名前: {gn_data.get('name', 'N/A')}\n"
f"- 最終確認日: {gn_data.get('last_seen', 'N/A')}"
)
note = Note(
content=note_content,
object_refs=[entity_id],
abstract=f"GreyNoise: {classification}",
allow_custom=True,
)
objects.append(note)
# 分類に基づいてラベルを追加
if classification == "malicious":
self.helper.api.stix_cyber_observable.add_label(
id=entity_id, label_name="greynoise:malicious"
)
elif riot:
self.helper.api.stix_cyber_observable.add_label(
id=entity_id, label_name="greynoise:benign-service"
)
except Exception as e:
self.helper.log_error(f"GreyNoiseエンリッチメントが失敗しました: {e}")
return objects
def _enrich_domain(self, domain, entity_id):
"""WHOISおよびDNSコンテキストでドメインをエンリッチします。"""
objects = []
try:
# ドメインエンリッチメントにSecurityTrails APIを使用
st_response = requests.get(
f"https://api.securitytrails.com/v1/domain/{domain}",
headers={"APIKEY": os.environ.get("SECURITYTRAILS_API_KEY")},
timeout=30,
)
if st_response.status_code == 200:
st_data = st_response.json()
current_dns = st_data.get("current_dns", {})
a_records = [
r.get("ip") for r in current_dns.get("a", {}).get("values", [])
]
note_content = (
f"## SecurityTrails Enrichment\n"
f"- Aレコード: {', '.join(a_records)}\n"
f"- Alexaランク: {st_data.get('alexa_rank', 'N/A')}\n"
f"- ホスト名: {st_data.get('hostname', 'N/A')}"
)
note = Note(
content=note_content,
object_refs=[entity_id],
abstract=f"SecurityTrails: {domain}",
allow_custom=True,
)
objects.append(note)
except Exception as e:
self.helper.log_error(f"SecurityTrailsエンリッチメントが失敗しました: {e}")
return objects
if __name__ == "__main__":
connector = CustomEnrichmentConnector()
次のステップ
- OpenCTIデプロイメントの監視とログ記録を構成する
- スコアリングのしきい値と自動応答ルールを定義する
- MITRE ATT&CKマッピングとの相関ルールを実装する
- エンリッチメントパイプラインをダッシュボードとレポートツールと統合する
- OpenCTI APIを使用したカスタムダッシュボードを構築する