def alarm_check()

in elkserver/docker/redelk-base/redelkinstalldata/scripts/modules/alarm_httptraffic/module.py [0:0]

• 36 lines of code
• 7 McCabe index (conditional complexity)

    def alarm_check(self, alarmed_ips):  # pylint: disable=no-self-use
        """This check queries for IP's that aren't listed in any iplist* but do talk to c2* paths on redirectors"""
        es_query = {
            "sort": [{"@timestamp": {"order": "desc"}}],
            "query": {
                "bool": {
                    "filter": [{"match": {"tags": "enrich_iplists"}}],
                    "must": {
                        "query_string": {
                            "fields": ["redir.backend.name"],
                            "query": "c2*",
                        }
                    },
                    "must_not": [
                        {"query_string": {"fields": ["tags"], "query": "iplist_*"}},
                        {"match": {"tags": info["submodule"]}},
                    ],
                }
            },
        }
        res = raw_search(es_query, index="redirtraffic-*")
        if res is None:
            not_enriched_hits = []
        else:
            not_enriched_hits = res["hits"]["hits"]

        # Created a dict grouped by IP address (from source.ip)
        ips = {}
        for not_enriched in not_enriched_hits:
            # pylint: disable=invalid-name
            ip = get_value("_source.source.ip", not_enriched)
            if ip in ips:
                ips[ip].append(not_enriched)
            else:
                ips[ip] = [not_enriched]

        hits = []

        # Now we check if the IPs have already been alarmed in the past timeframe defined in the config
        # pylint: disable=invalid-name
        for ip, ip_val in ips.items():
            # Not alarmed yet, process it
            if ip not in alarmed_ips:
                hits += ip_val

        # Return the array of new IP documents to be alarmed
        return hits