def alarm_check()

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

• 49 lines of code
• 5 McCabe index (conditional complexity)

    def alarm_check(self):
        """This check queries public sources given a list of md5 hashes. If a hash was seen we set an alarm"""
        es_query = "c2.log.type:ioc AND NOT tags:alarm_filehash AND ioc.type:file"
        alarmed_md5_q = {
            "aggs": {
                "interval_filter": {
                    "filter": {
                        "range": {
                            "alarm.last_checked": {
                                "gte": f"now-{self.interval}s",
                                "lt": "now",
                            }
                        }
                    },
                    "aggs": {"md5_interval": {"terms": {"field": "file.hash.md5"}}},
                },
                "alarmed_filter": {
                    "filter": {"terms": {"tags": ["alarm_filehash"]}},
                    "aggs": {"md5_alarmed": {"terms": {"field": "file.hash.md5"}}},
                },
            }
        }
        report = {}
        iocs = []
        self.logger.debug("Running query %s", es_query)

        # First, get all IOCs of type 'file' that have not been alarmed yet
        iocs = get_query(es_query, 10000, index="rtops-*")
        self.logger.debug("found ioc: %s", iocs)

        # Then we get an aggregation of all md5 alarmed within the last 'interval' time
        self.logger.debug("Running query %s", alarmed_md5_q)
        already_alarmed_result = raw_search(alarmed_md5_q, index="rtops-*")

        already_checked = []
        already_alarmed = []

        if already_alarmed_result:
            self.logger.debug(already_alarmed_result["aggregations"])

            # add md5 hashes that have been checked within the 'interval' in 'already_checked'
            for hit in already_alarmed_result["aggregations"]["interval_filter"][
                "md5_interval"
            ]["buckets"]:
                already_checked.append(hit["key"])

            # add md5 hashes that have been alarmed previously in 'already_alarmed'
            for hit in already_alarmed_result["aggregations"]["alarmed_filter"][
                "md5_alarmed"
            ]["buckets"]:
                already_alarmed.append(hit["key"])

        # Group all hits per md5 hash
        md5_dict = self.group_hits(iocs, already_alarmed, already_checked)

        # Create an array with all md5 hashes to send to the different providers
        # we now have an array with unique md5's to go test
        md5_list = []
        for md5 in md5_dict:
            md5_list.append(md5)

        self.logger.debug("md5 hashes to check: %s", md5_list)

        # Run the checks
        check_results = self.check_hashes(md5_list)

        # Get the alarmed hashes with their corresponding mutations
        alarmed_hashes = self.get_mutations(check_results)

        # Get the report
        report = self.build_report(md5_dict, alarmed_hashes)

        return report