Описание функций ClickHouse для работы с запросами: [https://clickhouse.com/docs/ru/sql-reference/functions/](https://clickhouse.com/docs/ru/sql-reference/functions/)
Поиск по атрибутам - [https://support.kaspersky.ru/kuma/4.2/294927](https://support.kaspersky.ru/kuma/4.2/294927)
Запрос из интерфейса пробрасывается в БД с добавлением границ временного промежутка и выбранных тенантов, пример: `... AND (Timestamp >= 1715689595208 AND Timestamp <= 1715689895208) AND (TenantID IN ('a1fbde7a-76d3-4bbc-a769-82126b41b56f')) ORDER BY ...`
## ## Время в событиях Если в событии присутствует информация о таймзоне (deviceTimeZone) или корректно парсится timestamp события в нормализаторе, где получаем ±hh:mm и маппится это значение в поле DeviceTimeZone - событие поступает в KUMA со значением этой таймзоны Если в событии нет deviceTimeZone или время из события не содержит таймзону, которую можно распарсить, то для такого события устанавливается таймзона сервера на котором установлен коллектор. Таймзону еще можно поправить следующим образом: - создать правило обогащения типа таймзона и скорректировать временную зону; - в поисковый запрос добавить константу с таймзоной (пример, SELECT '+0300' as deviceTimeZone), можно так же указать таймзону формата +03:00 или +03. ## Базовые запросы (поиск событий) --- ### Типы событий в KUMA  --- ### Подсчет событий по полю ```sql SELECT count(ID) as count_num, DeviceVendor FROM `events` GROUP BY DeviceVendor ORDER BY count_num DESC LIMIT 250 ```  --- ### Выполнение математических операций и сравнений ```sql SELECT DeviceProduct, SourceUserName, round(sum(BytesIn)/1024, 2) as KiloBytes FROM `events` WHERE BytesIn > '0' OR BytesOut > '0' GROUP by SourceUserName, DeviceProduct ORDER BY KiloBytes DESC LIMIT 250 ``` Иногда бОльшую производительность дает условие со скобками `WHERE (BytesIn > '0' OR BytesOut > '0')` --- ### По подстроке (регистрозависимый) с условием И ```sql SELECT * FROM `events` WHERE DeviceHostName like '%serv%' AND DeviceProduct = 'Windows' ORDER BY Timestamp DESC LIMIT 250 ```  --- ### По подстроке (регистроНЕзависимый) ```sql SELECT * FROM `events` WHERE DeviceEventCategory ilike '%auditing%' AND DeviceProduct = 'Windows' ORDER BY Timestamp DESC LIMIT 250 ```  --- ### По исходному / сырому событию ```sql SELECT * FROM `events` WHERE Raw ilike '%technique%' ORDER BY Timestamp DESC LIMIT 250 ```  --- ### По нескольким значениям, вместо OR можно использовать IN ```sql SELECT * FROM `events` WHERE DeviceEventClassID IN ('BROKER_USERLOGGEDOUT', 'BROKER_USERLOGGEDIN') AND DeviceProduct = 'Horizon' ORDER BY Timestamp DESC LIMIT 250 ```  --- ### По подсети ```sql SELECT * FROM `events` WHERE inSubnet(DeviceAddress, '10.68.85.70/32') ORDER BY Timestamp DESC LIMIT 250 ```  ### Внешние IPv4 адреса ```sql SELECT count(ID) as count, DestinationAddress FROM `events` WHERE NOT empty(DestinationAddress) AND NOT (inSubnet(DestinationAddress, '10.0.0.0/8') OR inSubnet(DestinationAddress, '172.16.0.0/12') OR inSubnet(DestinationAddress, '192.168.0.0/16') OR inSubnet(DestinationAddress, '127.0.0.0/8')) AND NOT isIPv6String(DestinationAddress) GROUP BY DestinationAddress ORDER BY count DESC LIMIT 250 ``` [](https://kb.kuma-community.ru/uploads/images/gallery/2024-12/gUdimage.png) ### По отсутствующему полю ```sql SELECT * FROM `events` WHERE empty(DeviceEventClassID) AND Raw ilike '%backdoor_user%' ORDER BY Timestamp DESC LIMIT 250 ```  --- ### По регулярному выражению ```sql SELECT * FROM `events` WHERE match(DestinationUserName, '^\w+\-\d\$$') ORDER BY Timestamp DESC LIMIT 250 ```  --- ### По регулярному выражению с выбором группы ```sql SELECT count(ID) AS `metric`, regexpExtract(FlexString1,'^((https?:\/\/)?([\da-z\.-]+\.[a-z\.]{2,6}|[\d\.]+)([\/:?=&#]{1}[\da-z\.-]+)*\S+)$', 1) AS `value` FROM `events` WHERE DeviceProduct = 'DFI' AND `value` != '' GROUP BY FlexString1 ORDER BY `metric` DESC LIMIT 250 ``` [](https://kb.kuma-community.ru/uploads/images/gallery/2025-09/kxbimage.png) --- ## Проверка работы обогащения --- ### Обогащение событий Активами ```sql SELECT * FROM `events` WHERE (DeviceAssetID != '' OR SourceAssetID != '' OR DestinationAssetID != '') ORDER BY Timestamp DESC LIMIT 250 ``` --- ### Обогащение событий с LDAP ```sql SELECT * FROM `events` WHERE (SourceAccountID != '' OR DestinationAccountID != '') ORDER BY Timestamp DESC LIMIT 250 ``` --- ### Обогащение событий данными из TI ```sql SELECT * FROM `events` WHERE NOT TI='' ORDER BY Timestamp DESC LIMIT 250 ``` --- ### Сортировка по текстовому значению (Severity) Сортировка от Critical до Low ```sql SELECT ExternalID AS `ID Обнаружения`, Severity AS `Критичность`, DeviceEventCategory AS `Категория`, FlexString1 AS `Объект`, JSONExtractString(Extra, 'name') AS `Описание`, DeviceCustomString5 AS `Рекомендация` FROM `events` WHERE DeviceProduct = 'DFI' ORDER BY indexOf(['Critical', 'High', 'Medium', 'Low'], Severity) LIMIT 250 ``` [](https://kb.kuma-community.ru/uploads/images/gallery/2025-09/image.png) --- ## Работа с Extra полем --- ### По полю Extra содержащие ключ ```sql SELECT * FROM `events` WHERE visitParamHas(Extra, 'memUsage') ORDER BY Timestamp DESC LIMIT 250 ``` [](https://kb.kuma-community.ru/uploads/images/gallery/2023-08/y7Pimage.png) --- ### По полю Extra содержащие ключ с определенным значением ```sql SELECT * FROM `events` WHERE visitParamExtractString(Extra, 'memUsage') = '61367' ORDER BY Timestamp DESC LIMIT 250 ``` ```sql SELECT * FROM `events` WHERE JSONExtractString(Extra, 'memUsage') = '61367' ORDER BY Timestamp DESC LIMIT 250 ``` [](https://kb.kuma-community.ru/uploads/images/gallery/2023-08/lhSimage.png) --- ### По полю Extra НЕ содержащие ключ с определенным значением ```sql SELECT * FROM `events` WHERE visitParamHas(Extra, 'memUsage') AND NOT visitParamExtractString(Extra, 'memUsage') = '61367' ORDER BY Timestamp DESC LIMIT 250 ```  ### Подсчет количества вхождений в массив ```sql SELECT sum(length(extractAll(Extra, '"tags\\.[0-9]+":'))) AS `metric` FROM `events` WHERE DeviceProduct = 'DFI' AND DeviceEventCategory = 'vulnerability' LIMIT 250 ``` [](https://kb.kuma-community.ru/uploads/images/gallery/2025-09/LtHimage.png) [](https://kb.kuma-community.ru/uploads/images/gallery/2025-09/YnNimage.png) --- ## Работа со временем Timestamp по умолчанию в формате epoch time числа с милисекундами (UnixTimestamp)Список тайм зон: [https://en.wikipedia.org/wiki/List\_of\_tz\_database\_time\_zones](https://en.wikipedia.org/wiki/List_of_tz_database_time_zones)
[](https://kb.kuma-community.ru/uploads/images/gallery/2023-08/qvUimage.png) --- ### Задание таймзоны ```sql SELECT Timestamp, fromUnixTimestamp64Milli(Timestamp, 'Europe/Moscow') as NormTime, DeviceProduct, Name, Message FROM `events` WHERE DeviceProduct = 'EDR' ORDER BY Timestamp DESC LIMIT 250 ``` [](https://kb.kuma-community.ru/uploads/images/gallery/2023-08/rHRimage.png) --- ### Указание своего формата времени ```sql SELECT Timestamp, formatDateTime(fromUnixTimestamp64Milli(Timestamp), '%d-%m-%Y %H:%i:%S') as NormTime, DeviceProduct, Name, Message FROM `events` WHERE DeviceProduct = 'EDR' ORDER BY Timestamp DESC LIMIT 250 ``` [](https://kb.kuma-community.ru/uploads/images/gallery/2023-08/u8bimage.png) --- ### Свой формат времени с таймзоной ```sql SELECT Timestamp, formatDateTime(fromUnixTimestamp64Milli(Timestamp), '%d-%m-%Y %H:%i:%S', 'Europe/Moscow') as NormTime, DeviceProduct, Name, Message FROM `events` WHERE DeviceProduct = 'EDR' ORDER BY Timestamp DESC LIMIT 250 ``` [](https://kb.kuma-community.ru/uploads/images/gallery/2023-08/YNHimage.png) --- ### События в "рабочее время" с 9 до 18 ```sql SELECT Timestamp, formatDateTime(fromUnixTimestamp64Milli(Timestamp), '%d-%m-%Y %H:%i:%S', 'Europe/Moscow') as NormTime, DeviceProduct, Name, Message FROM `events` WHERE DeviceProduct = 'EDR' AND toHour(fromUnixTimestamp64Milli(Timestamp, 'Europe/Moscow')) >= 9 AND toHour(fromUnixTimestamp64Milli(Timestamp, 'Europe/Moscow')) < 18 ORDER BY Timestamp ASC LIMIT 250 ``` [](https://kb.kuma-community.ru/uploads/images/gallery/2023-08/w7Jimage.png) --- ### Подсчет события по фильтру за период вчерашнего дня с 00 до 04 часов с группировкой по SourceAddress ```sql SELECT count(ID) as Count_Num, SourceAddress, groupArray(fromUnixTimestamp64Milli(Timestamp)) as time_stm FROM `events` WHERE DeviceEventClassID = '4624' AND DeviceProduct = 'Windows' AND Timestamp >= toUnixTimestamp(toStartOfDay((now() - INTERVAL 1 DAY)))*1000 AND Timestamp <= toUnixTimestamp(toStartOfDay((now() - INTERVAL 1 DAY)) + INTERVAL 4 HOUR)*1000 GROUP BY SourceAddress LIMIT 250 ``` [](https://kb.kuma-community.ru/uploads/images/gallery/2023-08/Ye0image.png) --- ### Подсчет уникальных адресов посетителей kb.kuma-community.ru ```sql SELECT count(DISTINCT SourceAddress) as uniqSrcIP FROM `events` WHERE Code = '200' AND DeviceCustomString2 != '-' AND DestinationServiceName = 'apache_access' AND DeviceCustomString3 = 'GET' AND DestinationHostName = 'kb' AND not inSubnet(SourceAddress, '91.103.66.0/24') ``` [](https://kb.kuma-community.ru/uploads/images/gallery/2023-10/psGimage.png) Можно использовать аналогичную более быструю функцию: ```sql SELECT uniq(SourceAddress) as uniqSrcIP FROM `events` WHERE Code = '200' AND DeviceCustomString2 != '-' AND DestinationServiceName = 'apache_access' AND DeviceCustomString3 = 'GET' AND DestinationHostName = 'kb' AND not inSubnet(SourceAddress, '91.103.66.0/24') ``` --- ### Подсчет уникальных адресов посетителей kb.kuma-community.ru по дням ```sql SELECT count(DISTINCT SourceAddress) as `metric`, formatDateTime(fromUnixTimestamp64Milli(Timestamp), '%d-%m-%Y', 'Europe/Moscow') as value FROM `events` WHERE Code = '200' AND DeviceCustomString2 != '-' AND DestinationServiceName = 'apache_access' AND DeviceCustomString3 = 'GET' AND DestinationHostName = 'kb' AND not inSubnet(SourceAddress, '91.103.66.0/24') GROUP BY value ``` [](https://kb.kuma-community.ru/uploads/images/gallery/2023-11/np2image.png) --- ### Подсчет среднего количесва переданных байт посетителей kb.kuma-community.ru по источнику IP c 00:00 по 08:00 ```sql SELECT avg(toInt32(BytesOut)), SourceAddress FROM `events` WHERE BytesOut!= 0 AND Timestamp >= toUnixTimestamp(toStartOfDay((now() - INTERVAL 1 DAY)))*1000 AND Timestamp <= toUnixTimestamp(toStartOfDay((now() - INTERVAL 1 DAY)) + INTERVAL 8 HOUR)*1000 GROUP BY SourceAddress DESC LIMIT 250 ``` [](https://kb.kuma-community.ru/uploads/images/gallery/2023-11/537image.png) ### Отображение количества переданных и принятых байт по внутренним адресам более 1 Гб ```sql SELECT StartTime , EndTime , SourceAddress AS `SOURSE ADDRESS`, DestinationPort AS `TO PORT`, ApplicationProtocol AS `APPLICATION`, DeviceCustomString1 AS `RULE`, formatReadableSize(BytesIn) AS `SENT`, formatReadableSize(BytesOut) AS `RECEIVED`, formatReadableSize(FlexNumber1) AS `TOTAL` FROM `events` WHERE FlexNumber1 > 1000000000 GROUP BY StartTime, EndTime, SourceAddress, DestinationPort, ApplicationProtocol, DeviceCustomString1, BytesIn, BytesOut, FlexNumber1 ORDER BY `EndTime` DESC LIMIT 250 ``` [](https://kb.kuma-community.ru/uploads/images/gallery/2024-06/pGZimage.png) --- ## Экстра запросы --- ### Склейка по времени подключений пользователей с одного адреса и на один VPN сервер ```sql SELECT SourceUserName, SourceAddress, SourceHostName, groupArray(FlexString1) as time FROM `events` WHERE SourceProcessName = 'Create session' GROUP BY SourceUserName, SourceAddress, SourceHostName LIMIT 10 ``` [](https://kb.kuma-community.ru/uploads/images/gallery/2023-08/rqIimage.png) --- ### Склейка различных адресов подключений от одного пользователя на VPN сервереОператор HAVING доступен с версии KUMA 3.0
```sql SELECT uniq(SourceAddress) as "Количество уникальных адресов", DestinationUserName as "Имя пользователя", groupUniqArray(SourceAddress) as "Список адресов" FROM `events` WHERE DeviceProduct = 'Ngate' AND Name = 'Create session' GROUP BY "Имя пользователя" HAVING "Количество уникальных адресов" > 1 ORDER BY "Количество уникальных адресов" DESC LIMIT 10 ``` [](https://kb.kuma-community.ru/uploads/images/gallery/2024-04/fBuimage.png) --- ### Склейка наименования ПО и его версии Пример строки Extra `desc: ## port 445 tcp / SMB version 1 is used` Также с помощью replaceRegexpOne удаляем в конце слеш если нет заполненного транспорта с портом. ```sql SELECT count(ID) as `metric`, replaceRegexpOne(concat(regexpExtract(JSONExtractString(Extra, 'desc'), '.+port\s(\d+)', 1), '/', regexpExtract(JSONExtractString(Extra, 'desc'), '.+port\s(\d+)\s+(tcp|udp)', 2)),'(.+)\/$','\\1') as `value` FROM `events` WHERE DeviceProduct = 'DFI' AND value NOT IN('', '/') GROUP BY value ORDER BY `metric` DESC LIMIT 250 ``` [](https://kb.kuma-community.ru/uploads/images/gallery/2025-09/orkimage.png) --- ### Обрезка доменов до второго уровня и условие с несколькими запросами ```sql SELECT count(ID) as cnt, cutToFirstSignificantSubdomain(DestinationHostName) as dstH FROM `events` WHERE DeviceCustomString1 = 'Q' AND DestinationProcessName = 'DNS' AND DeviceCustomString5 IN ('A', 'AAAA', 'HTTPS') GROUP BY dstH ORDER BY cnt DESC LIMIT 50 ``` [](https://kb.kuma-community.ru/uploads/images/gallery/2023-08/FGfimage.png) --- ### Обрезка доменов до второго уровня и исключение IP адресов с помощью регулярного выражения ```sql SELECT count(ID) as `metric`, cutToFirstSignificantSubdomain(RequestUrl) as `value`, match(`value`, '^([a-zA-Z0-9-]+\.)+[a-zA-Z]{2,}$') AS is_valid_domain FROM `events` WHERE DeviceProduct = 'UTM' AND FlexString2 = 'Instant Messaging' AND Type != 3 AND is_valid_domain = 1 GROUP BY `value` ORDER BY `metric` DESC LIMIT 250 ```  --- ### Группировка событий по категории из TI (Regex) ```sql SELECT count(ID) as cnt, extract(TI, '.+category\"\:\"([^"]+).+') as category FROM `events` WHERE TI !='' GROUP BY category ORDER BY cnt DESC ``` [](https://kb.kuma-community.ru/uploads/images/gallery/2023-10/r5Bimage.png) --- ### Все базовые события конкретного корреляционного события по его ID ```sql SELECT * FROM `events` WHERE ID IN ( SELECT arrayJoin(splitByChar(',', replaceRegexpAll(JSON_QUERY(BaseEvents, '$[*].ID'), '\\[|\\]|"| ', ''))) as TestID FROM `events` WHERE (Type = 3) AND (ID = '25e4eae3-ad6d-4114-b99e-019de3574d16') ) ``` [](https://kb.kuma-community.ru/uploads/images/gallery/2023-08/9Bbimage.png) --- ### Дерево процесса ```sql SELECT b.DeviceHostName AS HostName, b.SourceAccountID AS AccountID, b.SourceUserName AS UserName, a.DeviceCustomString3 AS GrandParentProcessID, a.SourceProcessName AS GrandParentProcessName, a.DeviceCustomString5 AS ParentProcessID, a.DestinationProcessName AS ParentProcessName, b.DeviceCustomString5 AS ProcessID, b.DestinationProcessName AS ProcessName, concat(a.SourceProcessName, ' -> ', a.DestinationProcessName, ' -> ', b.DestinationProcessName) AS ProcessTree FROM `events` AS a INNER JOIN ( SELECT * FROM `events` WHERE DeviceEventClassID='4688' ) AS b ON a.DeviceCustomString5 = b.DeviceCustomString3 WHERE DeviceEventClassID='4688' ```  ### Среднее время сессии Windows пользователей за 24 часа Рассматривается запрос на основе событий 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('Europe/Moscow') - 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('Europe/Moscow') - 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 ``` Несмотря на ошибку запрос выполняется корректно:  ### Частота и отношение неуспешных входов к успешным в Windows ```sql SELECT count(CASE WHEN DeviceEventClassID = '4625' THEN 1 END) as `Failed logins`, count(CASE WHEN DeviceEventClassID = '4624' THEN 1 END) as `Success logins`, `Failed logins`/`Success logins` as `Ratio` FROM 'events' HAVING `Ratio` > 0.03 ``` Или ```sql SELECT countIf(DeviceEventClassID = '4625') as F, countIf(DeviceEventClassID = '4624') as S FROM 'events' LIMIT 250 ``` [](https://kb.kuma-community.ru/uploads/images/gallery/2025-01/Lstimage.png) --- ## Работа с кастомными полями (SA. NA.) ### Вхождение элемента по полюПоиск проискодит по точному совпадению с элементом
```sql SELECT * FROM `events` WHERE has(SA.user_groups, 'system:serviceaccounts') ORDER BY Timestamp DESC LIMIT 250 ```  --- ### Вхождение по подстроке элемента по полю (где более 2 значений в массиве)Счет в массиве начинается с нуля
```sql SELECT * FROM `events` WHERE length(arrayFilter(x -> x LIKE '%serviceaccounts%', SA.user_groups)) > 1 ORDER BY Timestamp DESC LIMIT 250 ``` 