def enrich_beacon_data()

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

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

    def enrich_beacon_data(self):
        """Get all lines in rtops that have not been enriched yet (for CS)"""
        es_query = f'implant.id:* AND c2.program: cobaltstrike AND NOT c2.log.type:implant_newimplant AND NOT tags:{info["submodule"]}'
        not_enriched_results = get_query(es_query, size=10000, index="rtops-*")

        # Created a dict grouped by implant ID
        implant_ids = {}
        for not_enriched in not_enriched_results:
            implant_id = get_value("_source.implant.id", not_enriched)
            if implant_id in implant_ids:
                implant_ids[implant_id].append(not_enriched)
            else:
                implant_ids[implant_id] = [not_enriched]

        hits = []
        # For each implant ID, get the initial beacon line
        for implant_id, implant_val in implant_ids.items():
            initial_beacon_doc = self.get_initial_beacon_doc(implant_id)

            # If not initial beacon line found, skip the beacon ID
            if not initial_beacon_doc:
                continue

            for doc in implant_val:
                # Fields to copy: host.*, implant.*, process.*, user.*
                res = self.copy_data_fields(
                    initial_beacon_doc, doc, ["host", "implant", "user", "process"]
                )
                if res:
                    hits.append(res)

        return hits