This write-up explains how to implement Logstash with Wazuh to create lookup files and enrich incoming data. The implementation's topology is shown below.

First, install Logstash and follow this documentation: Wazuh Elastic Stack integration · Wazuh documentation

Download and install the Public Signing Key:

wget -qO - <https://artifacts.elastic.co/GPG-KEY-elasticsearch> | sudo gpg --dearmor -o /usr/share/keyrings/elastic-keyring.gpg

You may need to install the apt-transport-https package on Debian before proceeding:

sudo apt-get install apt-transport-https

Save the repository definition to /etc/apt/sources.list.d/elastic-8.x.list:

echo "deb [signed-by=/usr/share/keyrings/elastic-keyring.gpg] <https://artifacts.elastic.co/packages/8.x/apt> stable main" | sudo tee -a /etc/apt/sources.list.d/elastic-8.x.list

Run sudo apt-get update and the repository is ready for use. You can install it with:

sudo apt-get update && sudo apt-get install logstash

After installing Logstash install the opensearch-output plugin

sudo /usr/share/logstash/bin/logstash-plugin install logstash-output-opensearch

You only need Wazuh-indexer certificates for Logstash if you install Logstash on a different server. You can add to this path /etc/logstash/wazuh-indexer-certs and give Logstash user the necessary permissions sudo chmod -R 755 </PATH/TO/LOCAL/WAZUH_INDEXER/CERTIFICATE>/root-ca.pem .

The next step is to edit the filebeat.yml file to send logs to Logstash instead of directly to the Wazuh-indexer. The filebeat.yml file is located here etc/filebeat/filebeat.yml, and edit the file like this:

# Uncomment and configure either Elasticsearch or Logstash output. Only one should be active.

# Elasticsearch output (commented out in this case)
# output.elasticsearch:
#   hosts: ["127.0.0.1:9200"]
#   protocol: https
#   username: ${username}
#   password: ${password}
#   ssl.certificate_authorities:
#     - /etc/filebeat/certs/root-ca.pem
#   ssl.certificate: "/etc/filebeat/certs/wazuh-server.pem"
#   ssl.key: "/etc/filebeat/certs/wazuh-server-key.pem"
#   pipeline: "extract_fields_pipeline"

# Logstash output
output.logstash:
  hosts: ["127.0.0.1:5044"]

# Setup for Wazuh template
setup.template.json.enabled: true
setup.template.json.path: '/etc/filebeat/wazuh-template.json'
setup.template.json.name: 'wazuh'
setup.ilm.overwrite: true
setup.ilm.enabled: false

# Enable Wazuh Filebeat module
filebeat.modules:
  - module: wazuh
    alerts:
      enabled: true
    archives:
      enabled: false

# Logging configuration
logging.level: info
logging.to_files: true
logging.files:
  path: /var/log/filebeat
  name: filebeat
  keepfiles: 7
  permissions: 0644

logging.metrics.enabled: false

# Seccomp configuration
seccomp:
  default_action: allow
  syscalls:
    - action: allow
      names:
        - rseq

Now edit logstash.conf file under this directory /etc/logstash/conf.d to read logs coming from Filebeat, enrich the data with CSV files, and send the enriched logs to Wazuh-indexer. Here is the example configuration:

input {
  beats {
    port => 5044
  }
}

filter {
  json {
    source => "message"
    target => "parsed_message"
  }

  # Load the CSV file into a dictionary
  ruby {
    path => "/home/berkan/Documents/enrich.rb"
  }

  # Use the enriched data
  mutate {
    add_field => {
      "department" => "%{[parsed_message][data][department]}"
      "vlan_addr" => "%{[parsed_message][data][vlan_address]}"
      "role" => "%{[parsed_message][data][Role]}"
    }
  }

  mutate {
    remove_field => ["parsed_message"]
  }
}

output {
  # Send enriched logs to OpenSearch
  opensearch {
    hosts => ["<https://127.0.0.1:9200>"]
    index => "si_tcodes-%{+YYYY.MM.dd}"
    user => "replace it with your wazuh username"
    password => "replace it with your wazuh password"
    ssl => true
    cacert => "/etc/logstash/wazuh-indexer-certs/root-ca.pem"
    pipeline => "filebeat-7.10.2-wazuh-alerts-pipeline"
  }

Now I will explain the enrichment process. First You should use the JSON filter plugin to parse incoming logs I do this because we do lookups based on a field that already exists on the logs and if we try to parse a specific field we can’t do it on raw format. The output of this is, parsed_message, we can access the si_user field in this example. Then I wrote a ruby script to do the matching field between incoming logs and the CSV file, after the matching it will get the new values from the CSV file and format for us to use in the mutate plugin for generating new fields. And the other reason we use this script is we can read multiple columns in a CSV file. The ruby script and CSV file are below:

require 'csv'

# Load the CSV file into a hash
csv_file = '/home/berkan/Documents/test.csv'
$enrichment_data = {}
CSV.foreach(csv_file, headers: true) do |row|
  $enrichment_data[row['si_user']] = {
    'department' => row['Department'],
    'vlan_address' => row['vlan_address'],
    'Role' => row['Role']
  }
end

# Enrich the event
def register(params)
end

def filter(event)
    if event.get('[parsed_message][data][si_user]')
       si_user = event.get('[parsed_message][data][si_user]')
       if $enrichment_data.key?(si_user)
          enrichment = $enrichment_data[si_user]
          event.set('[parsed_message][data][department]', enrichment['department'])
          event.set('[parsed_message][data][vlan_address]', enrichment['vlan_address'])
          event.set('[parsed_message][data][Role]', enrichment['Role'])
       end
    end
  return [event]
end