1000_zeek_prep.conf

# Copyright (c) 2024 Battelle Energy Alliance, LLC.  All rights reserved.

filter {

  if (([message] =~ /^#/) or (![log][file][path])) {
    drop { id => "drop_zeek_invalid_logs" }
  }

  # - Tags may have been specified, like: conn(tagA,tagB,tagC).log, extract the log type (conn) and the tags (tagA,tagB,tagC).
  # - Normalize log types with - in their names to _ (e.g., opcua-binary -> opcua_binary).
  # - Zeek log files might be caught by filebeat right in the middle of being renamed/moved as
  #   part of log rotation (ie., renamed from conn.log to conn.2020-01-16-14-00-00.log or
  #   conn.2020_01_16_14_00_00.log). We don't care about that, ignore the date part and just process
  #   the log source as we normally would.
  ruby {
    id => "ruby_zeek_log_source_extract"
    #                                                                       ↓Type     ↓Tags   ↓Rotate Timestamp (discard)                                        ↓.log (discard)
    code => "
      if fileParts = event.get('[log][file][path]').split('/').last.match(/^(.*?)(?:\((.*)\))?(?:\.\d{4}[_:-]?\d{2}[_:-]?\d{2}[_:-]?\d{2}[_:-]?\d{2}[_:-]?\d{2})?\.log/i) then
        logType, tags  = fileParts.captures
        event.set('[log_source]', logType.gsub('-', '_').gsub(/_general\z/, '')) unless logType.nil?
        event.set('[@metadata][zeek_log_tags]', tags) unless tags.nil?
      end"
  }

  # Zeek logs we're going to ignore
  ruby {
    id => "ruby_zeek_log_type_determine_drop"
    init => "logtypesStr = ENV['LOGSTASH_ZEEK_IGNORED_LOGS'] || 'analyzer,broker,bsap_ip_unknown,bsap_serial_unknown,capture_loss,cluster,config,ecat_arp_info,loaded_scripts,packet_filter,png,print,prof,reporter,stats,stderr,stdout' ; @logtypes = logtypesStr.gsub(/\s+/, '').split(',')"
    code => "event.set('[@metadata][drop_zeek_log]', true) if @logtypes.include?(event.get('[log_source]').to_s)"
  }
  if [@metadata][drop_zeek_log] { drop { id => "drop_zeek_ignored_source" } }


  # remove some tags pulled from the filename we might not want
  if ([@metadata][zeek_log_tags]) {
    ruby {
      id => "ruby_zeek_prune_tags"
      code => "
        filenameTags = event.get('[@metadata][zeek_log_tags]').split(',')
        nbsiteid = filenameTags.find { |str| str[/NBSITEID(.+)/] }[/NBSITEID(.+)/, 1].to_s rescue nil
        filenameTags.delete_if{|v| ((v == nil) or (v == '') or (v !~ /\D/) or (v =~ /\A\s*NBSITEID/i) or (v =~ /\A\s*(autocarve)/i) or (v =~ /\A\s*(pcap|dmp|log|bro|zeek|suricata|m?tcpdump|m?netsniff|autozeek|autosuricata)s?\s*\z/i) or (v == event.get('[log_source]')))}
        event.set('[@metadata][zeek_log_tags]', filenameTags.uniq) unless (filenameTags.length == 0)
        event.set('[@metadata][nbsiteid]', nbsiteid) unless (nbsiteid.nil? || nbsiteid.empty?)
      "
    }
    if ([@metadata][zeek_log_tags]) { mutate { id => "mutate_merge_zeek_log_tags"
                                                  merge => { "[tags]" => "[@metadata][zeek_log_tags]" } } }
  }
  if (![@metadata][nbsiteid]) and ([netbox][site]) {
    # for forwarded logs, the NetBox site value arrives in netbox.site
    mutate {
      id => "mutate_rename_zeek_forwarded_site_id"
      rename => { "[netbox][site]" => "[@metadata][nbsiteid]" }
    }
  }

} # end Filter