#/*
#Copyright 2020-2021 IBM Inc. - All Rights Reserved.
#SPDX-License-Identifier: Apache-2.0
#*/

input{
	jdbc {
        jdbc_driver_class => "com.microsoft.sqlserver.jdbc.SQLServerDriver"
        jdbc_connection_string => "jdbc:sqlserver://<SERVER_NAME>:<PORT>;databaseName=msdb;"       #databaseName = msdb
        jdbc_user => "<login_name>"
        jdbc_password => "<password>"
        statement => "SELECT event_time, session_id, database_name, client_ip, server_principal_name, application_name, statement, succeeded, DATEDIFF_BIG(ns, '1970-01-01 00:00:00.00000', event_time) AS updatedeventtime FROM msdb.dbo.rds_fn_get_audit_file('D:\rdsdbdata\SQLAudit\*.sqlaudit', default, default ) Where schema_name not in ('sys') and object_name NOT IN ('dbo','syssubsystems','fn_sysdac_is_currentuser_sa','backupmediafamily','backupset','syspolicy_configuration','syspolicy_configuration_internal','syspolicy_system_health_state','syspolicy_system_health_state_internal','fn_syspolicy_is_automation_enabled','spt_values','sysdac_instances_internal','sysdac_instances') and database_principal_name not in('public') and ((succeeded =1) or (succeeded =0 and statement like '%Login failed%')) and statement != '' and DATEDIFF_BIG(ns, '1970-01-01 00:00:00.00000', event_time) > :sql_last_value order by event_time"
		    use_column_value => true
        tracking_column_type => "numeric"
        tracking_column => "updatedeventtime"
		    #This location stores the offset till which records are processed
		    last_run_metadata_path => "./.awsmssql_logstash_jdbc_last_run"
        type => "MSSQL-AWS"
	    schedule => "*/2 * * * *"
		#Insert the account id of the AWS account
		add_field => {"account_id" => "<ACCOUNT_ID>"}
		#Insert the server_name 
		add_field => {"server_name" => "<SERVER_NAME>"}
    }
}

filter{
	if [type]=="MSSQL-AWS" {
		grok { match => { "event_time" => "%{YEAR:year}-%{MONTHNUM:month}-%{MONTHDAY:day}T%{TIME:time}" } }
		mutate { add_field => { "mssqlTimeStamp" => "%{year}-%{month}-%{day}T%{time}Z" } }
		date{
			match => ["mssqlTimeStamp", "ISO8601"]
			target => "mssqlEventTime"
		}

		if [session_id]{
			mutate { add_field => { "[GuardRecord][sessionId]" => "%{session_id}" }}
		}
		else{
			mutate { add_field => { "[GuardRecord][sessionId]" => "" }}
		}

		if [database_name]{
			mutate { add_field => { "[GuardRecord][dbName]" => "%{account_id}:%{database_name}" }}

		}
		else{
			mutate { add_field => { "[GuardRecord][dbName]" => "NA" }}
		}

		mutate { add_field => { "[GuardRecord][appUserName]" =>  "AWSService" }}

		ruby { code => "event.set('[GuardRecord][time][timstamp]', event.get('mssqlEventTime').to_i * 1000)" }
		mutate { add_field => { "[GuardRecord][time][minOffsetFromGMT]" => "0" }}
		mutate { add_field => { "[GuardRecord][time][minDst]" => "0" }}

		if [client_ip] == "local machine" or [client_ip] == "Unknown"{
			mutate { add_field => { "[GuardRecord][sessionLocator][clientIp]" => "0.0.0.0"}}
		}else{
			mutate { add_field => { "[GuardRecord][sessionLocator][clientIp]" => "%{client_ip}" }}
		}
		mutate { add_field => { "[GuardRecord][sessionLocator][clientPort]" => "-1" }}
		mutate { add_field => { "[GuardRecord][sessionLocator][serverIp]" => "0.0.0.0" }}
		mutate { add_field => { "[GuardRecord][sessionLocator][serverPort]" => "-1" }}
		mutate { add_field => { "[GuardRecord][sessionLocator][isIpv6]" => "false" }}
		ruby { code => 'event.set("[GuardRecord][sessionLocator][clientIpv6]", nil)' }
		ruby { code => 'event.set("[GuardRecord][sessionLocator][serverIpv6]", nil)' }

		if [server_principal_name]{
			mutate { add_field => { "[GuardRecord][accessor][dbUser]" => "%{server_principal_name}" }}
		}
		else{
			mutate { add_field => { "[GuardRecord][accessor][dbUser]" => "NA" }}
		}
		mutate { add_field => { "[GuardRecord][accessor][dataType]" => "TEXT" }}
		mutate { add_field => { "[GuardRecord][accessor][osUser]" => "" }}
		mutate { add_field => { "[GuardRecord][accessor][serverType]" => "MSSQL" }}
		mutate { add_field => { "[GuardRecord][accessor][commProtocol]" => "AwsApiCall" }}
		mutate { add_field => { "[GuardRecord][accessor][dbProtocol]" => "MS SQL SERVER" }}
		mutate { add_field => { "[GuardRecord][accessor][language]" => "MSSQL" }} # MSSQL is a Guardium internal code for MSSQL
		mutate { add_field => { "[GuardRecord][accessor][serverOs]" => "" }}
		mutate { add_field => { "[GuardRecord][accessor][clientOs]" => "" }}

		if [server_name]
		{
			mutate { add_field => { "[GuardRecord][accessor][serverHostName]" => "%{account_id}_%{server_name}" }}
		}
		else
		{
			mutate { add_field => { "[GuardRecord][accessor][serverHostName]" => "%{account_id}_mssql" }}
		}

		mutate { add_field => { "[GuardRecord][accessor][clientHostName]" => "" }}
		mutate { add_field => { "[GuardRecord][accessor][dbProtocolVersion]" => "" }}
		mutate { add_field => { "[GuardRecord][accessor][clientMac]" => ""} }
		mutate { add_field => { "[GuardRecord][accessor][serverDescription]" => "" } }

		if [database_name]{
            mutate { add_field => { "[GuardRecord][accessor][serviceName]" => "%{account_id}:%{database_name}" }}
        }
        else{
            mutate { add_field => { "[GuardRecord][accessor][serviceName]" => "NA" }}
        }

		if [application_name]{
			mutate { add_field => { "[GuardRecord][accessor][sourceProgram]" => "%{application_name}"} }
		}
		else{
			mutate { add_field => { "[GuardRecord][accessor][sourceProgram]" => "NA"} }
		}

		mutate {add_field => { "isSuccess" => "%{succeeded}" }}

		if [isSuccess] == "true" {
			ruby { code => 'event.set("[GuardRecord][data][construct]", nil)' }
			mutate { add_field => { "[GuardRecord][data][originalSqlCommand]" => "%{statement}" }}

			ruby { code => 'event.set("[GuardRecord][exception]", nil)' }

		}
		else {
			mutate { add_field => { "[GuardRecord][exception][exceptionTypeId]" => "LOGIN_FAILED" }}
			mutate { add_field => { "[GuardRecord][exception][description]" => "%{statement}" }}
			mutate { add_field => { "[GuardRecord][exception][sqlString]" => "NA" }}

			ruby { code => 'event.set("[GuardRecord][data]", nil)' }

		}

		json_encode {
            source => "[GuardRecord]"
            target => "[GuardRecord]"
        }

		mutate {
			convert => {"[GuardRecord][sessionLocator][clientPort]" => "integer"}
			convert => {"[GuardRecord][sessionLocator][serverPort]" => "integer"}
			convert => {"[GuardRecord][sessionLocator][isIpv6]" => "boolean"}
			convert => {"[GuardRecord][time][minOffsetFromGMT]" => "integer"}
			convert => {"[GuardRecord][time][minDst]" => "integer"}
		}

		prune {
		    whitelist_names => [ "GuardRecord" ]
        }

	}
}