Разработка правил разбора и нормализации событий

Создание правил разбора

Управление правилами разбора осуществляется в разделе «Источники» → «Управление источниками». Далее выбрать вкладку «Правила разбора», после чего откроется страница управления правилами разбора (см. рисунок 1).

Страница управления правилами разбора

Рисунок 1 -- Страница управления правилами разбора

В разделе присутствует три вкладки:

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

Для опубликованных и неопубликованных правил разбора доступны общие функции:

Экспортировать - позволяет экспортировать выбранные правила разбора в файл архива формата ZIP.
Импортировать - позволяет импортировать правила разбора из файла архива формата ZIP.
Удалить - удаляет выбранные правила разбора.

На каждой вкладке с правилами разбора доступен поиск правила по его наименованию.

Для создания нового правила разбора необходимо указать название в поле «Добавление правила» и нажать на «+», после чего откроется форма создания правила разбора (см. рисунок 2).

Рисунок 2 -- Окно создания правила разбора

Рассмотрим процесс создания парсера на примере событий от продукта Micro Focus ArcSight SmartConnector.

Тип продукта: сетевое оборудование

Сырое событие:

{"rs_collector_hostname":"radar-balancer-01","rs_relay_fqdn":"arcsight-test","rs_relay_ip":"172.0.0.96","rs_collector_ts":"2021-06-09T10:41:02.253872+03:00","__rs_module":"3500-Arcsight-Smartconnector-Netflow-cef","message":"CEF:1|IP Flow|IP Flow|9|flow|NetFlow Event|Unknown| eventId=13252253246 start=1623223861208 end=1623223861272 proto=TCP in=1098 categoryBehavior=\/Communicate categoryDeviceGroup=\/Network Equipment catdt=Network Monitoring categoryOutcome=\/Attempt categoryObject=\/Host art=1623224462176 rt=1623223873000 deviceDirection=0 src=172.0.218.2 sourceZoneURI=\/All Zones\/ArcSight System\/Private Address Space Zones\/RFC1918: 10.0.0.0-10.255.255.255 spt=8787 dst=172.0.18.108 destinationZoneURI=\/All Zones\/ArcSight System\/Private Address Space Zones\/RFC1918: 10.0.0.0-10.255.255.255 dpt=53445 fileType=NAT Source IPv4 Address: fileHash=NAT Source Port: oldFileType=NAT Destination IPv4 Address: oldFileHash=NAT Destination Port: cs1=172.0.245.1 cs4=13 cs5=26 cn1=9 cn3=0 cs1Label=nexthop cs2Label=src_as cs3Label=dst_as cs4Label=src_mask cs5Label=dst_mask cs6Label=tcp_flags descr cn1Label=in_pkts cn2Label=out_pkts cn3Label=tcp_flags ahost=arcsight-test agt=172.0.6.96 agentZoneURI=\/All Zones\/ArcSight System\/Private Address Space Zones\/RFC1918: 10.0.0.0-10.255.255.255 amac=34-B3-54-BC-66-C6 av=7.14.0.8241.0 atz=Europe\/Moscow at=cisco_netflow dvchost=arcsight-test dvc=172.0.255.245 deviceZoneURI=\/All Zones\/ArcSight System\/Private Address Space Zones\/RFC1918: 10.0.0.0-10.255.255.255 dtz=Europe\/Moscow geid=0 _cefVer=1.0 ad.flow__sampler__id=0 ad.vendor__51=0 ad.DevicePort=61673 ad.interface__output__snmp=312 ad.src__tos=0 ad.pkthdr__uptime=444691076 ad.pkthdr__seq=787165105 ad.pkthdr__source__id=517 ad.pkthdr__count=32 ad.interface__input__snmp=153 aid=3hughqHkBABCBSulnxz6OxA\\=\\="}

Обратите внимание, что сырое событие представлено в формате JSON. Платформа Радар поддерживает работу с событиями, содержащими кириллицу.

3.. Сырое событие необходимо вставить в соответствующее поле.

Во вкладке «Тип» нужно указать «json».
После нажатия кнопки «Проверить» получаем результат разобранного JSON события (см. рисунок 3).

Рисунок 3 -- Результат разобранного события
Как видно из результата, одного этапа разбора недостаточно, потому что основная информация данного события находится в поле «message».
В качестве второго этапа разбора необходимо использовать этап CEF. Для этого следует нажать на кнопку «Добавить парсер» и выбрать «cef_nonstrict» (этот этап используется для разбора формата CEF версии 1).
Далее в поле «Цель» второго этапа разбора нужно указать название поля, которое необходимо разобрать, в случае рассматриваемого примера - это поле «message» (см. рисунок 4).

Рисунок 4 -- Добавление второго этапа разбора
Проверяем работоспособность этапов правила разбора нажатием на кнопку «Проверить».

Результатом проверки правила должен быть вывод полностью разобранного события:

{
"rs_collector_hostname": "radar-balancer-01",
"rs_relay_fqdn": "arcsight-test",
"rs_relay_ip": "172.0.0.96",
"rs_collector_ts": "2021-06-09T10:41:02.253872+03:00",
"__rs_module": "3500-Arcsight-Smartconnector-Netflow-cef",
  "cef_version": 1,
  "vendor": "IP Flow",
  "product": "IP Flow",
  "version": "9",
  "signature": "flow",
  "name": "NetFlow Event",
  "severity": "Unknown",
  "eventId": "13252253246",
  "start": "1623223861208",
  "end": "1623223861272",
  "proto": "TCP",
  "in": "1098",
  "categoryBehavior": "/Communicate",
  "categoryDeviceGroup": "/Network Equipment",
  "catdt": "Network Monitoring",
  "categoryOutcome": "/Attempt",
  "categoryObject": "/Host",
  "art": "1623224462176",
  "rt": "1623223873000",
  "deviceDirection": "0",
  "src": "172.0.218.2",
  "sourceZoneURI": "/All Zones/ArcSight System/Private Address Space Zones/RFC1918: 10.0.0.0-10.255.255.255",
  "spt": "8787",
  "dst": "172.0.18.108",
  "destinationZoneURI": "/All Zones/ArcSight System/Private Address Space Zones/RFC1918: 10.0.0.0-10.255.255.255",
  "dpt": "53445",
  "fileType": "NAT Source IPv4 Address:",
  "fileHash": "NAT Source Port:",
  "oldFileType": "NAT Destination IPv4 Address:",
  "oldFileHash": "NAT Destination Port:",
  "ahost": "arcsight-test",
  "agt": "172.0.6.96",
  "agentZoneURI": "/All Zones/ArcSight System/Private Address Space Zones/RFC1918: 10.0.0.0-10.255.255.255",
  "amac": "34-B3-54-BC-66-C6",
  "av": "7.14.0.8241.0",
  "atz": "Europe/Moscow",
  "at": "cisco_netflow",
  "dvchost": "arcsight-test",
  "dvc": "172.0.255.245",
  "deviceZoneURI": "/All Zones/ArcSight System/Private Address Space Zones/RFC1918: 10.0.0.0-10.255.255.255",
  "dtz": "Europe/Moscow",
  "geid": "0",
  "_cefVer": "1.0",
  "ad.flow__sampler__id": "0",
  "ad.vendor__51": "0",
  "ad.DevicePort": "61673",
  "ad.interface__output__snmp": "312",
  "ad.src__tos": "0",
  "ad.pkthdr__uptime": "444691076",
  "ad.pkthdr__seq": "787165105",
  "ad.pkthdr__source__id": "517",
  "ad.pkthdr__count": "32",
  "ad.interface__input__snmp": "153",
  "aid": "3hughqHkBABCBSulnxz6OxA==",
  "nexthop": "172.0.245.1",
  "src_mask": "13",
  "dst_mask": "26",
  "in_pkts": "9",
  "tcp_flags": "0"
}

После успешной проверки этапов разбора необходимо нажать кнопку «Сохранить» для сохранения правила и следом нажать кнопку «Опубликовать» для последующего его использования.

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

Создание правил нормализации

Управление правилами нормализации осуществляется в разделе «Источники» → «Управление источниками». Далее выбрать вкладку «Правила нормализации», после чего откроется страница создания, редактирования и просмотра правил нормализации (см. рисунок 5).

Страница управления правилами нормализации

Рисунок 5 -- Страница управления правилами нормализации

В разделе присутствует три вкладки:

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

Для опубликованных и неопубликованных правил нормализации доступны общие функции:

Экспортировать - позволяет экспортировать выбранные правила нормализации в файл архива формата ZIP.
Импортировать - позволяет импортировать правила нормализации из файла архива формата ZIP.
Удалить - удаляет выбранные правила нормализации.

На каждой вкладке с правилами нормализации доступен поиск правила по его наименованию.

Для создания нового правила нормализации необходимо указать название в поле "Добавление правила" и нажать на “+”, после чего откроется окно создания правила нормализации, изображенное на рисунке 6;

Рисунок 6 -- Окно создания правила нормализации

В качестве названия правила нормализации необходимо указывать уникальный идентификатор сообщения для данного источника, в случае с примером: arcsight_for_test
В поле для сырого события необходимо вставить разобранное событие (см. раздел результат создания правила разбора).
После создания правила, внутри него автоматически создается нормализатор root.yaml

root.yaml – файл содержит декларации преобразований, общих для всех событий данной системы. Преобразования, указанные в этом файле применяются ко всем событиям системы, прошедшим стадию парсинга. Как правило они содержат классификатор источника и данные, содержащиеся в заголовке события Данный нормализатор разрабатывается один на систему.
Для добавления поля нормализации нужно во вкладке "Добавить новое поле" выбрать необходимое поле и нажать на "+", в результате чего в нормализатор добавится выбранное поле, как изображено на рисунке 7. Можно использовать как преднастроенные системные поля нормализации, так и добавлять пользовательские поля;

Рисунок 7 -- Добавление поля нормализации

В поле "Значение из поля разбора" нужно указывать одно из разобранных полей, таким образом в поле нормализации будет записано значение поля из разбора. Также в данное поле можно записывать специальные функции для работы с полями (ниже будет представлено описание некоторых из них);

В поле "Фиксированное значение" нужно указывать произвольный набор символов соответствующий типу данного поля нормализации.

Описание и типы полей нормализации представлены в разделе Описание полей нормализации
Исходя из этого, в данном файле нормализации можно использовать поля, изображенные на рисунках ниже;

Рисунок 8 -- Поля нормализации для "root.yaml"

Рисунок 9 -- Поля нормализации для "root.yaml"
После добавления необходимых для нормализации полей - нужно нажать на кнопку "Проверить" для проверки, что во все поля нормализации записались нужные данные. Результат проверки изображен на рисунке 10;

Рисунок 10 -- Результат проверки нормализации
После настройки нормализатора root.yaml, необходимо перейти к созданию нормализатора для определенного типа событий от данного источника, в случае с примером - это Netflow;
Для этого в поле рядом с кнопкой "Добавить нормализатор" нужно ввести имя нормализатора и нажать на кнопку "+ Добавить нормализатор", как изображено на рисунке 11;

Рисунок 11 -- Добавление нового нормализатора
Далее необходимо добавить маршрутизацию для данного нормализатора. Это делается для того, чтобы не все события от данного источника нормализовались по данному "сценарию нормализации", а только те, которые нужны;
Для этого необходимо в поле "Маршрутизация события" ввести условия нормализации по данному сценарию. В качестве переменных в условии нужно использовать поля разобранного события. Заполненное поле маршрутизации изображено на рисунке 12;

Рисунок 12 -- Маршрутизация события

Таким образом, все события, которые подходят под условие: signature == 'flow' and version == '9' будут нормализованы по данному файлу нормализации
Для более гибкой, понятной и правильной нормализации в данном файле нормализации используются специальные функции, которые подробно описаны в разделе Специальные функции для работы с полями нормализации.
Также в данном файле нормализации используются дополнительные настройки, описание которых представлены ниже;
Функция Tapping (Поле "Настройка")

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

Пример использования:
```
tcp_flags = line.parsed['tcp']
      line.parsed['flow_tags'] = 
      [
          f"tcp_{flag}"
          for flag in
              ("syn", "fin", "rst", "psh", "ack", "cwr", "ecn", "urg")
          if tcp_flags.get(flag, False)
      ]
```
В результате выполнения данной настройки, в нормализации можно использовать поле "flow_tags"

ПРЕДУПРЕЖДЕНИЕ! Использование данного механизма может сказаться на производительности и скорости работы обработчиков событий.
Таким образом, таблица просмотра (функция lookup) для данного файла нормализации представлена на рисунке 13;

Рисунок 13 -- Таблицы просмотра
Дополнительная настройка нормализатора изображена на рисунке 14;

Рисунок 14 -- Дополнительная настройка нормализатора

Данная настройка необходима для того чтобы в разобранном событии найти поле "in" и/или "out" и присвоить их в новые поля "in\out_bytes" в формате целых чисел.
Поля нормализации используемые в данном файле нормализации представлены на рисунках ниже;

Рисунок 15 -- Поля нормализации для нормализатора "arcsight_test_flow.yaml"

Рисунок 16 -- Поля нормализации для нормализатора "arcsight_test_flow.yaml"

Рисунок 17 -- Поля нормализации для нормализатора "arcsight_test_flow.yaml"

Рисунок 18 -- Поля нормализации для нормализатора "arcsight_test_flow.yaml"

После добавления необходимых для нормализации полей, настроек и таблиц просмотра - нужно нажать на кнопку "Проверить" для проверки, что во все поля нормализации записались нужные данные;

Результат проверки:

{
  "event": {
    "uuid": "283f402ba7a04a3786ca8a52e19be872",
    "application": {
      "name": "smartconnector",
      "protocol": "TCP"
    },
    "bytes": {
      "received": 1098
    },
    "category": "connection",
    "description": "A connection was observed",
    "logsource": {
      "application": "flow",
      "name": "Microfocus ArcSight Smartconnector",
      "product": "arcsight",
      "subsystem": "communication",
      "vendor": "microfocus"
    },
    "packets": {
      "received": 9
    },
    "session": {
      "duration": 64,
      "endtime": "2021-06-09T07:31:01.272000+00:00",
      "flags": [
        "Unknown tcp flag"
      ],
      "starttime": "2021-06-09T07:31:01.208000+00:00"
    },
    "severity": 0,
    "subcategory": "connection_inbound",
    "timestamp": "2021-06-09T07:31:13+00:00"
  },
  "raw": null,
  "@timestamp": "2021-06-09T07:31:13+00:00",
  "action": "connect",
  "initiator": {
    "host": {
      "ip": [
        "172.0.218.2"
      ]
    },
    "socket": {
      "port": 8787
    }
  },
  "observer": {
    "event": {
      "id": "13252253246"
    },
    "host": {
      "hostname": [
        "arcsight-test"
      ],
      "ip": [
        "172.0.255.245"
      ]
    }
  },
  "reportchain": {
    "collector": {
      "timestamp": "2021-06-09T10:41:02.253872+03:00"
    },
    "relay": {
      "host": {
        "ip": [
          "172.0.0.96"
        ]
      }
    }
  },
  "target": {
    "host": {
      "ip": [
        "172.0.18.108"
      ]
    },
    "socket": {
      "port": 53445
    }
  }
}

После настройки основного нормализатора arcsight_test_netflow.yaml, необходимо перейти к созданию нормализатора по умолчанию parsed_only.yaml;

parsed_only.yaml – файл используется как «нормализатор по умолчанию». Для событий прошедших через этот нормализатор создается специализированных индекс в ElasticSearch, содержащий нормализованные данные. Данный нормализатор разрабатывается один на систему. В него попадают события, которые не прошли ни по одной из маршрутизаций в других нормализаторах
Для его создания, в поле названия нормализатора нужно ввести "parsed_only" и нажать на кнопку "+ Добавить нормализатор";
После чего, в маршрутизации события указать "fallback" и кликнуть на кнопку "Только разбор" чтобы она перешла в статус "Вкл", как изображено на рисунке 19;

Рисунок 19 -- Добавление нормализатора "parsed_only.yaml"
В данном нормализаторе не требуется добавления полей нормализации. В автоматически созданные поля необходимо указать "parsed" в поле "Фиксированное значение";
После - нужно еще раз провести проверку нормализации, нажав на кнопку "Проверить", если ошибки отсутствуют и в результате проверки ожидаемый результат можно перейти к сохранению нормализатора;
Для этого необходимо нажать кнопку "Сохранить" и следом кнопку "Опубликровать";
После чего в разделе "Правила нормализации" во вкладке "Опубликованные" должен появиться разработанный нормализатор. Это значит, что теперь его можно использовать.

Тестирование правил разбора и нормализации событий

Тестирование правил разбора и нормализации осуществляется в разделе «Источники» → «Управление источниками». Далее выбрать вкладку «Правила обработки», после чего откроется страница тестирования правил (см. рисунок 20).

Страница тестирования правил разбора и нормализации

Рисунок 20 -- Страница тестирования правил разбора и нормализации

В поле Сырое событие скопируйте сырое событие, полученное от источника.

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

Далее нажмите кнопку Запустить проверку, после чего откроется окно с результатами проверки (см. рисунок 21).

Результат проверки правил

Рисунок 21 -- Результат проверки правил