Logstash configuration for pfSense firewall logs (filterlog)

50-pfsense.conf

input {
	# output logs in pfsense to syslog
	tcp {
		port => 10514
		type => "pfsense"
		tags => ["unparsed"]
	}
	udp {
		port => 10514
		type => "pfsense"
		tags => ["unparsed"]
	}
}

filter {
	grok {
		match => {
			"message" => "<%{INT}>%{SYSLOGTIMESTAMP:timestamp} filterlog: %{GREEDYDATA:csv_source}"
		}
		tag_on_failure => ["grok_failure"]
	}
	date {
		locale => "en"
		match => ["timestamp", "MMM dd HH:mm:ss", "MMM  d HH:mm:ss"]
	}
}

filter {
	# only filterlog messages!
	if [csv_source] =~ /17/ and [csv_source] =~ /udp/ {
		csv {
			separator => ","
			source => "csv_source"
			columns => [
				"rule_number",
				"sub_rule_number",
				"anchor",
				"tracker",
				"if",
				"reason",
				"action",
				"direction",
				"ip_version",
				"tos",
				"ecn",
				"ttl",
				"id",
				"offset",
				"flags",
				"protocol_id",
				"protocol",
				"legth",
				"source_ip",
				"destination_ip",
				"source_port",
				"destination_port",
				"data_length"
			]
			remove_tag => ["unparsed"]
		}
	}

	if [csv_source] =~ /6/ and [csv_source] =~ /tcp/ {
		csv {
			separator => ","
			source => "csv_source"
			columns => [
				"rule_number",
				"sub_rule_number",
				"anchor",
				"tracker",
				"if",
				"reason",
				"action",
				"direction",
				"ip_version",
				"tos",
				"ecn",
				"ttl",
				"id",
				"offset",
				"flags",
				"protocol_id",
				"protocol",
				"legth",
				"source_ip",
				"destination_ip",
				"source_port",
				"destination_port",
				"data_length",
				"tcp_flags",
				"tcp_seq_num",
				"tcp_ack",
				"tcp_window",
				"tcp_urg",
				"tcp_opts"
			]
			remove_tag => ["unparsed"]
		}
	}

	if [csv_source] =~ /1/ and [csv_source] =~ /icmp/ and [csv_source] =~ /request/ {
		csv {
			separator => ","
			source => "csv_source"
			columns => [
				"rule_number",
				"sub_rule_number",
				"anchor",
				"tracker",
				"if",
				"reason",
				"action",
				"direction",
				"ip_version",
				"tos",
				"ecn",
				"ttl",
				"id",
				"offset",
				"flags",
				"protocol_id",
				"protocol",
				"legth",
				"source_ip",
				"destination_ip",
				"icmp_type",
				"icmp_request_id",
				"icmp_request_sequence"
			]
			remove_tag => ["unparsed"]
		}
	}

	# TODO: literal for 'protocol unreachable'
	#if [message] =~ /1/ and [message] =~ /icmp/ {
	#       csv {
	#
	#       }
	#}

	# TODO: literal for 'port unreachable'
	#if [message] =~ /1/ and [message] =~ /icmp/ {
	#       csv {
	#
	#       }
	#}
}

filter {
	if "unparsed" not in [tags] {
		# clean up
		mutate {
			remove_field => [
				#"rule_number",
				#"sub_rule_number",
				"anchor",
				"tracker",
				"protocol_id",
				"csv_source"
			]
		}
		# additional info
		geoip {
			tag_on_failure => ["geoip_source_error"]
			source => "source_ip"
			target => "source_geoip"
		}
		if [destination_ip] =~ /.+/ {
			geoip {
				tag_on_failure => ["geoip_destination_error"]
				source => "destination_ip"
				target => "destination_geoip"
			}
		}
	} else {
	}
}

output {
	elasticsearch {
		hosts => ["theelastichost:9200"]
		index => "logstash-%{+YYYY.MM.dd}"
	}
	stdout { codec => rubydebug }
}