やりたい事
定期的にエクスポートされている CSV 形式のログデータを Microsoft Sentinel に入れて分析ルールやハンティングに利用したい。
方針
Powreshell や Python など、csv から JSON 形式に変換する機能や、ライブラリは多くの言語で提供されているので Azure Function などを使って JSON 形式に変換し、ログ インジェスト API で取り込むことももちろんできますが、今回は Azure Data Factory を使って Sentinel にデータをコピーする方法を検証してみます。CSV ファイルの取り込み元は簡単に利用できる Azure Blob Storage にて検証を行いますが、 Azure Data Factory Source に設定できる場所であればどこでもよいはずです。
進め方
以下の流れで環境を構成します。
- Sentinel にデータを投入するためのカスタム テーブルを作成する
- 取り込み元の CSV ファイルを保存する Blob ストレージを作成する
- Azure Data Factory リソースを作成し、ログ インジェスト API 用のロールを付与する
- Azure Data Factory で Blob ストレージから REST API を使ったデータ コピーのパイプラインを作成する
実装
1.Sentinel にデータを投入するためのカスタム テーブルを作成する
投入したい Log Analytics ワークスペースにカスタム テーブルを作成します。カスタム テーブルの作成の際にはデータ収集エンドポイントの作成が必要ですので、先ずは、データ収集エンドポイント リソースから作成します。
データ収集エンドポイントの作成
Azure Portal サイトで、Azure Monitorのページを開き、[設定] > [データ収集エンドポイント] メニューで[作成]ボタンをクリックし新しいデータ収集エンドポイントを作成します。以下の公開情報も参照ください。
カスタム テーブルの作成
作成できた後は対象の Sentinel が有効化されている Log Analytics ワークスペースのページで[テーブル]のメニューから新しいカスタム ログ(DCR ベース)を選択します。
データ収集ルールはここで新規作成ができますので、新規作成をしておき、データ収集エンドポイントには、先ほど作成したデータ収集エンドポイントを設定しておきましょう。
[スキーマと変換]のタブでカスタム テーブルのスキーマを決めます。あらかじめ対象の CSV ファイルを JSON 形式に変換しておいて、そちらのサンプルをアップロードしてスキーマを定義しましょう。
以下の公開情報の手順も参照ください。
2. 取り込み元の CSV ファイルを保存する Blob ストレージを作成する
Azure Storage Account を作成して、Blob コンテナーを作成します。難しい点はないので、以下の公開情報の通りにやっていただければと思います。 取り込みのサンプル用の CSV ファイルもアップロードしておきましょう。
3. Azure Data Factory リソースを作成し、ログ インジェスト API 用のロールを付与する
Azure Data Factory リソースを作成します。初めての作成であれば、公開情報にもあるAzure Data Factory Studioからのクイック作成が簡単でした。
今回は Sentinel に API 経由でログの取り込みを行うため、Azure Data Factory の作成が完了した後に、マネージド ID に必要なロールを付与します。作成したリソースの画面から[設定] > [マネージド ID]を選択します。クイック作成で作った場合はシステム割り当てマネージド ID が有効になっていると思いますが、有効になっていなかった場合は状態を[オン]にしてください。[Azure ロールの割合]をクリックし、1. の手順で作成した、データ収集ルールのリソースへの「監視メトリック発行者」のロールを付与します。
4.Azure Data Factory で Blob ストレージから REST API を使ったデータ コピーのパイプラインを作成する
Azure Data Factory Studio の画面で、Blob ストレージ上の CSV からログ インジェスト API を使って Sentinel のワークスペースに取り込む設定をしていきます。
先ずは [New] > [Pipeline] で新規パイプラインを作成します。
作成した Pipeline の [Activities] の [Move and transform] にある [Copy Data] をデザイナー画面にドラッグ アンド ドロップをします。
Source の設定
[Source] タブの [Source dataset] で [New] を選択し、Azure Blob Storage を選択し、フォーマットに[DelimitedText]を選択します。
[Set properties]ペインで、Linked service で[New]を選択します。Blob ストレージへの接続を設定します。[Authentication type] は 「Account Key」、[Account selection method
]は[From Azure subscription] として、対象の Storage Account を選べば簡単に接続を構成できます。
Storage Account 都の設定を行った後に、CSV が格納されているファイル パスや、最初の行をヘッダーとして扱うか、スキーマをインポートするかを指定します。すべて指定が完了したら、[OK]を押しましょう。
Sink の設定
[Sink] タブの [Sink dataset] で [New] を選択し、 REST を選択します。
[Set properties]ペインで、Linked service で[New]を選択します。
| 項目 | 値 |
|---|---|
| Base URL | 作成したデータ収集エンドポイント / データ収集ルールで作成する URL 詳細は以下に記載 |
| Authentication type | System Assigned Managed Identity |
| Microsoft Entra ID resource | https://monitor.azure.com |
-
Base URL の確認
ログ インジェスト API の Base URL は以下の形式になっております。
{Data Collection Endpoint URI}/dataCollectionRules/{DCR Immutable ID}/streams/{Stream Name}?api-version=2023-01-01
例としては以下のような形です。
以下の公開情報も参照ください。
各環境での{Data Collection Endpoint URI}、{DCR Immutable ID}、{Stream Name}の確認方法を以下に記載します。
-
Data Collection Endpoint URI
データ収集エンドポイントの概要ページのログ インジェストの項目で記載されています。
-
DCR Immutable ID
データ収集ルールから確認しますが、JSON ビューでしか確認ができません。対象リソースの概要のページの右上の JSON ビューをクリックし、表示された "immutableId" の項目になります。
-
Stream Name
データ収集ルールの[Configuration] > [データ ソース]に表示されている、今回データを取り込みたいテーブルがターゲットに指定されている、データソースの名前になります。(以下の例だと「Custom-csvingestusingADF_CL」になります。)
[Sink dataset]の作成が完了したら[Request method] で [POST] を指定してください。
Mapping の設定
[Mapping] タブで Source の列名とTypeと、Sink(Sentinel のカスタムテーブル側)の列名とをマッピングします。私のテスト CSV ファイルでは以下のような3つの列の CSV を同名の Sentinel 側のテーブルの列名にマッピングしています。
テスト実行
ここまで設定すれば、設定としては完了です。Pipeline の画面で [Debag] をクリックすればテスト実行が行われます。
以下の通り、Activities Status が Succeeded で処理が終われば完了です。
取り込み後に以下の通り Sentinel のログから確認ができ、正しく取り込まれていることを確認できました。
発行
最後に設定したデータ コピーのパイプラインを発行しましょう。Data Factory は発行しておかないとパイプライン等の情報が保持されません。また、要件に合わせたファイルのアップロード時や定期的なスケジュールなどでのトリガーをセットしておくことで自動的な取り込みも構成できます。
最後に
Azure Data Factory を利用することで、コードを全く書くことなく、ログ インジェスト API を利用して Sentinel にデータ取込みを行うことができました。今回は細かくは確認していませんが、このパイプラインの処理のトリガーも様々な条件で設定ができますので、要件に応じて柔軟に継続的な CSV データの取り込みを実現できると思います。また、Logic Apps を利用した取り込みもできると思いますが、大規模や複数のデータソースからの取り込みなど Azure Data Factory で楽に取り込めるケースもあると思うので、標準コネクタがないデータソースからのSentinelログ取り込みの方法の一つとして検討いただければと思います。
Discussion