Relating two fields - using translate or mutate

Elastic Stack Logstash
1

Hi All,

I have a index for netflow data and recently I was trying to use the combine threat intelligence to see if there is any communication from any black listed ip's.

I have two tags one is netflow and other one is threat but same index.

Is it possible to relate the netflow data and entity threat field. Something like this in the logstash filter,

filter {
  if [netflow][ipv4_src_addr] == [entity_threat] {
    mutate {
      add_tag => "true"
    }
  } else {
  if [netflow][ipv4_dst_addr] == [entity_threat] {
    mutate {
      add_tag => "true"
    }
  }
}

Am not sure if am doing it correctly, please anyone help me in figuring it out.

Thanks,
Raj

2

I don't know how you are planning to use the translate filter here, but I can comment on your snippet above.

By tags above do you mean fields and by index above do you mean value?

Does it make sense to use a tag of "true" because you can't tell which branch caused that tag to be added? Surely the tags "source-address-is-entity-threat" and "destination-address-is-entity-threat" would be better.

Maybe add a field called [netflow][threat_detected] with a value of "source", "destination" or else "inconclusive" depending on the branch the conditional takes.

Also your else if nesting is not quite correct...

filter {
  if [netflow][ipv4_src_addr] == [entity_threat] {
    mutate {
      add_tag => "true"
    }
  } else if [netflow][ipv4_dst_addr] == [entity_threat] {
    mutate {
      add_tag => "true"
    }
  }
}
1 Like
3

Hi Guyboertje,

Thanks for the reply. Sorry for the confusion ,let me explain in another way , i have one index called logstash_netflow-* for netflow data and have another index called logstash_threat-* for threat intelligence feed and it has list of black listed ip's.

So am not sure whether we can relate the field (src_addr or dst_addr) from logstash_netflow-* to another field (entity_threat) from another index ie logstash_threat-*

So what I did is I used the threat intelligence feed to the logstash_netflow index. So now I have only one index which is logstash_netflow-*

Now, I want to match the field src_addr and dst_addr to the another field which is entitiy_threat ,example like this .

if [netflow][ipv4_dst_addr] == [entity_threat] ----------> Add field like source threat true or something
if if [netflow][ipv4_dst_addr] == [entity_threat] -----------> Add field like destination threat true or something

Thanks,
Raj

4

Ahhh. I see.

You can keep the two indexes separate.

Use the elasticsearch filter in your netflow stream to do a query on the logstash_threat-* index using the value from [netflow][ipv4_src_addr] in the query to find a matching document and add_tag will be invoked if there is a match. This means that you must have two elasticsearch filters one for each field.
If you only care that a threat is found on either src or dst then you might be able to use one elasticsearch filter with some sort of OR in the query.
Links:

  1. Stackoverflow
  2. Docs for ES filter
  3. ES Query Docs

Using translate will need the same two filter solution but you will have to export the threat IP addresses to a CVS file.

1 Like
5

Hi Guy,

Thanks for the reply, is it possible for you to create a sample translate and elasticsearch filter, to execute this function

Thanks,
Raj

6

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