def alarm_check()

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

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

    def alarm_check(self, alarmed_messages):
        """This check queries for C2 messages (input of eventlog) that contain 'REDELK_ALARM'"""
        es_query = {
            "sort": [{"@timestamp": {"order": "asc"}}],
            "query": {
                "bool": {
                    "must": {
                        "query_string": {
                            "query": "(c2.message:*REDELK_ALARM*) AND (((c2.log.type:implant_input) AND (tags:enrich_*)) OR (c2.log.type:events))"
                        }
                    },
                    "must_not": [{"match": {"tags": info["submodule"]}}],
                }
            },
        }
        res = raw_search(es_query, index="rtops-*")
        if res is None:
            not_enriched_hits = []
        else:
            not_enriched_hits = res["hits"]["hits"]

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

        hits = []

        # Now we check if the C2 messages have already been alarmed in the past timeframe defined in the config
        # pylint: disable=invalid-name
        for message, message_val in messages.items():
            # Not alarmed yet, process it
            if message not in alarmed_messages:
                hits += message_val

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