Microsoft Sentinel / Log Analytics で結合クエリー (JOIN) を使いこなす！

1. はじめに

ログクエリー分析を始めるユーザーにとって、応用編ともいえる結合クエリー (JOIN) は避けて通れないものになっています。SQL 言語に慣れた方でしたら、JOIN 結合の使い方に慣れている方も多いと思うのですが、筆者も SQL なんちゃって歴が長く JOIN と言われた瞬間に怯むことも多々ありました。
今回はドキュメントを見てもイマイチ何が書かれてあるのかよく分からない結合クエリーについて、本記事で取り上げてご紹介したいと思います。

2. 結合クエリー JOIN とは

SQL 言語では JOIN 句を用いることで複数のテーブルを結合条件（特定フィールドなど）で結びつけることが出来ます。以下イメージ例です。

2.1 INNER JOIN (内部結合)

内部結合では右テーブルの行数に合わせて、左テーブルの行数を複製します。
以下は左テーブルの id と右テーブルの user_id が同一のものを結合するサンプル例です。

なお、INNER JOIN では結合相手が居ない行は結果から削除されます。
以下の例だと左テーブルの id=4, Name=Ritsuko については、結合先の右テーブルのフィールドの user_id にマッチしていないため、結果から削除されます。

2.2 外部結合 (OUTER JOIN)

上記の通り、INNER JOIN を使うとどちらかのテーブルに存在しない場合は結合結果から削除されますが、ケースによってはテーブル結合しても結果から削除して欲しくない場合があります。そのような場合は外部結合 OUTER JOIN を使いましょう。

OUTER JOIN には以下の三つの種類が使えます。
OUTER JOIN を使うことは比較的少ないと思いますが、方法だけ覚えておくと良いと思います。

種類	別名	名称
LEFT JOIN	LEFT OUTER JOIN	左外部結合
RIGHT JOIN	RIGHT OUTER JOIN	右外部結合
FULL JOIN	FULL OUTER JOIN	完全外部結合

2.2.1 LEFT OUTER JOIN (左外部結合)

2.1 で紹介した INNER JOIN で削除される左テーブルを残す結合方法です。
結合した場合にマッチしないものは NULL として表示されます。

2.2.2 RIGHT OUTER JOIN (右外部結合)

結合先の右テーブルにある行でマッチしないものを残す結合方法です。
結合した場合にマッチしないものは NULL として表示されます。

2.2.3 FULL OUTER JOIN (完全外部結合)

双方のテーブルで結合しないものを NULL として表示させるクエリー方法です。

3. Log Analytics ワークスケースの KQL でやってみる

Azure Log Analytics ワークスペースでは JOIN 演算子として結合クエリーがサポートされています。JOIN 演算子のパラメータとして、ここまで挙げてきた結合方法のパラメータが紹介されています。

3.1 書式

Azure Log Analytics で JOIN 結合する書式は以下の通りです。
例を交えながらご紹介いたします！

Table1 | join kind=inner (Table2) on CommonColumn

パラメータ kind の部分で JOIN 結合する方式が選択できます。

3.2 KQL 例

3.2.1 INNERUNIQUE JOIN

デフォルトの JOIN 句では、 INNERUNIQUE JOIN = 内部結合になります。

let X = datatable(Key:string, Value1:long)
[
    'a',1,
    'b',2,
    'b',3,
    'c',4
];
let Y = datatable(Key:string, Value2:long)
[
    'b',10,
    'c',20,
    'c',30,
    'd',40
];
X | join Y on Key

3.2.2 INNER JOIN

左側重複を避ける場合、パラメータで kind=inner オプションを付けます

let X = datatable(Key:string, Value1:long)
[
    'a',1,
    'b',2,
    'b',3,
    'k',5,
    'c',4
];
let Y = datatable(Key:string, Value2:long)
[
    'b',10,
    'c',20,
    'c',30,
    'd',40,
    'k',50
];
X | join kind=inner Y on Key

3.2.3 LEFT OUTER JOIN

左側完全結合でテーブルを表示させたい場合は、kind=leftouterを付けます。

let X = datatable(Key:string, Value1:long)
[
    'a',1,
    'b',2,
    'b',3,
    'c',4
];
let Y = datatable(Key:string, Value2:long)
[
    'b',10,
    'c',20,
    'c',30,
    'd',40
];
X | join kind=leftouter Y on Key

3.2.4 RIGHT OUTER JOIN

右側完全結合でテーブルを表示させたい場合は、kind=rightouterを付けます。

let X = datatable(Key:string, Value1:long)
[
    'a',1,
    'b',2,
    'b',3,
    'c',4
];
let Y = datatable(Key:string, Value2:long)
[
    'b',10,
    'c',20,
    'c',30,
    'd',40
];
X | join kind=rightouter Y on Key

3.2.5 FULL OUTER JOIN

完全外部結合ですべてテーブルを表示させたい場合は、パラメータ kind=fullouter を付けます。

let X = datatable(Key:string, Value1:long)
[
    'a',1,
    'b',2,
    'b',3,
    'c',4
];
let Y = datatable(Key:string, Value2:long)
[
    'b',10,
    'c',20,
    'c',30,
    'd',40
];
X | join kind=fullouter Y on Key

3.2.6 LEFT ANTI JOIN

左反結合を用いると、右レコードに一致しない左テーブルを抽出します kind=leftanti

let X = datatable(Key:string, Value1:long)
[
    'a',1,
    'b',2,
    'b',3,
    'c',4
];
let Y = datatable(Key:string, Value2:long)
[
    'b',10,
    'c',20,
    'c',30,
    'd',40
];
X | join kind=leftanti Y on Key

3.2.7 RIGHT ANTI JOIN

右反結合を用いると、左レコードに一致しない右テーブルを抽出します kind=rightanti

let X = datatable(Key:string, Value1:long)
[
    'a',1,
    'b',2,
    'b',3,
    'c',4
];
let Y = datatable(Key:string, Value2:long)
[
    'b',10,
    'c',20,
    'c',30,
    'd',40
];
X | join kind=rightanti Y on Key

3.3 パラメータ / 結合条件のキー指定

左テーブルと右テーブルで結合するキー名が異なる場合は、onの後に$left.[キー名] == $right.[キー名] とキー名の結び付けを指定することが出来ます。

4.実際の JOIN 結合のユースケース

Microsoft Sentinel / Azure Log Analytics で JOIN 結合を用いたサンプル例です。

4.1 KQL 例 - ログインとログアウト回数をまとめて表示

WindowsSecurityEventから、ログインとログアウトの回数を同時に表示させます。
EventID == "4624" がログイン、EventID == "4634" がログアウトのイベントになります。

SecurityEvent
| where EventID == "4624"
| summarize LogOnCount=count() by EventID, Account
| project LogOnCount, Account
// JOIN でアカウントごとのログオフ回数出して、結合させる
| join kind = inner (
    SecurityEvent
    | where EventID == "4634"
    | summarize LogOffCount=count() by EventID, Account
    | project LogOffCount, Account
    ) on Account

4.2 KQL 例 - SecurityEvent から MMA エージェントの Heartbeat 情報を用いてメタデータを取得

WindowsSecurityEventの情報に対して、クライアント/サーバー端末に導入した MMA (Microsoft Monitoring Agent) の情報を抽出して、OS のバージョ情報を出力する例です。

SecurityEvent //table name
| join Heartbeat on Computer //joining SecurityEvent with Heartbeat on the common Computer column
| where EventID == "4688" //Looking for Event ID for new process
| project Computer, OSType, OSMajorVersion, Version //Displaying data from both tables

4.3 KQL 例 - Microsoft Sentinel の脅威インテリジェンス (TI) にマッチング

TI (脅威インテリジェンス) のテーブルに対して、カスタムログや通信ログをマッチングさせる例です。
テンプレートルールも提供されているが、特定ログなどをマッチングする場合は自作した方が分かりやすくなります！

AzureNetworkAnalytics_CL
| where TimeGenerated > ago(1d)
| where FlowType_s == "MaliciousFlow"

// 脅威インテリジェンスを結合させる（過去7日分のデータ）
| join kind=inner (
    ThreatIntelligenceIndicator
    | where TimeGenerated > ago(7d)
    | project ConfidenceScore,Description,ThreatType,Tags,NetworkSourceIP)
  on $left.SrcIP_s == $right.NetworkSourceIP

// 成型処理（見やすいように加工する）
| project TimeGenerated,FlowType_s,SrcIP_s,DestIP_s,DestPort_d,L7Protocol_s,NSGRule_s,ConfidenceScore,ThreatType,Country_s

4.4 KQL 例 - Microsoft Sentinel で UEBA を有効にして Azure Active Directory から AAD 情報を付与

Microsoft Sentinel で UEBA を有効にすることで Azure Active Directory (AAD) の情報を Log Analytics ワークスペースに取り込むことが出来ます。UPN などの情報が検知した場合、UEBA のテーブル IdentityInfo を JOIN 結合することで、組織情報を表示する例です。

SecurityAlert
| where TimeGenerated > ago(48h)
| where ProductName == "Azure Active Directory Identity Protection"
| mv-expand Entity = todynamic(Entities)
| where Entity.Type == "account"
| extend AadTenantId = tostring(Entity.AadTenantId)
| extend AadUserId = tostring(Entity.AadUserId)
| join kind=inner (
    IdentityInfo
    | where TimeGenerated > ago(14d)
    | distinct AccountTenantId, AccountObjectId, AccountUPN, AccountDisplayName, Department, GivenName, Surname
    | extend UserAccount = AccountUPN
    | extend UserName = AccountDisplayName
    | where isnotempty(AccountDisplayName) and isnotempty(UserAccount)
    | project AccountTenantId, AccountObjectId, UserAccount, UserName, Department, GivenName, Surname
    )
    on
    $left.AadTenantId == $right.AccountTenantId,
    $left.AadUserId == $right.AccountObjectId
| extend CompromisedEntity = iff(CompromisedEntity == "N/A" or isempty(CompromisedEntity), UserAccount, CompromisedEntity)
| project AadTenantId, AadUserId, AccountTenantId, UserAccount,UserName,GivenName,Surname

5.まとめ

JOIN 結合を使いこなすことで、過去データや他テーブルの情報を取り込んで分析に生かすことが出来るようになります。慣れないうちは個々のクエリーを試しながら、Azure Log Analytics / Microsoft Sentinel の KQL クエリーを実践して、スキルアップに生かしましょう。本記事がどなたかの参考になれば幸いです。

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