Перейти к содержанию

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

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

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

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

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

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

    Окно создания правила разбора

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

  3. Рассмотрим процесс создания парсера на примере событий от продукта 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. Платформа Радар поддерживает работу с событиями, содержащими кириллицу.

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

  5. Во вкладке «Тип» нужно указать «json».

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

    Результат разобранного события

    Рисунок 3 -- Результат разобранного события

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

  8. В качестве второго этапа разбора необходимо использовать этап CEF. Для этого следует нажать на кнопку «Добавить парсер» и выбрать «cef_nonstrict» (этот этап используется для разбора формата CEF версии 1).

  9. Далее в поле «Цель» второго этапа разбора нужно указать название поля, которое необходимо разобрать, в случае рассматриваемого примера - это поле «message» (см. рисунок 4).

    Добавление второго этапа разбора

    Рисунок 4 -- Добавление второго этапа разбора

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

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

    {
    "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"
    }
    
  12. После успешной проверки этапов разбора необходимо нажать кнопку «Сохранить» для сохранения правила и следом нажать кнопку «Опубликовать» для последующего его использования.

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

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

  1. Для создания правила нормализации необходимо перейти в раздел “Источники” → “Управление источниками”, выбрать вкладку “Правила нормализации”, после чего откроется страница создания, редактирования и просмотра правил нормализации.

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

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

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

    В качестве названия правила нормализации нужно указывать уникальный идентификатор сообщения для данного источника, в случае с примером: arcsight_for_test

  3. В поле для сырого события необходимо вставить разобранное событие (см. раздел результат создания правила разбора).

  4. После создания правила, внутри него автоматически создается нормализатор root.yaml

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

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

    Добавление поля нормализации

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

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

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

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

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

    Поля нормализации для root.yaml Поля нормализации для root.yaml

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

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

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

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

  8. После настройки нормализатора root.yaml, необходимо перейти к созданию нормализатора для определенного типа событий от данного источника, в случае с примером - это Netflow;

  9. Для этого в поле рядом с кнопкой "Добавить нормализатор" нужно ввести имя нормализатора и нажать на кнопку "+ Добавить нормализатор", как изображено на рисунке 10;

    Добавление нового нормализатора

    Рисунок 10 -- Добавление нового нормализатора

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

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

    Маршрутизация события

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

    Таким образом, все события, которые подходят под условие: signature == 'flow' and version == '9' будут нормализованы по данному файлу нормализации

  12. Для более гибкой, понятной и правильной нормализации в данном файле нормализации используются специальные функции, которые подробно описаны в разделе Специальные функции для работы с полями нормализации.

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

    Функция 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"

    ПРЕДУПРЕЖДЕНИЕ! Использование данного механизма может сказаться на производительности и скорости работы обработчиков событий.

  14. Таким образом, таблица просмотра (функция lookup) для данного файла нормализации представлена на рисунке 12;

    Таблицы просмотра

    Рисунок 12 -- Таблицы просмотра

  15. Дополнительная настройка нормализатора изображена на рисунке 13;

    Дополнительная настройка

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

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

  16. Поля нормализации используемые в данном файле нормализации представлены на рисунках 14-17;

    Поля нормализации

    Поля нормализации

    Поля нормализации

    Поля нормализации

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

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

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

    {
      "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
        }
      }
    }
    

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

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

  19. Для его создания, в поле названия нормализатора нужно ввести "parsed_only" и нажать на кнопку "+ Добавить нормализатор";

  20. После чего, в маршрутизации события указать "fallback" и кликнуть на кнопку "Только разбор" чтобы она перешла в статус "Вкл", как изображено на рисунке 18;

    Добавление нормализатора parsed_only

    Рисунок 18 -- Добавление нормализатора "parsed_only.yaml"

  21. В данном нормализаторе не требуется добавления полей нормализации. В автоматически созданные поля необходимо указать "parsed" в поле "Фиксированное значение";

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

  23. Для этого необходимо нажать кнопку "Сохранить" и следом кнопку "Опубликровать";

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