问题描述
我想将 Wazuh 服务器与 HELK 集成，但我做不到，logstash 也无法获取来自 kafka 的任何 Wazuh 警报或将 Wazuh 警报发送到 Elasticsearch.我使用wazuh-alerts"名称创建 kafka 主题并在 logstash 配置文件中设置我的配置.有什么问题?
I want to integrate Wazuh server with HELK but I can't do it and logstash cannot get any Wazuh alert from kafka or sending Wazuh alerts to Elasticsearch.I create kafka topic with "wazuh-alerts" name and set my configuration in logstash config files . What's problem ?
在 HELK 中，Logstash 配置放置在管道目录中，它包含以下文件:
In HELK , Logstash configuration placed on pipeline directory and it's contain below file :
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 文件:
I changed 0002-kafka-input.conf file from :
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"
}
}
下面是为 Wazuh 警报创建的新 kafka 主题输入:
to below for new kafka topic input that created for Wazuh alerts :
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 文件，其中包含以下内容:
and I created 1546-wazuh-alerts-filter.conf file with below content for input filter :
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"]
}
}
}
和用于 logstash 输出配置的 9963-wazuh-alerts-output.conf 文件:
and 9963-wazuh-alerts-output.conf file for logstash output config :
output {
if "wazuh-alerts" in [tags]{
elasticsearch {
hosts => ["helk-elasticsearch:9200"]
index => "mitre-attack-%{+YYYY.MM.dd}"
user => 'elastic'
#password => 'elasticpassword'
}
}
}
推荐答案
我将 logstash 配置更改为:
I changed logstash configuration to this :
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 配置在 这里.我应该使用 kafka 主题将 sendig wazuh 警报日志发送到 Helk，因为 HELK 使用 kafka(我将带有 filebeat 的 wazuh 警报日志发送到 kafka).现在我怎样才能改变 HELK 配置来做到这一点?谢谢
Please if possible see HELK configuration at here and wazuh logstash config at here. I should use kafka topic for sendig wazuh alerts log to Helk because HELK using kafka ( I sent wazuh alerts log with filebeat to kafka ). Now how can I change HELK Configuration to do it ?Thanks
这篇关于如何将 Wazuh 服务器与 HELK 集成的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持！