//
// Copyright 2021 - IBM Inc. All rights reserved
// SPDX-License-Identifier: Apache2.0
//

input {
    beats {
        port => <guc_input_param_port>
		include_codec_tag => false
		type => "postgres"
	}
}

filter {
	if [type]=="postgres"  and  "guc_postgres_param"  in [tags] {
		if ([message] =~ /^AUDIT:.*/ ){
			csv {
			separator => ","
			columns => [
				"log_header","class","ts","client_ip","remote_port","source_program","db_user","db_name","virtual_id","statement_id","substatement_id","command_tag","sql_state","object_type","oject_name","error_message","originalSqlCommand","data17"
			]
		
			autogenerate_column_names => false
			}
		}
		else {
			csv {
				separator => ","
				columns => [
					"ts","db_user","db_name","pid","client_ip","session_id","data1","data2","session_time","virtual_pid","data3","error_severity","sql_state","text","data4","data5","data6","data7","data8","command","data10","data11","source_program","data12","data13","data14","data15","data16","data17"
				]
		
				autogenerate_column_names => false
			}
		}
		
	mutate { add_field => { "TZ" => "%{[event][timezone]}" } }		
			
	grok { match => { "TZ" => "(?<minutes>[^:]*):(?<seconds>[^,]*)" } }
	grok { match => { "minutes" => "(?<offset_diff>[^[0-9]]*)%{GREEDYDATA:actual_minutes}" } }
	
	ruby { code => "event.set('minutes1', event.get('actual_minutes').to_i * 60)" }
	ruby { code => "event.set('offset1', event.get('minutes1') +  event.get('seconds').to_i)" }
	
	mutate { add_field => { "totalOffset" => "%{offset_diff}%{offset1}" } }
	mutate { add_field => { "minOffset" => "%{offset_diff}%{minutes1}" } }

	
	grok { match => { "ts" => "%{YEAR:year}-%{MONTHNUM:month}-%{MONTHDAY:day} (?<hourTs>[^:]*):(?<minutesTs>[^:]*):(?<secondsTs>[^.]*).(?<microseconds>[^.]*) (?<zone>[^:]*)" } }
	
	mutate { add_field => { "timestamp" => "%{year}-%{month}-%{day}T%{hourTs}:%{minutesTs}:%{secondsTs}Z" } } 
	
	
	date{
		match => ["timestamp", "ISO8601"]
		target => "finalTime"
	}
	
	ruby { code => "event.set('totalOffset1', event.get('totalOffset').to_i * 60000)" }
	ruby { code => "event.set('logTime', event.get('finalTime').to_i * 1000)" }
	ruby { code => "event.set('finalTimestamp', (event.get('logTime').to_i) - (event.get('totalOffset1').to_i))" }
	
	mutate { add_field => { "[GuardRecord][time][timstamp]" => "%{finalTimestamp}" } }
	mutate { add_field => { "[GuardRecord][time][minOffsetFromGMT]" => "%{minOffset}" } }
	mutate { add_field => { "[GuardRecord][time][minDst]" => "0" } }
		
		if [error_severity] =~ "AUDIT" {
		
			if [text] {
				mutate { gsub => ["text", "\"", ""] }
				ruby { code => "event.set('query', event.get('text').split(':', 2))" }
				mutate { gsub => ["[query][1]", "\"", ""] }
				if [query][1] =~ "pg_catalog" {
					drop {}
				}
				else {
					mutate { add_field => { "[GuardRecord][data][originalSqlCommand]" => "%{[query][1]}" } }
				}
			}
			else {
				drop { }
			}
			
			ruby { code => 'event.set("[GuardRecord][data][construct]", nil)' }
			ruby { code => 'event.set("[GuardRecord][exception]", nil)' }
		}
		else if [class] {
        	if [class] != "ERROR" {
			if [originalSqlCommand] =~ "pg_catalog" {
				drop {}
			}
			else {
				mutate { add_field => { "[GuardRecord][data][originalSqlCommand]" => "%{originalSqlCommand}" } }
			}	
			ruby { code => 'event.set("[GuardRecord][data][construct]", nil)' }
			ruby { code => 'event.set("[GuardRecord][exception]", nil)' }
			}
		}
		else {
			if [sql_state] =~ "28P01" {
				mutate { add_field => {"[GuardRecord][exception][exceptionTypeId]" => "LOGIN_FAILED"} }
			}
			else {
				mutate { add_field => {"[GuardRecord][exception][exceptionTypeId]" => "SQL_ERROR"} }
			}
			if [text] {
				mutate { 
					gsub => ["text", "\"", ""]
					add_field => {"[GuardRecord][exception][description]" => "%{text}" } 
				}
			}
			else if [error_message]{
				mutate { add_field => {"[GuardRecord][exception][description]" => "%{error_message}" } }
			} 
			else {
				mutate { add_field => {"[GuardRecord][exception][description]" => "" } }
			}
			
			if [command]{
				mutate { 
					add_field => {"[GuardRecord][exception][sqlString]" => "%{command}" } 
				}
			}
			else {
				mutate { add_field => {"[GuardRecord][exception][sqlString]" => "NA" } }
			}
		}
		if [client_ip] {
		
			mutate {add_field => { "[GuardRecord][sessionLocator][isIpv6]" => "false" } }
			ruby { code => 'event.set("[GuardRecord][sessionLocator][clientIpv6]", nil)' }
			ruby { code => 'event.set("[GuardRecord][sessionLocator][serverIpv6]", nil)' }
		
			if [client_ip] =~ "\[local\]"{
				mutate {add_field => {"[GuardRecord][sessionLocator][clientIp]" => "127.0.0.1" } }
				mutate { add_field => {"[GuardRecord][sessionLocator][clientPort]" => "-1" } }
			}
			else {
				#if clientIp contains : it is IPv6
				mutate { split => {"client_ip" => ":"}}
				ruby { code => "event.set('number_of_elements', event.get('client_ip').length)" }
			
				if [number_of_elements] > 2 {
					mutate {add_field => { "[GuardRecord][sessionLocator][isIpv6]" => "true" } }
			
					if [number_of_elements] == 4 {
						mutate { add_field => { "clientIp" => "%{[client_ip][0]}:%{[client_ip][1]}:%{[client_ip][2]}" }}
						mutate { add_field => {"clientPort" => "%{[client_ip][3]}"} }
					}
					else {
						mutate { add_field => { "clientIp" => "%{[client_ip][0]}:%{[client_ip][1]}:%{[client_ip][2]}:%{[client_ip][3]}:%{[client_ip][4]}:%{[client_ip][5]}:%{[client_ip][6]}:%{[client_ip][7]}" }}
						mutate { add_field => {"clientPort" => "%{[client_ip][8]}"} }
					}
				
					mutate {add_field => {"[GuardRecord][sessionLocator][clientIpv6]" => "%{clientIp}" } }
					mutate { add_field => {"[GuardRecord][sessionLocator][clientPort]" => "%{clientPort}" } }
				}
				else {
					if[number_of_elements] == 2 {
						mutate { add_field => { "clientIp" => "%{[client_ip][0]}" }}
						mutate { add_field => {"clientPort" => "%{[client_ip][1]}"} }
					}
					else {
						mutate { add_field => { "clientIp" => "%{[client_ip]}" }}
						mutate { add_field => {"clientPort" => "%{[remote_port]}"} }
					}
					mutate {add_field => {"[GuardRecord][sessionLocator][clientIp]" => "%{clientIp}" } }
					mutate { add_field => {"[GuardRecord][sessionLocator][clientPort]" => "%{clientPort}" } }
				}
			}
		}
		else {
			mutate {add_field => {"[GuardRecord][sessionLocator][clientIp]" => "0.0.0.0" } }
			mutate { add_field => {"[GuardRecord][sessionLocator][clientPort]" => "-1" } }
		}
		
		if [db_name] {
			mutate {add_field => { "[GuardRecord][dbName]" => "%{db_name}" } }
		}
		else {
			mutate {add_field => { "[GuardRecord][dbName]" => "NA" } }
		}
		mutate {add_field => { "server_ip" => "%{[@metadata][ip_address]}" } }
		
		if [session_id] {
			mutate {add_field => { "[GuardRecord][sessionId]" => "%{session_id}" } }
		}
		else {
			if [virtual_id]{
				fingerprint {
					source => ["client_ip", "remote_port", "server_ip"]
					concatenate_sources => true
				}
				mutate {add_field => { "[GuardRecord][sessionId]" => "%{fingerprint}" } }
			}else{ 
				mutate {add_field => { "[GuardRecord][sessionId]" => "NA" } }
			}
		}
		
		mutate {add_field => { "[GuardRecord][sessionLocator][serverIp]" => "%{server_ip}" } }
		mutate {add_field => { "[GuardRecord][sessionLocator][serverPort]" => "-1" } }
				
		if [db_user] {
			mutate { add_field => { "[GuardRecord][accessor][dbUser]" => "%{db_user}" } }
		}
		else {
			mutate { add_field => { "[GuardRecord][accessor][dbUser]" => "NA" } }
		}
		
		mutate { add_field => { "[GuardRecord][accessor][serverType]" => "POSTGRE"} }
		mutate { add_field => { "[GuardRecord][accessor][serverOs]" => "%{[host][os][name]}"} }
		mutate { add_field => { "[GuardRecord][accessor][clientOs]" => ""} }
		mutate { add_field => { "[GuardRecord][accessor][clientHostName]" => "" } }
		mutate { add_field => { "[GuardRecord][accessor][serverHostName]" => "%{[host][name]}"} }
		mutate { add_field => { "[GuardRecord][accessor][commProtocol]" => ""} }
		mutate { add_field => { "[GuardRecord][accessor][dbProtocol]" => "POSTGRESQL" } }
		mutate { add_field => { "[GuardRecord][accessor][dbProtocolVersion]" => ""} }
		mutate { add_field => { "[GuardRecord][accessor][osUser]" => "" } }
		if [source_program] != ""{
			mutate { add_field => { "[GuardRecord][accessor][sourceProgram]" => "%{source_program}"} }
		}
		else { drop {} }
		mutate { add_field => { "[GuardRecord][accessor][clientMac]" => ""} }
		mutate { add_field => { "[GuardRecord][accessor][serverDescription]" => "" } }
		mutate { add_field => { "[GuardRecord][accessor][serviceName]" => ""} }
		mutate { add_field => { "[GuardRecord][accessor][language]" => "PGRS" } } #PGRS is a Guardium internal code forPostreSQL
		mutate { add_field => { "[GuardRecord][accessor][dataType]" => "TEXT" } }
		
		json_encode {
			source => "[GuardRecord]"
			target => "[GuardRecord]"
		}
	
		mutate {
			convert => { "[GuardRecord][time][minOffsetFromGMT]" => "integer"}
			convert => { "[GuardRecord][time][minDst]" => "integer"}
			convert => { "[GuardRecord][sessionLocator][clientPort]" => "integer"}
			convert => { "[GuardRecord][sessionLocator][serverPort]" => "integer"}
			convert => { "[GuardRecord][sessionLocator][isIpv6]" => "boolean"}
		}
		
	       mutate { remove_field => [ "minutes", "TZ", "secondsTs", "seconds", "finalTimestamp", "totalOffset1", "actual_minutes", "ecs", "event", "type", "totalOffset", "@version", "log", "file", "offset", "message", "@timestamp", "offset_diff", "zone", "timestamp", "@timestamp", "offset_diff", "month", "offset1", "finalTime", "input", "microseconds", "minutesTs", "year", "agent", "hourTs", "logTime", "minutes1", "tags", "day", "query", "data3", "data13", "text", "data7", "data4", "source_program", "data12", "db_name", "data2", "error_severity", "ts", "number_of_elements", "pid", "session_time", "data1", "data8", "clientIp", "clientPort", "client_ip", "session_id", "data10", "virtual_pid", "command", "data6", "data5", "sql_state", "db_user", "data11", "host", "remote_ip", "command_tag", "object_type", "oject_name", "originalSqlCommand", "substatement_id", "statement_id", "virtual_id", "error_message", "class", "data17", "log_header" ] }
	}

}