Как интегрировать Wazuh Server с HELK - PullRequest
Новая
       17

Как интегрировать Wazuh Server с HELK

0 голосов
/ 27 апреля 2019

Я хочу интегрировать сервер Wazuh с HELK , но я не могу этого сделать, и logstash не может получить никакого предупреждения Wazuh от kafka или отправки предупреждений Wazuh в Elasticsearch. Я создаю тему kafka с именем "wazuh-alerts" и настраиваю свою конфигурацию в файлах конфигурации logstash. В чем проблема?

В HELK конфигурация Logstash помещается в каталог конвейера, и он содержит следующий файл: 

0002-kafka-input.conf                                       1534-winevent-application-filter.conf        8801-meta-command_line-enrichment_and_additions-filter.conf
0003-attack-input.conf                                      1535-winevent-wmiactivity-filter.conf        8802-meta-powershell-enrichment_and_additions-filter.conf
0004-beats-input.conf                                       1541-winevent-process-name-split.conf        8901-fingerprints-command_line-filter.conf
0005-nxlog-winevent-syslog-tcp-input.conf                   1542-winevent-process-ids-conversions.conf   8902-fingerprints-powershell.conf
0098-all-filter.conf                                        1543-winevent-user-ids-conversions.conf      9950-winevent-sysmon-output.conf
0099-all-fingerprint-hash-filter.conf                       1544-winevent-cleanup-other.conf             9951-winevent-security-output.conf
0301-nxlog-winevent-to-json.conf                            1545-winevent-security-conversions.conf      9952-winevent-system-output.conf
1010-winevent-winlogbeats-filter.conf                       9953-winevent-application-output.conf
1050-nxlog-winevent-to-winlogbeats-merge-filter.conf        2511-winevent-powershell-filter.conf         9954-winevent-powershell-output.conf
1216-attack-filter.conf                                     2512-winevent-security-schtasks-filter.conf  9955-winevent-wmiactivity-output.conf
1500-winevent-cleanup-no-dashes-only-values-filter.conf     8012-dst-ip-cleanups-filter.conf             9956-attack-output.conf
1521-winevent-conversions-ip-conversions-basic-filter.conf  8013-src-ip-cleanups-filter.conf             9957-winevent-sysmon-join-output.conf
1522-winevent-cleanup-lowercasing-windows-filter.conf       8014-dst-nat-ip-cleanups-filter.conf         9958-osquery-output.conf
1523-winevent-process-name-filter.conf                      8015-src-nat-ip-cleanups-filter.conf         9959-winevent-codeintegrity-output.conf
1524-winevent-process-ids-filter.conf                       8112-dst-ip-filter.conf                      9960-winevent-bits-output.conf
1531-winevent-sysmon-filter.conf                            8113-src-ip-filter.conf                      9961-winevent-dns-client-output.conf
1532-winevent-security-filter.conf                          8114-dst-nat-ip-filter.conf                  9962-winevent-firewall-advanced-output.conf
1533-winevent-system-filter.conf                            8115-src-nat-ip-filter.conf

Я изменил файл 0002-kafka-input.conf из: 

input {
  kafka {
    bootstrap_servers => "helk-kafka-broker:9092"
    topics => ["winlogbeat", "SYSMON_JOIN","filebeat"]
    decorate_events => true
    codec => "json"
    auto_offset_reset => "latest"
    ############################# HELK Optimizing Throughput & Latency #############################
    fetch_min_bytes => "1"
    request_timeout_ms => "40000"
    ############################# HELK Optimizing Durability #############################
    enable_auto_commit => "false"
    ############################# HELK Optimizing Availability #############################
    connections_max_idle_ms => "540000"
    session_timeout_ms => "30000"
    max_poll_interval_ms => "300000"
    #############################
    max_poll_records => "500"
  }
}

ниже для ввода новой темы kafka, созданной для предупреждений Wazuh: 

input {
  kafka {
    bootstrap_servers => "helk-kafka-broker:9092"
    topics => ["winlogbeat", "SYSMON_JOIN","filebeat"]
    decorate_events => true
    codec => "json"
    auto_offset_reset => "latest"
    tags => [ "winlog-sysmon" ]
    ############################# HELK Optimizing Throughput & Latency #############################
    fetch_min_bytes => "1"
    request_timeout_ms => "40000"
    ############################# HELK Optimizing Durability #############################
    enable_auto_commit => "false"
    ############################# HELK Optimizing Availability #############################
    connections_max_idle_ms => "540000"
    session_timeout_ms => "30000"
    max_poll_interval_ms => "300000"
    #############################
    max_poll_records => "500"
  }
  kafka {
    bootstrap_servers => "helk-kafka-broker:9092"
    topics => ["wazuh-alerts"]
    decorate_events => true
    codec => "json_lines"
    tags => [ "wazuh-alerts" ]
    auto_offset_reset => "latest"
  }
}

и я создал файл 1546-wazuh-alerts-filter.conf с содержимым ниже для фильтра ввода: 

filter {
  if "wazuh-alerts" in [tags]{
    if [data][srcip] {
      mutate {
        add_field => [ "@src_ip", "%{[data][srcip]}" ]
    }
    }
    if [data][aws][sourceIPAddress] {
      mutate {
        add_field => [ "@src_ip", "%{[data][aws][sourceIPAddress]}" ]
      }
    }
    geoip {
      source => "@src_ip"
      target => "GeoLocation"
      fields => ["city_name", "country_name", "region_name", "location"]
    }
    date {
      match => ["timestamp", "ISO8601"]
      target => "@timestamp"
    }
    mutate {
      remove_field => [ "timestamp", "beat", "input_type", "tags", "count", "@version", "log", "offset", "type", "@src_ip", "host"]
    }
  }
}
Файл

и 9963-wazuh-alerts-output.conf для конфигурации вывода logstash: 

output {
  if "wazuh-alerts" in [tags]{
    elasticsearch {
      hosts => ["helk-elasticsearch:9200"]
      index => "mitre-attack-%{+YYYY.MM.dd}"
      user => 'elastic'
      #password => 'elasticpassword'
    }
  }
}

Ответы [ 2 ]

0 голосов
/ 28 апреля 2019

Я изменил конфигурацию logstash на это: 

input {
  kafka {
    bootstrap_servers => "helk-kafka-broker:9092"
    topics => ["winlogbeat", "SYSMON_JOIN","filebeat"]
    decorate_events => true
    codec => "json"
    auto_offset_reset => "latest"
    ############################# HELK Optimizing Throughput & Latency #############################
    fetch_min_bytes => "1"
    request_timeout_ms => "40000"
    ############################# HELK Optimizing Durability #############################
    enable_auto_commit => "false"
    ############################# HELK Optimizing Availability #############################
    connections_max_idle_ms => "540000"
    session_timeout_ms => "30000"
    max_poll_interval_ms => "300000"
    #############################
    max_poll_records => "500"
  }
  kafka {
    bootstrap_servers => "helk-kafka-broker:9092"
    topics => ["wazuh-alerts"]
    decorate_events => true
    codec => "json_lines"
    auto_offset_reset => "latest"
  }
}


filter {
  if [@metadata][kafka][topic] == "wazuh-alerts" {
    if [data][srcip] {
      mutate {
        add_field => [ "@src_ip", "%{[data][srcip]}" ]
    }
    }
    if [data][aws][sourceIPAddress] {
      mutate {
        add_field => [ "@src_ip", "%{[data][aws][sourceIPAddress]}" ]
      }
    }
    geoip {
      source => "@src_ip"
      target => "GeoLocation"
      fields => ["city_name", "country_name", "region_name", "location"]
    }
    date {
      match => ["timestamp", "ISO8601"]
      target => "@timestamp"
    }
    mutate {
      remove_field => [ "timestamp", "beat", "input_type", "tags", "count", "@version", "log", "offset", "type", "@src_ip", "host"]
    }
  }
}

output {
  if [@metadata][kafka][topic] == "wazuh-alerts" {
    file {
      path => "/var/log/greatlog.log"
    }
  }
}

Пожалуйста, если возможно, смотрите конфигурацию HELK на здесь и wazuh logstash config на здесь , Я должен использовать тему kafka для отправки предупреждений wazuh в Helk, потому что HELK использует kafka (я отправил журнал предупреждений wazuh с filebeat в kafka). Теперь, как я могу изменить конфигурацию HELK, чтобы сделать это? Спасибо

0 голосов
/ 27 апреля 2019

Ваш вывод имеет условие, используя поле tags, но вы удаляете его в блоке фильтров, оно не работает, поскольку поле не существует, когда событие входит в выходной блок.

Оставьте поле tags в своем сообщении и повторите попытку.

...