ClickHouse: настройка источника

Данное руководство описывает процесс настройки сбора и отправки событий ClickHouse в R-vision SIEM.

Настройка ClickHouse

Настройте передачу событий из файла. Для этого отредактируйте файл /etc/clickhouse-server/config.xml:

Раскомментируйте таблицы, название которых оканчивается на _log.
Задайте нужный уровень: level = 8.

СУБД ClickHouse сохраняет логи в разные таблицы системной БД system, например:

query_log — лог запросов к СУБД.
backup_log — логи резервного копирования. Подробнее о настройке резервного копирования см. в официальной документации ClickHouse.
session_log — логи входа в систему и выхода из системы.

С полным списком системных таблиц можно ознакомиться в официальной документации ClickHouse.

Настройка в R-Vision SIEM

Настройка подключения ClickHouse в качестве источника на конвейере R-Vision SIEM сводится к подключению в качестве точек входа нескольких системных таблиц.

Для настройки сбора и нормализации событий источника в R-Vision SIEM выполните следующие шаги:

В интерфейсе R-Vision SIEM создайте секрет со строкой подключения. Для этого:
1. Перейдите в раздел Ресурсы → Секреты.
2. Нажмите на кнопку Создать ().
3. В раскрывшемся окне создания секрета заполните поля:
  - Название: введите название секрета.
  - Описание (опционально): опишите, для чего будет использоваться секрет.
  - Тип секрета: выберите вариант Строка подключения.
  - Строка подключения: введите строку вида:
    
    jdbc:clickhouse://<address>:<port>/default?user=<user>&password=<password>&compress_algorithm=gzip
    
    Здесь:
    
    <address> — адрес сервера базы данных.
    
    <port> — порт подключения.
    
    <user> — логин пользователя базы данных.
    
    <password> — пароль пользователя базы данных.
    
    Пример 1. Пример строки подключения:
    
    jdbc:clickhouse://clc03.sea.land:8123/default?user=user&password=password&compress_algorithm=gzip
4. Нажмите на кнопку Создать.
В интерфейсе R-Vision SIEM создайте новый конвейер в коллекторе.

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

Название: введите название точки входа.
Тип точки входа: выберите вариант Database.
Драйвер базы данных: выберите вариант ClickHouse.
Адрес подключения: выберите секрет, созданный ранее.

SQL-запрос: для каждой таблицы введите отдельный SQL-запрос:

Таблица system.query_log:

SELECT FQDN() AS hostname, 'clickhouse' AS appname, 'query_log' AS at, type, event_time AS timer, query_start_time,current_database, query, query_kind, databases, tables, user, query_id, address, port, initial_user, initial_query_id, initial_port, interface, is_secure, os_user, client_hostname, client_name, http_user_agent FROM system.query_log WHERE timer > FROM_UNIXTIME(toInt32(?)) ORDER BY timer ASC LIMIT 50;

Таблица system.session_log:

SELECT FQDN() AS hostname, 'clickhouse' AS appname, 'session_log' AS at, type, auth_id, event_time AS timer, user, auth_type, profiles, roles, client_address, client_port, interface, failure_reason FROM system.session_log WHERE timer > FROM_UNIXTIME(toInt32(?)) ORDER BY timer ASC LIMIT 50;

Таблица system.backup_log:

SELECT FQDN() AS hostname, 'clickhouse' AS appname, 'backup_log' AS at, hostname AS dvchost, id AS externalId, base_backup_name AS objPath, event_time_microseconds AS timer, 'Info' AS severity, status AS act, error AS reason, name AS objName FROM system.backup_log WHERE timer > FROM_UNIXTIME(toInt32(?)) ORDER BY timer ASC LIMIT 10;

Оператор LIMIT в запросах — обязательный параметр, определяющий выводимое число строк. Конкретное число строк для ограничения значения LIMIT подбирается индивидуально по загруженности СУБД. Если на эксплуатируемой СУБД количество возвращаемых строк в результате выполнения запроса больше, чем то, которое указано в запросе, то примерно оценить LIMIT можно через запрос, описанный ниже.

В приложении DBeaver выполните SQL-запрос, чтобы узнать, сколько примерно строк возвращается за 15 секунд, именно такой интервал указывается при настройке точки входа. Если у вас иной интервал, нужно учесть его в запросе. Для этого измените значение toInt32(INTEGER) на свой интервал.

SELECT event_time AS rt, level AS severity, logger_name AS aid, message AS msg FROM system.text_log WHERE rt > FROM_UNIXTIME(toInt32(1712239973)) AND rt <= FROM_UNIXTIME(toInt32(1712239988)) ORDER BY rt DESC;

Поле идентификатора: введите ключ timer со значением, указывающим момент времени в формате Unix time. Можно воспользоваться конвертером.
Интервал запроса, секунд: введите значение 15. Можно выставить иной интервал.
Максимальное количество строк (опционально): введите значение 100. Можно выставить иное значение.

Добавьте на конвейер элемент Нормализатор с правилом ClickHouse (идентификатор правила: RV-N-193).
Соедините нормализатор с точками входа.
Добавьте на конвейер элемент Конечная точка типа Хранилище событий.
Соедините конечную точку с нормализатором.
Сохраните и установите конфигурацию конвейера.