text: # Standard library imports
  import json
  import threading
  import hashlib
  import logging
  
  # Third party imports
  from kafka import KafkaConsumer
  from elasticsearch import Elasticsearch
  
  # Local application imports
  from Config.config import KafkaTopicNames, KafkaConfig, KafkaGroupIds, ElasticConfig
  
  class ElasticSendConsumers(threading.Thread):
      def __init__(self):
          threading.Thread.__init__(self)
          self.consumer = KafkaConsumer(
                          KafkaTopicNames.ELASTICSEND,
                          bootstrap_servers       = KafkaConfig.BOOTSTRAPSERVER,
                          auto_offset_reset       = 'earliest',
                          enable_auto_commit      = False,
                          group_id                = KafkaGroupIds.ELASTICSEND,
                          value_deserializer      = lambda x: json.loads(x.decode('utf-8')))
          self.logger = logging.getLogger(__name__)
  
      def connect_elasticsearch(self):
          # Connect to cluster over SSL using auth for best security:
          es_header = [{
                  'host': ElasticConfig.HOSTNAME,
                  'port': ElasticConfig.PORT,
                  'use_ssl': ElasticConfig.USESSL,
                  'http_auth': (ElasticConfig.USERNAME, ElasticConfig.PASSWORD)
                  }]
          es = Elasticsearch(es_header)
          return es
  
      def run(self):
          try:
              es = self.connect_elasticsearch()
  
              while True:
                  msg_pack = self.consumer.poll(timeout_ms=2000)
                  for message in msg_pack.items():
                      for consumer_record in message[1]:
                          try:
                              data = consumer_record.value
                              self.logger.info('Recieved {}_{}'.format(data.get("target"), data.get("hostname")))
  
                              if (data.get("elastic_type") == "nmapOutput"):
                                  nmap_elastic_id =  data.get('scan_id') + "_" + data.get('target')
                                  data['id'] = nmap_elastic_id
                                  outcome = es.index(index = ElasticConfig.NMAPINDEX, id = nmap_elastic_id, body=data)
                                  self.logger.info(outcome)
                              elif (data.get("elastic_type") == "cveSearchOutput"):
                                  cveSearch_elastic_id =  data.get('scan_id') + "_" + data.get('target') + "_" + data.get('source')
                                  data['id'] =  data.get('scan_id') + "_" + data.get('target')
                                  outcome = es.index(index = ElasticConfig.CVESEARCHINDEX, id = cveSearch_elastic_id, body=data)
                                  self.logger.info(outcome)
  
                                  # Vulnerbilites
                                  cveSearchVuls = self.genCVESearchVuls(data, cveSearch_elastic_id)
                                  for cveSearchVul in cveSearchVuls:
                                      outcomeScanDetail = es.index(index = ElasticConfig.VULNERABILITY_INDEX, body=cveSearchVul)
                                      self.logger.info(outcomeScanDetail)
                              elif (data.get("elastic_type") == "nseOutput"):
                                  nse_elastic_id =  data.get('scan_id') + "_" + data.get('target')
                                  data['id'] = nse_elastic_id
                                  outcome = es.index(index = ElasticConfig.NSEINDEX, id = nse_elastic_id, body=data)
                                  self.logger.info(outcome)
  
                                  # Vulnerbilites
                                  nseVuls = self.genNSEVuls(data, nse_elastic_id)
                                  for nseVul in nseVuls:
                                      outcomeScanDetail = es.index(index = ElasticConfig.VULNERABILITY_INDEX, body=nseVul)
                                      self.logger.info(outcomeScanDetail)
                              elif (data.get("elastic_type") == "wappalyzerOutput"):
                                  wappalyzer_elastic_id =  data.get('scan_id') + "_" + data.get('target')
                                  data['id'] = wappalyzer_elastic_id
                                  outcome = es.index(index = ElasticConfig.WAPPALYZERINDEX, id = wappalyzer_elastic_id, body=data)
                                  self.logger.info(outcome)
                              elif (data.get("elastic_type") == "acunetixOutput"):
                                  (acunetixSumaryData, acunetixDetailData) = self.normalizeAcunetixData(data)
                                  acunetix_sumary_elastic_id =  data.get('scan_id') + "_" + data.get('target') + "_" + data.get('portScanned')
                                  acunetixSumaryData['id'] =  data.get('scan_id') + "_" + data.get('target')
  
                                  outcomeScanSumary = es.index(index = ElasticConfig.ACUNETIX_SUMARY_INDEX, id = acunetix_sumary_elastic_id, body=acunetixSumaryData)
                                  self.logger.info(outcomeScanSumary)
                                  for detailData in acunetixDetailData:
                                      acunetix_detail_elastic_id = detailData.get('vuln_id')
                                      detailData['target'] = outcomeScanSumary.get('target')
                                      outcomeScanDetail = es.index(index = ElasticConfig.ACUNETIX_DETAIL_INDEX, id = acunetix_detail_elastic_id, body=detailData)
                                      self.logger.info(outcomeScanDetail)
  
                                  # Vulnerbilites
                                  acunetixVuls = self.genAcunetixVuls(acunetixSumaryData, acunetix_sumary_elastic_id)
                                  for acunetixVul in acunetixVuls:
                                      outcomeScanDetail = es.index(index = ElasticConfig.VULNERABILITY_INDEX, body=acunetixVul)
                                      self.logger.info(outcomeScanDetail)
                              elif (data.get("elastic_type") == "niktoOutput"):
                                  nikto_elastic_id =  data.get('scan_id') + "_" + data.get('target') + "_" + data.get('portScanned')
                                  data['id'] =  data.get('scan_id') + "_" + data.get('target')
                                  outcome = es.index(index = ElasticConfig.NIKTO_INDEX, id = nikto_elastic_id, body=data)
                                  self.logger.info(outcome)
  
                                  # Vulnerbilites
                                  niktoVuls = self.genNiktoVuls(data, nikto_elastic_id)
                                  for niktoVul in niktoVuls:
                                      outcomeScanDetail = es.index(index = ElasticConfig.VULNERABILITY_INDEX, body=niktoVul)
                                      self.logger.info(outcomeScanDetail)
                              elif (data.get("elastic_type") == "nessusOutput"):
                                  nessus_elastic_id =  data.get('scan_id') + "_" + data.get('target')
                                  data['id'] =  nessus_elastic_id
                                  outcome = es.index(index = ElasticConfig.NESSUS_INDEX, id = nessus_elastic_id, body=data)
                                  self.logger.info(outcome)
  
                                  # Vulnerbilites
                                  nessusVuls = self.genNessusVuls(data, nessus_elastic_id)
                                  for nessusVul in nessusVuls:
                                      outcomeScanDetail = es.index(index = ElasticConfig.VULNERABILITY_INDEX, body=nessusVul)
                                      self.logger.info(outcomeScanDetail)
                          except:
                              self.logger.exception(consumer_record.value)
  
                          self.consumer.commit()
          except:
              self.logger.exception("Thread " + __name__ + " terminated")
  
      def normalizeAcunetixData(self, acunetixData):
          def normalizeSumaryData(acunetixData):
              acunetixSumaryData = dict()
              acunetixSumaryData['target'] = acunetixData.get('target')
              acunetixSumaryData['hostname'] = acunetixData.get('hostname')
              acunetixSumaryData['scanstat'] = acunetixData.get('scanstat')
              acunetixSumaryData['vuln_stats'] = acunetixData.get('vuln_stats')
              acunetixSumaryData["portScaned"] = acunetixData.get("portScaned")
              acunetixSumaryData['root_scan_id'] = acunetixData.get('root_scan_id')
              acunetixSumaryData['scan_type'] = acunetixData.get('scan_type')
              acunetixSumaryData['scan_id'] = acunetixData.get('scan_id')
  
              scan_details = acunetixData.get('scan_details')
  
              acunetixSumaryData['scan_details'] = list()
  
              for scan_detail in scan_details:
                  sumary_scanDetail = dict()
                  sumary_scanDetail['response_info'] = scan_detail.get('response_info')
                  sumary_scanDetail['affects_url'] = scan_detail.get('affects_url')
                  sumary_scanDetail['vt_name'] = scan_detail.get('vt_name')
                  sumary_scanDetail['criticality'] = scan_detail.get('criticality')
                  sumary_scanDetail['description'] = scan_detail.get('description')
                  sumary_scanDetail['vuln_id'] = scan_detail.get('vuln_id')
                  sumary_scanDetail['recommendation'] = scan_detail.get('recommendation')
                  sumary_scanDetail['affects_detail'] = scan_detail.get('affects_detail')
                  sumary_scanDetail['cvss_score'] = scan_detail.get('cvss_score')
                  sumary_scanDetail['severity'] = scan_detail.get('severity')
                  acunetixSumaryData['scan_details'].append(sumary_scanDetail)
  
              return acunetixSumaryData
  
          def normalizeDetailData(acunetixData):
              acunetixDetailData = acunetixData.get('scan_details')
              return acunetixDetailData
  
  
          acunetixSumaryData = normalizeSumaryData(acunetixData)
          acunetixDetailData = normalizeDetailData(acunetixData)
          return (acunetixSumaryData, acunetixDetailData)
  
      def genAcunetixVuls(self, acunetixData, acunetixID):
          vuls = []
          scan_details = acunetixData.get("scan_details")
          for scan_detail in scan_details:
              vul = {}
              vul['index'] = "Acunetix"
              vul['id'] = acunetixID
              vul['time_scan'] = acunetixData.get('scanstat').get('startTime')
              vul['target'] = acunetixData.get('target')
              vul['hostname'] = acunetixData.get('hostname')
              vul['root_scan_id'] = acunetixData.get('root_scan_id')
              vul['scan_name'] = acunetixData.get('scan_name')
              vul['scan_type'] = "Full Scan" if acunetixData.get('scan_type') == "full_scan" else "Non-commercial"
  
              vul['name'] = scan_detail.get('vt_name')
              vul['level'] = scan_detail.get('cvss_score')
              vul['link'] = acunetixID + '/' + scan_detail.get('vuln_id')
              vuls.append(vul)
          return vuls
  
      def genNiktoVuls(self, niktoData, niktoID):
          vuls = []
          vulnerabilities = niktoData.get("vulnerabilities")
          for vulnerabilitiy in vulnerabilities:
              vul = {}
              vul['index'] = "Nikto"
              vul['id'] = niktoID
              vul['time_scan'] = niktoData.get('scanstat').get('startTime')
              vul['target'] = niktoData.get('target')
              vul['hostname'] = niktoData.get('hostname')
              vul['root_scan_id'] = niktoData.get('root_scan_id')
              vul['scan_name'] = niktoData.get('scan_name')
              vul['scan_type'] = "Full Scan" if niktoData.get('scan_type') == "full_scan" else "Non-commercial"
  
              vul['name'] = vulnerabilitiy.get("OSVDB") if vulnerabilitiy.get("OSVDB") != "0" else "None"
              vul['level'] = vulnerabilitiy.get('cvss')
              vul['link'] = niktoID
              vuls.append(vul)
          return vuls
  
      def genNessusVuls(self, nessusData, nessusID):
          vuls = []
          scan_details = nessusData.get("scan_details")
          for scan_detail in scan_details:
              vul = {}
              vul['index'] = "Nessus"
              vul['id'] = nessusID
              vul['time_scan'] = nessusData.get('scanstat').get('startTime')
              vul['target'] = nessusData.get('target')
              vul['hostname'] = nessusData.get('hostname')
              vul['root_scan_id'] = nessusData.get('root_scan_id')
              vul['scan_name'] = nessusData.get('scan_name')
              vul['scan_type'] = "Full Scan" if nessusData.get('scan_type') == "full_scan" else "Non-commercial"
  
              vul['name'] = scan_detail.get('pluginname')
              vul['level'] = scan_detail.get('cvss')
              vul['link'] = nessusID
              vuls.append(vul)
          return vuls
  
      def genCVESearchVuls(self, cveSearchData, CVESearchID):
          vuls = []
          weaknesses = cveSearchData.get("weakness")
          for weakness in weaknesses:
              vul = {}
              vul['index'] = "CVE-Search"
              vul['id'] = CVESearchID
              vul['time_scan'] = cveSearchData.get('scanstat').get('startTime')
              vul['target'] = cveSearchData.get('target')
              vul['hostname'] = cveSearchData.get('hostname')
              vul['root_scan_id'] = cveSearchData.get('root_scan_id')
              vul['scan_name'] = cveSearchData.get('scan_name')
              vul['scan_type'] = "Full Scan" if cveSearchData.get('scan_type') == "full_scan" else "Non-commercial"
  
              vul['name'] = weakness.get('cve')
              vul['level'] = weakness.get('cvss')
              vul['link'] = CVESearchID
              vuls.append(vul)
          return vuls
      
      def genNSEVuls(self, nseData, nseID):
          vuls = []
          nseOutputs = nseData.get("nseOutputs")
          for nseOutput in nseOutputs:
              vul = {}
              vul['index'] = "NSE"
              vul['id'] = nseID
              vul['time_scan'] = nseData.get('scanstat').get('startTime')
              vul['target'] = nseData.get('target')
              vul['hostname'] = nseData.get('hostname')
              vul['root_scan_id'] = nseData.get('root_scan_id')
              vul['scan_name'] = nseData.get('scan_name')
              vul['scan_type'] = "Full Scan" if nseData.get('scan_type') == "full_scan" else "Non-commercial"
  
              vul['name'] = nseOutput.get('script')
              vul['level'] = 0.0
              vul['link'] = nseID
              vuls.append(vul)
          return vuls