# Correlation of IPs within ELK

**URL:** <https://discuss.elastic.co/t/correlation-of-ips-within-elk/151189>\
**Category:** Logstash\
**Created:** [October 5, 2018, 10:07am UTC](https://discuss.elastic.co/t/correlation-of-ips-within-elk/151189 "2018-10-05T10:07:39Z")\
**Posts on this page:** 8\
**Page:** 1

<div class="post-metadata">

**Author:** ![Alessandro\_89](https://avatars.discourse-cdn.com/v4/letter/a/3ec8ea/32.png) [@Alessandro\_89](https://discuss.elastic.co/u/Alessandro_89)\
**Post date:** [October 5, 2018, 10:07am UTC](https://discuss.elastic.co/t/correlation-of-ips-within-elk/151189/1 "2018-10-05T10:07:39Z")

</div>

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.

---

<div class="post-metadata">

**Author:** ![warkolm](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/warkolm/32/39224_2.png) [@warkolm](https://discuss.elastic.co/u/warkolm)\
**Post date:** [October 6, 2018, 7:49am UTC](https://discuss.elastic.co/t/correlation-of-ips-within-elk/151189/2 "2018-10-06T07:49:29Z")

</div>

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.

---

<div class="post-metadata">

**Author:** ![Alessandro\_89](https://avatars.discourse-cdn.com/v4/letter/a/3ec8ea/32.png) [@Alessandro\_89](https://discuss.elastic.co/u/Alessandro_89)\
**Post date:** [October 6, 2018, 12:54pm UTC](https://discuss.elastic.co/t/correlation-of-ips-within-elk/151189/3 "2018-10-06T12:54:28Z")

</div>

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.

---

<div class="post-metadata">

**Author:** ![warkolm](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/warkolm/32/39224_2.png) [@warkolm](https://discuss.elastic.co/u/warkolm)\
**Post date:** [October 8, 2018, 2:20am UTC](https://discuss.elastic.co/t/correlation-of-ips-within-elk/151189/4 "2018-10-08T02:20:04Z")

</div>

Showing your config would be super handy 🙂

---

<div class="post-metadata">

**Author:** ![Alessandro\_89](https://avatars.discourse-cdn.com/v4/letter/a/3ec8ea/32.png) [@Alessandro\_89](https://discuss.elastic.co/u/Alessandro_89)\
**Post date:** [October 8, 2018, 6:53am UTC](https://discuss.elastic.co/t/correlation-of-ips-within-elk/151189/5 "2018-10-08T06:53:06Z")

</div>

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.

---

<div class="post-metadata">

**Author:** ![Alessandro\_89](https://avatars.discourse-cdn.com/v4/letter/a/3ec8ea/32.png) [@Alessandro\_89](https://discuss.elastic.co/u/Alessandro_89)\
**Post date:** [October 8, 2018, 6:59am UTC](https://discuss.elastic.co/t/correlation-of-ips-within-elk/151189/6 "2018-10-08T06:59:19Z")

</div>

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.

---

<div class="post-metadata">

**Author:** ![warkolm](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/warkolm/32/39224_2.png) [@warkolm](https://discuss.elastic.co/u/warkolm)\
**Post date:** [October 8, 2018, 8:38am UTC](https://discuss.elastic.co/t/correlation-of-ips-within-elk/151189/7 "2018-10-08T08:38:35Z")

</div>

If you map the IP field [correctly](https://www.elastic.co/guide/en/elasticsearch/reference/6.4/ip.html), then you can run subnet queries in Kibana 🙂

---

<div class="post-metadata">

**Author:** ![system](https://us1.discourse-cdn.com/elastic/original/3X/1/a/1ac57faf039f6b580b3f104ef42a2a89e41014de.png) [@system](https://discuss.elastic.co/u/system)\
**Post date:** [November 5, 2018, 8:38am UTC](https://discuss.elastic.co/t/correlation-of-ips-within-elk/151189/8 "2018-11-05T08:38:40Z")

</div>

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