Watcher conditional input

[admlko]

I'm trying to set up a watch to join data from multiple indices.

Here's how I would like it to work: first, I look up if there are anti-virus alerts from the firewall log, if there are, then look up the IP address from DHCP log and finally, look up DNS records which were made from the IP address around the time of the alert.

This is what I have succeeded to do so far:

{
  "trigger": {
    "schedule": {
      "interval": "1m"
    }
  },
  "input": {
    "chain": {
      "inputs": [
        {
          "firewall": {
            "search": {
              "request": {
                "search_type": "query_then_fetch",
                "indices": [
                  "<company-firewall-{now/d}>"
                ],
                "types": [],
                "body": {
                  "query": {
                    "bool": {
                      "filter": [
                        {
                          "term": {
                            "event_id": 809
                          }
                        },
                        {
                          "range": {
                            "@timestamp": {
                              "from": "{{ctx.trigger.scheduled_time}}||-1m",
                              "to": "{{ctx.trigger.triggered_time}}"
                            }
                          }
                        }
                      ]
                    }
                  }
                }
              }
            }
          }
        },
        {
          "dhcp": {
            "search": {
              "request": {
                "search_type": "query_then_fetch",
                "indices": [
                  "<company-dhcp-{now/M{YYYY.MM}}>"
                ],
                "types": [],
                "body": {
                  "query": {
                    "bool": {
                      "must": [
                        {
                          "term": {
                            "ID": 11
                          }
                        },
                        {
                          "exists": {
                            "field": "{{ctx.payload.firewall.hits.hits.0._source}}"
                          }
                        }
                      ],
                      "should": [
                        {
                          "term": {
                            "IP": "{{ctx.payload.firewall.hits.hits.0._source.src_ip}}"
                          }
                        },
                        {
                          "term": {
                            "IP": "{{ctx.payload.firewall.hits.hits.0._source.dst_ip}}"
                          }
                        }
                      ],
                      "minimum_should_match": 1
                    }
                  },
                  "size": 1,
                  "sort": [
                    {
                      "@timestamp": {
                        "order": "desc"
                      }
                    }
                  ]
                }
              }
            }
          }
        },
        {
          "dns": {
            "search": {
              "request": {
                "search_type": "query_then_fetch",
                "indices": [
                  "<company-dns-{now/d}>"
                ],
                "types": [],
                "body": {
                  "query": {
                    "bool": {
                      "filter": [
                        {
                          "term": {
                            "client_ip": "{{ctx.payload.dhcp.hits.hits.0._source.IP}}"
                          }
                        },
                        {
                          "range": {
                            "@timestamp": {
                              "from": "{{ctx.payload.firewall.hits.hits.0._source.@timestamp}}||-1m",
                              "to": "{{ctx.payload.firewall.hits.hits.0._source.@timestamp}}||+1m"
                            }
                          }
                        }
                      ]
                    }
                  }
                }
              }
            }
          }
        }
      ]
    }
  },
  "condition": {
    "script": "ctx.payload.firewall.hits.total > 0 && ctx.payload.dhcp.hits.total > 0 && ctx.payload.dns.hits.total > 0"
  },
  "actions": {
    "cmk_ec": {
      "webhook": {
        "scheme": "http",
        "host": "logstash.company.tld",
        "port": 6557,
        "method": "post",
        "path": "/alarm",
        "params": {},
        "headers": {},
        "auth": {
          "basic": {
            "username": "user1",
            "password": "hunter2"
          }
        },
        "body": "{\"_type\":\"ids\", \"msg\":\"{{ctx.payload.firewall.hits.hits.0._source.msg}}\", \"src_ip\": \"{{ctx.payload.firewall.hits.hits.0._source.src_ip}}\", \"dst_ip\": \"{{ctx.payload.firewall.hits.hits.0._source.dst_ip}}\", \"MAC\": \"{{ctx.payload.dhcp.hits.hits.0._source.MAC}}\", \"HostName\": \"{{ctx.payload.dhcp.hits.hits.0._source.HostName}}\", \"dns\": [\"{{#ctx.payload.dns.hits.hits}}{{_source.dns.question.etld_plus_one}}, {{/ctx.payload.dns.hits.hits}}\"]}"
      }
    }
  }
}

Now of course the dhcp and dns queries will fail if there are no virus alerts.
Is there a way to make it conditional, because now it floods the log with index out of bound exceptions.

I tried to tacle this problem with:
"exists": {
"field": "{{ctx.payload.firewall.hits.hits.0._source}}"
}

But it didn't do what I expected.

Thank you.

[spinscale (Alexander Reelsen)]

Hey,

maybe you can spent some more time what you are trying to achieve here and talk about your use-case in more detail. Even if the out of bounds exception did not happen, you would still be able to only get data for the first hit - but what if you got twenty hits?

Can you solve this problem differently? Can you maybe use aggregations or run several queries and then use the script in the condition to decide if you want to trigger an alert?

--Alex

[admlko]

Great!

So what I am trying to achieve is that if there is a virus alert (or basically any other type of event, which should trigger an action and piggyback more data from other indices), it should trigger an alarm. Now the virus alert alone is not enough to get to the bottom of the incident, but I would like to add the IP address of the host involved in my network, so I would know to which device does the alert relate to. And of course, it is necessary to understand which website or server was accessed when the virus alert was generated.

It should also be noted that I still need a way to alarm about virus alerts even though there won't be data in DHCP log nor DNS log.

This use-case can be expanded to almost infinite number of things related to incident response field, but this is the current case I am working on.

Surely, as you suggested, I have not yet thought of multiple alerts within trigger schedule. This is working only when there is one, otherwise it will always trigger to the first one, and discard everything else (which is bad)

I wrote this post because I am not sure what would be the optimal approach to this type of an issue. I didn't thought of using aggregations as previously when I tried to use them to detect multiple login attempts, it was a pain to try to parse them in order to trigger multiple alarms into our monitoring system.
Would you elaborate what do you mean by running several queries and then using the script in the condition?

I would appreciate to hear what do you think would be the way to go in this use-case now that you know what I am trying to achieve.

Cheers Alex!

[admlko]

I have been banging my head against the wall for a week and just cannot come up with a solution purely based on Watcher.

The most challenging part is how to handle multiple events. With any programming language it would be trivial to use a loop but I haven't found a way to do the same with Watcher.

So the part I am not understanding is how to handle multiple virus alerts in a way that each of them should trigger new queries to the DHCP and DNS logs and then at last trigger an action which would send them forward with webhook, one by one or all together in an array.

EDIT:
I guess one way could be to use Logstash to piggyback DHCP and DNS data, using logstash-filter-elasticsearch. So watcher would trigger all virus alerts and forward them to Logstash for furher processing and forwarding to our NMS.
But I would still be interested to understand if this could be done with Watcher