Correlation of IPs within ELK

Elastic Stack Logstash
1

Greetings,

I am trying to automate/ease a procedure to review firewall rules within ELK (ElasticSearch, Logstash, Kibana). I have some data obtained from a CSV, which is structured like this:

Source;Destination;Service;Action;Comment
10.0.0.0/8 172.16.0.0/16 192.168.0.0/24 23.2.20.6;10.0.0.1 10.0.0.2 10.0.0.3;udp:53 
tcp:53;accept;No.10: ID: INC0000000001

My objective is to import this data within ELK by parsing each field (for subnet and/or IP address) and, if possible, add a sequential field (IP_Source1,IP_Destination2,etc) containing each one. This, in order to correlate each IP and (for example) see if it's contained within one of the subnets (aggregation of rules), and enrich the data with other inputs.

Is this possible, to your knowledge? How?

Thanks a lot for any input you might have.

2

Yes this can be done with a mix of the various filters Logstash has.

I'd start with the CSV filter to parse the events, and go from there.

3

Hello warkolm, thanks a lot for your reply. I was looking for a bit more detail, if possible.
I beautified the CSV and created the relative filter within logstash, in order to separate the source, destination, service, action and comment, but I have no idea on how to progress from there in order to parse the IPs within the source, destination fields and see how to correlate them.

4

Showing your config would be super handy :slight_smile:

5

Hello warkolm, below is my actual configuration for logstash:

input {
	file {
		path => "/home/alessandro/Desktop/Exports/Test/test.csv"
		start_position => "beginning"
		type => "test"
	}
}

filter {
csv {
	separator => ";"
	columns => ["Source","Destination","Service","Action","Comment"]
	}
}
	
output {
	elasticsearch { hosts => ["localhost:9200"] }
	stdout { codec => rubydebug }
}

The input has already been beautified wherever possible, and I also mocked up some ruby code (I was helped also in this because I'm not brilliant with such programming language). You may find the snippet below, but I don't know how to use it:

filter { 
    ruby { 
       code => "
	  import ipaddr
          arr = event.get('source').to_s.split(' ')
          arr.each.with_index(1) do |a, index|
          puts event.set(ip_source+index, a)
          end
       "
    }
}

The script should scan the IP address contained within the source, and split it in different subsets of a list. then, it should be printed out.

Really appreciate your help.

6

Also, an idea is to use Kibana in order to accomplish my objective, but I'm not sure also how to use it. I'd appreciate any input also on this if you have it.

7

If you map the IP field correctly, then you can run subnet queries in Kibana :slight_smile:

8

This topic was automatically closed 28 days after the last reply. New replies are no longer allowed.