Императивные правила

В данном разделе представлен обзор структуры, принципа работы, процесса создания и расширения конфигурации императивного правила корреляции R-Vision SIEM. Управление правилами корреляции в системе осуществляется в разделе Экспертиза.

Об императивных правилах корреляции

Императивное правило корреляции в системе R-Vision SIEM представляет собой объект данных в формате RObject (.ro), содержащий детально определенные команды и процедуры для анализа и корреляции событий ИБ. Такие правила характеризуются точной спецификацией действий, определенным порядком операций, контрольными структурами и явным управлением состоянием.

Коллектор, настроенный для работы в режиме распределенной корреляции, не поддерживает работу императивных правил корреляции.

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

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

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

Императивные правила корреляции поддерживают три режима работы окна корреляции (window_mode):

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

    Если поле start_when не указано, окно будет открываться для каждого события, прошедшего через фильтр. В таком случае событие также будет добавлено в каждое открытое корреляционное окно.
  • Без перекрывания (non-overlapping) — это режим работы корреляционного окна, при котором правило имеет одновременно не более одного активного окна, таким образом открываемые правилом окна не пересекаются во времени. Окно создается при получении первого события, которое удовлетворяет условиям открытия окна. Последующее окно может быть открыто только после закрытия текущего окна. Таким образом, каждое событие может находиться не более чем в одном окне одновременно. Этот подход предотвращает дублирование анализа событий и обеспечивает предсказуемое распределение ресурсов для обработки данных.

  • Без состояния (stateless) — это режим работы корреляционного окна, при котором создаваемые правилом окна хранят данные только об одном событии. На каждое событие, удовлетворившем условию, создается индивидуальное корреляционное окно, которое сразу закрывается, генерируя корреляционное событие. Таким образом, каждое событие анализируется независимо от предыдущего, и их контекст не учитывается при анализе. Этот режим обеспечивает простоту реализации и минимальную нагрузку на систему, но ограничивает способность обнаружения сложных атак и аномалий, требующих анализа последовательности событий в контексте периода времени.

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

Режим работы окна корреляции определяется в поле window_mode. Каждое правило может функционировать только в одном выбранном режиме. Если режим не задан, система ожидает декларативный подход к написанию правила.

Структура правила корреляции

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

Строки, начинающиеся с символа $ ($foo), интерпретируются в VRL-блоках элементов экспертизы как переменные окружения. Чтобы избежать этого, экранируйте такие строки символом $ ($$foo).

Общие поля

Поля верхнего уровня:

Поле Описание Тип данных Обязательное поле

id

Уникальный идентификатор правила корреляции.

строка

да

name

Название правила, которое обозначает его цель или функцию.

строка

да

version

Версия правила корреляции, представленная в формате Semantic Versioning.

строка

да

author

Имя и контактные данные автора правила корреляции.

строка

нет

window_mode

Режим окна корреляции, определяющий каким образом система будет обрабатывать корреляционные события. Доступные режимы: parallel, non-overlapping, stateless. Если значение не задано, система ожидает декларативный подход к написанию правила.

строка

да

description

Описание правила корреляции, где можно указать дополнительную информацию о его назначении и функциональности.

строка

нет

data_source

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

массив строк

массив объектов

нет

tags

Список тегов, присвоенных правилу корреляции.

массив строк

нет

group_by

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

массив строк

нет

type

Тип правила. Всегда имеет значение correlation_rule для правила этого типа.

строка

да

severity

Уровень угрозы корреляционного события, испускаемого корреляционным правилом. Доступны значения: low, medium, high, critical.

строка

да

filter

Блок правил на языке VRL, определяющий условия (фильтр), при которых правило корреляции будет применяться к событиям.

строка

нет

on_correlation

Блок правил на языке VRL, определяющий логику обработки корреляционного события после его создания.

строка

нет

throttle_time_sec

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

целое число

нет

tests

Блок тестирования правила корреляции, содержащий массив тестовых событий и условия для проверки срабатывания правила. Обязательные поля: name, events, assertion.

объект

нет

Структура data_source

Поле data_source, представленное массивом строк, имеет структуру:

data_source:
- <platform> # Платформа или информационная система
- <source> # Источник событий.
# События (EventId) или путь к журналу событий:
- <EventID_1>
- <EventID_2>

Поле data_source, представленное массивом объектов, имеет следующую структуру:

Поле Описание Тип данных Обязательное поле

platform

Платформа или информационная система.

строка

нет

source

Источник событий.

строка

нет

events

События (EventId) или путь к журналу событий.

массив строк

нет

Структура tests

Поле Описание Тип данных Обязательное поле

name

Название теста. Позволяет идентифицировать конкретный тест в тестовой выдаче.

строка

да

events

Поток тестовых событий, поступающих в тестируемый элемент конвейера.

строка

да

assertion

Блок правил на языке VRL, который будет вызываться на каждое событие, испускаемое полем events. Используя встроенные функции assert и assert_eq, можно строить проверки на соответствие испускаемого события тестовым условиям.

строка

да

Дополнительные поля режима parallel

Поле Описание Тип данных Обязательное поле

on_event

Блок правил на языке VRL, определяющий логику обновления состояния окна корреляции при получении событий.

строка

да

correlate_when

Блок правил на языке VRL, определяющий условие закрытия корреляционного окна и генерации корреляционного события.

строка

да

start_when

Блок правил на языке VRL, определяющий условие открытия корреляционного окна. Если условие не задано, новое окно будет открываться на каждое событие прошедшее через фильтр.

строка

нет Language.adoc

expire_after_sec

Время жизни корреляционного окна, если не происходит обновление его состояния (в секундах).

целое число

нет

windows_limit

Максимально допустимое количество корреляционных окон, которое можно открыть одновременно.

целое число

нет

Дополнительные поля режима non-overlapping

Поле Описание Тип данных Обязательное поле

on_event

Блок правил на языке VRL, определяющий логику обновления состояния окна корреляции при получении событий.

строка

да

correlate_when

Блок правил на языке VRL, определяющий условие закрытия корреляционного окна и испускания корреляционного события.

строка

да

expire_after_sec

Указывает время автоматического закрытия корреляционного окна (по умолчанию 60 сек).

целое число

нет

Дополнительные поля режима stateless

Поле Описание Тип данных Обязательное поле

on_event

Блок правил на языке VRL, определяющий логику обновления состояния окна корреляции при получении событий. Если событие не обновляет окно, необходимо вызвать инструкцию abort.

строка

нет

correlate_when

Блок правил на языке VRL, определяющий условие закрытия корреляционного окна и испускания корреляционного события.

строка

да

Составление правила

Управление корреляционными правилами в системе осуществляется в разделе Экспертиза.

Чтобы составить правило корреляции:

  1. Заполните общие поля метаданных корреляционного правила:

    • id: задайте уникальный идентификатор правила.

    • name: задайте название правила.

    • window_mode (обязательное): определите режим работы корреляционного окна (parallel, non-overlapping или stateless).

    • type (обязательное): укажите correlation_rule в поле, чтобы определить тип правила.

    • severity (обязательное): задайте уровень угрозы испускаемого события. Доступны значения: low, medium, high, critical.

    • version: укажите версию правила в формате Semantic Versioning.

    • author (опциональное): укажите имя и контактные данные автора.

    • description (опциональное): предоставьте краткое описание назначения и функциональности правила.

    • tags (опциональное): добавьте теги для классификации и поиска правила.

  2. Заполните общие поля корреляционного правила:

    • filter (опциональное): создайте блок правил на языке VRL, который определит, к каким событиям будет применяться правило.

    • group_by (опциональное): определите поля, по которым будет производиться группировка событий для корреляции.

    • on_correlation (опциональное): создайте блок правил на языке VRL, которые необходимо выполнить после корреляции событий.

    • throttle_time_sec (опциональное): установите интервал в секундах, который определяет, как часто могут генерироваться корреляционные события для каждой группы в поле group_by.

  3. Заполните специфические поля в зависимости от выбранного режима корреляции (window_mode):

    Для правила parallel:

    • start_when (опциональное): создайте блок правил на языке VRL, определяющий условие открытия корреляционного окна. Если условие не задано, новое окно будет открываться на каждое событие прошедшее через фильтр.

    • on_event (обязательное): создайте блок правил на языке VRL, определяющий логику обновления состояния окна корреляции при получении событий. Если событие не обновляет окно, необходимо вызвать правил abort.

    • correlate_when (обязательное): создайте блок правил на языке VRL, определяющий условие закрытия корреляционного окна и генерации корреляционного события. Если условие удовлетворено, окно будет закрыто, и по нему будет сформировано корреляционное событие.

    • expire_after_sec (обязательное): задайте период времени, по истечении которого корреляционное окно будет закрыто автоматически.

    • windows_limit (опциональное): ограничьте количество корреляционных окон, которые можно открыть правилом (по умолчанию 100,000).

    Для правила non-overlapping:

    • on_event (обязательное): создайте блок правил на языке VRL, определяющий логику обновления состояния окна корреляции при получении событий. Если событие не обновляет окно, необходимо вызвать инструкцию abort.

    • correlate_when (обязательное): создайте блок правил на языке VRL, определяющий условие закрытия корреляционного окна и генерации корреляционного события. если условие удовлетворено, окно будет закрыто, и по нему будет сформировано корреляционное событие.

    • expire_after_sec (обязательное): зЗадайте период времени, по истечении которого корреляционное окно будет закрыто автоматически.

    Для правила stateless:

    • on_event (опциональное): создайте блок правил на языке VRL, определяющий логику обновления состояния окна корреляции при получении событий. Если событие не обновляет окно, необходимо вызвать инструкцию abort.

    • correlate_when (обязательное): создайте блок правил на языке VRL, определяющий условие закрытия корреляционного окна и генерации корреляционного события. Если условие удовлетворено, окно будет закрыто, и по нему будет сформировано корреляционное событие.

  4. Создайте блок условий тестирования работы правила корреляции в поле tests.

Тестирование правила

Тесты в правиле корреляции позволяют проверить корректность работы правила и его способность реагировать на заданные типы событий.

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

В случае, если правило корреляции использует данные из активных списков, необходимо инициализировать MOCK данные через поле setup_tests. Это позволит тестировать правило в условиях, максимально приближенных к реальной эксплуатации, без необходимости доступа к продуктивным данным.

Поле setup_tests

Для тестирования правил, которые используют внешние зависимости (например, взаимодействуют с активными списками или таблицами обогащения), использование поля setup_tests является обязательным.

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

setup_tests используется для:

  • Имитации взаимодействия правила с активными списками и таблицами обогащения.

  • Проверки логики правила с учетом ожидаемых данных обогащения.

Процедура настройки setup_tests:

  1. В корне конфигурации правила указывается секция setup_tests.

  2. Внутри setup_tests через ключ enrichment_tables или active_lists задаются MOCK таблицы обогащения и активные списки, используемые правилом.

  3. Для каждого MOCK элемента указывается его структура и набор MOCK данных, которые будут использоваться при тестировании.

  4. MOCK данные должны соответствовать ожидаемым данным из реальных таблиц обогащения/активных списков и позволять полноценно проверить работу правила.

Example 1. Структура setup_tests для активного списка:
setup_tests:
  active_lists:
    <имя_списка>:
      <поле>:
        - [<значение_поля1>]
        - [<значение_поля2>]
Example 2. Структура setup_tests для таблицы обогащения:
setup_tests:
  enrichment_tables:
    <имя_таблицы>:
      fields:
        - <поле1>
        - <поле2>
      data:
        - [<значение1_поля1>, <значение1_поля2>]
        - [<значение2_поля1>, <значение2_поля2>]

Поле tests

Поле tests включает набор тестов для проверки и верификации правила корреляции. Эти тесты помогают убедиться, что правило работает корректно и соответствует ожидаемым результатам.

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

  • В name укажите название теста.

  • В events создайте массив событий, которые будут использоваться для тестирования. Эти события должны имитировать поток реальных исходных событий, которые поступают на коррелятор.

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

Тестовые события

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

Если в правиле настроена корреляция по полю с типом DateTime, например timestamp, то значение этого поля в tests необходимо задавать в формате t'<DateTime-значение>'. Например: t'15.05.2024 06:28:15'. Это важно при корреляции на отсутствие событий.

В противном случае при запуске тестов правила система обработает поле в виде строки и корреляция по дате-времени будет работать некорректно.

Условия проверки

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

Пример:

tests:
  - name: "Тестирование обнаружения DDoS атаки"
    events:
      - { source_ip: "192.168.1.1", event_type: "traffic_anomaly", traffic_volume: "10000" }
    assertion: !vrl |
        assert_eq!(.alert_type, "DDoS", "Ожидается обнаружение DDoS атаки")

Обогащение через внешние зависимости

Обогащение корреляционных событий представляет собой процесс дополнения событий, сгенерированных в ходе корреляции, дополнительными данными. В декларативных правилах корреляции можно настроить логику обогащения в полях on_event и on_correlation.

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

Активные списки

Активные списки представляют собой структуры данных, которые используются для хранения и управления как динамически изменяющимися наборами информации, так и статическими списками и таблицами, такими как IP-адреса, доменные имена, идентификаторы устройств или пользователей. В контексте императивных правил корреляции, поля on_correlation и on_event могут включать вызовы функций для взаимодействия с активными списками, используя язык VRL. Это позволяет добавлять, извлекать и удалять записи из активных списков на основе логики обработки корреляционных событий.

Функции для взаимодействия с активными списками:

Функция Описание Параметры и Результат

add_active_record(list, key, value)

Добавляет запись в активный список. Если запись с таким ключом уже существует, значение обновляется. В противном случае создается новая запись.

list: название активного списка, в который добавляется запись. key: уникальный ключ для записи. value: значение, ассоциированное с ключом.

get_active_record(list, key)

Извлекает значение из активного списка по указанному ключу. Возвращает значение, ассоциированное с ключом, или null, если запись отсутствует.

list: активный список, из которого нужно извлечь запись. key: уникальный ключ записи.

remove_active_record(list, key)

Удаляет запись из активного списка. Если запись с таким ключом отсутствует, ничего не происходит.

list: активный список, из которого необходимо удалить запись. key: уникальный ключ записи, которую необходимо удалить.

Example 3. Пример использования в поле on_correlation правила корреляции:
on_correlation: !vrl |
  if .threat_level == "High" {
    add_active_record("suspicious_ips", .source_ip, true)
  }

  if .event_type == "authentication_failure" {
    let failure_count = get_active_record("auth_failure_counts", .user_id) || 0
    add_active_record("auth_failure_counts", .user_id, failure_count + 1)
  }

  if .event_type == "authentication_success" {
    remove_active_record("auth_failure_counts", .user_id)
  }

В этом примере:

  • IP-адреса с высоким уровнем угрозы добавляются в активный список подозрительных IP-адресов.

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

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

Example 4. Правило корреляции с активным списком, схема RObject
# Уникальный идентификатор правила.
id: example/ddos_detection_rule

# Название правила.
name: DDoSDetectionRule

# Режим работы корреляционного окна.
window_mode: parallel

# Тип правила.
type: correlation_rule

# Уровень угрозы, присваиваемый корреляционному правилу.
severity: high

# Версия правила.
version: 1.0.0

# Описание назначения правила.
description: Обнаружение атаки DDoS на основе трафика и взаимодействия с активным списком подозрительных IP.

# Имя и контактные данные автора правила.
author: John Doe <johndoe@example.com>

# Список совместимых с правилом источников данных.
data_source:
  - platform: Windows
  - source: Sysmon
  - events:
    - EventID_13
    - EventID_1

# Теги, связанные с правилом.
tags:
  - ddos
  - security

# Фильтр применяется к событиям перед входом в правило.
filter: !vrl |
  .event_type == "HTTP_REQUEST" &&
  .source_ip != null &&
  .request_path != null

# Логика группировки событий.
group_by:
  - source_ip

# Время жизни корреляционного окна (в секундах).
expire_after_sec: 600

# Частота генерации корреляционных событий в рамках каждой группы.
throttle_time_sec: 60

# Логика обновления состояния окна корреляции при получении событий.
on_event: !vrl |
  if int!(%state.request_count) >= 10 {
    abort
  }
  %state.request_count = int(%state.request_count) + 1

# Условие открытия корреляционного окна.
start_when: !vrl |
  .event_type == "HTTP_REQUEST" && .source_ip != null

# Условие закрытия корреляционного окна и генерации корреляционного события.
correlate_when: !vrl |
  int!(%state.request_count) >= 10

# Логика, применяемая после генерации корреляционного события.
on_correlation: !vrl |
  let is_suspicious = get_active_record("suspicious_ips", .source_ip)
  if is_suspicious != null {
    .message = "Обнаружена атака DDoS с источника: " + .source_ip + ". IP помечен как подозрительный."
  } else {
    .message = "Обнаружена атака DDoS с источника: " + .source_ip + ". IP не найден в списке подозрительных."
  }

# Инициализация MOCK данных для активных списков.
setup_tests:
  active_lists:
    suspicious_ips:
      - source_ip: "192.0.2.1"
        is_suspicious: true
      - source_ip: "198.51.100.10"
        is_suspicious: false

# Тесты для проверки работы правила.
tests:
  - name: "Detect DDoS from suspicious IP"
    events:
      - { "event_type": "HTTP_REQUEST", "source_ip": "192.0.2.1", "request_path": "/api" }
    assertion: !vrl |
      assert_eq!(.message, "Обнаружена атака DDoS с источника: 192.0.2.1. IP помечен как подозрительный.")
   - name: "Detect DDoS from non-suspicious IP"
     events:
       - { "event_type": "HTTP_REQUEST", "source_ip": "198.51.100.10", "request_path": "/api" }
     assertion: !vrl |
       assert_eq!(.message, "Обнаружена атака DDoS с источника: 198.51.100.10. IP не найден в списке подозрительных.")

Логика правила описывается в блоках on_event, start_when, correlate_when, on_correlation, где:

  • on_event учитывает количество HTTP-запросов от каждого источника IP.

  • start_when инициирует корреляционное окно при поступлении HTTP-запроса.

  • correlate_when активирует генерацию корреляционного события, когда количество запросов достигает порога в 10.

  • on_correlation дополняет корреляционное событие информацией о том, был ли источник IP помечен как подозрительный в активном списке.

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

Тесты Detect DDoS from suspicious IP и Detect DDoS from non-suspicious IP демонстрируют работу правила в различных условиях, проверяя его способность корректно реагировать на атаки с подозрительных и не подозрительных IP-адресов соответственно. Успешное прохождение этих тестов подтверждает, что правило корреляции обнаруживает DDoS-атаки и правильно взаимодействует с активными списками для обогащения корреляционных событий дополнительной информацией об источнике атаки.

Таблицы обогащения

Таблицы обогащения представляют собой структурированные наборы данных, используемые для добавления контекстной информации к событиям в процессе их корреляции. Это может включать данные о геолокации, информацию об устройствах, базы данных угроз и другие внешние источники данных. В контексте императивных правил корреляции, поля on_correlation и on_event могут включать вызовы функции VRL для запроса и извлечения данных из таблиц обогащения, тем самым обогащая события дополнительными деталями для более глубокого анализа.

Функции для взаимодействия с таблицами обогащения:

  • find_enrichment_table_records(table, condition, [select, case_sensitive]): позволяет выполнить поиск записей в таблице обогащения, которые соответствуют заданным условиям. Возвращает массив записей, удовлетворяющих условиям поиска.

  • get_enrichment_table_record(table, condition, [select, case_sensitive]): используется для извлечения одной записи из таблицы обогащения, которая точно соответствует заданному условию. В случае, если запись не найдена или найдено более одной записи, функция возвращает ошибку.

Пример использования в поле on_correlation:

on_correlation: !vrl |
  let geo_info = find_enrichment_table_records("geoip", {"ip": .source_ip}, select: ["country_name", "city_name"])
  if !geo_info.is_empty() {
    .source_country = geo_info[0].country_name
    .source_city = geo_info[0].city_name
  }

  let threat_info = get_enrichment_table_record("threat_intelligence", {"ip": .source_ip}, select: ["category", "score"])
  if !is_null(threat_info) {
    .threat_category = threat_info.category
    .threat_score = threat_info.score
  }

В этом примере:

  • Выполняется поиск геолокационной информации по IP-адресу источника события в таблице geoip, и результаты добавляются к событию как .source_country и .source_city.

  • Затем производится запрос к таблице threat_intelligence для извлечения информации о категории угрозы и оценке уровня угрозы, ассоциированных с IP-адресом источника. Эти данные добавляются к событию как .threat_category и .threat_score.

Тестирование правила с внешними зависимостями

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

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

Процесс тестирования

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

  2. Конфигурация setup_tests: в корне конфигурации правила корреляции необходимо добавить секцию setup_tests, где указываются все используемые таблицы обогащения и активные списки с их MOCK данными. Имя каждого внешнего элемента в секции должно точно совпадать с именем, используемым в теле правила.

  3. Заполнение MOCK данных: внутри внешнего элемента в секции setup_tests указывается список полей и соответствующих им MOCK данных. Данные задаются в формате, соответствующем ожидаемым запросам к активному списку и/или таблице обогащения в рамках теста.

Примеры:

Example 5. Структура MOCK данных для активного списка:
setup_tests:
  active_lists:
    suspicious_ips:
      - source_ip: "192.168.1.1"
        is_suspicious: true
      - source_ip: "10.10.10.10"
        is_suspicious: false
    auth_failure_counts:
      - user_id: "user1"
        failure_count: 1
      - user_id: "user2"
        failure_count: 2
Example 6. Структура MOCK данных для таблицы обогащения:
  enrichment_tables:
    geoip:
      fields:
        - ip
        - country
        - city
      data:
        - ["8.8.8.8", "US", "Mountain View"]
        - ["9.9.9.9", "GB", "London"]

Создание и выполнение тестов

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

  1. Описание теста (name): краткое описание цели теста.

  2. События (events): список событий, которые будут подаваться на вход правила в процессе тестирования.

  3. Утверждения (assertions): логика проверки результата работы правила на основе предоставленных событий и MOCK данных активных списков.

Пример теста:

tests:
  - name: "Test for Suspicious IP Detection"
    events:
      - { source_ip: "192.168.1.1", event_type: "login_attempt" }
    assertion: !vrl |
        let is_suspicious = get_active_record("suspicious_ips", .source_ip);
        assert_eq!(is_suspicious, true, "IP should be marked as suspicious")
  - name: "Authentication Failure Count Test"
    events:
      - { user_id: "user2", event_type: "login_failure" }
    assertion: !vrl |
        let failure_count = get_active_record("auth_failure_counts", .user_id);
        assert_eq!(failure_count, 2, "User2 should have 2 authentication failures")

В разделе событий (events) указываются события, которые подаются на вход правила корреляции для проверки его срабатывания. В первом тесте это попытка входа с IP-адреса, который должен быть помечен как подозрительный. Во втором тесте — неудачная попытка входа от пользователя, для которого уже зафиксированы подобные инциденты.

Утверждения (assertions) содержат логику проверки, выполняемую после обработки событий правилом. В первом тесте проверяется, что IP-адрес из события корректно идентифицирован как подозрительный на основе данных активного списка suspicious_ips. Во втором — что количество неудачных попыток входа для пользователя соответствует ожидаемому значению, основанному на данных активного списка auth_failure_counts.

Если утверждение (assertion) выполнено успешно, это означает, что правило корреляции корректно обрабатывает входные данные и правильно взаимодействует с активными списками. В противном случае, если результат не соответствует ожидаемому, это указывает на проблемы в логике правила или в его взаимодействии с активными списками.

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

  • Тщательно проверяйте соответствие имен активных списков и таблиц обогащения в setup_tests именам, используемым в правиле корреляции.

  • Используйте результаты тестирования для уточнения и оптимизации логики правила корреляции и взаимодействия с внешним элементом.

Пример правила корреляции

Example 7. Правило корреляции, схема RObject
# Уникальный идентификатор правила.
id: example/ddos_detection_rule

# Название правила.
name: DDoSDetectionRule

# Режим работы корреляционного окна.
window_mode: parallel

# Тип правила.
type: correlation_rule

# Уровень угрозы, присваиваемый корреляционному правилу.
severity: high

# Версия правила.
version: 1.0.0

# Описание назначения правила.
description: Обнаружение атаки DDoS на основе трафика

# Имя и контактные данные автора правила нормализации.
author: John Doe <johndoe@rvision.ru>

# Список совместимых с правилом источников данных.
data_source:
  - platform: Windows
  - source: Sysmon
  - events:
    - EventID_13
    - EventID_1

# Теги, связанные с правилом.
tags:
  - ddos
  - security

# Фильтр применяется к событиям перед входом в правило.
filter: !vrl |
  .event_type == "HTTP_REQUEST" &&
  .source_ip != null &&
  .request_path != null

# Логика группировки событий.
group_by:
  - source_ip

# Время жизни корреляционного окна (в секундах).
expire_after_sec: 600

# Частота генерации корреляционных событий в рамках каждой группы.
throttle_time_sec: 60

# Логика обновления состояния окна корреляции при получении событий.
on_event: !vrl |
  if int!(%state.request_count) >= 10 {
    abort
  }
  %state.request_count = int(%state.request_count) + 1

# Условие открытия корреляционного окна.
start_when: !vrl |
  .event_type == "HTTP_REQUEST" && .source_ip != null

# Условие закрытия корреляционного окна и генерации корреляционного события.
correlate_when: !vrl |
  int!(%state.request_count) >= 10

# Логика, применяемая после генерации корреляционного события.
on_correlation: !vrl |
  .message = "Обнаружена атака DDoS с источника: " + .event.source_ip
  .additional_info = "Обнаружено " + string(%state.request_count) + " запросов с одного IP за короткий промежуток времени"

# Тесты для проверки работы правила.
tests:
  - name: DDoS Attack Detected
    events:
      - { "event_type": "HTTP_REQUEST", "source_ip": "192.0.2.1", "request_path": "/api" }
      - { "event_type": "HTTP_REQUEST", "source_ip": "192.0.2.1", "request_path": "/api" }
      - { "event_type": "HTTP_REQUEST", "source_ip": "192.0.2.1", "request_path": "/api" }
      - { "event_type": "HTTP_REQUEST", "source_ip": "192.0.2.1", "request_path": "/api" }
      - { "event_type": "HTTP_REQUEST", "source_ip": "192.0.2.3", "request_path": "/api" }
      - { "event_type": "HTTP_REQUEST", "source_ip": "192.0.2.1", "request_path": "/api" }
      - { "event_type": "HTTP_REQUEST", "source_ip": "192.0.2.1", "request_path": "/api" }
      - { "event_type": "HTTP_REQUEST", "source_ip": "192.0.2.2", "request_path": "/api" }
      - { "event_type": "HTTP_REQUEST", "source_ip": "192.0.2.1", "request_path": "/api" }
      - { "event_type": "HTTP_REQUEST", "source_ip": "192.0.2.1", "request_path": "/api" }
      - { "event_type": "HTTP_REQUEST", "source_ip": "192.0.2.4", "request_path": "/api" }
      - { "event_type": "HTTP_REQUEST", "source_ip": "192.0.2.1", "request_path": "/api" }
      - { "event_type": "HTTP_REQUEST", "source_ip": "192.0.2.1", "request_path": "/api" }
    assert_events_count: 1
    assert_event: !vrl |
      assert_eq!(.message, "Обнаружена атака DDoS с источника: 192.0.2.1")

Интерпретация

Правило корреляции DDoSDetectionRule разработано с целью выявления DDoS-атак на основе анализа входящего сетевого трафика. Вот подробное разъяснение каждого компонента этого правила:

Метаданные:

  • id: example/ddos_detection_rule — уникальный идентификатор правила.

  • name: DDoSDetectionRule — идентификационное название правила.

  • version: 1.0.0 — версия правила.

  • author: John Doe <johndoe@rvision.ru> — создатель или команда, отвечающая за правило.

  • description: уточняет цель правила — обнаружение DDoS-атак, основанное на анализе трафика.

  • tags: помечено как ddos и security, что подсказывает об основной функциональности и сфере применения правила.

Параметры корреляции:

  • window_mode: parallel — режим корреляции, где события анализируются параллельно.

  • expire_after_sec: 600 — время жизни окна корреляции в секундах (10 минут). Окно закроется не сгенерировав корреляционное событие, если в течение указанного времени не произойдет обновления состояния.

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

  • type: correlation_rule — тип правила.

  • severity: high — уровень серьезности, присеваемый правилу.

Логика правила:

  • filter: правило активируется только для событий с типом HTTP_REQUEST, имеющих определенный source_ip и request_path.

  • group_by: события группируются по полю source_ip, что позволяет отслеживать активность каждого отдельного IP-адреса.

  • on_event: если количество запросов (request_count) для данного источника IP достигает или превышает 10, то выполнение правила прерывается с помощью директивы abort. Это означает, что атака DDoS уже обнаружена, и состояние перестает меняться. В противном случае значение request_count увеличивается на 1.

  • start_when: определяет условие, при котором начинается новое корреляционное окно. В данном случае это любое событие HTTP_REQUEST с непустым полем source_ip.

  • correlate_when: корреляционное событие генерируется, когда количество запросов от одного IP достигает 10.

  • on_correlation: после обнаружения атаки DDoS формируется корреляционное событие со следующими аттрибутами:

    • type: "correlation_rule",

    • severity: "high",

    • event_type: "DDoS Detection",

    • source_ip: "192.168.1.1",

    • datetime: "2023-09-28T12:45:00Z",

    • message: "Обнаружена атака DDoS с источника: 192.168.1.1",

    • additional_info: "Обнаружено 10 запросов с одного IP за короткий промежуток времени."

Тестирование правила:

Тест DDoS Attack Detected имитирует поток событий, включая 13 HTTP-запросов. Из них 10 запросов исходят от одного IP-адреса "192.168.1.1". После обработки этого потока событий правило должно генерировать одно корреляционное событие с сообщением о DDoS-атаке от IP-адреса "192.168.1.1".

Таким образом, тест DDoS Attack Detected проверяет, что правило корреляции правильно реагирует на атаку DDoS, а именно, что оно обнаруживает атаку, когда количество запросов от одного IP-адреса достигает 10 и более.

Таким образом, правило корреляции DDoSDetectionRule предназначено для отслеживания атаки DDoS на основе количества HTTP-запросов от одного источника IP. Как только количество запросов достигает 10 или более за короткий промежуток времени (10 минут), правило обнаруживает атаку и генерирует соответствующее корреляционное событие.

Обработка событий правилом корреляции

Продолжим анализ работы правила DDoSDetectionRule на примере исходного события и его обработки.

Предположим, что в течение 10 минут на коррелятор, на котором установлено правило DDoSDetectionRule, поступило 10 JSON-событий с источника "192.168.1.1" и они имеют следующий вид:

Example 8. Исходное событие, JSON
{
  "event_type": "HTTP_REQUEST",
  "source_ip": "192.168.1.1",
  "datetime": "2023-09-28T12:34:56Z",
  "request_path": "/api",
  "user_agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/93.0.4472.124 Safari/537.36"
}

При обработке событий происходит срабатывание правила корреляции DDoSDetectionRule. Правило создает корреляционное окно каждый раз, когда событие проходит его фильтр. Как только для источника "192.168.1.1" накапливается 10 событий, правило генерирует корреляционное событие, которое будет иметь следующий вид:

Example 9. Корреляционное событие, генерируемое правилом DDoSDetectionRule
{
  "type": "correlation_rule",
  "severity": "high",
  "event_type": "DDoS Detection",
  "source_ip": "192.168.1.1",
  "datetime": "2023-09-28T12:45:00Z",
  "message": "Обнаружена атака DDoS с источника: 192.168.1.1",
  "additional_info": "Обнаружено 10 запросов с одного IP за короткий промежуток времени."
}

Таким образом, правило отслеживает аномально высокую активность отдельных IP-адресов на основе входящих HTTP-запросов.

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

Расширение конфигурации существующего правила

Рассмотрим сценарий, в котором необходимо расширить конфигурацию существующего правила корреляции DDoSDetectionRule с помощью дополнительного условия. Данное условие будет учитывать количество HTTP-запросов, поступающих с одного источника, и ранжировать эти источники в зависимости от этого параметра. В зависимости от количества коррелируемых событий, корреляционному событию будет присвоен один из четырех уровней угрозы: "low" (низкий), "medium" (средний), "high" (высокий) или "critical" (критический).

Исходное правило корреляции "DDoSDetectionRule", схема RObject
# Уникальный идентификатор правила.
id: example/ddos_detection_rule

# Название правила.
name: DDoSDetectionRule

# Режим работы корреляционного окна.
window_mode: parallel

# Тип правила.
type: correlation_rule

# Уровень угрозы, присваиваемый корреляционному правилу.
severity: high

# Версия правила.
version: 1.0.0

# Описание назначения правила.
description: Обнаружение атаки DDoS на основе трафика

# Имя и контактные данные автора правила нормализации.
author: John Doe <johndoe@rvision.ru>

# Список совместимых с правилом источников данных.
data_source:
  - platform: Windows
  - source: Sysmon
  - events:
    - EventID_13
    - EventID_1

# Теги, связанные с правилом.
tags:
  - ddos
  - security

# Фильтр применяется к событиям перед входом в правило.
filter: !vrl |
  .event_type == "HTTP_REQUEST" &&
  .source_ip != null &&
  .request_path != null

# Логика группировки событий.
group_by:
  - source_ip

# Время жизни корреляционного окна (в секундах).
expire_after_sec: 600

# Частота генерации корреляционных событий в рамках каждой группы.
throttle_time_sec: 60

# Логика обновления состояния окна корреляции при получении событий.
on_event: !vrl |
  if int!(%state.request_count) >= 10 {
    abort
  }
  %state.request_count = int(%state.request_count) + 1

# Условие открытия корреляционного окна.
start_when: !vrl |
  .event_type == "HTTP_REQUEST" && .source_ip != null

# Условие закрытия корреляционного окна и генерации корреляционного события.
correlate_when: !vrl |
  int!(%state.request_count) >= 10

# Логика, применяемая после генерации корреляционного события.
on_correlation: !vrl |
  .message = "Обнаружена атака DDoS с источника: " + .event.source_ip
  .additional_info = "Обнаружено " + string(%state.request_count) + " запросов с одного IP за короткий промежуток времени"

# Тесты для проверки работы правила.
tests:
  - name: DDoS Attack Detected
    events:
      - { "event_type": "HTTP_REQUEST", "source_ip": "192.0.2.1", "request_path": "/api" }
      - { "event_type": "HTTP_REQUEST", "source_ip": "192.0.2.1", "request_path": "/api" }
      - { "event_type": "HTTP_REQUEST", "source_ip": "192.0.2.1", "request_path": "/api" }
      - { "event_type": "HTTP_REQUEST", "source_ip": "192.0.2.1", "request_path": "/api" }
      - { "event_type": "HTTP_REQUEST", "source_ip": "192.0.2.3", "request_path": "/api" }
      - { "event_type": "HTTP_REQUEST", "source_ip": "192.0.2.1", "request_path": "/api" }
      - { "event_type": "HTTP_REQUEST", "source_ip": "192.0.2.1", "request_path": "/api" }
      - { "event_type": "HTTP_REQUEST", "source_ip": "192.0.2.2", "request_path": "/api" }
      - { "event_type": "HTTP_REQUEST", "source_ip": "192.0.2.1", "request_path": "/api" }
      - { "event_type": "HTTP_REQUEST", "source_ip": "192.0.2.1", "request_path": "/api" }
      - { "event_type": "HTTP_REQUEST", "source_ip": "192.0.2.4", "request_path": "/api" }
      - { "event_type": "HTTP_REQUEST", "source_ip": "192.0.2.1", "request_path": "/api" }
      - { "event_type": "HTTP_REQUEST", "source_ip": "192.0.2.1", "request_path": "/api" }
    assert_events_count: 1
    assert_event: !vrl |
      assert_eq!(.message, "Обнаружена атака DDoS с источника: 192.0.2.1")
Example 10. Конфигурация расширенного правила, схема RObject
# Уникальный идентификатор правила.
id: example/enhanced_ddos_detection_rule

# Название правила.
name: EnhancedDDoSDetectionRule

# Режим работы корреляционного окна.
window_mode: parallel

# Тип правила.
type: correlation_rule

# Уровень угрозы, присваиваемый корреляционному правилу.
severity: high

# Версия правила.
version: 1.0.0

# Описание назначения правила.
description: Расширенное правило обнаружения атаки DDoS с классификацией уровней угрозы

# Имя и контактные данные автора правила нормализации.
author: John Doe <johndoe@example.com>

# Список совместимых с правилом источников данных.
data_source:
  - platform: Windows
  - source: Sysmon
  - events:
    - EventID_13
    - EventID_1

# Теги, связанные с правилом.
tags:
 - ddos
 - security

# Фильтр применяется к событиям перед входом в правило.
filter: !vrl |
  .event_type == "HTTP_REQUEST" &&
  .source_ip != null &&
  .request_path != null

# Логика группировки событий.
group_by:
  - source_ip

# Время жизни корреляционного окна (в секундах).
expire_after_sec: 600

# Частота генерации корреляционных событий в рамках каждой группы.
throttle_time_sec: 60

# Логика обновления состояния окна корреляции при получении событий.
on_event: !vrl |
  if int!(%state.request_count) >= 10 {
    abort
  }
  %state.request_count = int(%state.request_count) + 1

# Условие открытия корреляционного окна.
start_when: !vrl |
  .event_type == "HTTP_REQUEST" && .source_ip != null

# Условие закрытия корреляционного окна и генерации корреляционного события.
correlate_when: !vrl |
  int!(%state.request_count) >= 10

# Логика, применяемая после генерации корреляционного события.
on_correlation: !vrl |
  .message = "Обнаружена атака DDoS с источника: " + .event.source_ip

  # Расширяем правило для классификации уровней угрозы в зависимости от количества запросов
  .severity = select(
    int!(%state.request_count) >= 10 => "critical",
    int!(%state.request_count) >= 6 => "high",
    int!(%state.request_count) >= 3 => "medium",
    true => "low"
  )
  .alert_type = "DDoS Attack"
  .additional_info = "Обнаружено " + string(%state.request_count) + " запросов с одного IP за короткий промежуток времени"

# Тесты для проверки работы правила.
tests:
 - name: DDoS Attack Detected
   events:
    - { "event_type": "HTTP_REQUEST", "source_ip": "192.0.2.1", "request_path": "/api" }
    - { "event_type": "HTTP_REQUEST", "source_ip": "192.0.2.1", "request_path": "/api" }
    - { "event_type": "HTTP_REQUEST", "source_ip": "192.0.2.1", "request_path": "/api" }
    - { "event_type": "HTTP_REQUEST", "source_ip": "192.0.2.1", "request_path": "/api" }
    - { "event_type": "HTTP_REQUEST", "source_ip": "192.0.2.3", "request_path": "/api" }
    - { "event_type": "HTTP_REQUEST", "source_ip": "192.0.2.1", "request_path": "/api" }
    - { "event_type": "HTTP_REQUEST", "source_ip": "192.0.2.1", "request_path": "/api" }
    - { "event_type": "HTTP_REQUEST", "source_ip": "192.0.2.2", "request_path": "/api" }
    - { "event_type": "HTTP_REQUEST", "source_ip": "192.0.2.1", "request_path": "/api" }
    - { "event_type": "HTTP_REQUEST", "source_ip": "192.0.2.1", "request_path": "/api" }
    - { "event_type": "HTTP_REQUEST", "source_ip": "192.0.2.4", "request_path": "/api" }
    - { "event_type": "HTTP_REQUEST", "source_ip": "192.0.2.1", "request_path": "/api" }
    - { "event_type": "HTTP_REQUEST", "source_ip": "192.0.2.1", "request_path": "/api" }
   assert_events_count: 1
   assert_event: !vrl |
     assert_eq!(.message, "Обнаружена атака DDoS с источника: 192.0.2.1")

Интерпретация

В поле severity расширенного правила устанавливается один из следующих уровней серьезности в зависимости от значения %state.request_count:

  • Если значение %state.request_count больше или равно 10, то уровень серьезности будет установлен как critical.

  • Если значение %state.request_count больше или равно 6, но меньше 10, то уровень серьезности будет установлен как high.

  • Если значение %state.request_count больше или равно 3, но меньше 6, то уровень серьезности будет установлен как medium.

  • В противном случае, если ни одно из вышеперечисленных условий не выполняется (true), то уровень серьезности будет установлен как low.

Предположим, что в течение 10 минут на правило EnhancedDDoSDetectionRule поступает 10 событий в формате JSON с источника "192.168.1.1", и они имеют следующий вид:

Example 11. Исходное событие, JSON
"event_type": "HTTP_REQUEST",
"source_ip": "192.168.1.1",
"datetime": "2023-08-15T12:34:56Z",
"request_path": "/api/resource",
"user_agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36"

Тогда в результате срабатывании правила корреляции EnhancedDDoSDetectionRule, генерируемое корреляционное событие будет иметь следующий вид:

Example 12. Корреляционное событие, генерируемое правилом EnhancedDDoSDetectionRule
"type": "correlation_rule",
"severity": "critical", # Уровень угрозы установлен в значение `critical` согласно расширенному условию классификации уровней угрозы
"event_type": "DDoS Detection",
"source_ip": "192.168.1.1",
"datetime": "2023-09-28T12:45:00Z",
"message": "Обнаружена атака DDoS с источника: 192.168.1.1",

Улучшенное правило EnhancedDDoSDetectionRule представляет собой расширенную версию исходного правила DDoSDetectionRule. Оно расширяет конфигурацию исходного правила так, чтобы при обнаружении DDoS-атак использовалось дополнительное условие, которое учитывает количество HTTP-запросов приходящих от источников. Условие ранжирует источники по количеству запросов, чтобы присвоить событию один из 4 уровней угрозы: low, medium, high или critical. Таким образом, улучшенное правило классифицирует источники атаки DDoS на основе их активности.