Log Analytics を用いた Azure NSG フローログの分析方法 (Network Watcher)

# 1. はじめに
Azure では Vnet 環境の通信をモニターする機能として、NSG フローログ機能が提供されています。
NSG フローログは Azure Storage やトラフィック分析を有効にして Log Analytics に出力することが出来るのですが、何か問題が発生した際にフローログを分析する方法が分からない！といったご相談も多いように思います。
本記事では、NSG フローログを Log Analytics に出力し、Azure 利用ユーザーが自ら通信ログを調査するまでをご紹介したいと思います。

# 2. 構成方法
NSG フローログを分析するために、対象の NSG に対してフローログを有効化し、フローログ設定にて「トラフィック分析」機能を有効化して Log Analytics ワークスペースにログを出力する設定が必要になります。
トラフィック分析の詳細については以下 Docs を参照下さい。

https://learn.microsoft.com/ja-jp/azure/network-watcher/traffic-analytics

構成イメージは以下になります。NSG に対してフローログが生成されるため、AWS の [VPC フローログ](https://docs.aws.amazon.com/ja_jp/vpc/latest/userguide/flow-logs.html)とはちょっと異なります。
もし、AWS 同様に Vnet に対する通信監査を行いたい場合は、Subnet に適用できる NSG を別途作成して、フローログを取得すると良いのではと思います。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/653790/8d4289c2-1346-185f-aad6-19bad9e8c9b1.png)

# 3. 設定イメージ
NSG フローログの設定を以下添付します。
Log Analytics で分析を行う場合は、トラフィック分析 (Traffic Analytics) を有効化し、対象の Log Analytics ワークスペース宛に出力を行います。2022.12 現在、フロー分析の処理時間間隔として 1 時間 / 10 分が選べるため、1 時間以内での通信処理を行うような粒度であれば 10 分、長期間の分析目的であれば 1 時間を選ぶなど、用途に応じて切り分けるのがお勧めです。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/653790/978a5205-a598-a0d3-9372-2715d655e1db.png)

# 4. [参考] 可視化
NSG フローログの可視化だけであれば、二つの方法が提供されています。
どちらも傾向などの全体可視化であって、特定の IP やポートを調査するようなハンティング用途では無いのでご注意下さい。
## 4.1 Network Watcher トラフィック分析
Network Watcher のサービスからトラフィック分析で可視化された画面が表示されます。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/653790/2e1afbb7-3ea7-f525-d013-e81049c6434b.png)

## 4.2 Microsoft Sentinel - Network Watcher ブック
Sentinel のブックで提供されているものです。
個人的には Network Wacher よりは見やすいような気がします。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/653790/c67f7a34-2cae-0e4b-cd8a-4fd25e45d168.png)

# 5. Log Analytics から検索を行う
それでは本目的である Log Analytics を用いて見たいフロー通信を調査してみます。
## 5.1 最低限把握しておく必要があるフィールド情報
NSG フローログが生成するスキーマ情報は以下公式 Docs に公開されています。

https://learn.microsoft.com/ja-jp/azure/network-watcher/traffic-analytics-schema#fields-used-in-traffic-analytics-schema

ここでは最低限理解しておくフィールド情報をご紹介します。
5-Tuples (送受信 IP、ポート、プロトコル）を抑えておくことが重要です。

| Field | Format(例) | 意味 |
| ---- | ---- | ---- |
| FlowType_s |IntraVNet<BR>InterVNet<BR>S2S<BR>P2S<BR>AzurePublic<BR>ExternalPublic<BR>MaliciousFlow<BR>Unknown Private<BR>Unknown|通信の接続の方向が分かる。詳しくは公式Docsを参照。<BR>IntraVNet – フロー内の両方の IP アドレスが、同じ Azure 仮想ネットワーク内に存在している。<BR>InterVNet - フロー内の IP アドレスが、2 つの異なる Azure 仮想ネットワーク内に存在している。<BR>S2S – (Site To Site) 一方の IP アドレスは Azure 仮想ネットワークに属していて、もう一方の IP アドレスは、VPN ゲートウェイまたは Express Route を介して Azure 仮想ネットワークに接続された、お客様のネットワーク (サイト) に属している。<BR> |
| SrcIP_s | 10.0.0.1 | 送信元 IP アドレス。<BR>Azure Public と External Public の場合は、``PublicIP_S``が使われる。|
| DestIP_s | 10.0.0.2 | 宛先 IP アドレス。<BR>Azure Public と External Public の場合は、``PublicIP_S``が使われる。|
| DestPort_d | 80 | 宛先ポート |
| L4Protocol_s |T<BR>U | TCP/UDP 通信判定 |
| L7Protocol_s | http | プロトコル/アプリケーション名 |
| FlowDirection_s | I<BR>O | I -- Inbound 通信<BR>O -- Outbound 通信 |
| FlowStatus_s | A<BR>D | A -- Allow(許可)<BR>D -- Deny(拒否) |
| InboundBytes_d | 数値 | フロー受信のデータ量。 |
| OutboundBytes_d | 数値 | フロー送信のデータ量。 |
| InboundPackets_d | 数値 | フロー受信のパケット数。 |
| OutboundPackets_d | 数値 | フロー送信のデータ数。 |

## 5.2 基本的なフィルタリング
KQL (Kusto Query) のサンプルは色々なサイトで紹介されています。

https://qiita.com/YoshiakiOi/items/f2cb2b5626c2f7f71dba

ここでは私が良く用いる基本クエリー例を載せておきます。
### 5.2.1 時間でフィルタリング
#### 指定時間でフィルタしたい
```sql
AzureNetworkAnalytics_CL
| where TimeGenerated >= todatetime('2022-11-13T00:00:00') and TimeGenerated <= todatetime('2022-11-14T00:00:00')
| project TimeGenerated,NSGRule_s,FlowDirection_s,L4Protocol_s,SrcIP_s,SrcPublicIPs_s,DestIP_s,DestPort_d,L7Protocol_s,FlowCount_d
```
#### 過去 XX 時間内で検索したい
クエリーの先頭に let 宣言で時間を宣言しておくと、探す際に手間暇省けます。
Log Analytics の GUI で時間範囲をポータルから変える場合は省きます。
```sql
let timeframe = 1d;
AzureNetworkAnalytics_CL 
| where TimeGenerated > ago(timeframe)
| project TimeGenerated,NSGRule_s,FlowDirection_s,L4Protocol_s,SrcIP_s,SrcPublicIPs_s,DestIP_s,DestPort_d,L7Protocol_s,FlowCount_d
```
### 5.2.2 サブネットでフィルタしたい
IP アドレスやサブネットレンジでフィルタリングする場合は、``ipv4_is_match``関数を用います。
```sql
let timeframe = 1d;
AzureNetworkAnalytics_CL
| where TimeGenerated > ago(timeframe)
| where ipv4_is_match(DestIP_s, '10.1.1.0/24') // Subnet でフィルタリング
| project TimeGenerated,NSGRule_s,FlowDirection_s,L4Protocol_s,SrcIP_s,SrcPublicIPs_s,DestIP_s,DestPort_d,L7Protocol_s,FlowCount_d
```
### 5.2.3 通信方向（インバウンド/アウトバウンド）および通信許可・拒否
NSG フローログの方向指定をする場合は、``FlowDirection``、許可・拒否判定は``FlowStatus_s``を用います。
```sql
let timeframe = 1d;
AzureNetworkAnalytics_CL
| where SubType_s == "FlowLog" and FASchemaVersion_s == "2"
| where TimeGenerated > ago(timeframe)
| where ipv4_is_match(DestIP_s, '10.1.1.0/24')
| where FlowDirection_s == "I"　// I = Inbound, O = Outbound
| where FlowStatus_s == "D" // D = Deny, A = Allow
| project TimeGenerated,NSGRule_s,FlowDirection_s,L4Protocol_s,SrcIP_s,SrcPublicIPs_s,DestIP_s,DestPort_d,L7Protocol_s,FlowCount_d
```
### 5.2.4 Vnet 内通信のみを抽出する
``FlowType_s``を用いることで、Azure 内外の通信条件をフィルタリングすることが出来ます。
```sql
let timeframe = 1d;
AzureNetworkAnalytics_CL
| where SubType_s == "FlowLog" and FASchemaVersion_s == "2"
| where TimeGenerated > ago(timeframe)
| where FlowType_s in ("IntraVNet","InterVNet") // FlowType でVnetのみに制限
| project TimeGenerated,NSGRule_s,FlowDirection_s,L4Protocol_s,SrcIP_s,SrcPublicIPs_s,DestIP_s,DestPort_d,L7Protocol_s,FlowCount_d
```

## 5.3 可視化してみる
Log Anaytics 側からもクエリー一発でグラフ化が出来ます。
Azure Monitor のブックを作る方法もありますが、検索しながら統計でグラフを出して調べるといった使い方が多いので、以下ご紹介します。

### 5.3.1 円グラフで表示する (通信許可・拒否の統計)
円グラフで表示する場合は ``summarize``関数を用いて集約し、``render``を用いて円グラフを生成します。
```sql
let timeframe = 1d;
AzureNetworkAnalytics_CL
| where SubType_s == "FlowLog" and FASchemaVersion_s == "2"
| where TimeGenerated > ago(timeframe)
| extend FlowAction = case (
 FlowStatus_s == 'A', 'Allow',
 FlowStatus_s == 'D', 'Deny', 'Deny')
| summarize count() by FlowAction
| sort by count_ desc
| render piechart with(title='NSG Flow Log Action "Allow/Deny"')
```
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/653790/45c1c8cc-af08-2b59-1c4c-05d64a597b2c.png)

### 5.3.2 円グラフで表示する (フロー種別)
``summarize`` 関数を用いて FlowType_s を集約し、``render`` を用いて円グラフ指定を行います。
```sql
let timeframe = 1d;
AzureNetworkAnalytics_CL
| where SubType_s == "FlowLog" and FASchemaVersion_s == "2"
| where TimeGenerated > ago(timeframe)
| summarize count() by FlowType_s
| sort by count_ desc
| render piechart with(title='NSG Flow Log FlowType')
```
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/653790/3dae1bb8-818c-3c87-7207-9db365674208.png)

### 5.3.3 円グラフで表示する (L7 アプリケーション)
``summarize`` 関数を用いて L7Protocol_s を集約し、``render`` を用いて円グラフ指定を行います。
```sql
let timeframe = 1d;
AzureNetworkAnalytics_CL
| where SubType_s == "FlowLog" and FASchemaVersion_s == "2"
| where TimeGenerated > ago(timeframe)
| summarize count() by L7Protocol_s
| sort by count_ desc
| render piechart with(title='NSG Flow Log Target Protocols')
```
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/653790/b9b06b3c-e16e-8795-781f-8a8b46cc9943.png)

### 5.3.4 時系列チャートで表示する (セッションカウント)
時系列チャートを作成する場合は、``bin(TimeGenrated, 1h)`` のように時間単位での集計を行います。
```sql
let timeframe = 1d;
AzureNetworkAnalytics_CL
| where SubType_s == "FlowLog" and FASchemaVersion_s == "2"
| where TimeGenerated > ago(timeframe)
| where FlowStatus_s == "A" // D = Deny, A = Allow
| summarize TotalFlowCounts = sum(FlowCount_d) by bin(TimeGenerated, 1h)
| render columnchart with(title='NSG Flow Log Traffic', ytitle='Flow(session) Counts')
```
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/653790/d30e8bd1-2975-8eca-6839-2e49523c3aea.png)

### 5.3.5 時系列チャートで表示する (入力・出力の転送量　MBytes で計算する)
時系列チャートを作成する場合は、``bin(TimeGenrated, 1h)`` のように時間単位での集計を行います。
``Summarize`` 集計の際に、入出力のフィールドを MBytes にするために計算式を入れています。
```sql
let timeframe = 7d;
AzureNetworkAnalytics_CL
| where SubType_s == "FlowLog" and FASchemaVersion_s == "2"
| where TimeGenerated > ago(timeframe)
| summarize TotalInboundMByte = sum(InboundBytes_d)/(1024*1024),TotalOutboundMByte = sum(OutboundBytes_d)/(1024*1024) by bin(TimeGenerated, 1h)
| render areachart with(title='NSG Flow Log Traffic (許可) ', ytitle='Mbytes')
```
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/653790/1cbbe372-b53a-7b18-5a25-aa028d23d1f0.png)

## 5.4 ユースケース例
実際に試すようなクエリー例です。

### 5.4.1 特定通信を探す
Where 条件で送信/宛先IPを絞ることで、特定通信をハンティングします。
``ipv4_is_match`` は、サブネット、単一IPも含めてマッチしてくれるのでとても便利です！
```sql
let timeframe = 7d;
AzureNetworkAnalytics_CL
| where SubType_s == "FlowLog" and FASchemaVersion_s == "2"
| where TimeGenerated > ago(timeframe)
| where FlowType_s in ("IntraVNet","InterVNet") // VNet間通信に制限
| where ipv4_is_match(SrcIP_s, '10.1.0.99') // 送信元
| where ipv4_is_match(DestIP_s,'10.100.10.10/24') // 宛先
| project TimeGenerated,NSGRule_s,FlowDirection_s,L4Protocol_s,SrcIP_s,SrcPublicIPs_s,DestIP_s,DestPort_d,L7Protocol_s,FlowCount_d
```
### 5.4.2 時間あたりに急激にセッションが増加を検知
外部公開のセグメントに対して、通信セッションの閾値を抽出する例です。
クエリーの最後にコメントアウトしている``where``条件を外して、閾値を見ながら判断すると良いと思います。

```sql
let timeframe = 1h;
AzureNetworkAnalytics_CL
| where SubType_s == "FlowLog" and FASchemaVersion_s == "2"
| where TimeGenerated > ago(timeframe)
//| where ipv4_is_match(DestIP_s, '10.1.10.0/24') // 特定サブネット
| where FlowStatus_s == "A" // D = Deny, A = Allow
| extend DstPublicIP = tostring(split(PublicIPs_s, "|",0)[0]) // パブリックIPアドレスを抽出
| summarize TotalFlowCounts = sum(FlowCount_d) by SrcIP_s,DstPublicIP,L7Protocol_s
//| where TotalFlowCounts > 50 // 閾値
```

### 5.4.3 時間あたりに急激に通信量が増加を検知
5.4.2 の条件の通信量版です。インバウンド/アウトバウンドの通信量に対して、時間フレーム内での通信量の異常を判定するのに使えます。
クエリーの最後にコメントアウトしている``where``条件を外して、閾値を見ながら判断すると良いと思います。

```sql
let timeframe = 1h;
AzureNetworkAnalytics_CL
| where SubType_s == "FlowLog" and FASchemaVersion_s == "2"
| where TimeGenerated > ago(timeframe)
//| where ipv4_is_match(DestIP_s, '10.1.10.0/24') // 特定サブネット
| where FlowStatus_s == "A" // D = Deny, A = Allow
| extend DstPublicIP = tostring(split(PublicIPs_s, "|",0)[0]) // パブリックIPアドレスを抽出
| summarize TotalInboundMByte = sum(InboundBytes_d)/(1024*1024),TotalOutboundMByte = sum(OutboundBytes_d)/(1024*1024) by SrcIP_s,DstPublicIP,L7Protocol_s
//| where TotalInboundMByte > 10 //閾値(Mbytes)
```

# 6. クエリーを保存しておく - Log Analytics / Sentinel の活用
調査に使ったクエリーは、LogAnalaytics に保存したり、Microsoft Sentinel のハンティングクエリーで一括で分析にかけることが出来ます。
## 6.1 Log Analytics への保存
よく使うクエリーは、Log Analytics に保存しておいて読み出せるようにしましょう。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/653790/e76c880e-8c71-3118-0317-2d1894e89533.png)
クエリーは分かりやすいように、「**お気に入り**」に登録しておくと、上記の画面で表示されるようになります。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/653790/c2ecc212-ef37-fa15-4079-c8a9ce7056c8.png)

## 6.2 Microsoft Sentinel でハンティングクエリーに入れておく
Microsoft Sentinel を使うと、よく用いるクエリーは「ハンティング」機能に登録して、一括でクエリー検索を行ったり、時間間隔での比較値を容易に出したり、インシデントアラートとして登録することが出来るようになります。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/653790/8469abb6-6044-d9d2-adfb-9e345b6c2568.png)
実際の画面例です。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/653790/af5aa38e-3008-6358-0de4-a2ba66a6588e.png)
ハンティングクエリーの作成では、カスタムクエリーや抽出したいエンティティ属性などを設定しておきます。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/653790/e2017d70-39ba-1ca2-384e-6a7cf5bdd36f.png)
ハンティングクエリーを実施すると、保存しておいた複数のクエリーに足して一括で結果を出すことが出来ます。
また、事前のクエリーに対しての結果比較を見ることが出来るため、例えば 1 日と 1 時間での比較などを判断がし易いように作られています。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/653790/f0622874-77e2-2355-0ba3-8504700dd94c.png)

# 7.まとめ
以上、NSG フローログの分析方法のご紹介でした。お客様では何かの障害や問題の際に通信ログを調査することが多いと思いますが、NSG フローログを有効にしておくことで、何かの際に通信ログからの調査を Azure Log Analytics から実施できることが分かるのではと思います。
本記事がどなたかの参考になれば幸いです。

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