Linux ログを Logstash を用いて、カスタムログとして Microsoft Sentinel (LogAnalytics) に基本ログで取り込む！

# 1. はじめに
Microsoft Sentinel / Log Analytics ワークスペースを運用していく中で、ミドルウェアやアプリケーションのログを取り込んで分析・解析に用いたいといった要望が出てきます。Microsoft Sentinel では[データコネクタ](https://learn.microsoft.com/ja-jp/azure/sentinel/connect-data-sources)と呼ばれるモジュールで簡単に接続する仕組みが提供されているのですが、データコネクタに対応していない製品についてはカスタムログとして取り込む方法を考える必要があります。
本記事では、Linux のアプリケーションログを想定して、以下を実施してみました。

- Linux のアプリケーションログを [Logstash](https://www.elastic.co/jp/logstash) で取り込み、Microsoft Sentinel / Log Analytics ワークスペースに転送する
    - 本記事では Linux 側のログは syslog 受信を想定しているが、Logstash のプラグインを用いることで様々な受信（例:ローカルに吐き出される text ファイルを tail するなど）が可能
    - Microsoft が提供する Logstash の Microsoft Sentinel 用プラグイン [**Microsoft Sentinel output plugin for Logstash**](https://github.com/Azure/Azure-Sentinel/tree/master/DataConnectors/microsoft-sentinel-log-analytics-logstash-output-plugin) を用いて、Azure Monitor DCR (**データ収集ルール**) で正規化を行い、DCE (**データ収集エンドポイント**) を通じてログ転送を行う
- (オプション) [Log Analytics 基本ログ](https://learn.microsoft.com/ja-jp/azure/sentinel/basic-logs-use-cases)を用いて、取り込み時のコスト削減を行う
    - 基本ログでは、インシデント発生時の二次的な解析や、コンプライアンス要件、「何かあった場合に検索する」際のデータ保管に用いる

キーワードとして、Linux 環境で「**カスタムログ**の取り込み」「**基本ログ**の活用」といったテーマとなります。なお Windows アプリケーションログについては以下記事を参照下さい。

https://qiita.com/hisnakad/items/c6e83e3989509e9401fb

# 2. やりたいこと
改めて、今回やりたいこと (Before/After) は以下になります。
- Before : Linux ログ（アプリケーションログなど）
- After : Microsoft Sentinel / LogAnalytics に正規化して取り込み、KQLで解析出来るようにする

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/653790/66070f15-6e00-3d87-b692-ecd6e513377f.png)

# 3.実装のイメージ図
今回の実装は Logstash を用いて、データ収集ルール (DCR) を用いてデータ収集エンドポイント (DCE) を経由することで、カスタムログとして取り込む方法になります。
以下イメージ図を参考としてください。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/653790/ad178f60-7586-3ad2-ce2a-c0833dcef883.png)

# 4. やってみる
実際の設定の流れを以下まとめておきます。
## 4.1 Logstash / Sentinel 用プラグインのインストール
- Linux に Logstash をインストールします。手順の詳細は [Elastic 社の公式ドキュメント](https://www.elastic.co/guide/en/logstash/current/working-with-plugins.html)まで。
- java のインストール
```
[azureuser@vmlinuxcustomlog share]$ yum install java-1.8.0-openjdk
[azureuser@vmlinuxcustomlog share]$ java -version
openjdk version "1.8.0_382"
OpenJDK Runtime Environment (build 1.8.0_382-b05)
OpenJDK 64-Bit Server VM (build 25.382-b05, mixed mode)
```
- Logstash 用 公式 repo の登録
```
[azureuser@vmlinuxcustomlog share]$ sudo rpm --import https://artifacts.elastic.co/GPG-KEY-elasticsearch
[root@vmLinuxCustomLog ~]# cat /etc/yum.repos.d/logstash.repo 
[logstash-8.x]
name=Elastic repository for 8.x packages
baseurl=https://artifacts.elastic.co/packages/8.x/yum
gpgcheck=1
gpgkey=https://artifacts.elastic.co/GPG-KEY-elasticsearch
enabled=1
autorefresh=1
type=rpm-md
[root@vmLinuxCustomLog ~]# 
```
- [**Microsoft Sentinel plugin** で対応している logstash バージョン](https://github.com/Azure/Azure-Sentinel/tree/master/DataConnectors/microsoft-sentinel-log-analytics-logstash-output-plugin#1-install-the-plugin)のインストール
(2023.9 時点の対応 latest は 8.8.1 のため、```yum install logstash``` は用いず指定バージョンの導入を行う。
```
[root@vmLinuxCustomLog /]# sudo yum install logstash-8.8.1-1.x86_64
Loaded plugins: langpacks
Resolving Dependencies
--> Running transaction check
---> Package logstash.x86_64 1:8.8.1-1 will be installed
--> Finished Dependency Resolution

Dependencies Resolved

====================================================================================================================================================
 Package                           Arch                            Version                              Repository                             Size
====================================================================================================================================================
Installing:
 logstash                          x86_64                          1:8.8.1-1                            logstash-8.x                          329 M

Transaction Summary
====================================================================================================================================================
Install  1 Package

Total download size: 329 M
Installed size: 574 M
Is this ok [y/d/N]:
```
- [**microsoft-sentinel-logstash-output-plugin** プラグイン](https://github.com/Azure/Azure-Sentinel/tree/master/DataConnectors/microsoft-sentinel-log-analytics-logstash-output-plugin)のインストール
    - 本記事執筆時のバージョンは ``v1.1.0`` (2023-7-23リリース版)
    - Azure Monitor DCE / DCR に対応
```
[root@vmLinuxCustomLog ~]$ sudo /usr/share/logstash/bin/logstash-plugin install microsoft-sentinel-logstash-output-plugin
Using bundled JDK: /usr/share/logstash/jdk
Validating microsoft-sentinel-logstash-output-plugin
Resolving mixin dependencies
Installing microsoft-sentinel-logstash-output-plugin
Installation successful
```
- プラグインを最新にしておく
```
[root@vmLinuxCustomLog /]$ sudo /usr/share/logstash/bin/logstash-plugin update
Using bundled JDK: /usr/share/logstash/jdk
Updated logstash-filter-elasticsearch 3.15.1 to 3.15.2
Updated logstash-filter-translate 3.4.1 to 3.4.2
Updated logstash-input-azure_event_hubs 1.4.4 to 1.4.5
Updated logstash-input-beats 6.6.1 to 6.6.3
Updated logstash-input-http 3.7.1 to 3.7.2
Updated logstash-input-snmp 1.3.1 to 1.3.3
Updated logstash-integration-aws 7.1.1 to 7.1.5
Updated logstash-integration-jdbc 5.4.3 to 5.4.4
Updated logstash-integration-rabbitmq 7.3.2 to 7.3.3
Updated logstash-output-csv 3.0.8 to 3.0.9
Updated logstash-output-tcp 6.1.1 to 6.1.2
```

## 4.2 Logstash サンプルログの生成
Logstash では output プラグインで転送する際に用いる JSON ファイルのサンプルファイルを作成することが出来ます。Azure Monitor DCR の Transform 機能で正規化する際に元となる JSON データをそのまま利用することが出来るため、サンプルアプリ用の JSON を以下手順で作成します。

- json サンプルログ生成用のフォルダを作成する
```
sudo mkdir /tmp/logstash
```
- vim で logstash 用のサンプルデータ生成用の構成ファイルを作成する
```
sudo vim /etc/logstash/conf.d/syslog-to-dcr-based-sentinel-sample.conf
```
- 構成ファイル用 json ファイル ``/etc/logstash/conf.d/syslog-to-dcr-based-sentinel-sample.conf``
    - input は syslog 受信データ
    - output は ``/tmp/logstash`` に書き込む
```json
input {
     syslog {
         port => 514
    }
}

output {
    microsoft-sentinel-logstash-output-plugin {
      create_sample_file => true
      sample_file_path => "/tmp/logstash/"
    }
}
```
- 上記を設定した後、手動で logstash を構成ファイル指定で読み込みます。
正しく実行できると logstash が起動し、syslog を ``tmp/logstash`` にサンプルログとして書き込みます。
```
sudo /usr/share/logstash/bin/logstash -f /etc/logstash/conf.d/syslog-to-dcr-based-sentinel-sample.conf
```
- サンプルログを実際に送るか、logger コマンドを用いてログを生成します。
```
logger -p local4.warn --tcp -t CEF: "0|NETWORK|YOURFAVORITEFIREWALL|DnsProxyLog|Datacenter|proxy|10.73.4.7|48454|113.1.15.87|443|123|OUT|eu-v20.events.data.microsoft.com|TCP|ALLOW" -P 514 -d -n 127.0.0.1
```
- 実際に出力されたサンプルファイル（以下例だと ``sampleFile1693801696.json``）をローカルに落としておきます。(事後の Azure Monitor DCR ルールの設定で用います）

```bash
[root@vmlinuxcustomlog ~]# cd /tmp/logstash/
[root@vmlinuxcustomlog logstash]# ll
total 4
-rw-r--r--. 1 root root 1819 Sep  4 13:28 sampleFile1693801696.json
```

## 4.3 Azure Monitor データ収集エンドポイント(DCE) / データ収集ルール(DCR) の準備
Logstash で取り込まれる際に API にて転送する DCE/DCR を作成します。
### 4.3.1 データ収集エンドポイント(DCE) 
Azure Monitor 「モニター」より DCE を作成します。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/653790/af2762ba-0b89-fdd5-9d1d-dc89a03bf9f5.png)
作成後に生成されるログインジェスト用 URL をコピーしておきます。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/653790/daf1288b-ba7b-9498-47e9-bf7e3f0a2707.png)

## 4.3.2 データ収集ルール(DCR) 
- 次に DCR を作成します。Azure Portal の「テーブル」画面より、「新しいカスタムログ(DCRベース)」を選択します。 
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/653790/91fb85eb-80dc-5404-8bcd-2fdae77421d6.png)
- Logstash から取り込まれる Linux アプリケーションログ用のカスタムログ (xxx_CL) 名を作成します。データ収集ルール (DCR) は「新しいデータ収集ルールの作成」より新規作成します。前で作成した DCE を選択します。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/653790/530bf404-093f-a044-bc21-9c4d3153bd4a.png)

- 「スキーマと変換」より、「サンプルデータのアップロード」をクリックし、前述で作成した logstash のサンプルファイル (json) をアップロードします。
    - ここでは時刻フィールドである ``Timegenerated`` が反映されていないためアラートが表示されていますが、次の ``Transform`` 機能で時刻を抽出します。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/653790/142f74a6-c0b5-cca1-833a-058ce929b46b.png)

- 「変換エディター」を用いて Log Analytics ワークスペースに取り込まれるデータの正規化を設定します。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/653790/50f47d94-dd9f-8772-0ccb-111f7e63f75f.png)
ここではサンプルの CEF フォーマットのログから以下のような **transform** を KQL で設定しましたが、実際のアプリケーションに沿ったKQL を記載すれば OK です。

```
source
| extend TimeGenerated = ls_timestamp
| parse message
    with    MessageId: int
    "|"     DeviceType: string
    "|"     DeviceName: string
    "|"     EventType: string
    "|"     Location: string
    "|"     NetworkType: string
    "|"     SrcIp: string
    "|"     SrcPort: int
    "|"     DstIp: string
    "|"     DstPort: int
    "|"     MessageSize: int
    "|"     Direction: string
    "|"     DstAddress: string
    "|"     Protocol: string
    "|"     Result: string
| project-away message
```
- 「適用」を押すと実際に変換したデータが反映されているかどうかがテストデータから分かるようになります。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/653790/981f9a37-ecae-d54d-5791-e82590350d9f.png)
- 最後に設定反映を押して **DCR/DCE** 設定は完了です！
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/653790/84140530-e160-6b92-38e3-3bec6ca254f0.png)

## 4.4 (オプション) Log Analytics ワークスペーステーブルの「基本ログ」の設定
Azure Monitor DCR/DCE を経由する場合、基本ログを用いて Ingestion コストを削減することが出来ます。**summarize** 集計が使えないなどの部分的な KQL のみであることや、Microsoft Sentinel の分析ルールを利用できないデメリットがあるため、用途に応じて変更してください。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/653790/9704689d-1697-33c7-8f5c-22e84ed1a2e2.png)

こちらに基本ログ (Basic Log) を用いる際の注意事項をまとめておりますので合わせて参考としてください。

https://qiita.com/hisnakad/items/c6e83e3989509e9401fb#5-%E3%82%AA%E3%83%97%E3%82%B7%E3%83%A7%E3%83%B3-%E5%9F%BA%E6%9C%AC%E3%83%AD%E3%82%B0%E3%82%92%E7%94%A8%E3%81%84%E3%82%8B%E5%A0%B4%E5%90%88%E3%81%AE%E6%B3%A8%E6%84%8F%E7%82%B9

## 4.5 Azure Monitor DCE API 接続のための Azure AD サービスプリンシパルの登録
Logstash から Azure Monitor DCE に接続するため、AAD サービスプリンシパルの認証を用います。
AAD アプリケーションを登録し、Azure Monitor DCR に対して RBAC 権限で「監視メトリック発行者」の権限が必要です。以下簡単な手順です。
### 4.5.1 AAD アプリ登録 / サービスプリンシパルの発行
- AAD からアプリの登録
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/653790/cc6a4f17-5d5f-16fb-2605-8e8450ba41b4.png)
- アプリの Client ID とテナントID をコピーする
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/653790/26d3514b-7c61-1888-2ec3-c28b417ade22.png)
- シークレットの発行
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/653790/580c7223-b900-55c4-0f02-a100225e3ff4.png)

### 4.5.2 モニター データ収集ルール / RBAC 「監視メトリック発行者」の追加
- 設定した Azure Monidor DCR (データ収集ルール)に対して、外部から利用できるように発行した AAD アプリに対して「監視メトリック発行者」の権限を付与します。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/653790/8bf19841-9dc4-6e43-8fb3-1b79e49e5707.png)

## 4.6 その他情報の取得
Logstash 側で他に必要な情報として以下を確認します。
- DCR ``Immutable_id`` の取得 
    - DCR のリソース JSON より確認が出来ます。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/653790/a499914b-98db-acdd-ad9a-c6f203fb2bbc.png)

- streams 先の Custom テーブル名を取得
    - 以下例だと ``Custom_SampleSyslogLogstash_CL``
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/653790/caa860ca-5cba-2bac-5389-a7bc47f894ac.png)

## 4.7 Logstash 構成ファイル (Azure Monitor DCR/DCE 接続用) の作成
4.1 ～ 4.6 で通じた構成を元に、最後に Logstash の構成ファイルを Azure Monitor / DCR (データ収集ルール) を用いるように構成を追加します。

```
vim /etc/logstash/conf.d/syslog-to-dcr-based-sentinel.conf
```

以下は Logstash の input / output の設定例です。
今回は Syslog 受信を設定例としていますが、環境に合わせて変更してください。

```json
input {
     syslog {
         port => 514
    }
}

output {
    microsoft-sentinel-logstash-output-plugin {
        client_app_Id => "(AAD アプリの Client ID)"
        client_app_secret => "(AAD アプリの Secret)"
        tenant_id => "(AAD Tenant ID)"
        data_collection_endpoint => "https://dce-uri.japaneast.ingest.monitor.azure.com"
        dcr_immutable_id => "dcr-XXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXX"
        dcr_stream_name => "Custom_SampleSyslogLogstash_CL"
    }
}
```

## 4.8 Logstash 起動
構成した conf ファイルを読み込んで logstash を起動します。
```
sudo /usr/share/logstash/bin/logstash -f /etc/logstash/conf.d/syslog-to-dcr-based-sentinel.conf
```

# 5. テストしてみる
- サンプルログを実際に送るか、logger コマンドを用いてログを生成します。
```
logger -p local4.warn --tcp -t CEF: "0|NETWORK|YOURFAVORITEFIREWALL|DnsProxyLog|Datacenter|proxy|10.73.4.7|48454|113.1.15.87|443|123|OUT|eu-v20.events.data.microsoft.com|TCP|ALLOW" -P 514 -d -n 127.0.0.1
```
- Logstash 側のログを確認します
    - sentinel プラグインが正常に動作していると、以下のようなログが確認できます。
```
[INFO ] 2023-09-04 18:16:39.904 [[main]>worker0] microsoftsentineloutput - Successfully posted 1 logs into log analytics DCR stream [Custom-SampleSyslogLogstash_CL].
[INFO ] 2023-09-04 18:16:39.922 [[main]>worker1] microsoftsentineloutput - Successfully posted 1 logs into log analytics DCR stream [Custom-SampleSyslogLogstash_CL].
[INFO ] 2023-09-04 18:16:42.363 [Ruby-0-Thread-19: /usr/share/logstash/vendor/bundle/jruby/2.6.0/gems/microsoft-sentinel-logstash-output-plugin-1.0.1/lib/logstash/sentinel/customSizeBasedBuffer.rb:117] microsoftsentineloutput - Successfully posted 3 logs into log analytics DCR stream [Custom-SampleSyslogLogstash_CL].
```
- Log Analytics ワークスペースにカスタムログが作成されているか確認します。
    - 成功すると、transform で正規化されたテーブルが自動生成され、カスタムログが確認できます。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/653790/f2fa6995-87ce-bd45-2764-ee96b39c2db3.png)

# 6. まとめ
本記事では Linux アプリケーションログ（Syslog）を Microsoft Sentinel に取り込み、基本ログや正規化を行う具体的な方法をご紹介いたしました。Microsoft の公式ドキュメントでは分かり難い部分ですが、一度設定を試していただくと比較的簡単にカスタムログを取り込み、分析に必要なフィールドを自由に抽出することが出来ることがお分かりになるのではと思います。
ドキュメントだけでは分かり難い部分も多いので、ご興味ある方は本記事を参考に試していただければ幸いです。

※本稿は、個人の見解に基づいた内容であり、所属する会社の公式見解ではありません。また、いかなる保証を与えるものでもありません。正式な情報は、各製品の販売元にご確認ください

# 参考
https://learn.microsoft.com/ja-jp/azure/sentinel/connect-logstash-data-connection-rules

https://koosg.medium.com/ingest-dcr-based-custom-logs-in-microsoft-sentinel-with-logstash-f94c79e69b93

https://github.com/Azure/Azure-Sentinel/tree/master/DataConnectors/microsoft-sentinel-log-analytics-logstash-output-plugin