Рекомендации по использованию запросов расширенного выслеживания

Получите результаты быстрее и избежать превышения времени ожидания при выполнении сложных запросов за счет оптимизации запросов. Рекомендации по повышению производительности запросов:

• Общие советы по оптимизации — в этой статье
• join Оптимизация оператора — в этой статье
• summarize Оптимизация оператора — в этой статье
• Сценарии запросов — в этой статье
• Рекомендации по запросам Kusto — включает несколько сценариев повышения эффективности запроса.
• Оптимизация запросов журнала в Azure Monitor — содержит дополнительные рекомендации по оптимизации запросов.
• Оптимизация запросов KQL (видео) — наиболее распространенные способы улучшения запросов

Общие сведения о квотах ресурсов ЦП

В зависимости от размера каждый клиент имеет доступ к заданному объему ресурсов ЦП, выделенных для выполнения расширенных запросов охоты. Подробные сведения о различных параметрах использования см. в статье Дополнительные квоты охоты и параметры использования.

После выполнения запроса вы увидите время выполнения и использование ресурсов (низкий, средний, высокий). Высокий уровень указывает на то, что выполнение запроса потребовало больше ресурсов и что запрос можно оптимизировать, чтобы результаты возвращались более эффективно.

Клиенты, которые регулярно выполняют несколько запросов, должны отслеживать потребление и применять рекомендации по оптимизации, приведенные в этой статье, чтобы свести к минимуму сбои в результате превышения квот или параметров использования.

Общие советы по оптимизации

• Размер новых запросов. Если вы подозреваете, что запрос вернет большой результирующий набор, сначала оцените его с помощью оператора count. Используйте ограничение или его синоним 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)
  

• Оператор has beats contains—Чтобы не искать подстроки внутри слов без необходимости, используйте оператор has вместо contains. Сведения о строковых операторах

• Ограничьте область поиска — избегайте выполнения запросов search или union без ограничения области поиска, так как они охватывают все таблицы в схеме и могут превышать ограничения по размеру запросов в средах с большим количеством таблиц. Используйте search in, чтобы указать таблицы, в которых нужно выполнить поиск. Например, вместо search "email" используйте search in (EmailEvents, EmailAttachmentInfo, IdentityInfo) "email".

• Поиск в определенных столбцах — поиск в определенном столбце вместо выполнения полнотекстового поиска. Не используйте *, чтобы проверить все столбцы.

• Учитывайте регистр для повышения скорости — поиск с учетом регистра обычно точнее и работает быстрее. Имена строковых операторов, чувствительных к регистру, таких как has_cs и contains_cs, обычно заканчиваются на _cs. Кроме того, вместо =~ можно использовать оператор равенства с учетом регистра ==.

• Синтаксический анализ, не извлекайте. По возможности используйте оператор синтаксического анализа или функцию синтаксического анализа , например parse_json(). matches regex Избегайте оператора string или функции extract(), оба из которых используют регулярное выражение. Зарезервируйте использование регулярных выражений для более сложных сценариев. Дополнительные сведения о функциях синтаксического анализа

• Фильтруйте таблицы, а не выражения — Не применяйте фильтр к вычисляемому столбцу, если можно фильтровать по столбцу таблицы.

• Нет трехсимвных терминов. Избегайте сравнения или фильтрации терминов с тремя символами или меньше. Эти термины не индексируются, и для их сопоставления потребуется больше ресурсов.

• Выборочное проецирование— упростите результаты, проецируя только нужные столбцы. Проецирование определенных столбцов перед выполнением операций соединения или аналогичных операций также помогает повысить производительность.

Оптимизируйте оператор join

Оператор join объединяет строки из двух таблиц путем сопоставления значений в указанных столбцах. Примените эти советы для оптимизации запросов, использующих этот оператор.

• Меньшая таблица слева. Оператор join сопоставляет записи в таблице слева от оператора join с записями справа. Если меньшую таблицу расположить слева, потребуется сопоставить меньше записей, что ускорит выполнение запроса.

  В таблице ниже мы уменьшаем левую таблицу DeviceLogonEvents , чтобы охватить только три конкретных устройства, прежде чем присоединить ее к IdentityLogonEvents идентификаторам безопасности учетной записи.

  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
  

• Используйте тип inner join—тип соединения по умолчанию или innerunique-join удаляет дубликаты строк в левой таблице по ключу соединения, прежде чем вернуть строку для каждого совпадения в правой таблице. Если левая таблица содержит несколько строк с одинаковым значением join для ключа, эти строки будут дедуплицированы, чтобы оставить одну случайную строку для каждого уникального значения.

  Это поведение по умолчанию может оставить важные сведения из левой таблицы, которые могут предоставить полезные аналитические сведения. Например, в приведенном ниже запросе будет отображаться только одно сообщение, содержащее определенное вложение, даже если это же вложение было отправлено с помощью нескольких сообщений электронной почты:

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

  Чтобы устранить это ограничение, мы применяем вариант inner-join, указав 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 производительность. Приведенный ниже запрос применяется 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 , чтобы указать серверной части распределять нагрузку при выполнении ресурсоемких операций. Подробнее о подсказках соединения.

  Например, подсказка shuffle помогает повысить производительность запросов при соединении таблиц по ключу с высокой кардинальностью — ключу с множеством уникальных значений, такому как AccountObjectId в приведенном ниже запросе:

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

  Подсказка broadcast полезна, когда левая таблица невелика (до 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, поскольку идентификатор сетевого сообщения для отдельного электронного письма всегда связан с уникальным адресом отправителя.

  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, вы также можете применить подсказку shuffle с summarize, чтобы распределить вычислительную нагрузку и, возможно, повысить производительность при работе со столбцами с высокой кардинальностью.

  В приведенном ниже запросе для подсчета уникальных адресов электронной почты получателей используется summarize, число которых в крупных организациях может исчисляться сотнями тысяч. Для повышения производительности он включает в себя hint.shufflekey:

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

Сценарии запросов

Определение уникальных процессов с помощью идентификаторов процессов

Идентификаторы процессов (PID) в Windows перерабатываются и используются для новых процессов. Они не могут служить уникальными идентификаторами для определенных процессов сами по себе.

Как правило, единственным способом уникальной идентификации процесса на конкретном устройстве является объединение идентификатора процесса с временем создания процесса вместе с идентификатором устройства (или DeviceIdDeviceName). Например, следующий пример запроса находит процессы, которые обращаются к более чем 10 IP-адресам через порт 445 (SMB), возможно, для проверки общих папок.

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, чтобы рассматривать отдельный процесс, не смешивая несколько процессов с одним и тем же идентификатором процесса.

Этот подход по-прежнему действителен, особенно для систем, отличных от Windows. Однако в Windows существует более прямой метод, использующий ProcessUniqueId поле. Хотя и предыдущий, и описанный ниже метод дают уникальные экземпляры процесса, рекомендуется использовать ProcessUniqueId при наличии, так как он упрощает запросы и устраняет необходимость в обработке сценариев повторного использования PID.

В этом запросе показано, как использовать поля ProcessUniqueId и InitiatingProcessUniqueId, чтобы связать определённый родительский процесс с его дочерними процессами. Сопоставляя InitiatingProcessUniqueId каждого дочернего процесса с ProcessUniqueId родительского процесса, он выделяет только те дочерние процессы, которые были запущены именно этим экземпляром родительского процесса, даже если идентификаторы процессов со временем используются повторно.

Пример запроса

// 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, чтобы рассматривать один процесс, не смешивая несколько процессов с одинаковым идентификатором процесса.

Командные строки запроса

Создать командную строку для выполнения задачи можно разными способами. Например, злоумышленник может ссылаться на файл изображения без пути, без расширения файла, с помощью переменных среды или с кавычками. Злоумышленник также может изменить порядок параметров или добавить несколько кавычек и пробелов.

Чтобы создать более устойчивые запросы к командным строкам, примените следующие методики:

• Определите известные процессы (например ,net.exe или psexec.exe) путем сопоставления в полях имени файла вместо фильтрации по самой командной строке.
• Анализ разделов командной строки с помощью функции parse_command_line()
• При запросе аргументов командной строки искать точное совпадение для нескольких несвязанных аргументов в определенном порядке не имеет смысла. Вместо этого используйте регулярные выражения или несколько отдельных операторов contains.
• Используйте совпадения без учёта регистра. Например, используйте =~, in~и contains вместо ==, inи contains_cs.
• Чтобы устранить проблемы с методами маскировки командной строки, рассмотрите возможность удаления кавычек, замены запятых пробелами и замены нескольких последовательных пробелов одним пробелом. Существуют более сложные методы маскирования, требующие других подходов, но эти настройки могут помочь в решении распространенных.

В следующих примерах показаны различные способы создания запроса, который ищет файлnet.exe для остановки службы брандмауэра "MpsSvc".

// 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 или других форматах. В приведенном ниже примере показано, как использовать обширный список хэшей вредоносных программ SHA-256, предоставляемых MalwareBazaar (abuse.ch), для проверка вложений в сообщениях электронной почты:

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

Синтаксический анализ строк

Существуют различные функции, которые можно использовать для эффективной обработки строк, требующих синтаксического анализа или преобразования.

Чтобы узнать обо всех поддерживаемых функциях синтаксического анализа, ознакомьтесь со строковыми функциями Kusto.

Статьи по теме

• Документация по языку запросов Kusto
• Квоты и параметры использования
• Устранение ошибок расширенного поиска угроз
• Обзор расширенной охоты
• Изучение языка запросов