I have spent the past several days working with Bro-IDS and Logstash
parsing and wanted to share this with anyone else who may be doing the
same and needs some decent parsing. Some pieces of this I have taken
from other sources online but have been modifying to suit my needs as
well as adding additional functionality. This will be updated as time
goes on and will be hosted on GitHub.
# 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"
# }
#} 