bro-ids_logstash.conf

# Bro-IDS Logstash parser
# Parts of this taken from http://www.appliednsm.com/wp-content/uploads/logstash-SObro22-parse.conf_.txt

#Logs being parsed:
#app_stats.log
#conn.log
#dns.log
#dpd.log
#files.log
#http.log
#notice.log
#software.log
#dhcp.log
#known_certs.log
#known_hosts.log
#known_services.log
#ssh.log
#ssl.log
#weird.log

input {

#Production Logs#############################
  file {
    type => "BRO_httplog"
    path => "/usr/local/bro/logs/current/http.log"
    sincedb_path => "/var/log/.bro_http_sincedb"
  }
  file {
    type => "BRO_known_certslog"
    path => "/usr/local/bro/logs/current/known_certs.log"
    sincedb_path => "/var/log/.bro_known_certs_sincedb"
  }
  file {
    type => "BRO_noticelog"
    path => "/usr/local/bro/logs/current/notice.log"
    sincedb_path => "/var/log/.bro_notice_sincedb"
  }
  file {
    type => "BRO_known_hostslog"
    path => "/usr/local/bro/logs/current/known_hosts.log"
    sincedb_path => "/var/log/.bro_known_hosts_sincedb"
  }
  file {
    type => "BRO_known_serviceslog"
    path => "/usr/local/bro/logs/current/known_services.log"
    sincedb_path => "/var/log/.bro_known_services_sincedb"
  }
  file {
    type => "BRO_sshlog"
    path => "/usr/local/bro/logs/current/ssh.log"
    sincedb_path => "/var/log/.bro_ssh_sincedb"
  }
  file {
    type => "BRO_dpdlog"
    path => "/usr/local/bro/logs/current/dpd.log"
    sincedb_path => "/var/log/.bro_dpd_sincedb"
  }
  file {
    type => "BRO_connlog"
    path => "/usr/local/bro/logs/current/conn.log"
    sincedb_path => "/var/log/.bro_conn_sincedb"
  }
  file {
    type => "BRO_weirdlog"
    path => "/usr/local/bro/logs/current/weird.log"
    sincedb_path => "/var/log/.bro_weird_sincedb"
  }
  file {
    type => "BRO_app_statslog"
    path => "/usr/local/bro/logs/current/appstats.log"
    sincedb_path => "/var/log/.bro_appstats_sincedb"
  }
  file {
    type => "BRO_dhcplog"
    path => "/usr/local/bro/logs/current/dhcp.log"
    sincedb_path => "/var/log/.bro_dhcp_sincedb"
  }
  file {
    type => "BRO_fileslog"
    path => "/usr/local/bro/logs/current/files.log"
    sincedb_path => "/var/log/.bro_files_sincedb"
  }
  file {
    type => "BRO_ssllog"
    path => "/usr/local/bro/logs/current/ssl.log"
    sincedb_path => "/var/log/.bro_ssl_sincedb"
  }
  file {
    type => "BRO_noticelog"
    path => "/usr/local/bro/logs/current/notice.log"
    sincedb_path => "/var/log/.bro_notice_sincedb"
  }
  file {
    type => "BRO_softwarelog"
    path => "/usr/local/bro/logs/current/software.log"
    sincedb_path => "/var/log/.bro_software_sincedb"
  }
  file {
    type => "BRO_dnslog"
    path => "/usr/local/bro/logs/current/dns.log"
    sincedb_path => "/var/log/.bro_dns_sincedb"
  }
  file {
    type => "BRO_intellog"
    path => "/usr/local/bro/logs/current/intel.log"
    sincedb_path => "/var/log/.bro_intel_sincedb"
  }
}

filter {
  if [message] =~ /^#/ {
    drop {  }
     }
    else {
 	
    # BRO_app_statslog ######################
      if [type] == "BRO_app_statslog" {
        grok {
          match => [ "message", "(?<ts>(.*?))\t(?<ts_delta>(.*?))\t(?<app>(.*?))\t(?<uniq_hosts>(.*?))\t(?<hits>(.*?))\t(?<bytes>(.*))" ]
        }
      }
    
    # BRO_connlog ######################
      if [type] == "BRO_connlog" {
        grok {
    	    match => [ 
    		    "message", "(?<ts>(.*?))\t(?<uid>(.*?))\t(?<id.orig_h>(.*?))\t(?<id.orig_p>(.*?))\t(?<id.resp_h>(.*?))\t(?<id.resp_p>(.*?))\t(?<proto>(.*?))\t(?<service>(.*?))\t(?<duration>(.*?))\t(?<orig_bytes>(.*?))\t(?<resp_bytes>(.*?))\t(?<conn_state>(.*?))\t(?<local_orig>(.*?))\t(?<missed_bytes>(.*?))\t(?<history>(.*?))\t(?<orig_pkts>(.*?))\t(?<orig_ip_bytes>(.*?))\t(?<resp_pkts>(.*?))\t(?<resp_ip_bytes>(.*?))\t(?<tunnel_parents>(.*?))\t(?<orig_cc>(.*?))\t(?<resp_cc>(.*?))\t(?<sensorname>(.*))",
    		    "message", "(?<ts>(.*?))\t(?<uid>(.*?))\t(?<id.orig_h>(.*?))\t(?<id.orig_p>(.*?))\t(?<id.resp_h>(.*?))\t(?<id.resp_p>(.*?))\t(?<proto>(.*?))\t(?<service>(.*?))\t(?<duration>(.*?))\t(?<orig_bytes>(.*?))\t(?<resp_bytes>(.*?))\t(?<conn_state>(.*?))\t(?<local_orig>(.*?))\t(?<missed_bytes>(.*?))\t(?<history>(.*?))\t(?<orig_pkts>(.*?))\t(?<orig_ip_bytes>(.*?))\t(?<resp_pkts>(.*?))\t(?<resp_ip_bytes>(.*?))\t(%{NOTSPACE:tunnel_parents})"
    	    ]
        }
      }
    
    # BRO_noticelog ######################
      if [type] == "BRO_noticelog" {
        grok { 
          match => [ "message", "(?<ts>(.*?))\t(?<uid>(.*?))\t(?<id.orig_h>(.*?))\t(?<id.orig_p>(.*?))\t(?<id.resp_h>(.*?))\t(?<id.resp_p>(.*?))\t(?<fuid>(.*?))\t(?<file_mime_type>(.*?))\t(?<file_desc>(.*?))\t(?<proto>(.*?))\t(?<note>(.*?))\t(?<msg>(.*?))\t(?<sub>(.*?))\t(?<src>(.*?))\t(?<dst>(.*?))\t(?<p>(.*?))\t(?<n>(.*?))\t(?<peer_descr>(.*?))\t(?<actions>(.*?))\t(?<suppress_for>(.*?))\t(?<dropped>(.*?))\t(?<remote_location.country_code>(.*?))\t(?<remote_location.region>(.*?))\t(?<remote_location.city>(.*?))\t(?<remote_location.latitude>(.*?))\t(?<remote_location.longitude>(.*))" ]
        }
      }
    
    
    # BRO_dhcplog ######################
      if [type] == "BRO_dhcplog" {
        grok { 
          match => [ "message", "(?<ts>(.*?))\t(?<uid>(.*?))\t(?<id.orig_h>(.*?))\t(?<id.orig_p>(.*?))\t(?<id.resp_h>(.*?))\t(?<id.resp_p>(.*?))\t(?<mac>(.*?))\t(?<assigned_ip>(.*?))\t(?<lease_time>(.*?))\t(?<trans_id>(.*))" ]
        }
      }
    
    # BRO_dnslog ######################
      if [type] == "BRO_dnslog" {
        grok {
          match => [ "message", "(?<ts>(.*?))\t(?<uid>(.*?))\t(?<id.orig_h>(.*?))\t(?<id.orig_p>(.*?))\t(?<id.resp_h>(.*?))\t(?<id.resp_p>(.*?))\t(?<proto>(.*?))\t(?<trans_id>(.*?))\t(?<query>(.*?))\t(?<qclass>(.*?))\t(?<qclass_name>(.*?))\t(?<qtype>(.*?))\t(?<qtype_name>(.*?))\t(?<rcode>(.*?))\t(?<rcode_name>(.*?))\t(?<AA>(.*?))\t(?<TC>(.*?))\t(?<RD>(.*?))\t(?<RA>(.*?))\t(?<Z>(.*?))\t(?<answers>(.*?))\t(?<TTLs>(.*?))\t(?<rejected>(.*))" ]
        }
      }
    
    # BRO_softwarelog ######################
      if [type] == "BRO_softwarelog" {
        grok { 
          match => [ "message", "(?<ts>(.*?))\t(?<bro_host>(.*?))\t(?<host_p>(.*?))\t(?<software_type>(.*?))\t(?<name>(.*?))\t(?<version.major>(.*?))\t(?<version.minor>(.*?))\t(?<version.minor2>(.*?))\t(?<version.minor3>(.*?))\t(?<version.addl>(.*?))\t(?<unparsed_version>(.*))" ]
        }
      }
    
    # BRO_dpdlog ######################
      if [type] == "BRO_dpdlog" {
        grok {
          match => [ "message", "(?<ts>(.*?))\t(?<uid>(.*?))\t(?<id.orig_h>(.*?))\t(?<id.orig_p>(.*?))\t(?<id.resp_h>(.*?))\t(?<id.resp_p>(.*?))\t(?<proto>(.*?))\t(?<analyzer>(.*?))\t(?<failure_reason>(.*))" ]
        }
      }
    
    # BRO_fileslog ######################
      if [type] == "BRO_fileslog" {
        grok {
          match => [ "message", "(?<ts>(.*?))\t(?<fuid>(.*?))\t(?<tx_hosts>(.*?))\t(?<rx_hosts>(.*?))\t(?<conn_uids>(.*?))\t(?<source>(.*?))\t(?<depth>(.*?))\t(?<analyzers>(.*?))\t(?<mime_type>(.*?))\t(?<filename>(.*?))\t(?<duration>(.*?))\t(?<local_orig>(.*?))\t(?<is_orig>(.*?))\t(?<seen_bytes>(.*?))\t(?<total_bytes>(.*?))\t(?<missing_bytes>(.*?))\t(?<overflow_bytes>(.*?))\t(?<timedout>(.*?))\t(?<parent_fuid>(.*?))\t(?<md5>(.*?))\t(?<sha1>(.*?))\t(?<sha256>(.*?))\t(?<extracted>(.*))" ]
        }
      }
    
    # BRO_httplog ######################
      if [type] == "BRO_httplog" {
        grok {
          match => [ "message", "(?<ts>(.*?))\t(?<uid>(.*?))\t(?<id.orig_h>(.*?))\t(?<id.orig_p>(.*?))\t(?<id.resp_h>(.*?))\t(?<id.resp_p>(.*?))\t(?<trans_depth>(.*?))\t(?<method>(.*?))\t(?<bro_host>(.*?))\t(?<uri>(.*?))\t(?<referrer>(.*?))\t(?<user_agent>(.*?))\t(?<request_body_len>(.*?))\t(?<response_body_len>(.*?))\t(?<status_code>(.*?))\t(?<status_msg>(.*?))\t(?<info_code>(.*?))\t(?<info_msg>(.*?))\t(?<filename>(.*?))\t(?<http_tags>(.*?))\t(?<username>(.*?))\t(?<password>(.*?))\t(?<proxied>(.*?))\t(?<orig_fuids>(.*?))\t(?<orig_mime_types>(.*?))\t(?<resp_fuids>(.*?))\t(?<resp_mime_types>(.*))" ]
        }
      }
    
    # BRO_known_certslog ######################
      if [type] == "BRO_known_certslog" {
        grok {
          match => [ "message", "(?<ts>(.*?))\t(?<bro_host>(.*?))\t(?<port_num>(.*?))\t(?<subject>(.*?))\t(?<issuer_subject>(.*?))\t(?<serial>(.*))" ]
        }
      }
    
    # BRO_known_hostslog ######################
      if [type] == "BRO_known_hostslog" {
        grok {
          match => [ "message", "(?<ts>(.*?))\t(?<bro_host>(.*))" ]
        }
      }
    
    # BRO_known_serviceslog ######################
      if [type] == "BRO_known_serviceslog" {
        grok {
          match => [ "message", "(?<ts>(.*?))\t(?<bro_host>(.*?))\t(?<port_num>(.*?))\t(?<port_proto>(.*?))\t(?<service>(.*))" ]
        }
      }
    
    # BRO_sshlog ######################
      if [type] == "BRO_sshlog" {
        grok {
          match => [ "message", "(?<ts>(.*?))\t(?<uid>(.*?))\t(?<id.orig_h>(.*?))\t(?<id.orig_p>(.*?))\t(?<id.resp_h>(.*?))\t(?<id.resp_p>(.*?))\t(?<status>(.*?))\t(?<direction>(.*?))\t(?<client>(.*?))\t(?<server>(.*?))\t(?<remote_location.country_code>(.*?))\t(?<remote_location.region>(.*?))\t(?<remote_location.city>(.*?))\t(?<remote_location.latitude>(.*?))\t(?<remote_location.longitude>(.*))" ]
        }
      }
    
    # BRO_ssllog ######################
      if [type] == "BRO_ssllog" {
        grok {
          match => [ "message", "(?<ts>(.*?))\t(?<uid>(.*?))\t(?<id.orig_h>(.*?))\t(?<id.orig_p>(.*?))\t(?<id.resp_h>(.*?))\t(?<id.resp_p>(.*?))\t(?<version>(.*?))\t(?<cipher>(.*?))\t(?<server_name>(.*?))\t(?<session_id>(.*?))\t(?<subject>(.*?))\t(?<issuer_subject>(.*?))\t(?<not_valid_before>(.*?))\t(?<not_valid_after>(.*?))\t(?<last_alert>(.*?))\t(?<client_subject>(.*?))\t(?<client_issuer_subject>(.*?))\t(?<cert_hash>(.*?))\t(?<validation_status>(.*))" ]
        }
      }
    
    # BRO_weirdlog ######################
    if [type] == "BRO_weirdlog" {
    	grok {
    		match => [ "message", "(?<ts>(.*?))\t(?<uid>(.*?))\t(?<id.orig_h>(.*?))\t(?<id.orig_p>(.*?))\t(?<id.resp_h>(.*?))\t(?<id.resp_p>(.*?))\t(?<name>(.*?))\t(?<addl>(.*?))\t(?<notice>(.*?))\t(?<peer>(.*))" ]
        	}
    }
    if [type]== "BRO_intellog" {
      grok {
        match => [ "message", "(?<ts>(.*?))\t%{DATA:uid}\t(?<id.orig_h>(.*?))\t(?<id.orig_p>(.*?))\t(?<id.resp_h>(.*?))\t(?<id.resp_p>(.*?))\t%{DATA:fuid}\t%{DATA:file_mime_type}\t%{DATA:file_desc}\t(?<seen.indicator>(.*?))\t(?<seen.indicator_type>(.*?))\t(?<seen.where>(.*?))\t%{NOTSPACE:sources}" ]
     }
   }
  }
  date {
    match => [ "ts", "UNIX" ]
  }
}
filter {
  if [bro_host] {
    mutate {
      replace => [ "host", "%{bro_host}" ]
    }
  }
}
filter {
  if "BRO" in [type] {
    if [id.orig_h] {
      mutate {
        add_field => [ "senderbase_lookup", "http://www.senderbase.org/lookup/?search_string=%{id.orig_h}" ]
        add_field => [ "CBL_lookup", "http://cbl.abuseat.org/lookup.cgi?ip=%{id.orig_h}" ]
        add_field => [ "Spamhaus_lookup", "http://www.spamhaus.org/query/bl?ip=%{id.orig_h}" ]
      }
    }
    mutate {
      add_tag => [ "BRO" ]
    }
    mutate {
      convert => [ "id.orig_p", "integer" ]
      convert => [ "id.resp_p", "integer" ]
      convert => [ "orig_bytes", "integer" ]
      convert => [ "resp_bytes", "integer" ]
      convert => [ "missed_bytes", "integer" ]
      convert => [ "orig_pkts", "integer" ]
      convert => [ "orig_ip_bytes", "integer" ]
      convert => [ "resp_pkts", "integer" ]
      convert => [ "resp_ip_bytes", "integer" ]
    }
  }
}
filter {
  if [type] == "BRO_connlog" {
    #The following makes use of the translate filter (logstash contrib) to convert conn_state into human text. Saves having to look up values for packet introspection
    translate {
      field => "conn_state"
      destination => "conn_state_full"
      dictionary => [ 
        "S0", "Connection attempt seen, no reply",
        "S1", "Connection established, not terminated",
        "S2", "Connection established and close attempt by originator seen (but no reply from responder)",
        "S3", "Connection established and close attempt by responder seen (but no reply from originator)",
        "SF", "Normal SYN/FIN completion",
        "REJ", "Connection attempt rejected",
        "RSTO", "Connection established, originator aborted (sent a RST)",
        "RSTR", "Established, responder aborted",
        "RSTOS0", "Originator sent a SYN followed by a RST, we never saw a SYN-ACK from the responder",
        "RSTRH", "Responder sent a SYN ACK followed by a RST, we never saw a SYN from the (purported) originator",
        "SH", "Originator sent a SYN followed by a FIN, we never saw a SYN ACK from the responder (hence the connection was 'half' open)",
        "SHR", "Responder sent a SYN ACK followed by a FIN, we never saw a SYN from the originator",
        "OTH", "No SYN seen, just midstream traffic (a 'partial connection' that was not later closed)" 
      ]
    }
  }
}
# Resolve @source_host to FQDN if possible if missing for some types of logging using source_host_ip from above
filter {
  if [id.orig_h] {
    if ![id.orig_h-resolved] {
      mutate {
        add_field => [ "id.orig_h-resolved", "%{id.orig_h}" ]
      }
      dns {
        reverse => [ "id.orig_h-resolved" ]
        action => "replace"
      }
    }
  }
}
filter {
  if [id.resp_h] {
    if ![id.resp_h-resolved] {
      mutate {
        add_field => [ "id.resp_h-resolved", "%{id.resp_h}" ]
      }
      dns {
        reverse => [ "id.resp_h-resolved" ]
        action => "replace"
      }
    }
  }
}
# Change to the output that fits your needs below

#output {
#  elasticsearch {
#    embedded => true
#  }
#}
#output {
#  elasticsearch {
#    cluster => "logstash-cluster"
#    flush_size => "5000"
#    manage_template => true
#    template_overwrite => true
#    template => "/opt/logstash/lib/logstash/outputs/elasticsearch/elasticsearch-template.json"
#    workers => "3"
#  }
#}
output {
  elasticsearch {
    cluster => "logstash-cluster"
    flush_size => 1
#    manage_template => true
#    template => "/opt/logstash/lib/logstash/outputs/elasticsearch/elasticsearch-template.json"
  }
}
#output {
#  redis {
#    host => logstash
#    data_type => "list"
#    key => "logstash"
#  }
#}

Hi there,
Thank you for your sharing this configuration.
And I just want to ask how can I filter some fields from all *.log file then save it to a table?