高度な捜索クエリのベスト プラクティス

クエリを最適化することで、複雑なクエリの実行中に結果をより速く取得し、タイムアウトを回避します。 クエリパフォーマンスの向上に関するガイダンス:

• 一般的な最適化のヒント - この記事の
• join 演算子を最適化する - この記事の
• summarize 演算子を最適化する - この記事の
• クエリ シナリオ - この記事の
• Kusto クエリのベスト プラクティス - クエリ をより効率的にするためのいくつかのシナリオが含まれています
• Azure Monitor でログ クエリを最適化 する - クエリの最適化に関する追加のガイダンスが含まれています
• KQL クエリの最適化 (ビデオ) - クエリを改善するための最も一般的な方法

CPU リソース クォータについて

各テナントは、そのサイズに応じて、高度なハンティング クエリを実行するために割り当てられた CPU リソースのセット量にアクセスできます。 さまざまな使用パラメーターの詳細については、 高度なハンティング クォータと使用パラメーターに関するページを参照してください。

クエリを実行すると、実行時間とそのリソース使用量 (低、中、高) が表示されます。 高は、クエリの実行に必要なリソースが増え、結果をより効率的に返すために改善される可能性があることを示します。

複数のクエリを定期的に実行するお客様は、使用量を追跡し、クォータまたは使用パラメーターの超過による中断を最小限に抑えるために、この記事の最適化ガイダンスを適用する必要があります。

一般的な最適化のヒント

• 新しいクエリのサイズ - クエリが大きな結果セットを返すと思われる場合は、最初に count 演算子を使用して評価します。 大きな結果セットを回避するには、 limit またはそのシノニム take を使用します。

• フィルターを早期に適用する - 時間フィルターやその他のフィルターを適用して、特に substring()、replace()、trim()、toupper()、parse_json()などの変換および解析関数を使用する前に、データ セットを減らします。 次の例では、演算子のフィルター処理によってレコードの数が減った後、解析関数 extractjson() が使用されます。

  DeviceEvents
  | where Timestamp > ago(1d)
  | where ActionType == "UsbDriveMount"
  | where DeviceName == "user-desktop.domain.com"
  | extend DriveLetter = extractjson("$.DriveLetter", AdditionalFields)
  

• 拍に含まれるものがある - 単語内の部分文字列を不必要に検索しないようにするには、containsの代わりに has 演算子を使用します。 文字列演算子について学習する

• 特定の列を検索する - すべての列でフルテキスト検索を実行するのではなく、特定の列を検索します。 *を使用してすべての列をチェックしないでください。

• 速度に応じて大文字と小文字が区別されます。大文字と小文字を区別する検索は、より具体的で、一般的によりパフォーマンスが高くなります。 has_csやcontains_csなど、大文字と小文字を区別する文字列演算子の名前は、一般に_csで終わる。 =~ではなく、大文字と小文字を区別する equals 演算子==を使用することもできます。

• 解析、抽出しない - 可能な限り、 解析演算子 または parse_json()のような解析関数を使用します。 matches regex文字列演算子または extract() 関数は使用しないでください。どちらも正規表現を使用します。 より複雑なシナリオでは、正規表現の使用を予約します。 関数の解析の詳細を確認する

• 式ではなくテーブルをフィルター処理する - テーブル列でフィルター処理できる場合は、計算列でフィルター処理しないでください。

• 3 文字の用語なし - 3 文字以下の用語を使用して比較やフィルター処理を行わないようにします。 これらの用語はインデックス付けされず、一致する場合は、より多くのリソースが必要になります。

• [選択的にプロジェクト] - 必要な列のみを投影することで、結果をわかりやすくします。 結合または同様の操作を実行する前に特定の列を投影すると、パフォーマンスの向上にも役立ちます。

join 演算子を最適化する

結合演算子は、指定された列内の値を照合することで、2 つのテーブルの行をマージします。 これらのヒントを適用して、この演算子を使用するクエリを最適化します。

• 左側の小さいテーブル — join 演算子は、join ステートメントの左側にあるテーブル内のレコードを右側のレコードと照合します。 左側に小さいテーブルを配置すると、一致する必要があるレコードが少なくなるため、クエリが高速化されます。

  次の表では、左側のテーブル DeviceLogonEvents を減らして、アカウント SID で IdentityLogonEvents に参加させる前に、3 つの特定のデバイスのみをカバーします。

  DeviceLogonEvents
  | where DeviceName in ("device-1.domain.com", "device-2.domain.com", "device-3.domain.com")
  | where ActionType == "LogonFailed"
  | join
      (IdentityLogonEvents
      | where ActionType == "LogonFailed"
      | where Protocol == "Kerberos")
  on AccountSid
  

• 内部結合フレーバーを使用する - 既定の 結合フレーバー または innerunique-join は、右のテーブルに一致する行を返す前に、結合キーによって左側のテーブルの行を重複除去します。 左のテーブルに join キーの値が同じ複数の行がある場合、それらの行は重複除去され、一意の値ごとに 1 つのランダムな行が残されます。

  この既定の動作では、役に立つ分析情報を提供できる重要な情報を左側のテーブルから除外できます。 たとえば、次のクエリでは、同じ添付ファイルが複数のメール メッセージを使用して送信された場合でも、特定の添付ファイルを含む電子メールは 1 つだけ表示されます。

  EmailAttachmentInfo
  | where Timestamp > ago(1h)
  | where Subject == "Document Attachment" and FileName == "Document.pdf"
  | join (DeviceFileEvents | where Timestamp > ago(1h)) on SHA256
  

  この制限に対処するために、右側に一致する値を持つ左側のテーブルのすべての行を表示するkind=innerを指定して、内部結合フレーバーを適用します。

  EmailAttachmentInfo
  | where Timestamp > ago(1h)
  | where Subject == "Document Attachment" and FileName == "Document.pdf"
  | join kind=inner (DeviceFileEvents | where Timestamp > ago(1h)) on SHA256
  

• 時間枠からレコードを結合する - セキュリティ イベントを調査するときに、アナリストは、同じ期間に発生する関連イベントを検索します。 joinを使用するときに同じ方法を適用すると、チェックするレコードの数を減らすことでパフォーマンスも向上します。

  次のクエリでは、悪意のあるファイルを受信してから 30 分以内にログオン イベントがチェックされます。

  EmailEvents
  | where Timestamp > ago(7d)
  | where ThreatTypes has "Malware"
  | project EmailReceivedTime = Timestamp, Subject, SenderFromAddress, AccountName = tostring(split(RecipientEmailAddress, "@")[0])
  | join (
  DeviceLogonEvents
  | where Timestamp > ago(7d)
  | project LogonTime = Timestamp, AccountName, DeviceName
  ) on AccountName
  | where (LogonTime - EmailReceivedTime) between (0min .. 30min)
  

• 両側に時間フィルターを適用する - 特定の時間枠を調査していない場合でも、左右のテーブルにタイム フィルターを適用すると、レコードの数を減らしてチェックし、joinパフォーマンスを向上させることができます。 次のクエリは、過去 1 時間のレコードのみを結合するように、両方のテーブルに Timestamp > ago(1h) を適用します。

  EmailAttachmentInfo
  | where Timestamp > ago(1h)
  | where Subject == "Document Attachment" and FileName == "Document.pdf"
  | join kind=inner (DeviceFileEvents | where Timestamp > ago(1h)) on SHA256
  

• パフォーマンスにヒントを使用する - join 演算子でヒントを使用して、リソースを集中的に消費する操作を実行するときに負荷を分散するようにバックエンドに指示します。 結合ヒントの詳細については、こちらをご覧ください。

  たとえば、 シャッフル ヒント は、カーディナリティの高いキー (以下のクエリの AccountObjectId など、一意の値が多いキー) を使用してテーブルを結合するときのクエリ パフォーマンスを向上させるのに役立ちます。

  IdentityInfo
  | where JobTitle == "CONSULTANT"
  | join hint.shufflekey = AccountObjectId
  (IdentityDirectoryEvents
      | where Application == "Active Directory"
      | where ActionType == "Private data retrieval")
  on AccountObjectId
  

  ブロードキャスト ヒントは、左側のテーブルが小さく (最大 100,000 レコード)、右側のテーブルが非常に大きい場合に役立ちます。 たとえば、次のクエリでは、特定の件名を持ついくつかの電子メールと、EmailUrlInfo テーブルにリンクを含むすべてのメッセージを結合しようとしています。

  EmailEvents
  | where Subject in ("Warning: Update your credentials now", "Action required: Update your credentials now")
  | join hint.strategy = broadcast EmailUrlInfo on NetworkMessageId
  

summarize 演算子を最適化する

summarize 演算子は、テーブルの内容を集計します。 これらのヒントを適用して、この演算子を使用するクエリを最適化します。

• 個別の値を検索する - 一般に、 summarize を使用して、繰り返し可能な個別の値を見つけます。 これを使用して、繰り返し値を持たない列を集計する必要はありません。

  1 つの電子メールが複数のイベントの一部になる場合があります。次の例は、個々のメールのネットワーク メッセージ ID に常に一意の送信者アドレスが付属しているため、summarizeの効率的な使用ではありません。

  EmailEvents
  | where Timestamp > ago(1h)
  | summarize by NetworkMessageId, SenderFromAddress
  

  summarize演算子は、projectに簡単に置き換えることができるので、リソースを少なくしながら同じ結果が得られる可能性があります。

  EmailEvents
  | where Timestamp > ago(1h)
  | project NetworkMessageId, SenderFromAddress
  

  次の例は、同じ受信者アドレスに電子メールを送信する送信者アドレスの個別のインスタンスが複数存在する可能性があるため、 summarize のより効率的な使用です。 このような組み合わせはあまり区別されておらず、重複する可能性があります。

  EmailEvents
  | where Timestamp > ago(1h)
  | summarize by SenderFromAddress, RecipientEmailAddress
  

• クエリをシャッフルする - summarize は繰り返し値を持つ列で最もよく使用されますが、同じ列の カーディナリティが高い 場合や一意の値が多い場合もあります。 join演算子と同様に、summarizeでシャッフル ヒントを適用して処理負荷を分散させ、カーディナリティの高い列で操作するときのパフォーマンスを向上させることもできます。

  次のクエリでは、 summarize を使用して個別の受信者の電子メール アドレスをカウントします。これは、大規模な組織で数十万人で実行できます。 パフォーマンスを向上させるために、次の hint.shufflekeyが組み込まれています。

  EmailEvents
  | where Timestamp > ago(1h)
  | summarize hint.shufflekey = RecipientEmailAddress count() by Subject, RecipientEmailAddress
  

クエリ シナリオ

プロセス ID を使用して一意のプロセスを識別する

Windows では、プロセス ID (PID) はリサイクルされ、新しいプロセス用に再利用されます。 そのため、それ単体では特定のプロセスの一意の識別子として機能しません。

通常、特定のデバイス上のプロセスを一意に識別する唯一の方法は、プロセス ID とそのプロセス作成時間とデバイス識別子 ( DeviceId または DeviceName) を組み合わせることでした。 たとえば、次のクエリ例では、ポート 445 (SMB) 経由で 10 個を超える IP アドレスにアクセスするプロセスを検索し、ファイル共有をスキャンする可能性があります。

DeviceNetworkEvents
| where RemotePort == 445 and Timestamp > ago(12h) and InitiatingProcessId !in (0, 4)
| summarize RemoteIPCount=dcount(RemoteIP) by DeviceName, InitiatingProcessId, InitiatingProcessCreationTime, InitiatingProcessFileName
| where RemoteIPCount > 10

上記のクエリは、 InitiatingProcessId と InitiatingProcessCreationTime の両方によって要約されるため、同じプロセス ID を持つ複数のプロセスを混在させることなく、1 つのプロセスを確認できます。

この方法は、特に Windows 以外のシステムでは有効です。 ただし、Windows では、 ProcessUniqueId フィールドを使用したより直接的なメソッドがあります。 前の方法と以下で説明する方法の両方が一意のプロセス インスタンスを生成しますが、使用可能な場合は ProcessUniqueId を使用することをお勧めします。クエリが簡素化され、PID 再利用シナリオを処理する必要がなくなるためです。

このクエリでは、 ProcessUniqueId フィールドと InitiatingProcessUniqueId フィールドを使用して、特定の親プロセスを子プロセスにリンクする方法を示します。 各子の InitiatingProcessUniqueId を親の ProcessUniqueIdと照合することで、プロセス ID が時間の経過と同時に再利用された場合でも、その親インスタンスによって起動された子プロセスのみを分離します。

クエリ例:

// Step 1: Select a specific parent process instance (for instance, powershell.exe). 
let parentProcess = 
    DeviceProcessEvents
    | where FileName =~ "powershell.exe" // For your specific use case, consider modifying the FileName and adding more identifying properties to specify your query.
    | where isnotempty(ProcessUniqueId)
    | top 1 by Timestamp asc 
    | project DeviceId, DeviceName, ParentProcessUniqueId = ProcessUniqueId, ParentFileName = FileName;
// Step 2: Find all child processes started by this unique parent.
DeviceProcessEvents
| where isnotempty(InitiatingProcessUniqueId)
| join kind=inner (
    parentProcess
) on DeviceId
| where InitiatingProcessUniqueId == ParentProcessUniqueId
| project 
    DeviceName,
    ParentProcessUniqueId,
    ParentFileName,
    ChildProcessName = FileName,
    ChildProcessId = ProcessId,
    ChildProcessUniqueId = ProcessUniqueId,
    Timestamp

同様に、クエリは InitiatingProcessId と InitiatingProcessCreationTime の両方で集計されるため、同じプロセス ID を持つ複数のプロセスを混在させることなく、1 つのプロセスを確認できます。

コマンド ラインのクエリ

タスクを実行するためのコマンド ラインは、さまざまな方法で構築できます。 たとえば、攻撃者は、パスのないイメージ ファイル、ファイル拡張子、環境変数、引用符を使用して参照する可能性があります。 攻撃者は、パラメーターの順序を変更したり、複数の引用符とスペースを追加したりすることもできます。

コマンド ラインを中心に持続性の高いクエリを作成するには、次のプラクティスを適用します。

• コマンド ライン自体でフィルター処理するのではなく、ファイル名フィールドで照合することで、既知のプロセス ( net.exe や psexec.exeなど) を特定します。
• parse_command_line() 関数を使用してコマンド ライン セクションを解析する
• コマンドライン引数をクエリする際は、関連性のない複数の引数で完全な一致を特定の順序で検索しないようにします。 代わりに、正規表現を使用するか、複数の個別の contains 演算子を使用します。
• 大文字と小文字を区別しない一致を使用します。 たとえば、==、in、contains_csの代わりに、=~、in~、containsを使用します。
• コマンド ラインの難読化手法を軽減するには、引用符を削除し、コンマをスペースに置き換え、複数の連続するスペースを 1 つのスペースに置き換えることを検討してください。 他のアプローチを必要とするより複雑な難読化手法がありますが、これらの調整は一般的な方法に対処するのに役立ちます。

次の例では、ファイアウォール サービス "MpsSvc" を停止するために net.exe ファイルを検索するクエリを作成するさまざまな方法を示します。

// Non-durable query - do not use
DeviceProcessEvents
| where ProcessCommandLine == "net stop MpsSvc"
| limit 10

// Better query - filters on file name, does case-insensitive matches
DeviceProcessEvents
| where Timestamp > ago(7d) and FileName in~ ("net.exe", "net1.exe") and ProcessCommandLine contains "stop" and ProcessCommandLine contains "MpsSvc"

// Best query also ignores quotes
DeviceProcessEvents
| where Timestamp > ago(7d) and FileName in~ ("net.exe", "net1.exe")
| extend CanonicalCommandLine=replace("\"", "", ProcessCommandLine)
| where CanonicalCommandLine contains "stop" and CanonicalCommandLine contains "MpsSvc"

外部ソースからデータを取り込む

長いリストまたは大きなテーブルをクエリに組み込むには、 externaldata 演算子 を使用して、指定された URI からデータを取り込みます。 TXT、CSV、JSON、またはその 他の形式のファイルからデータを取得できます。 次の例は、MalwareBazaar (abuse.ch) によって提供されるマルウェア SHA-256 ハッシュの広範な一覧を利用して、電子メールに添付ファイルをチェックする方法を示しています。

let abuse_sha256 = (externaldata(sha256_hash: string)
[@"https://bazaar.abuse.ch/export/txt/sha256/recent/"]
with (format="txt"))
| where sha256_hash !startswith "#"
| project sha256_hash;
abuse_sha256
| join (EmailAttachmentInfo
| where Timestamp > ago(1d)
) on $left.sha256_hash == $right.SHA256
| project Timestamp,SenderFromAddress,RecipientEmailAddress,FileName,FileType,
SHA256,ThreatTypes,DetectionMethods

文字列の解析

解析または変換が必要な文字列を効率的に処理するために使用できるさまざまな関数があります。

String	機能	使用例
コマンド ライン	parse_command_line()	コマンドとすべての引数を抽出します。
Paths	parse_path()	ファイルまたはフォルダー パスのセクションを抽出します。
バージョン番号	parse_version()	セクションごとに最大 4 つのセクションと最大 8 文字のバージョン番号を分解します。 解析されたデータを使用して、バージョンの有効期間を比較します。
IPv4 アドレス	parse_ipv4()	IPv4 アドレスを長整数に変換します。 IPv4 アドレスを変換せずに比較するには、 ipv4_compare()を使用します。
IPv6 アドレス	parse_ipv6()	IPv4 または IPv6 アドレスを標準の IPv6 表記に変換します。 IPv6 アドレスを比較するには、 ipv6_compare()を使用します。

サポートされているすべての解析関数については、 Kusto 文字列関数に関するページを参照してください。

注:

この記事の一部のテーブルは、Microsoft Defender for Endpointでは使用できない場合があります。 Microsoft Defender XDRをオンにして、より多くのデータ ソースを使用して脅威を探します。 高度なハンティング ワークフローをMicrosoft Defender for EndpointからMicrosoft Defender XDRに移動するには、「高度なハンティング クエリをMicrosoft Defender for Endpointから移行する」の手順に従います。

関連項目

• Kusto クエリ言語のドキュメント
• クォータと使用パラメータ
• 高度なハンティング エラーを処理する
• 高度な追求の概要
• クエリ言語の説明

ヒント

さらに多くの情報を得るには、 Tech Community 内の Microsoft Security コミュニティにご参加ください: 「Microsoft Defender XDR Tech Community」。