spam.conf

input {
	file {
		path => "/media/johnny/C66E87A36E878B3F/Spam data/*"
		start_position => beginning
		sincedb_path => "/home/johnny/Documents/spam-analysis/record.db"

		codec => multiline { #用「Delivered-To:」做每筆資料的切割依據
			pattern => "^Delivered-To:"
			negate => true
			what => "previous"
		}
	}
} 

filter {

	#Delivered-To
	grok {
		match => { "message" => "^Delivered-To:[ \t]*%{DATA}(?<Delivered-to>[-0-9a-zA-Z.+_]+@[-0-9a-zA-Z.+_]+\.[a-zA-Z]{2,4})"}
	}

	#fetch ip
	if "X-Originating-IP:" in [message] {
		grok {
			match => { "message" => "^X-Originating-IP:[ \t]*%{IP:X-Originating-IP}"}
		}
	} else {
		grok { #抓取public IP(0.0.0.0~9.255.255.255, 11.0.0.0~172.15.255.255, 172.32.0.0~192.167.255.255, 192.169.0.0~223.255.255.255)
			match => [ "message", "(?<X-Originating-IP>(?<![0-9])(?:(?:1[0-2][0-6]|[1-9][1-9]|[0-9])[.](?:25[0-5]|2[0-4][0-9]|[0-1]?[0-9]{1,2})[.](?:25[0-5]|2[0-4][0-9]|[0-1]?[0-9]{1,2})[.](?:25[0-5]|2[0-4][0-9]|[0-1]?[0-9]{1,2})|(?:12[8-9]|1[3-6][0-9]|17[13-9]|18[0-9]|19[01])[.](?:25[0-5]|2[0-4][0-9]|[0-1]?[0-9]{1,2})[.](?:25[0-5]|2[0-4][0-9]|[0-1]?[0-9]{1,2})[.](?:25[0-5]|2[0-4][0-9]|[0-1]?[0-9]{1,2})|172[.](?:[0-9]|1[0-5]|3[2-9]|[4-9][0-9]|25[0-5]|2[0-4][0-9]|1[0-9][0-9])[.](?:25[0-5]|2[0-4][0-9]|[0-1]?[0-9]{1,2})[.](?:25[0-5]|2[0-4][0-9]|[0-1]?[0-9]{1,2})|(?:19[3-9]|2[0-2][0-3])[.](?:25[0-5]|2[0-4][0-9]|[0-1]?[0-9]{1,2})[.](?:25[0-5]|2[0-4][0-9]|[0-1]?[0-9]{1,2})[.](?:25[0-5]|2[0-4][0-9]|[0-1]?[0-9]{1,2})|192[.](?:25[0-5]|2[0-4][0-9]|1[0-57-9][0-9]|16[0-79]|[1-9][0-9]|[0-9])[.](?:25[0-5]|2[0-4][0-9]|[0-1]?[0-9]{1,2})[.](?:25[0-5]|2[0-4][0-9]|[0-1]?[0-9]{1,2}))(?![0-9]))"]
		}
	}

	#Date
	grok { #時間格式：d MMM yyyy hh:mm:ss Z
		match => [ "message", "(?<Date>%{MONTHDAY}[ \t]*%{MONTH}[ \t]*%{YEAR}[ \t]*%{TIME}[ \t]*[+-]\d{4})"]
	}

	#Subject(Maybe multiline)
	if "Subject:" in [message] {
		grok {
			match => [ "message", "^Subject:[ \t]*(?<Subject>((\S+[\S ]*)(\n[\t ]+(\S+[\S ]*))*))"]
			add_tag => [ "has subject" ]
		}
	}
	
	if "has subject" in [tags]{
		#Do nothing
	}else{
		grok {
			match => [ "message", "%{DATA}"]
			add_field => { "Subject" => "No Subject" }
		}
	}

	#Link count
	mutate {
			add_field => {"link_count" => "%{message}"}
	}

	mutate {
			split => ["link_count", "http"]
			add_tag => "split_http"
	}

	if "split_http" in [tags] {
		mutate {
			gsub => ["link_count", "(.|\n)+", "*"] 
		}
	}


	#From
	grok {
		match => [ "message", "^From:[ \t]*%{DATA}(?<From>[-0-9a-zA-Z.+_]+@[-0-9a-zA-Z.+_]+\.[a-zA-Z]{2,4})"]
	}

	#To
	grok {
		match => [ "message", "^To:[ \t]*(?<To>((\S+[\S ]*)(\n[\t ]+(\S+[\S ]*))*))"]
		add_tag => [ "valid" ]
	}
	
	
	#Reply-To
	grok {
		match => [ "message", "^Reply-To:[ \t]*(?<Reply-To>((\S+[\S ]*)(\n[\t ]+(\S+[\S ]*))*))"]
	}
	if "valid" in [tags] {
		#Do nothing
	}else{
		#如果To欄位不存在，則將Reply-To的內容也定義為To欄位的內容
		grok {
			match => [ "message", "^Reply-To:[ \t]*(?<To>((\S+[\S ]*)(\n[\t ]+(\S+[\S ]*))*))"]
			add_tag => [ "valid" ]
		}
	}
	
	mutate {
		split => ["Reply-To", ","]
		gsub => ["Reply-To", "(([\w ]*<)|\s|>)", ""]
	}
	
	#Cc
	grok {
		match => [ "message", "^Cc:[ \t]*(?<CC>((\S+[\S ]*)(\n[\t ]+(\S+[\S ]*))*))"]
	}
	if "valid" in [tags] {
		#Do nothing
	}else{
		#如果To欄位不存在，則將Cc的內容也定義為To欄位的內容
		grok {
			match => [ "message", "^Cc:[ \t]*(?<To>((\S+[\S ]*)(\n[\t ]+(\S+[\S ]*))*))"]
			add_tag => [ "valid" ]
		}
	}
	mutate {
		split => ["CC", ","]
		gsub => ["CC", "(([\w ]*<)|\s|>)", ""]
	}
	
	#Bcc
	grok {
		match => [ "message", "^Bcc:[ \t]*(?<BCC>((\S+[\S ]*)(\n[\t ]+(\S+[\S ]*))*))"]
	}
	if "valid" in [tags] {
		#Do nothing
	}else{
		#如果To欄位不存在，則將Bcc的內容也定義為To欄位的內容
		grok {
			match => [ "message", "^Bcc:[ \t]*(?<To>((\S+[\S ]*)(\n[\t ]+(\S+[\S ]*))*))"]
			add_tag => [ "valid" ]
		}
	}
	mutate { #把密件副本切割成陣列
		split => ["BCC", ","]
		gsub => ["BCC", "(([\w ]*<)|\s|>)", ""]
	}
	
	mutate { #把收件者切割成陣列
		split => ["To", ","]
		gsub => ["To", "(([\w ]*<)|\s|>)", ""]
	}

	#time format
	date { #將match的時間欄位format成ISO 8601格式，並存入@sendTime欄位中
		match => ["Date", "d MMM yyyy HH:mm:ss Z"]
		locale => "en"
		target => "@sendTime"
	}

	#convert ip to location info
	geoip {
		source=> "X-Originating-IP"
		database => "/home/johnny/Documents/logstash-1.4.2/vendor/geoip/GeoLiteCity.dat"
		#remove_field => ["[geoip][location]"]
	}
	
	#我們將Header中有收件者訊息的Email認定為合法的Email Header，否則將其drop掉
	if "valid" in [tags] {
		#remove useless fields
		mutate {
			remove_field => [ "message", "tags", "host", "path", "@version", "_version" ]
		}
	} else {
		drop {}
	}
}

output {
	#stdout {
	#	codec => rubydebug
	#}
	
	elasticsearch { 
		host => localhost
		codec => json
	}
}