このガイドでは、外部インフラを使用せずに Push 配信（Webhook 2.0） を使って Axur Platform のフィードを Elastic Security に取り込む方法を説明します。
本手順では、Elastic Stack のネイティブ機能（インデックス、Ingest Pipeline、Watcher）だけを使用して、データをアイテム単位のドキュメントに分割します。
重要: これはあくまで一例の統合手法です。インデックス設計やデータ変換は、お使いの環境や要件に合わせてカスタマイズ可能です。以下の手順は、Axur フィードから直接データを取り込み、エンティティごとに特定のインデックス（例：チケットや認証情報）に分割することを目的として設計されています。
注記: このチュートリアルでは、すでに Axur Platform へのアクセス権と必要な API キーをお持ちであることを前提としています。
前提条件
Elastic Stack（Elasticsearch + Kibana） バージョン 8.x
Kibana への管理者アクセス権
Axur インフラから Elastic クラスターへの ネットワークアクセス
対象インデックスへの書き込み権限を持つ Elastic API キー
コストおよび課金に関する考慮事項
重要: この統合では Elastic の組み込み REST API と Watcher を使用します。Axur 側での課金は発生しません。
主なコスト要因:
使用する Elastic ライセンス／機能（Watcher は適切なライセンスが必要）
取り込んだイベントおよび派生インデックスのストレージ
ホスティング環境に応じたネットワークの送受信量
コストを抑えるためのヒント:
初期はより絞り込んだフィードフィルターを使用してデータ量を制限
ここで作成するインデックスに ILM（Index Lifecycle Management）ポリシー を適用
取り込み状況を監視し、必要に応じてフィードフィルターを調整
ヒント：このチュートリアルで使用されている Elastic 用語の用語集は、ドキュメントの末尾に掲載されています。
1）Axur API キーを作成する
API キーを作成し、後の手順で使用できるよう値を控えておきます。
API キーの生成方法:
Axur Platform で API Keys 設定に移動
新しい API キーを作成し、API キーの値を安全にコピー
注記:フィードは、API キーのユーザーがアクセス権を持つデータのみを返します。
スクリーンショット:
この時点（ステップ 2～6）以降のすべての設定は、Elastic/Kibana コンソール内で実施 します。
2）Elastic で：入力用インデックスを作成する
HTTP POST 経由で Axur データを直接受信するためのインデックスを作成します。
動的フィールドを扱うために、flattened マッピングを使用してください。
PUT axur-feed-input
{
"mappings": {
"dynamic": true,
"dynamic_templates": [
{
"collectionData_flattened": {
"path_match": "collectionData.*",
"mapping": { "type": "flattened" }
}
}
]
}
}
スクリーンショット:
3）Elastic で：Ingest Pipeline を作成する
入力インデックスに届いた各ドキュメントに、取り込みタイムスタンプ を追加するパイプラインを作成します。
PUT _ingest/pipeline/axur-feed-input-pipeline
{
"processors": [
{ "set": { "field": "ingested_at", "value": "{{_ingest.timestamp}}" } }
]
}
重要: Axur フィードの Push URL には、必ず ?pipeline=axur-feed-input-pipeline を付加して、このパイプラインを適用してください。
スクリーンショット:
4）Elastic で：出力用インデックスを作成する
これらのインデックスは、エンティティごとに分割されたアイテム単位のドキュメント を格納します。
チケット用インデックス:
PUT axur-feed-ticket_splitted
{
"mappings": {
"dynamic": true,
"properties": {
"item": { "type": "flattened" },
"feedData": { "type": "flattened" }
}
}
}
認証情報（detections）用インデックス:
PUT axur-feed-credential_splitted
{
"mappings": {
"dynamic": true,
"properties": {
"item": { "type": "flattened" },
"feedData": { "type": "flattened" }
}
}
}
スクリーンショット:
5）Elastic で：チェックポイント用インデックスを作成する
最後に処理したタイムスタンプを保存し、データの再処理を防ぐ ためのインデックスです。
PUT axur-feed-checkpoint
{
"mappings": {
"properties": {
"last_ts": { "type": "date" }
}
}
}
スクリーンショット:
6）Elastic で：アイテムを分割する Watcher を作成する
この Watcher は、新規の入力ドキュメントのみを処理 し、分割したアイテムを出力インデックスに格納するとともに、チェックポイントを更新します。
PUT _watcher/watch/axur_split_batch
{
"trigger": { "schedule": { "interval": "1m" } },
"input": {
"chain": {
"inputs": [
{
"cp": {
"search": {
"request": {
"indices": ["axur-feed-checkpoint"],
"body": {
"size": 1,
"query": { "ids": { "values": ["checkpoint"] } }
}
}
}
}
},
{
"src": {
"search": {
"request": {
"indices": ["axur-feed-input"],
"body": {
"size": 200,
"query": {
"range": {
"ingested_at": {
"gt": "{{#ctx.payload.cp.hits.hits}}{{ctx.payload.cp.hits.hits.0._source.last_ts}}{{/ctx.payload.cp.hits.hits}}{{^ctx.payload.cp.hits.hits}}1970-01-01T00:00:00Z{{/ctx.payload.cp.hits.hits}}"
}
}
},
"sort": [{ "ingested_at": "asc" }]
}
}
}
}
}
]
}
},
"condition": {
"script": {
"lang": "painless",
"source": "def h = ctx?.payload?.src?.hits; if (h == null) return false; def t = h.total; if (t == null) return false; if (t instanceof Map) return (t.value != null && t.value > 0); return (t > 0);"
}
},
"transform": {
"script": {
"lang": "painless",
"source": """
def tics = new ArrayList();
def creds = new ArrayList();
def maxTs = null;
for (def hit : ctx.payload.src.hits.hits) {
def src = hit._source;
if (src == null) continue;
def ts = src.containsKey('ingested_at') ? src.ingested_at : null;
if (ts != null) {
if (maxTs == null) { maxTs = ts; }
else if (java.time.ZonedDateTime.parse(ts).isAfter(java.time.ZonedDateTime.parse(maxTs))) { maxTs = ts; }
}
def feed = src.containsKey('feedData') ? src.feedData : null;
if (!src.containsKey('collectionData')) continue;
def cd = src.collectionData;
if (cd != null && cd.containsKey('tickets') && cd.tickets != null) {
for (def it : cd.tickets) {
if (it == null) continue;
tics.add(['feedData': feed, 'item': it]);
}
}
if (cd != null && cd.containsKey('detections') && cd.detections != null) {
for (def it2 : cd.detections) {
if (it2 == null) continue;
creds.add(['feedData': feed, 'item': it2]);
}
}
}
return ['tics': tics, 'creds': creds, 'next_checkpoint': maxTs];
"""
}
},
"actions": {
"index_tics": {
"foreach": "ctx.payload.tics",
"max_iterations": 500,
"transform": { "script": { "lang": "painless", "source": "return ctx.payload._value;" } },
"index": { "index": "axur-feed-ticket_splitted", "execution_time_field": "indexed_at", "refresh": "wait_for" }
},
"index_creds": {
"foreach": "ctx.payload.creds",
"max_iterations": 500,
"transform": { "script": { "lang": "painless", "source": "return ctx.payload._value;" } },
"index": { "index": "axur-feed-credential_splitted", "execution_time_field": "indexed_at", "refresh": "wait_for" }
},
"update_checkpoint": {
"condition": { "script": { "lang": "painless", "source": "return ctx.payload.next_checkpoint != null;" } },
"transform": { "script": { "lang": "painless", "source": "return ['last_ts': ctx.payload.next_checkpoint];" } },
"index": { "index": "axur-feed-checkpoint", "doc_id": "checkpoint", "execution_time_field": "updated_at", "refresh": "wait_for" }
}
}
}
スクリーンショット:
7）Axur PUSH フィードのエンドポイントを設定する
Axur フィードを Elastic の入力インデックス に向け、パイプラインを適用します。
エンドポイント URL:
https://YOUR_ELASTIC_CLUSTER/axur-feed-input/_doc?pipeline=axur-feed-input-pipeline追加するヘッダー:
Authorization: ApiKey <YOUR_ENCODED_ELASTIC_API_KEY>（Elastic API Keys UI から取得した Encoded 形式 を使用）
Secret Key: フィード内で安全なシークレットを設定 Axur はリクエストを HMAC 署名 で送信 必要に応じて Elastic 側で検証可能
スクリーンショット:
検証
設定後、入力インデックスにテストドキュメントを送信 して動作を確認します。
（Kibana → Dev Tools を使用）
POST axur-feed-input/_doc?pipeline=axur-feed-input-pipeline
{
"feedData": { "feedType": "ticket", "customerKey": "TEST" },
"collectionData": {
"tickets": [ { "ticket": { "id": "TCK123", "type": "leak" } } ],
"detections": [ { "credential": { "user": "alice" } } ]
}
}
約 1 分待って、以下を確認します：
GET axur-feed-ticket_splitted/_search
GET axur-feed-credential_splitted/_search
GET axur-feed-checkpoint/_doc/checkpoint
各出力インデックスにアイテム単位のドキュメントが作成され、チェックポイントには last_ts が更新されていることを確認します。
その後の実行では、同じ入力ドキュメントが再処理されないことを確認してください。
スクリーンショット:
トラブルシューティング
出力インデックスにドキュメントがない場合:
Watcher が有効でエラーが発生していないか確認
Watcher > History をチェック
401 / 403 インデックスエラー:
Elastic API キーの権限（書き込み、インデックス作成）を確認
重複が発生する場合:
チェックポイントドキュメントが更新されているか確認
クラスターの時刻同期を検証
データが全く取り込まれない場合:
Axur から Push URL への到達性を確認
パス内の pipeline パラメータが正しいか確認
完了：データの確認場所とアラート例
チケット:
axur-feed-ticket_splittedインデックスに格納認証情報（detections）:
axur-feed-credential_splittedインデックスに格納
例：Kibana → Discover でのクエリ
GET axur-feed-ticket_splitted/_search
{
"query": { "term": { "item.detection.status": "open" } },
"size": 20,
"sort": [{ "indexed_at": { "order": "desc" } }]
}
GET axur-feed-credential_splitted/_search
{
"query": { "exists": { "field": "item.credential.user" } },
"size": 20,
"sort": [{ "indexed_at": { "order": "desc" } }]
}
検出ルールの作成 (Elastic Security → Rules → Create rule → Custom query)
インデックスパターン:
axur-feed-credential_splittedカスタムクエリ（KQL）:
item.credential.user:* and item.access.domain:"example.com"
スケジュール: 5 分ごと
重大度（Severity）: Medium
アクション: 通知を追加（メール、Slack など）
Elastic API キーの生成方法
Axur Push リクエストを認証するために、Elastic で API キーを作成します。
手順（Kibana 内）:
Stack Management → Security → API Keys → Create API key に移動
名前:
axur-feed-ingest権限:
Index privileges:
axur-feed-*（または作成した特定のインデックス）に対して write と create_index を付与
作成後、Encoded 形式 のキーをコピーして安全に保管
スクリーンショット:
用語集
Elasticsearch: Elastic Stack の分散型検索・分析エンジン
Kibana: Elasticsearch データの管理・可視化用 Web UI
Index: Elasticsearch 内のドキュメントの論理的な名前空間
Ingest Pipeline: ドキュメントをインデックス化する前に実行されるプロセッサ
Watcher: 定期的な分割などのアラート／自動化を実装する機能
ILM (Index Lifecycle Management): データ保持やパフォーマンスを管理するポリシー
API Key: Elasticsearch の認証メカニズム
Push (Webhook 2.0): Axur の配信モードで、イベントを直接指定したエンドポイントに送信
ご不明点があれば、いつでも [email protected] までお気軽にご連絡ください 😊