チュートリアル: 検知から AI 分析まで

このチュートリアルでは、ファイアウォールログを題材に、CSIRT-Pro の一連の流れを実際の画面で体験します。

1. Pipeline を作成し、まとまったサンプルデータを取り込む
2. 検索で異常（SSH ブルートフォース）を見つける
3. 結果を可視化する
4. AI に自然言語で可視化させる
5. Playbook で検知してアラートを発報する
6. 発報したアラートを AI で分析する
7. 同種の誤検知（FP）を自動でクローズする SOAR を AI に生成させ、承認フロー（申請 → 承認 → 適用）で適用する

クイックスタート を一度通したあとに取り組むと理解しやすくなります。

本チュートリアルの取り込み・検索・可視化は実機で検証しています

Step 1〜5 の手順・クエリ・結果は、稼働中の環境で実際に操作して確認したものです（結果の件数はサンプルデータ量により変わります）。

AI 機能と Playbook の前提

Step 4（AI による可視化）と Step 6 / 7 / 8（Playbook・AI 分析・SOAR 自動化）は、AI バックエンドと自動化（ワークフロー実行）基盤が稼働している必要があります。 検索・取り込みを担うコア機能とは別サービスのため、環境によっては利用できないことがあります（動作前提・制約 を参照）。

---

シナリオ

外部の攻撃者 203.0.113.66 が SSH ポート（22）に対してブルートフォースを試み、別の送信元 198.51.100.23 が複数ポートをスキャンしている、という想定のファイアウォールログを扱います。 正常な通信に少数の不審な通信が紛れている状態から、不審な送信元を特定し、検知と分析を行うまでの流れをたどります。

ログ 1 行の形式は次のとおりです。

2026-06-24T10:30:00Z DENY 203.0.113.66 10.0.0.50 22 64
（時刻 アクション 送信元IP 宛先IP 宛先ポート バイト数）

ログイン後のホーム画面です。左メニューから各機能にアクセスします。

---

Step 1: Pipeline を作成する

1. サイドメニューから Pipeline を開き、Create New をクリックします。
2. Pipeline 名を入力します（例: quickstart_demo。名前は英数字とアンダースコアのみ使えます）。
3. Settings タブで Parse Settings（正規表現） と Parse Fields（フィールド名） を設定します。

Parse Settings（正規表現）:

^\S+ (\S+) (\S+) (\S+) (\S+) (\S+)

Parse Fields（フィールド名）:

action, src_ip, dst_ip, dst_port, bytes

1. 設定を保存 をクリックします。

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

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

抽出は検索時に行われます（schema-on-read）

パースルールは取り込み時ではなく検索時に適用されます。 あとから Parse Settings を変更しても、すでに取り込み済みのデータに新しいルールが反映されます。

---

Step 2: サンプルデータを取り込む（Ingest Data）

1 行だけでは検索や可視化の練習になりません。 Pipeline 画面の Ingest Data タブから、正常通信に攻撃を混ぜたデータをまとめて投入します。

1. Pipeline 画面の Ingest Data タブを開きます。
2. Input format で Plain Text を選びます。
3. ログを貼り付けます（下記のサンプル、または後述のスクリプトで生成したもの）。
4. 今すぐ投入する をクリックします。

投入前に確認ダイアログが表示されます。 本番の Pipeline へ即時書き込みされ取り消せないため、内容を確認して 投入する を押します。

投入後、数秒〜数分で検索できるようになります。

代表的なサンプル

2026-06-24T10:00:05Z ALLOW 192.168.1.21 10.0.0.20 443 1820
2026-06-24T10:00:11Z ALLOW 192.168.1.34 10.0.0.10 443 980
2026-06-24T10:00:19Z ALLOW 192.168.1.52 10.0.0.80 80 540
2026-06-24T10:01:02Z DENY 203.0.113.66 10.0.0.50 22 64
2026-06-24T10:01:03Z DENY 203.0.113.66 10.0.0.50 22 64
2026-06-24T10:01:05Z DENY 203.0.113.66 10.0.0.50 22 64
2026-06-24T10:01:21Z ALLOW 192.168.1.40 10.0.0.20 53 210
2026-06-24T10:02:09Z DENY 198.51.100.23 10.0.0.50 3389 60
2026-06-24T10:02:14Z DENY 198.51.100.23 10.0.0.80 445 60
2026-06-24T10:03:00Z ALLOW 192.168.1.28 10.0.0.10 443 3120

より実データに近い量を生成する

可視化と検知を体験するには、ある程度の件数があると分かりやすくなります。 次のスクリプトは、現在時刻に近いタイムスタンプで「正常通信＋ブルートフォース＋ポートスキャン」を生成して送信します（取り込み API を利用）。

python3 - <<'PY'
import requests, random, datetime
PID = "<pipeline_id>"; KEY = "<your-api-key>"; DOMAIN = "<your-domain>"
now = datetime.datetime.now(datetime.timezone.utc)
def ts(m): return (now - datetime.timedelta(minutes=m, seconds=random.randint(0,59))).strftime("%Y-%m-%dT%H:%M:%SZ")
lines = []
for _ in range(120):
    lines.append(f"{ts(random.randint(1,120))} {'ALLOW' if random.random()<0.9 else 'DENY'} "
                 f"192.168.1.{random.randint(10,60)} 10.0.0.{random.choice([10,20,50,80])} "
                 f"{random.choice([443,443,80,53])} {random.randint(200,5000)}")
for _ in range(35):                       # 203.0.113.66 による SSH ブルートフォース
    lines.append(f"{ts(random.randint(1,10))} DENY 203.0.113.66 10.0.0.50 22 64")
for _ in range(12):                       # 198.51.100.23 によるポートスキャン
    lines.append(f"{ts(random.randint(1,60))} DENY 198.51.100.23 10.0.0.{random.choice([50,80])} "
                 f"{random.choice([23,3389,445,1433,8080])} 60")
random.shuffle(lines)
r = requests.post(f"https://{DOMAIN}/api/v2/ingest?pipeline_id={PID}",
                  data="\n".join(lines).encode(), headers={"X-API-Key": KEY, "Content-Type": "text/plain"})
print(r.status_code, r.text[:200])
PY

---

Step 3: 検索で異常を見つける

Search 画面を開き、PRQL でデータを確認します。 まず、DENY が多い送信元 IP を集計します。

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

実行すると、結果が表とグラフで表示されます。 203.0.113.66 の DENY が突出していることが一目で分かります。

結果（例）:

src_ip	cnt
203.0.113.66	53
198.51.100.23	19
192.168.1.26	2
192.168.1.14	1

この送信元が何をしているかを確認します。

from `quickstart_demo`
filter src_ip == "203.0.113.66"
group {dst_port} (
  aggregate {cnt = count this}
)
sort {-cnt}

宛先ポートが 22（SSH）に集中していれば、SSH ブルートフォースの兆候です。

フィールドの型は Parse Fields で指定できる

Parse Fields は既定では各フィールドを文字列として抽出するため、本チュートリアルの quickstart_demo では dst_port == "22" のように文字列で比較しています。 ただし、Parse Fields に dst_port(Int32) や bytes(Int64) のように フィールド名(型) と併記すると、その型で抽出されます。 型を指定すれば dst_port == 22 のように数値で比較でき、sum bytes などの集計も変換なしで行えます。 型指定は正規表現モード・JSON モードのどちらでも有効で、未指定のフィールドは文字列のままです（その場合はクエリ側で to_int32 等で都度変換することも可能）。 詳細は Pipeline（ログ取り込み） を参照してください。

---

Step 4: 結果を可視化する

検索結果はそのままチャートになります。 上の検索結果（DENY 上位 IP）は棒グラフとして表示され、203.0.113.66 が突出した棒として現れます。

時間帯ごとの推移を見るには、時間バケットで集計します。

from `quickstart_demo`
filter action == "DENY"
group {h = to_start_of_hour __time} (
  aggregate {cnt = count this}
)
sort {h}

この結果を 折れ線（Line） で可視化すると、攻撃が発生した時間帯が分かります。 チャートエリアのコンテキストメニューから種類を切り替えられます。

可視化の種類と保存方法は 検索ガイド と Dashboard を参照してください。

---

Step 5: AI に自然言語で可視化させる

PRQL を書かずに、自然言語から可視化することもできます。

1. サイドメニューから AI Investigate を開きます。

2. 次のように依頼します。

   quickstart_demo で DENY が多い送信元 IP を上位5件、棒グラフにして
   

3. AI が PRQL を生成して実行し、可視化カードとして棒グラフを返します。 203.0.113.66 が突出したグラフが、Step 3 の検索と同じ結果として表示されます。 カードには、AI が実際に実行したツール（例: 取り込み定義の一覧取得や詳細取得）も併記され、何を根拠に答えたかが分かります。

AI 機能の位置づけ

AI が返すのは検索条件と可視化の提案です。 AI が対応やケースのクローズを自動実行することはありません。 詳細は AI Investigate を参照してください。 本機能には AI バックエンドの稼働が必要です。

---

Step 6: Playbook で検知してアラートを発報する

毎回手で検索する代わりに、Step 3 の検知ロジックを Playbook にして定期実行します。

この検知 Playbook は、トリガー（CRON 5 分）→ 検索 → 条件 → ケース作成 の 4 ノードで構成します。 ビジュアルエディタでノードを追加し、各ノード右端の 「＋」 で次のノードへつないでいきます。 ノードの種別はコードの先頭キーワードで決まります（filter = 条件、search = 検索、action <関数>(…) = アクション）。 後続のノードは、前のノードに付けた名前でその結果を参照します。

#	ノード種別	名前	中身（コード）	役割
1	検索	detection	下記の PRQL	ブルートフォースを検知して送信元を返す
2	条件	has_hits	len(detection) > 0	検索結果が 1 件以上なら True 側へ分岐
3	ケース作成	make_case	create_case('…')	アラート（ケース）を起票する

手順 1: Playbook を新規作成する

Playbook → Create New で新しい Playbook を開き、最初のフロー（ノード）を追加します。

手順 2: 検索ノード detection を作る

検索ノードに、ブルートフォースを検知する PRQL を設定します（検索ノードはコード先頭を search にします）。

from `quickstart_demo`
filter action == "DENY"
filter dst_port == "22"
filter __time > (now) - (to_interval_minute 5)
group {src_ip} (
  aggregate {cnt = count this}
)
filter cnt > 10

このクエリは、直近 5 分で SSH への DENY が 10 件を超える送信元だけを返します。

デモデータで試すときの注意（5 分ウィンドウ）

この検知は「直近 5 分」で評価するため、取り込みから 5 分以上経過した静的なデモデータでは 0 件になります（クエリ自体は正しくても結果は返りません）。 すぐに発火を確認したい場合は、検知の直前に同じ送信元のサンプルを再投入してください（Step 2 と同じ取り込み手順）。 テスト目的に限り、時間条件を一時的に広げる（例: (to_interval_hour 24)）方法もあります。 本番では対象ログが継続的に流入するため、5 分ウィンドウのままで自然に発火します。

手順 3: 条件ノード has_hits を作る

検索ノードの右の「＋」で条件ノードを追加し、検索結果が 1 件以上かを判定します。

filter len(detection) > 0

detection は手順 2 の検索ノードの名前で、その結果（ヒットした送信元の行リスト）を参照しています。 1 件以上ヒットしたとき（True）だけ、次のケース作成ノードへ分岐します。

手順 4: ケース作成ノード make_case を作る

条件ノードの True 側に、ケース作成ノードを追加してアラート（ケース）を起票します。

action create_case('SSH ブルートフォースの疑い: quickstart_demo の 22/tcp への DENY が直近5分で閾値(10件)を超過。要調査。')

第 1 引数がアラート本文です（第 2 引数でタグ、classification= / status= で分類・状態も指定できます）。 本文に半角カンマは使えません（引数の区切りに使われるため。句読点は全角「、」「。」を使用してください）。

手順 5: トリガーを設定して保存する

このフローのトリガーを 5 分間隔の CRON（*/5 * * * *）に設定して保存します（または UEBA のタスクとして登録して定期実行します）。 これで 5 分ごとに「検索 → 判定 →（ヒット時）ケース起票」が自動実行されます。 ノードの追加・接続やトリガーの詳細は Playbook を参照してください。

下図は、上記の手順で実際に作成・実行した検知 Playbook です（トリガーは 5 分間隔の CRON）。

検知ロジックは Playbook、スケジュールは UEBA

CSIRT-Pro では、検知のロジックは Playbook として作り、UEBA がそれを定期実行する役割を担います。 UEBA 自体は統計・機械学習の検知エンジンではなく、スケジューラです。 実行のスケジューリング（起動）は UEBA が担うため、周期は Playbook の CRON トリガーか UEBA タスクの実行間隔のどちらか一方で設定すれば十分で、同じ周期を二重に指定する必要はありません。 定期実行する検知では、時間条件に絶対日時ではなく相対時間（(now) - (to_interval_minute 5)）を使ってください。 Playbook の作り方は Playbook、検知の考え方は 脅威ハンティング を参照してください。

対応の自動実行について

Playbook は対応案を組み立てますが、外部システムへの書き込みを伴う対応は承認フロー（申請→承認→適用）を経て実行されます。 検知してケースを起票する流れは自動化できますが、遮断などの対応そのものが無断で実行されることはありません。

---

Step 7: 発報したアラートを AI で分析する

Playbook が起票したケース（アラート）を、AI に調査させます。

1. Case Management で、起票されたケースを開きます。
2. ケース詳細の右パネルで AI チャット タブを開き、「このアラートを分析して」 を選びます（自由に質問することもできます）。

3. AI が PRQL でアラートを自律的に調査し、実行したクエリも提示しながら、誤検知（FP）か真の検知（TP）かを根拠付きでまとめます。

   実際に AI が自動で実行した調査クエリの例:

   # 送信元のアクション・宛先・ポートを集計
   from `quickstart_demo`
   filter src_ip == "203.0.113.66"
   group {action, dst_ip, dst_port} (aggregate {cnt = count this})
   sort {-cnt}
   
   # SSH 標的（10.0.0.50:22）の確認
   from `quickstart_demo`
   filter dst_ip == "10.0.0.50" && dst_port == "22"
   group {src_ip, action} (aggregate {cnt = count this})
   

   判定の例:

   FP 寄り（無害化済み）/ confidence = medium IR フェーズ: Identification で完結 / Tier1 で完結可（クローズ提案）

   すべての通信が DENY（＝ファイアウォールが遮断済み）で侵入は成立していない、と AI が推論した例です。

4. 結果を確認し、必要に応じて スレッドへ投稿 します（投稿はユーザーのボタン操作で、自動投稿ではありません）。TP / FP の分類ボタンで判定を確定できます。

AI 分析は読み取り専用

ケース内の AI 機能はすべて読み取り専用で、検索・相関・類似ケース照合などの読み取りツールのみを使います。 対応・修復・ケースのクローズを自動実行することはありません。 詳細は ケース内 AI 機能 を参照してください。 本機能には AI バックエンドの稼働が必要です。

---

Step 8: AI に SOAR を生成させ、承認して適用する

調査の次は、繰り返し起こる同種の誤検知（FP）を自動でクローズする SOAR（自動対応ルール） を AI に提案させ、承認フロー（申請 → 承認 → 適用） を通して適用します。 対応の自動化は、必ず人手の承認を最終ゲートとして通します。

1. ケース詳細の AI チャット タブで、「このアラートの SOAR を提案して」 を選びます。

2. AI が、同種の過去ケース（特に FP）と対応履歴（reply）を調べ、自動クローズ用のルール（ゲート条件 + クローズ動作） を生成します。 今回の例では、次のゲート（signature × 送信元 CIDR レンジ）が生成されました。送信元の個別 IP は焼き込まず、レンジで一般化します。

   tags.get('ThreatID') == 'SSH Brute Force'
     and tags.get('action') == 'DENY'
     and tags.get('DirectionOfAttack') == 'inbound'
     and tags.get('dst_port') == '22'
     and tags.get('SourceAddress', '0.0.0.0') in cidr('203.0.113.0/24')
   

   あわせて バックテストが走り、過去の FP に一致し（例: FP 2/2 一致）、過去の真の検知（TP）には 1 件も誤マッチしない（TP=0） ことが確認されます。クローズ動作は「ステータス=Resolved/分類=FP/クローズ理由を返信」です。

   AI は根拠が無ければ安全に「保留」する

   SOAR 自動クローズ生成は、根拠が不十分なときはゲートを作らず保留します（生ログ文字列を推測でタグに使わない＝捏造しない設計）。 submittable なルールを得るには、対象ケースに 意味のあるタグ（signature/送信元/通信方向 など） が付いており、かつ 同種の FP がいくつか「なぜ FP か」の返信付きでクローズ済み である必要があります。 バックテストの TP=0 は、その過去履歴に対する評価です。新しい検知種別では、まず人手で数件 FP を分類・根拠付けしてから生成してください。

3. 内容に問題なければ、チャットで 「申請して」 と伝えます。AI がケース対応用 Playbook への編集申請を作成します（ステータス: 承認待ち）。この時点ではまだ適用されません。

4. 承認 します。Playbook 編集申請（承認フロー）の画面を開くと、「承認待ち」の申請が一覧表示されます。

5. 申請を開き、変更内容の差分（追加されるノード） と 影響範囲（impact） を確認します。問題なければ 「承認して反映」 します。

6. 承認されると、ケース対応用 Playbook が新しいバージョンとして更新され（例: v0 → v1）、申請は 反映済 になります。 以降、同種のアラートは起票時に自動で FP クローズされます。

承認時の競合（rebase conflict）

申請を出してから承認するまでの間に、対象の Playbook が別の変更で更新されていると、申請が前提としたバージョンと一致せず、競合（rebase conflict）になります。 このとき承認は適用されず（競合エラーが返り）、申請は承認待ちのまま残ります。 その場合は、最新のバージョンに対して差分を組み直し、申請をやり直してください（古いバージョン基準の変更が、知らないうちに上書き適用されることはありません）。 複数人で同じ Playbook を編集・承認する運用では、この競合を前提に、申請はこまめに承認して長時間放置しないことを推奨します。

承認は人の最終ゲート（二者承認）

承認は対応の最終ゲートで、自分が出した申請は自分で承認しない運用（二者承認）が基本です（画面でも自分の申請には「承認して反映」ボタンが出ません）。 遮断など外部システムへの書き込みを伴う対応も同じ承認フローを通り、無断で適用されることはありません。 また、承認は今後の同種ケースに作用するもので、現在のケース自体は自動では閉じません。 詳細は Playbook と 責任分界点 を参照してください。

---

まとめ

このチュートリアルでは、次の一連の流れを体験しました。

1. Pipeline を作成し、まとまったログを取り込む（schema-on-read）
2. 検索で DENY の多い送信元（203.0.113.66）を特定する
3. 棒グラフ・時系列で可視化する
4. AI に自然言語で可視化させる
5. 検知ロジックを Playbook 化し、UEBA で定期実行してアラートを発報する
6. 起票されたアラートを AI で分析する
7. 同種 FP の自動クローズ SOAR を AI に生成させ、承認フロー（申請→承認→適用）で適用する

次のステップ

• 検索ガイド：PRQL と可視化の詳細
• PRQL リファレンス：関数カタログと実行例
• Playbook：自動化フローの作り方
• UEBA：検知 Playbook タスクのスケジュール実行
• ケース内 AI 機能 / AI Investigate：AI による調査