securityonion/salt/logstash/pipelines/config/so/6000_bro.conf

# Original Author: Justin Henderson
#         SANS Instructor and author of SANS SEC555: SIEM and Tactical Analytics
# Updated by: Doug Burks
# Last Update: 2/10/2018
#
filter {
  if "bro" in [tags] {

	# Bro logs have a high quality timestamp, so let's copy that to @timestamp.
	# Before we do, let's copy the existing logstash @timestamp to timestamp.
	mutate {
		add_field => { "logstash_timestamp" => "%{@timestamp}" }
	}
	mutate {
		convert => { "logstash_timestamp" => "string" }
	}
	mutate {
		convert => { "timestamp" => "string" }
	}
	# New Bro JSON logs use ISO8601 timestamps.
	# Old Bro TSV logs use UNIX timestamps.
	date {
		match => [ "timestamp", "ISO8601", "UNIX" ]
	}
	mutate {
		rename => { "logstash_timestamp" => "timestamp" }
	}

    if [duration] == "-" {
      mutate {
        replace => [ "duration", "0" ]
      }
    }
    if [original_bytes] == "-" {
      mutate {
        replace => [ "original_bytes", "0" ]
      }
    }
    # If MissedBytes is unspecified set it to zero so it is an integer
    if [missed_bytes] == "-" {
      mutate {
        replace => [ "missed_bytes", "0" ]
      }
    }
    # If OriginalIPBytes is unspecified set it to zero so it is an integer
    if [original_ip_bytes] == "-" {
      mutate {
        replace => [ "original_ip_bytes", "0" ]
      }
    }
    # If RespondBytes is unspecified set it to zero so it is an integer
    if [respond_bytes] == "-" {
      mutate {
        replace => [ "respond_bytes", "0" ]
      }
    }
    # If RespondIPBytes is unspecified set it to zero so it is an integer
    if [respond_ip_bytes] == "-" {
      mutate {
        replace => [ "respond_ip_bytes", "0" ]
      }
    }
    if [request_body_length] == "-" {
      mutate {
        replace => [ "request_body_length", "0" ]
      }
    }
    if [response_body_length] == "-" {
      mutate {
        replace => [ "response_body_length", "0" ]
      }
    }
    if [source_port] == "-" {
      mutate {
        remove_field => ["source_port"]
      }
    }
    if [destination_port] == "-" {
      mutate {
        remove_field => ["destination_port"]
      }
    }
    if [virtual_host] == "-" {
      mutate {
        remove_field => ["virtual_host"]
      }
    }
    if [x_originating_ip] == "-" {
      mutate {
        remove_field => ["x_originating_ip"]
      }
    }
    if [basic_constraints_path_length] == "-" {
      mutate {
        remove_field => ["basic_constraints_path_length"]
      }
    }
    if [data_channel_source_ip] == "-" {
      mutate {
        remove_field => ["data_channel_source_ip"]
      }
    }
    if [data_channel_destination_ip] == "-" {
      mutate {
        remove_field => ["data_channel_destination_ip"]
      }
    }
    if [desktop_width] == "-" {
      mutate {
        remove_field => ["desktop_width"]
      }
    }
    if [desktop_height] == "-" {
      mutate {
        remove_field => ["desktop_height"]
      }
    }
    if [height] == "-" {
      mutate {
        remove_field => ["height"]
      }
    }
 

    # I renamed conn_uids to uid so that it is easy to pivot to all things tied to a connection
    mutate {
       rename => [ "connection_uids", "uid" ]
    }
    # If total_bytes is set to "-" change it to 0 so it is an integer
    if [total_bytes] == "-" {
      mutate {
        replace => [ "total_bytes", "0" ]
      }
    }
    # If seen_bytes is set to "-" change it to 0 so it is an integer
    if [seen_bytes] == "-" {
      mutate {
        replace => [ "seen_bytes", "0" ]
      }
    }
    # If missing_bytes is set to "-" change it to 0 so it is an integer
    if [missing_bytes] == "-" {
      mutate {
        replace => [ "missing_bytes", "0" ]
      }
    }
    # If overflow_bytes is set to "-" change it to 0 so it is an integer
    if [overflow_bytes] == "-" {
      mutate {
        replace => [ "overflow_bytes", "0" ]
      }
    }
    if [dcc_file_size] == "-" {
      mutate {
        replace => [ "dcc_file_size", "0" ]
      }
    }
    if [authentication_attempts] == "-" {
      mutate {
        replace => [ "authentication_attempts", "0" ]
      }
    }
    if [file_size] == "-" {
      mutate {
        replace => [ "file_size", "0" ]
      }
    }
    if [original_ip_bytes] == "-" {
      mutate {
        replace => [ "original_ip_bytes", "0" ]
      }
    }
    
    # I recommend changing the field types below to integer or floats so searches can do greater than or less than
    # and also so math functions can be ran against them
    mutate {
      convert => [ "bound_port", "integer" ]
      convert => [ "data_channel_destination_port", "integer" ]
      convert => [ "destination_port", "integer" ]
      convert => [ "depth", "integer" ]
      #convert => [ "duration", "float" ]
      convert => [ "info_code", "integer" ]
      convert => [ "missed_bytes", "integer" ]
      convert => [ "missing_bytes", "integer" ]
      convert => [ "n", "integer" ]
      convert => [ "original_bytes", "integer" ]
      convert => [ "original_packets", "integer" ]
      convert => [ "original_ip_bytes", "integer" ]
      convert => [ "overflow_bytes", "integer" ]
      convert => [ "p", "integer" ]
      convert => [ "query_class", "integer" ]
      convert => [ "query_type", "integer" ]
      convert => [ "rcode", "integer" ]
      convert => [ "request_body_length", "integer" ]
      convert => [ "request_port", "integer" ]
      convert => [ "respond_bytes", "integer" ]
      convert => [ "respond_packets", "integer" ]
      convert => [ "respond_ip_bytes", "integer" ]
      convert => [ "response_body_length", "integer" ]
      convert => [ "seen_bytes", "integer" ]
      convert => [ "source_port", "integer" ]
      convert => [ "status_code", "integer" ]
      #convert => [ "suppress_for", "float" ]
      convert => [ "total_bytes", "integer" ]
      convert => [ "trans_depth", "integer" ]
      convert => [ "transaction_id", "integer" ]
      # convert the following boolean to text for now
      convert => [ "local_respond", "string" ]
      convert => [ "tc", "string" ]
      convert => [ "is_orig", "string" ]
      convert => [ "local_orig", "string" ]
      lowercase => [ "query" ]
      #remove_field => [ "timestamp" ]
    }

    # Combine OriginalBytes and RespondBytes and save the value to total_bytes
    if [original_bytes] {
      if [respond_bytes] {
        ruby {
          code => "event.set('total_bytes', event.get('original_bytes') + event.get('respond_bytes'))"
        }
      }
    }
	mutate {
		#add_tag => [ "conf_file_6000"]
	}
  }
}