Правила корреляции в KUMA (CookBook)
Этакий "CookBook" по правилам корреляции в KUMA
- Типы правил корреляции
- Активные листы в корреляторе
- Приемы в правилах корреляции
- Тестирование правил корреляции
- Производительность правил корреляции
- Сегментация правил корреляции
- Правила сбора и анализа данных (Data Mining)
Типы правил корреляции
Простое правило (simple)
"Обновить параметры" нужно делать в корреляторе, когда какое-либо правило меняется, чтобы подтянулись актуальные изменения в правилах в коррелятор.
Простое правило (simple) — срабатывает при обнаружении каждого события, удовлетворяющего условиям в одном селекторе.
Типовой пример правила:
Параметр Наследуемые поля (Identical fields) имеет разный смысл, в зависимости от типа правила. В простом правиле он просто перечисляет, какие поля базового события коррелятор скопирует в корреляционное событие при срабатывании правила. Этот параметр обязательный, поэтому хотя бы одно такое поле нужно задать.
Например, если простое правило срабатывает на события о сетевых атаках, в идентичных полях уместно будет перечислить поля с информацией, характеризующие атаку: адрес злоумышленника, адрес жертвы, тип атаки. Аналитику следует посмотреть, в каких полях базовых событий содержится эта информация и перечислить эти поля в Наследуемых полях.
Если аналитик планирует создавать другие правила корреляции, которые реагируют не только на базовые, но и на корреляционные события, то от того, какие поля будут скопированы в корреляционное событие, будет зависеть, какие условия аналитик сможет использовать для такого события.
Простое правило используется, когда нужно создать алерт при обнаружении любого события, которое соответствует определенным условиям. В этом правиле есть только один селектор, который определяет эти условия.
Селектор работает как фильтр. В настройках селектора можно выбрать фильтр из существующих ресурсов или создать новый прямо в этом правиле. Также, как и в других фильтрах, можно использовать ссылки на другие фильтры в более сложных условиях. Например, можно задать условие: "Если поле события равно Х" или "Если выполняются условия фильтра Y".
При срабатывании правила аналитик может настроить одно или несколько из следующих действий:
- output — создать корреляционное событие, которое будет передано в настроенные точки назначения (обычно это хранилище), и по которому будет создано (или дополнен) алерт
- loop — переслать корреляционное событие на вход этого же коррелятора для рекурсивной обработки
- пополнить активные списки — добавить в активный список (или удалить из списка) запись на основании содержимого полей события
- обогатить корреляционное событие по словарю, по данным исходного события, константой или по шаблону, без запросов во внешние системы (т.е. такое же обогащение как в нормализаторе на коллекторе). Обогащение правилами можно задать в корреляторе отдельно, точно так же как в коллекторе
Необходимо указывать все поля и переменные участвующие в селекторах в наследуемых полях.
Стандартное правило (standard)
"Обновить параметры" нужно делать в корреляторе, когда какое-либо правило меняется, чтобы подтянулись актуальные изменения в правилах в коррелятор.
Стандартное правило (standard) — срабатывает при достижении определенного порогового значения группы событий, которые удовлетворяют условиям селектора, полей группировки событий (на основе значений поля создается группа) и времени жизни контейнера для группы.
Если частота срабатывания (Rate limiting) явна не указана, то устанавливается лимит умолчанию - 100 срабатываний в секунду. При превышении лимита правило ничего не делает.
Политика хранения базовых событий (Base events keep policy) - указание, какие из базовых событий должны сохраняться в корреляционном. Возможно указать одно из значений:
- first (по умолчанию) - сохранять только первое базовое событие от каждого селектора в корреляционном событии
- last - сохранять только последнее базовое событие от каждого селектора в корреляционном событии
- all - сохранять все базовые события в корреляционном событии
Типовой пример правила (обнаружение сканирования портов, перебор портов >30 назначения, от одного адреса источника и назначения в течение 60 секунд):
Другие подходящие примеры для стандартных правил:
- просто много обращений к опасным URL: с разных компьютеров и к разным URL
- много обращений к одному и тому же опасному URL с разных компьютеров
- много обращений к опасным URL, в том числе разным, с одного компьютера
- много обращений с одного и того же компьютера к одному и тому же опасному URL
Стандартные правила разбивают все анализируемые события на группы (так называемые «корзины», buckets) с совпадающими значениями полей, перечисленных в параметре Группируемые поля (Identical fields) и затем обрабатывает каждую группу независимо от других. Критерии срабатывания применяются отдельно в каждой такой группе. Состояние всех групп хранится в памяти коррелятора.
Бакет (Окно корреляции):
- Бакет открывается на событие из любого селектора, не важно в каком они порядке в правиле, порядок проверяется после наполнения бакета!
- Для каждого набора Identical Fields создается свой бакет.
- Когда событие подпадает под селектор, коррелятор смотрит, есть ли уже бакет с нужным набором полей Identical Fields, если нет - создает, если есть - событие отправляется в существующий.
- Когда под селектор с Unique fields подпадает событие, то проверяется, есть ли уже в бакете события с таким же набором значений для Unique Fields, если есть, то событие не учитывается.
Пример для Identical Fields с полями RequestUrl и SourceAddress:
В более общем смысле параметр Время жизни контейнера (Window) определяет время жизни группы. Принцип такой:
- identical fields определяет, на какие группы разбивать события
- селекторы определяют, какие событие будут включены в группы (если событие не соответствует ни одному селектору, оно не попадет ни в одну группу)
- если событие соответствует селектору и уже есть группа с таким же набором значений идентичных полей, как в событии, событие добавляется в эту группу
- если событие соответствует селектору, но его значения идентичных полей не соответствуют ни одной группе, создается новая группа с временем жизни, заданным параметром Window
- по истечении времени жизни группы, группа удаляется
- если позже поступает новое событие с набором значений идентичных параметров группы, которой больше нет, создается новая группа и отсчет времени жизни начинается заново
Т.е. все свое время жизни (или окно наблюдения) группа накапливает события, после чего удаляется, и накопление событий может начаться заново. Для каждой комбинации значений идентичных полей это происходит параллельно и независимо.
Правило срабатывает, если группа за время жизни накапливает число событий, указанное в параметре Порог срабатывания (Threshold) селектора.
В общих настройках стандартного правила есть еще параметр Уникальные поля (Unique fields). Он не является обязательным, но он позволяет считать в группах только события с уникальными значениями выбранных полей. При добавлении событий в группу правило будет сравнивать значение уникальных полей нового события со значениями уникальных полей событий, которые уже есть в группе. Если комбинация значений уникальных полей нового событий уже встречается у одного из событий в группе, новое событие отбрасывается и в группу не попадает.
Возможные действия (допускается указать одно или более) правила:
- On first threshold — создавать корреляционное событие только после первого превышения порога, а двукратное, трехкратное и т.д. превышение порога за время жизни группы игнорировать. Например, если при пороге 3 за 30 секунд группа накопит 10 событий, корреляционное событие все равно будет одно - после третьего накопленного события
- On every threshold — создавать корреляционное событие после каждого превышения порога за время жизни группы. Если группа накопит 10 событий при пороге 3, будет создано 3 корреляционных события: после 3-го, 6-го и 9-го события в группе
- On subsequent threshold — создавать корреляционные событие при всех превышениях порога, кроме первого. Например, при пороге 3 после 6-го, 9-го и т.д. события. Так можно по-разному реагировать на первое переполнение порога и последующие. Например, можно настроить отправлять на вход коррелятора только первое корреляционное событие, но в хранилище писать все. Или пополнять активный список только данными из первого события, а при последующих превышениях порога этого не делать, так как в последующих событиях нет новых артефактов для списка
- On timeout — в стандартных правилах есть еще возможность настройки действий по окончании времени жизни группы. Это действие используется в связке с опцией Recovery (Обнуление) в настройках селектора, в каких случаях это уместно и как именно это работает рассматривается ниже. Обнуляющие селекторы можно использовать и не только с действием onTimeout.
Можно также использовать несколько селекторов. Например, несколько неудачных попыток брутфорса (ловится на основе сработки другого правила корреляции) и успешный вход. В общем случае правило, в котором задано несколько селекторов, срабатывает при одновременном превышении порогов во всех селекторах.
Пример правила с несколькими селекторами:
Можно также рекавери правило. Например, когда событие типа «Вредоносное ПО удалено» не обнаружено в течение 5 минут после получения события «Вредоносное ПО обнаружено».
Recovery селектор (Обнуление):
- Бакет открывается только на событие из обычного селектора, на событие из recovery-селектора бакет не открывается никогда!
- Место нахождения селектора с recovery не имеет значения, как только в бакет попадут все нужные recovery-события бакет будет закрыт!
- На recovery-селектор не влияет настройка фильтра Order By.
- Если нужно, чтобы произошло событие А, и не произошло событие Б, при этом событие Б может произойти раньше А, нужно использовать активные листы, т.к. с помощью recovery-селектора такой логики не достичь (см п.1).
Операционное правило (operational)
"Обновить параметры" нужно делать в корреляторе, когда какое-либо правило меняется, чтобы подтянулись актуальные изменения в правилах в коррелятор.
Операционное правило (operational) — наполняет активные листы без создания корреляционного события, механика работы аналогична простому правилу корреляции. С помощью списков можно отслеживать закономерности на длительных промежутках времени. Активный список хранится в памяти коррелятора (и кешируется на диск), так что состояние списка не теряется при перезагрузке службы или сбое.
Наполнять список могут любые правила. Но простые и стандартные правила всегда создают корреляционные события и предупреждения. Чтобы просто наполнять список и не создавать предупреждений, предусмотрены операционные правила.
В операционном правиле доступны только две операции над списком: set и del. Операция get не имеет смысла, поскольку она призвана обогащать корреляционное событие, а операционное правило не создает таких событий.
Пример правила:
Если при выполнении операции set окажется, что запись с таким же ключом уже есть в списке, она будет перезаписана новым значением на основании полей нового события.
При записи в активный лист, в качестве ключевого поля могут быть несколько занчений полей события, для составления комбинированного ключа, при этом в значении ключа это быдет выглядеть так: поле1|поле2|поле3 сравнивать с этим впоследствии можно будет только с целым ключом, а не с какой-то его частью.
Атрибуты записей из активного списка можно использовать для обогащения корреляционных событий. Для этого его нужно сохранить в виде атрибута записи в активном списке операционным правилом. А затем в корреляционном правиле в разделе действий нужно будет выполнить операцию get над активным списком и записать в какое-нибудь поле корреляционного события содержимое атрибута записи из активного списка.
Если есть несколько служб коррелятора, использующих один и тот же ресурс списка, у каждой будет свое состояние этого списка.
Чтобы просмотреть содержимое списка нужно открыть список активных сервисов (Active services), выбрать службу типа Correlator (поставить галочку слева) и у нее появится активная кнопка Go to active lists (Перейти в активные листы).
Аналитик может экспортировать и импортировать, а также очистить содержимое списка. Аналитик также может отобразить содержимое списка, увидеть, какие в нем есть записи, когда они были созданы или обновлены, и когда у них истекает время жизни, если для списка настроено время жизни. Аналитик может вручную удалять (но не редактировать) записи.
Каждую запись можно открыть и изучить ее дополнительные атрибуты.
Активные листы в корреляторе
Активный лист – это контейнер для данных (представляет собой структуру ключ и значение), предназначенный для быстрой записи/чтения динамических данных, доступных всем фильтрам и корреляционным правилам в рамках одного сервиса Коррелятора. Чтобы их посмотреть, нужно из активных сервисов нажать кнопку Смотреть активные листы.
Взаимодействовать с активным листом могут не только компоненты коррелятора, но и пользователи, с помощью Web-консоли и API.
Активные листы работают в памяти коррелятора, также при работе активного листа используется Write-Ahead Log (WAL), который предполагает сохранение каждого изменения состояния в виде двоичного файла на жестком диске. Каждой записи журнала присваивается уникальный идентификатор, позволяющий выполнять дополнительные операции с журналом, такие как сегментация журнала и очистка. Уникальность записей журнала также помогает применять обновления журнала с использованием единой очереди обновлений, обеспечивая последовательные и согласованные обновления.
Синхронизация активного листа между несколькими корреляторами. ВАЖНО это не поддерживаемый и не официальный сценарий. Можно попробовать сделать так: WAL записывается на сетевую папку, примонтиованную к двум корреляторам. И если один из корреляторов упал, скрипт (заранее написанный) запускает службу второго коррелятора. Второй коррелятор перечитает WAL и импортирует данные в лист.
Приемы в правилах корреляции
Сравнение с константой
Сравнение с листом/списком
Аналогично =константе ИЛИ =константе
Содержит список констант регистронезависимый
Ищется заданная подстрока “whoami” или "ipconfig" и др в занчении поля DestinationProcessName
Соответствие регулярному выражению (REGEX)
Должно быть условие регулярного выражения в формате RE2
Работа с подсетями
Содержит любое значение (не пустое)
Активный лист содержит ключ
Ключ листа должен совпадать со значением поля Message
Сравнение по экстра данным в активном листе (неключевому полю)
Где threat_score > 70
Сравнение переменной с числовым значением
Из строки с REGEX вырезается число и кладется его в переменную, например, для получения SID пользователя из "SID: 1-21-1231-500", получаем "500", кладем в переменную $temp, чтобы сделать сравнение необходимо $temp привести к числовому типу это можно сделать новой переменной $usersid = $temp + 0
и далее сравнивать, например, $usersid > 1000
Условие с полем TI, сравнение с категорией
Значение feed это именование фидов из CyberTrace, например для фидов Kaspersky бывают такие значения (зависит какие фиды приобретены/подключены):
Фильтр по количеству записей по ключу в активном листе
Где количество записей > 1, используется служебная переменная _count
Другие служебные поля активных листов:
- _count (счетчик количества записей)
- _created (время создания записи UnixTime, в наносекундах)
- _updated (время обновления записи UnixTime, в наносекундах)
- _expires (время окончания жизни записи UnixTime, в наносекундах)
- _key (значение ключевой записи)
Событие истечения времени жизни в активном листе
Возникает служебное событие active list record expired, помимо этого необходимо указать UUID активного листа в поле DeviceExternalID
Значение ключевого поля передатся в DevicePayloadID служебного события.
Работа с активным листом не по его ID
Если указывать ID листа неудобно (а это обычно так), то можно в ключевое поле писать уникальный префикс типа "failed login attempts|username|1.1.1.1" и в события ловить по полю devicePayloadID функцией startsWith "failed login attempts" такой вариант реализации правила не зависит от инсталляции.
Такие события существуют только в рамках коррелятора (служебные события) и не сохраняются в сторадже, их можно поймать только правилом корреляции.
Работа с группами AD
Необходимо указывать полный DN, пример:
Условие фильтра:
Актив находится в определенной категории
Работа с полем Extra в селекторе
Работа с полями типа SA (массив строк) KUMA 3.0+
При операции match к полям типа SA применяется как к строке, т.е. массив представляется в виде строки ["a1", "b2", "c3"] и т.д. Т.е. ко всему массиву сразу, а не к отдельным его элементам по очереди.
При операции contains применяется именно к элементам массива. Т.е. contains [ для массива вернет false, как и contains " или '. Но при этом же, если в массиве есть элемент abc, то contains abc вернет true и contains ab тоже вернет true
Работа с переменными (KUMA 2.1+)
Вход в рабочее время
Документация по функциям переменных.
Сначала извлекается час из таймштемпа, с помощью функции: extract_from_timestamp(Timestamp, ‘h’, ‘Europe/Moscow’)
Условие в селекторе:
Переменные необходимо указывать в группирующих полях:
Работа с Extra
Здесь вместо Event.System.Channel нужно указать интересующее вас поле экстра. Регулярка: ."Event\.System\.Channel":"([^"]+)".
Работа с контекстной таблицей (KUMA 3.0+)
Исходная задача: Необходимо отслеживать, на каких отличных друг от друга устройствах производится вход одной УЗ. (активный лист будет менее удобен т.к. различных устройств может быть > 1)
Пример Контестной таблицы:
Примеры переменных:
Извлечение содержимого поля с массивом pc
из контекстной таблицы в тенанте Shared
(только для этого тенанта нужно в переменной это указывать) по ключевому полю user
и его значением DestinationUserName
, назовем переменную ct_value
:
context_table('pc-user@Shared', 'pc', 'user', DestinationUserName)
В событии выглядит это так:
Получение индекса (номер символа) по содержимому поля массива pc
по значению поля DestinationHostName
из контекстной таблицы в тенанте Shared (только для этого тенанта нужно в переменной это указывать) по ключевому полю user и его значением DestinationUserName, назовем переменную ct_contains
:
index_of(DestinationHostName, $ct_value)
Возвращает первую позицию символа или подстроки в строке, расчет индекса начинается с 0. Если в результате работы функции подстрока не была найдена, функция вернёт значение -922337203685477580
Вот как выглядит это в событии, значение ct_contains
в поле FlexNumber1.
Получение количества элементов в поле с массивом pc
из контекстной таблицы (пример в собтии на рисунке выыше в поле FlexNumber2), назовем переменную ct_len
:
len($ct_value)
Если в содержимом поля с массивом pc
из контекстной таблицы есть подстока см. выше описание переменной ct_contains
, то вернуть true
назовем переменную ct_item_exist
:
conditional(`$ct_contains LIKE '-.*'`, 'false', 'true')
Вот как выглядит это в событии, значение ct_item_exist
в поле FlexString2.
Как записать значение из нескольких событий в одно поле в правиле корреляции типа Standard
В обогащении можнонаписать шаблон, в котором можно пройтись по всем подсобытиям и получить список всех полей. Дальше уже с помощью функционала Go template можете сделать что хотите. Для правильной работы метода keep event policy в правиле корреляции должна иметь значение all.
В данном примере можно получить в строку все значения SourceAddress из базовых событий через ";" в корреляционном событии.
На выходе получается примерно следующее
Тестирование правил корреляции
Для тестирования правил можно использовать ретроскан (из раздела “События”), предварительно это правило нужно добавить в коррелятор и осуществить выборку интересующих событий запросом (выбрать временной диапазон):
В нашем случае, если правило сработает создастся алерт, заполнятся листы, если это есть в действиях правила корреляции. Также можно включить опцию запуска реагирования.
ыыаыа
Производительность правил корреляции
Написание правил
В правилах корреляции очередность условий в селекторах имеет значение
Уникальные условия надо поднимать вверх в правиле корреляции:
Еще, например, есть правило, в котором в переменную кладется значение из активного листа, а затем эта переменная сравнивается в условии. Так вот в этом случае очередность условий имеет большое значение, так как поменяв условия местами и отодвинув проверку по активному листу в конец, в метриках количество OPS с активным листом уменьшилось со 100000 OPS до 1,1 OPS.
При наличии условия с листами, словарями и т.д., отодвигайте их в конец.
Мониторинг произвоительности
Для мониторинга производительности по корреляции есть метрики, градации веса по операциям в продукте нет, все выполняется быстро благодаря GoLang. Метрики по правилам можно увидеть в разделе метрики, нажав на название “KUMA Collectors” затем выбрав “KUMA Correlators”:
Пример метрик по корреляции:
Сегментация правил корреляции
По умолчанию, если в корреляторе какое-то правило корреляции сработает несколько раз, все созданные в результате этого корреляционные события будут присоединены к одному алерту. Правила сегментации алертов дают возможность определить условия, при которых на основе таких однотипных корреляционных событий будут создаваться разные алерты.
Порядок применения правил сегментации соответсвует порядку правил сегментации созданным в интерфейсе KUMA (могут примениться и несколько сегментаций, если сработка правила корреляции соответствует нескольким правилам сегментации)
Блок-схема работы сегментации (спасибо за наработку интегратору):
Пример работы, есть правило корреляции, которое срабатывает на событие с полем Code равное "777" с разными занчениями SourceUserName, все сработки правила складваются в один Алерт, мы хотим создать отдельные алерты для отдельных пользователей:
Для этог онужно создать правило сегментации ипривязать его к правилу корреляции. Перейдите в Ресурсы - Правила сегментации и нажмите на кнопку Добавить правило сегментации. В нашем случае подойдет тип По группирующим полям:
Указываем свой шаблон именования и группирующее поле - используем поля имени пользователя и Сохраняем:
Далее необходимо привязать правило сегментации к нашему правилу корреляции. Это делается в Параметры - Алерты - вкладка Сегментация. Выбираем необходимый тенант, нажимаем кнопку Добавить, Указываем название, Выбираем правило корреляции и добавлем правило сегментации, затем на каждом шаге все Сохраняем.
По итогу, при появлении событий удовлетворяющих правилу корреляции мы получим отдельные алерты на основе SourceUserName.
Аналогичным образом можно использовать и другие типы правил сегментации.
Правила сбора и анализа данных (Data Mining)
В отличие от потоковой корреляции, работающей в режиме реального времени, Data Mining правила позволяют с помощью языка SQL и функций ClickHouse (примеры запросов, почти все возможно использовать) распознавать и анализировать события, сохраненных в хранилище KUMA (можно указать и конкретный спейс хранилища).
Для работы необходимо указать, рассмотрим на примере:
- В Ресурсах - Правила сбора и анализа данных Создать правило
- В правиле указать:
-
- Интервал (частота) выполнения SQL-запроса можно указать в минутах, часах и днях (минимум 1 минута)
- SQL-запрос должен содержать функцию агрегации (примеры) и/или группировку (GROUP BY) данных c обязательным указанием ограничения LIMIT (от 1 до 10 000)
Каждое выполнение такого правила происходит в виде запроса в Хранилище, а это значит неосторожным движением в виде частого или тяжелого правила можно нагрузить базу больше чем хотелось бы
В примере рассматривается запрос на основе событий Windows по пользователям (DestinationUserName) событиям входа (EventID 4624) и выхода (EventID 4634) с расчетом среднего времени сесии пользователя за последние 24 часа.
Посмотреть SQL запрос (пример)
SELECT
login_events.DestinationUserName AS destination_user_name,
round(AVG(logout_events.logout_time - login_events.login_time)/1000) AS avg_time_diff_s,
COUNT(DISTINCT login_events.login_time) AS total_logins,
COUNT(DISTINCT logout_events.logout_time) AS total_logouts,
concat(
toString(floor(avg_time_diff_s / 86400)), ' days, ',
toString(floor((avg_time_diff_s % 86400) / 3600)), ' hours, ',
toString(floor((avg_time_diff_s % 3600) / 60)), ' minutes, ',
toString(avg_time_diff_s % 60), ' seconds'
) AS human_readable_diff
FROM
(SELECT
DestinationUserName,
toUnixTimestamp(EndTime) AS login_time,
FlexString1 AS logon_id
FROM `events`
WHERE DeviceEventClassID = '4624'
AND EndTime >= now() - INTERVAL 24 HOUR
AND DestinationUserName NOT LIKE '%$%') AS login_events
INNER JOIN
(SELECT
DestinationUserName,
toUnixTimestamp(EndTime) AS logout_time,
FlexString1 AS logon_id
FROM `events`
WHERE DeviceEventClassID = '4634'
AND EndTime >= now() - INTERVAL 24 HOUR
AND DestinationUserName NOT LIKE '%$%') AS logout_events
ON login_events.DestinationUserName = logout_events.DestinationUserName
AND logout_events.logon_id = login_events.logon_id
WHERE logout_events.logout_time >= login_events.login_time
GROUP BY login_events.DestinationUserName
ORDER BY avg_time_diff_s DESC
LIMIT 100
-
- Добавить маппинг (сопоставление) по полям запроса и модели KUMA
- В Ресурсах - Сбор и анализ данных добавить ранее созданное правило
- Открыть правило и установить связи:
- Привязать хранилище по которому будет осуществляться поиск на вкладке Привязанные хранилища
- Привязать коррелятор с соответвующим правилом корреляции для сработки на вкладке Привязанные корреляторы
- Для ручного запуска нажмите кнопку Запустить
- По результатам запроса на выходе будут какие-то данные, которые не будут нигде сохраняться, но на них можно настроить правило корреляции. В нашем случае правило ловит события, где время сессии меньше 5 секунд:
Корреляционное событие выглядит следующим образом:
А событие на основе которого произошла сработка:
Еще пример:
Работу правил можно отслеживать с помощью метрик в разделе KUMA Core: