クイックスタート

このガイドは CSIRT-Pro の基本操作を 6 ステップでたどります。 Pipeline の作成、ログの取り込み、検索、ケース作成、Playbook の設定までを順に試します。

一連の流れをまとまったデータで試したい場合

検索、可視化、AI による可視化、Playbook によるアラート発報、アラートの AI 分析までを、攻撃を混ぜたサンプルデータで通して体験するには、チュートリアル: 検知から AI 分析までを参照してください。

前提条件

CSIRT-Pro アカウントが作成済みであること
curl コマンドが利用可能な環境があること

Step 1: Pipeline を作成する

Pipeline はログを取り込む受け口です。ログソースの種類ごとに Pipeline を作成します。

Step 1: Pipeline 作成画面 図: Step 1 Pipeline の新規作成

UI から作成

サイドメニューから Pipeline を選択
「新規作成」 をクリック
Pipeline 名を入力（例: firewall_logs）
保存をクリック

パースルールの設定

ログの形式に応じてパースルールを設定します。抽出は検索時に行われます（schema-on-read 方式）。

サンプルログ:

2025-01-15T10:30:00Z DENY 192.168.1.100 10.0.0.50 443 1024

Parse Settings（正規表現）:

^\S+ (\S+) (\d+\.\d+\.\d+\.\d+) (\d+\.\d+\.\d+\.\d+) (\d+) (\d+)$

Parse Fields（フィールド名）:

action, src_ip, dst_ip, dst_port, bytes

数値・日時で扱うフィールドは型を指定する

Parse Fields は既定で各フィールドを文字列として抽出します。数値として比較・集計したいフィールドは dst_port(Int32)、bytes(Int64) のように フィールド名(型) と書くと、その型で抽出されます（正規表現モードと JSON モードの両方で有効）。指定しなければ文字列のままで、クエリ側で to_int32 等を使うこともできます。詳細は Pipeline（ログ取り込み）を参照してください。

__time（タイムスタンプ）は Parse Fields に含めない

先頭のタイムスタンプはキャプチャせず（^\S+）、システムの自動推定に任せます。こうすると __time が日時型として扱われ、検索画面の既定の時間範囲フィルタ（Last 24 hours など）が正しく動作します。 __time を Parse Fields に含めると文字列型で保存され、既定の時間範囲フィルタで型エラーになります。

AI によるパースルール生成

サンプルログを「AI 生成」ボタンに入力すると、生成 AI が正規表現とフィールド名の候補を提示します。候補はそのまま確定するのではなく、内容を確認してから適用してください。フィールド名は ECS スキーマに沿って正規化されます。

JSON ログの場合

JSON 形式のログでは、パースルールの設定は不要です。 Parse Settings と Parse Fields を空のまま保存すると、検索時に JSON のキーがフィールドとして抽出されます。キーを数値・日時などの型で抽出したい場合は、Parse Fields にキー名と型を書きます（例: dst_port(Int32)、ネストキーは _source.MsgNum(Int64) のようにドット表記）。型を指定しないキーは文字列として抽出されます。

Step 2: API キーを発行する

ログの送信や外部ツールとの連携には API キーが必要です。

Step 2: API キー発行画面 図: Step 2 API キーの発行と表示

発行手順

画面右上のユーザーアイコンをクリック
「設定」 を選択
左メニューから 「API キー」 を選択
「新規作成」 をクリック
キーの名前を入力（例: log-forwarder）
「作成」 をクリック

API キーの保管

API キーは作成時に一度だけ表示されます。必ず安全な場所に保存してください。紛失した場合は、既存のキーを無効化して新しいキーを発行してください。

API キーには scope（権限の範囲）を設定できます。発行された API キーは X-API-Key ヘッダーに設定して使用します。

X-API-Key: sk-xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx

Step 3: ログを送信する

Pipeline と API キーの準備ができたらログを送信します。ログはテキスト形式 (text/plain) でそのまま送信でき、Pipeline に設定したパースルールに従って検索時にフィールドが抽出されます。取り込みエンドポイントでは pipeline_id の指定が必須です。

送信先は POST https://<your-domain>/api/v2/ingest?pipeline_id=<pipeline_id>、認証は X-API-Key ヘッダーです。本文は テキスト（text/plain。改行区切りで 1 行 = 1 レコード） か JSON（application/json）。パースは Pipeline 側で行うため、送信側でフィールド抽出を書く必要はありません。代表的な送信方法をタブで示します。

curlPythonLogstashVectorSyslog

# 1 行（テキスト）
curl -X POST "https://<your-domain>/api/v2/ingest?pipeline_id=<pipeline_id>" \
  -H "X-API-Key: <your-api-key>" \
  -H "Content-Type: text/plain" \
  -d '2025-01-15T10:30:00Z DENY 192.168.1.100 10.0.0.50 443 1024'

# 複数行（改行区切り。1 行ごとに 1 レコードになります）
curl -X POST "https://<your-domain>/api/v2/ingest?pipeline_id=<pipeline_id>" \
  -H "X-API-Key: <your-api-key>" \
  -H "Content-Type: text/plain" \
  --data-binary @firewall.log

# JSON ログ
curl -X POST "https://<your-domain>/api/v2/ingest?pipeline_id=<pipeline_id>" \
  -H "X-API-Key: <your-api-key>" \
  -H "Content-Type: application/json" \
  -d '{"timestamp":"2025-01-15T10:30:00Z","action":"DENY","src_ip":"192.168.1.100"}'

import requests

URL = "https://<your-domain>/api/v2/ingest"
API_KEY, PIPELINE = "<your-api-key>", "<pipeline_id>"

# テキストログを 1 行ずつ送る
with open("firewall.log", encoding="utf-8") as f:
    for line in f:
        line = line.rstrip("\n")
        if line:
            requests.post(
                URL, params={"pipeline_id": PIPELINE},
                headers={"X-API-Key": API_KEY, "Content-Type": "text/plain"},
                data=line.encode("utf-8"), timeout=10,
            )

# 複数行をまとめて送る（バッチ）
# batch は messages（文字列の配列）を body に、pipeline_id はクエリで渡す
lines = [
    "2025-01-15T10:30:00Z DENY 192.168.1.100 10.0.0.50 443 1024",
    "2025-01-15T10:30:01Z ALLOW 192.168.1.101 10.0.0.51 80 2048",
]
requests.post(
    f"{URL}/batch", params={"pipeline_id": PIPELINE},
    headers={"X-API-Key": API_KEY, "Content-Type": "application/json"},
    json={"messages": lines}, timeout=30,
)

# logstash.conf （filter は不要 — パースは Pipeline 側）
input {
  file {
    path => "/var/log/firewall/*.log"
    start_position => "beginning"
    mode => "tail"
  }
}
output {
  http {
    url => "https://<your-domain>/api/v2/ingest?pipeline_id=<pipeline_id>"
    http_method => "post"
    format => "message"
    message => "%{message}"       # 元のログ行をそのまま送信（format=message には message 指定が必須）
    content_type => "text/plain"
    headers => { "X-API-Key" => "<your-api-key>" }
    retry_non_idempotent => true
  }
}

上の内容を logstash.conf として保存し、ローカルにインストールせず Docker で起動します。

docker run --rm \
  -e XPACK_MONITORING_ENABLED=false \
  -v "$PWD/logstash.conf:/usr/share/logstash/pipeline/logstash.conf:ro" \
  -v "/var/log/firewall:/var/log/firewall:ro" \
  docker.elastic.co/logstash/logstash:8.15.0

Logstash の http 出力の注意

format => "message" を使うときは message => "%{message}" の指定が必須です（ないと「message must be set if message format is used」でパイプラインが起動しません）。 -e XPACK_MONITORING_ENABLED=false は、Elasticsearch 未接続時に出る X-Pack monitoring の接続エラーログを抑止するためのもので、転送自体には影響しません。

# vector.toml （file を読んで HTTP 転送）
[sources.firewall]
type = "file"
include = ["/var/log/firewall/*.log"]

[sinks.csirt_pro]
type = "http"
inputs = ["firewall"]
uri = "https://<your-domain>/api/v2/ingest?pipeline_id=<pipeline_id>"
method = "post"
encoding.codec = "text"           # .message をそのまま送信（1 行 = 1 レコード）

[sinks.csirt_pro.request.headers]
"X-API-Key" = "<your-api-key>"
"Content-Type" = "text/plain"

上の内容を vector.toml として保存し、Docker で起動します。

docker run --rm \
  -v "$PWD/vector.toml:/etc/vector/vector.toml:ro" \
  -v "/var/log/firewall:/var/log/firewall:ro" \
  timberio/vector:latest-alpine --config /etc/vector/vector.toml

CSIRT-Pro は Syslog を直接受信しません。機器・サーバの Syslog はフォワーダで受けて HTTP に変換します。

rsyslog（omhttp モジュール）:

# /etc/rsyslog.d/csirt-pro.conf
module(load="omhttp")
action(
  type="omhttp"
  server="<your-domain>"
  restpath="api/v2/ingest?pipeline_id=<pipeline_id>"
  httpheaders=["X-API-Key: <your-api-key>"]
  template="RSYSLOG_FileFormat"
)

Vector（syslog を受けて HTTP 転送）:

[sources.syslog_in]
type = "syslog"
address = "0.0.0.0:514"
mode = "udp"

[sinks.csirt_pro]
type = "http"
inputs = ["syslog_in"]
uri = "https://<your-domain>/api/v2/ingest?pipeline_id=<pipeline_id>"
method = "post"
encoding.codec = "text"

[sinks.csirt_pro.request.headers]
"X-API-Key" = "<your-api-key>"

上の内容を vector.toml として保存し、syslog 受信用に Docker で起動します（UDP 514 を公開）。

docker run --rm \
  -p 514:514/udp \
  -v "$PWD/vector.toml:/etc/vector/vector.toml:ro" \
  timberio/vector:latest-alpine --config /etc/vector/vector.toml

送信に成功すると、status と取り込み件数 messages_count を含む JSON が返ります。

{
  "status": "ok",
  "organization_uuid": "xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx",
  "table_name": "firewall_logs",
  "messages_count": 1
}

messages_count の数え方

messages_count は API が受け付けた payload 数です。 text/plain では本文全体が 1 payload として数えられるため、複数行を送っても messages_count は 1 です（各行はサーバ側で「1 行 = 1 レコード」に分割・格納されます）。送った件数をそのまま数えたい場合は、batch エンドポイント（/api/v2/ingest/batch、body は messages 配列）を使うと、配列の要素数が messages_count として返ります。

レスポンスの確認

送信後、"status": "ok" が返ることを確認してください。エラーが返る場合は、Pipeline ID と API キーが正しいかを確認してください。

フォワーダは Docker ですぐ起動できます（インストール不要）

Logstash / Vector はローカルにインストールせず、上記の docker run で設定ファイルをマウントして起動できます。監視対象のログを置いたディレクトリ（例の /var/log/firewall）をコンテナにマウントしてください。 Windows の PowerShell ではカレントディレクトリ参照を ${PWD} と書き、Docker Desktop のファイル共有で対象フォルダを許可してください。

Vector を 1 コマンドで起動する docker-compose.yml の例:

services:
  vector:
    image: timberio/vector:latest-alpine
    command: ["--config", "/etc/vector/vector.toml"]
    volumes:
      - ./vector.toml:/etc/vector/vector.toml:ro
      - /var/log/firewall:/var/log/firewall:ro

docker compose up で起動します（本番ではイメージのバージョンを固定してください）。

もっとデータを入れて検索・可視化・検知まで試す

1 行や数行だけでは検索や可視化の練習になりません。正常通信に攻撃を混ぜたまとまったサンプルデータで、検索から AI 分析までの一連の流れを試すには、チュートリアル: 検知から AI 分析までを参照してください。サンプルデータの生成スクリプトも掲載しています。

Syslog は直接受信しません

CSIRT-Pro は UDP/TCP のネイティブ Syslog をポートで直接受信しません（syslog リスナーを持ちません）。取り込みは HTTPS の取り込み API と HEC 互換取り込みのみです。 Syslog 送信元は、上記タブの rsyslog（omhttp）や Vector（syslog source）などのフォワーダで受信し、HTTP で取り込み API へ転送してください（取り込み後、Pipeline のパースルールで Syslog ヘッダーを含めた正規表現を設定します）。

フォワーダ運用のヒント

ファイル読み取り位置を自動管理するツール（Vector / Logstash / Fluent Bit 等）なら、再起動時もログの重複・欠損が起きにくいです
一定件数・一定時間でまとめて送るバッチ設定で API 呼び出し回数を抑えられます（text/plain は改行区切りで 1 行 = 1 レコード）
既存の収集基盤がある場合は、その HTTP 出力の宛先に CSIRT-Pro を追加すれば、従来の保存先と並行運用できます

Step 4: ログを検索する

取り込んだログを PRQL クエリで検索します。

図: Step 4 Search 画面でのクエリ実行と結果表示

UI から検索

サイドメニューから Search を選択
クエリエディタに PRQL クエリを入力
Ctrl + Enter で実行

基本クエリ

全件取得:

from `firewall_logs`

条件フィルタ

DENY アクションのログに絞り込み:

from `firewall_logs`
filter action == "DENY"

送信元 IP ごとの集計

from `firewall_logs`
filter action == "DENY"
group {src_ip} (
  aggregate {cnt = count this}
)
sort {-cnt}
take 10

時間範囲を指定した検索

from `firewall_logs`
filter __time > @2025-01-15
filter __time < @2025-01-16
filter action == "DENY"
sort {-__time}

部分一致検索

from `firewall_logs`
filter src_ip ~= "192.168"

可視化

クエリ結果は、棒、折れ線、円、タイムライン、KPI、ヒートマップなど 20 種類以上のチャートで可視化できます。チャートエリアのコンテキストメニューからタイプを切り替えられます。一覧と Markdown（Jinja2 テンプレート）の詳しい書き方は検索ガイドを参照してください。

集計クエリの応答時間

生ログを保持し検索時に抽出する方式のため、フィールドを指定した絞り込み（point lookup）は高速に返ります。一方、広い時間範囲を対象にした大規模な集計（GROUP BY）は本文列のスキャンを伴うため、データ量に応じて応答に時間がかかることがあります。集計対象は時間範囲や条件で絞ると応答が速くなりやすいです。

Step 5: ケースを作成する

検索で不審なアクティビティを見つけたら、ケースとして登録します。

Step 5: ケース作成画面 図: Step 5 新規ケースの作成

UI から作成

サイドメニューから Case Management を選択
「新規作成」 をクリック
ケースの詳細を入力:
- タイトル: 例）「192.168.1.100 からの大量 DENY イベント」
- メッセージ: 発見した内容の詳細説明
- タグ: 関連するタグ（例: firewall, brute-force）
保存をクリック

Playbook から自動作成する

手動作成のほかに、Playbook の「ケース作成（Create Case）ノード」でケースを自動起票できます。検知用の Playbook を定期実行し、ヒットしたときに自動でアラート（ケース）を立てる、という運用です。

検知 Playbook のフロー（CRON から検索、条件、ケース作成へ） 図: トリガー（CRON 5 分）から検索（detection）、条件（has_hits）、ケース作成（make_case）へつないだ検知 Playbook

作成の流れ

サイドメニューから Playbook を開き、新規 Playbook を作成します。
ノードを追加し、種別のドロップダウンから ケース作成（Create Case） を選びます。
コード欄に、ケース本文（第 1 引数）を入力します。
```
'SSH ブルートフォースの疑い: 22/tcp への DENY が短時間に多発。要調査。'
```
コードを action create_case('...') の形で書いて、種別を自動判定させることもできます。
検索ノード（検知クエリ）→ 条件ノード（ヒット判定）→ ケース作成ノード の順につなぎます。
トリガーを CRON（例: 5 分間隔）に設定して保存します。条件にヒットしたときだけケースが起票されます。

ケース作成ノードの編集パネル。右側の USAGE に引数とオプションが表示される 図: ケース作成ノードの編集パネル。種別「Create Case」、本文を入れるコード欄、引数の説明（USAGE）が並ぶ

引数とオプション

ノード編集パネル右側の USAGE に、そのまま使える引数の一覧が表示されます。

位置 / キー	内容
第 1 引数	ケース本文（content）
第 2 引数	タグ（tags、dict）
`classification=`	分類。`TP` / `FP` / `Unknown`
`status=`	ステータス。`Open` / `Resolved` / `In Progress` / `Ignore`
`created_at=`	作成日時（`%Y-%m-%d %H:%M:%S` 形式）

引数は半角カンマ（,）で区切られます。そのためケース本文に半角カンマは使えません（読点「、」は使えます）。タグに複数のキーを渡すときは、リテラルの dict ではなくトリガーの値（例: trigger.tags）を渡してください（dict 内の , も区切りと解釈されるため）。

検知から自動起票、AI 分析までの一連の流れはチュートリアル: 検知から AI 分析までを参照してください。 Playbook の作り方は本ガイドの Step 6 と Playbook ガイドを参照してください。

ケースの活用

ケースを作成すると、以下の機能が利用できます。

機能	説明
スレッドビュー	会話形式でインシデントの調査状況を記録
類似ケース検索	過去の類似ケースを提示
対応推奨	類似度とアクセス履歴を加重したスコアに基づき、対応手順の候補を提示
ステータス管理	Open から In Progress、Resolved へワークフロー管理
URL 共有	ケースの URL をチームメンバーに共有

AI による調査の位置づけ

類似ケース検索と対応推奨は候補の提示にとどまります。対応の実行、通信の遮断、ケースのクローズを自動では行いません。

Step 6: Playbook で自動化する

繰り返し発生するインシデントへの対応を自動化します。

Step 6: Playbook エディタ 図: Step 6 Playbook ビジュアルフローエディタでの自動化構築

自動トリアージ Playbook の作成例

新しいケースが作成されたとき、送信元 IP のレピュテーションを照会し、結果に応じて通知する Playbook を作成します。

サイドメニューから Playbook を選択
「新規作成」 をクリック
トリガーを 「ケース作成」 に設定

ノードの追加

Step 6-1: IP レピュテーション照会（Action ノード）

import requests

ip = input_data.get('src_ip', '')
if ip:
    response = requests.get(
        "https://api.abuseipdb.com/api/v2/check",
        params={"ipAddress": ip, "maxAgeInDays": 90},
        headers={"Key": env['ABUSEIPDB_API_KEY']}
    )
    result = response.json()['data']
    output = {
        "ip": ip,
        "abuse_score": result['abuseConfidenceScore'],
        "country": result['countryCode'],
        "total_reports": result['totalReports']
    }
else:
    output = {"ip": "", "abuse_score": 0}

Step 6-2: 高リスク判定（Condition ノード）

output = input_data.get('abuse_score', 0) > 80

Step 6-3: Slack 通知（Action ノード True パス）

import requests

requests.post(
    env['SLACK_WEBHOOK_URL'],
    json={
        "text": f":rotating_light: 高リスク IP を検知\n"
                f"IP: {input_data['ip']}\n"
                f"スコア: {input_data['abuse_score']}\n"
                f"国: {input_data['country']}\n"
                f"報告件数: {input_data['total_reports']}"
    }
)
output = {"notified": True}

環境変数の設定

Playbook エディタの「環境変数」タブで、以下を設定します。

キー	値
`ABUSEIPDB_API_KEY`	AbuseIPDB の API キー
`SLACK_WEBHOOK_URL`	Slack Webhook URL

テスト実行

「テスト」タブを開く
サンプル入力を設定:

{
  "src_ip": "203.0.113.50",
  "title": "テストケース",
  "severity": "high"
}

「テスト実行」 をクリック
各ノードの入出力を確認

対応実行は承認制

通信遮断やルール変更など、外部システムへの書き込みを伴う対応は、申請から承認を経て適用する運用を前提とします。 Playbook が自動でこれらを適用することはありません。

Playbook の保存も承認制

Playbook を保存すると、変更は即時反映されず 編集申請 として登録されます。別のメンバーが影響プレビュー（自動処理の対象になるケース件数など）を確認して承認すると反映されます。 AI チャットに作成を依頼した場合も、同じ承認フローを通ります。詳細は Playbook ガイドを参照してください。

まとめ

以上の 6 ステップで、CSIRT-Pro の基本的なワークフローを体験しました。

ステップ	内容
Step 1	Pipeline を作成し、ログの受け口を準備
Step 2	API キーを発行
Step 3	API、ログ転送ツール、既存パイプラインからログを送信して取り込み
Step 4	PRQL でログを検索・分析
Step 5	不審なアクティビティをケースとして登録
Step 6	Playbook で対応を自動化

次のステップ

検索・クエリの詳細 PRQL クエリ言語の完全ガイド
Pipeline の詳細 パースルール設計のベストプラクティス
Playbook の詳細 高度な自動化ワークフローの構築
UEBA スケジュール実行される検知 Playbook タスクの管理
API リファレンス REST API の詳細仕様