# Watcher: IP abuse

**URL:** <https://discuss.elastic.co/t/watcher-ip-abuse/204887>\
**Category:** Elasticsearch\
**Tags:** elastic-stack-alerting\
**Created:** [October 23, 2019, 2:01pm UTC](https://discuss.elastic.co/t/watcher-ip-abuse/204887 "2019-10-23T14:01:24Z")\
**Posts on this page:** 8\
**Page:** 1

<div class="post-metadata">

**Author:** ![gerard1](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/gerard1/32/53507_2.png) [@gerard1](https://discuss.elastic.co/u/gerard1)\
**Post date:** [October 23, 2019, 2:01pm UTC](https://discuss.elastic.co/t/watcher-ip-abuse/204887/1 "2019-10-23T14:01:24Z")

</div>

I'm trying to create a Watcher alert to Slack, every time an IP is generating more than 10 logs per minute, of a certain type (for example, ModSecurity).

This is what I have so far:

```
{
  "trigger": {
    "schedule": {
      "interval": "1m"
    }
  },
  "input": {
    "search": {
      "request": {
        "search_type": "query_then_fetch",
        "indices": [
          "filebeat-*"
        ],
        "rest_total_hits_as_int": true,
        "body": {
          "query": {
            "bool": {
              "filter": [
                {
                  "range": {
                    "@timestamp": {
                      "gte": "now-1min",
                      "lt": "now"
                    }
                  }
                },
                {
                  "term": {
                    "event.dataset": "modsecurity.log"
                  }
                }
              ]
            }
          },
          "aggs": {
            "sources": {
              "terms": {
                "field": "source.ip"
              }
            }
          }
        }
      }
    }
  },
    "condition" : {
      "script": "return ctx.payload.hits.total > ctx.metadata.min_hits"
    },
  "actions": {
    "log_hits": {
      "foreach": "ctx.payload.buckets.hits",
      "max_iterations": 500,
      "slack": {
        "account": "security",
        "message": {
          "from": "Watcher - ModSecurity",
          "text": "{{ctx.payload._source.source.ip}} generated *{{ctx.payload.hits.total}}* logs in the last minute"
        }
      }
    }
  },
  "metadata": {
    "min_hits": 10
  }
}

```

So far, it generates some Slack notifications, but not what I'm expecting. Some times the source.ip is empty, and the total is less than 10.

There is any easiest way to do this? This seems to me a pretty common watcher, but I couldn't find anything similar in the docs.

---

<div class="post-metadata">

**Author:** ![spinscale](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/spinscale/32/25011_2.png) [@spinscale](https://discuss.elastic.co/u/spinscale)\
**Post date:** [October 25, 2019, 1:18pm UTC](https://discuss.elastic.co/t/watcher-ip-abuse/204887/2 "2019-10-25T13:18:22Z")

</div>

If you are trying to check if there is an ip with more than 10 hits in a minute, then checking the total hit count will not help you (that's what you checked in the condition).

You can add a `min_doc_count` to the aggregation, and then check if any aggregation bucket exists or if the first count is more than ten.

Also `ctx.payload._source.source.ip` will not exist, as the payload resembles the structure of a search response, and a search response contains a field of a document within the `ctx.payload.hits.hits` array.

hope this helps as a start!

---

<div class="post-metadata">

**Author:** ![gerard1](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/gerard1/32/53507_2.png) [@gerard1](https://discuss.elastic.co/u/gerard1)\
**Post date:** [October 28, 2019, 2:31pm UTC](https://discuss.elastic.co/t/watcher-ip-abuse/204887/3 "2019-10-28T14:31:51Z")

</div>

Thanks for the feedback! I managed to build the Watcher:

```
{
  "trigger": {
    "schedule": {
      "interval": "1m"
    }
  },
  "input": {
    "search": {
      "request": {
        "search_type": "query_then_fetch",
        "indices": [
          "filebeat-*"
        ],
        "rest_total_hits_as_int": true,
        "body": {
          "query": {
            "bool": {
              "filter": [
                {
                  "range": {
                    "@timestamp": {
                      "gte": "now-1m",
                      "lt": "now"
                    }
                  }
                },
                {
                  "term": {
                    "event.dataset": "modsecurity.log"
                  }
                }
              ]
            }
          },
          "aggs": {
            "sources": {
              "terms": {
                "field": "source.ip",
                "min_doc_count": 10
              }
            }
          }
        }
      }
    }
  },
  "condition": {
    "compare": {
      "ctx.payload.hits.total": {
        "gt": 0
      }
    }
  },
  "actions": {
    "notify-slack": {
      "slack": {
        "account": "security",
        "message": {
          "from": "Watcher - ModSecurity",
          "text": "`{{ctx.payload.aggregations.sources.buckets.0.key}}` generated *{{ctx.payload.aggregations.sources.buckets.0.doc_count}}* logs in the last minute"
        }
      }
    }
  }
}

```

What I'm not sure is what will happen if two different IPs are abusing at the same time, as as far as I know, the action will only notify about the first one found in the aggregation. Correct?

Maybe a foreach action is needed?

---

<div class="post-metadata">

**Author:** ![gerard1](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/gerard1/32/53507_2.png) [@gerard1](https://discuss.elastic.co/u/gerard1)\
**Post date:** [October 29, 2019, 2:59pm UTC](https://discuss.elastic.co/t/watcher-ip-abuse/204887/4 "2019-10-29T14:59:57Z")

</div>

I realised that the correct condition should be:

```
"condition": {
  "script": {
    "source": "return ctx.payload.aggregations.sources.buckets.size() > 0",
    "lang": "painless"
  }
}

```

Otherwise, if there are no aggregation buckets in the results, will fail.

---

<div class="post-metadata">

**Author:** ![spinscale](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/spinscale/32/25011_2.png) [@spinscale](https://discuss.elastic.co/u/spinscale)\
**Post date:** [October 31, 2019, 12:31pm UTC](https://discuss.elastic.co/t/watcher-ip-abuse/204887/5 "2019-10-31T12:31:58Z")

</div>

In order to properly notify about more than one bucket you can loop through results in mustache like this

```auto
{{#ctx.payload.aggregations.sources.buckets}}
Found {{key}}
{{/ctx.payload.aggregations.sources.buckets}}

```

--Alex

---

<div class="post-metadata">

**Author:** ![gerard1](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/gerard1/32/53507_2.png) [@gerard1](https://discuss.elastic.co/u/gerard1)\
**Post date:** [October 31, 2019, 1:14pm UTC](https://discuss.elastic.co/t/watcher-ip-abuse/204887/6 "2019-10-31T13:14:13Z")

</div>

But in that way, it will encapsulate all the results in a single slack message. I think the "foreach" definition would work better, however I'm not able to make it working...

---

<div class="post-metadata">

**Author:** ![spinscale](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/spinscale/32/25011_2.png) [@spinscale](https://discuss.elastic.co/u/spinscale)\
**Post date:** [October 31, 2019, 1:19pm UTC](https://discuss.elastic.co/t/watcher-ip-abuse/204887/7 "2019-10-31T13:19:08Z")

</div>

then please go ahead and share the full watch and the execute watch output with the `foreach` field.

Thanks!

---

<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 28, 2019, 1:19pm UTC](https://discuss.elastic.co/t/watcher-ip-abuse/204887/8 "2019-11-28T13:19:20Z")

</div>

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