Initial commit

parsedmarc/cli.py

@@ -13,10 +13,9 @@ import json
 from elasticsearch.exceptions import ElasticsearchException
 
 from parsedmarc import logger, IMAPError, get_dmarc_reports_from_inbox, \
-    parse_report_file, elastic, splunk, save_output, watch_inbox, \
+    parse_report_file, elastic, kafkaclient, splunk, save_output, watch_inbox, \
     email_results, SMTPError, ParserError, __version__
 
-
 def _main():
     """Called when the module is executed"""
     def process_reports(reports_):
@@ -25,6 +24,12 @@ def _main():
                                                indent=2))
         if not args.silent:
             print(output_str)
+        if args.kafka_hosts:
+            try:
+              kafkaClient = kafkaclient.KafkaClient(args.kafka_hosts)
+              # dont do this
+            except Exception as error:
+              logger.error("Kafka Error: {0}".format(error.__str__()))
         if args.save_aggregate:
             for report in reports_["aggregate_reports"]:
                 try:
@@ -37,6 +42,14 @@ def _main():
                     logger.error("Elasticsearch Error: {0}".format(
                         error_.__str__()))
                     exit(1)
+                try:
+                     if args.kafka_hosts:
+                         kafkaClient.save_aggregate_reports_to_kafka(
+                         report, kafka_aggregate_topic)
+                         # dont do this catch specific exceptions
+                except Exception as error_:
+                     logger.error("Kafka Error: {0}".format(
+                         error_.__str__()))
             if args.hec:
                 try:
                     aggregate_reports_ = reports_["aggregate_reports"]
@@ -55,6 +68,14 @@ def _main():
                 except ElasticsearchException as error_:
                     logger.error("Elasticsearch Error: {0}".format(
                         error_.__str__()))
+                try:
+                       if args.kafka_hosts:
+                           kafkaClient.save_forensic_reports_to_kafka(
+                           report, kafka_forensic_topics)
+                           # dont do this
+                except Exception as error_:
+                       logger.error("Kafka Error: {0}".format(
+                           error_.__str__()))
             if args.hec:
                 try:
                     forensic_reports_ = reports_["forensic_reports"]
@@ -122,6 +143,12 @@ def _main():
                             default=False,
                             help="Skip certificate verification for Splunk "
                                  "HEC")
+    arg_parser.add_argument("-K", "--kafka-hosts", nargs="*",
+                            help="A list of one or more Kafka hostnames or URLs")
+    arg_parser.add_argument("--kafka-aggregate-topic",
+                            help="The Kafka topic to publish aggregate reports to.")
+    arg_parser.add_argument("--kafka-forensic_topic",
+                            help="The Kafka topic to publish forensic reports to.")
     arg_parser.add_argument("--save-aggregate", action="store_true",
                             default=False,
                             help="Save aggregate reports to search indexes")
@@ -191,7 +218,7 @@ def _main():
         es_forensic_index = "{0}_{1}".format(es_forensic_index, suffix)
 
     if args.save_aggregate or args.save_forensic:
-        if args.elasticsearch_host is None and args.hec is None:
+        if args.elasticsearch_host is None and args.hec  and args.kafka_hosts is None:
             args.elasticsearch_host = ["localhost:9200"]
         try:
             if args.elasticsearch_host:
@@ -214,6 +241,15 @@ def _main():
                                       args.hec_index,
                                       verify=verify)
 
+    kafka_aggregate_topic = "dmarc_aggrregate"
+    kafka_forensic_topic = "dmarc_forensic"
+
+    if args.kafka_aggregate_topic:
+      kafka_aggregate_topic = args.kafka_aggregate_topic
+
+    if args.kafka_forensic_topic:
+      kafka_forensic_topic = args.kafka_forensic_topic
+
     file_paths = []
     for file_path in args.file_path:
         file_paths += glob(file_path)

parsedmarc/kafkaclient.py

@@ -0,0 +1,51 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+
+from kafka import KafkaProducer
+import json
+
+class KafkaError(RuntimeError):
+    """Raised when a Kafka error occurs"""
+
+class KafkaClient(object):
+  def __init__(self, kafka_hosts):
+      """ Right now lets just do one host"""
+      self.host = kafka_hosts
+      self.producer = KafkaProducer(bootstrap_servers=kafka_hosts, value_serializer=lambda v: json.dumps(v).encode('utf-8'))
+
+  def save_aggregate_reports_to_kafka(self, aggregate_reports, aggregate_topic):
+        """
+        Saves aggregate DMARC reports to Splunk
+
+        Args:
+            aggregate_reports (list):  A list of aggregate report dictionaries
+            to save to kafka
+
+        """
+        if type(aggregate_reports) == dict:
+            aggregate_reports = [aggregate_reports]
+
+        if len(aggregate_reports) < 1:
+            return
+
+        for report in aggregate_reports:
+            self.producer.send(aggregate_topic, report)
+
+
+  def save_forensic_reports_to_kafka(self, forensic_reports, forensic_topic):
+            """
+            Saves forensic DMARC reports to Kafka
+
+            Args:
+                forensic_reports (list):  A list of forensic report dictionaries
+                to save to kafka
+
+            """
+            if type(forensic_reports) == dict:
+                forensic_reports = [forensic_reports]
+
+            if len(forensic_reports) < 1:
+                return
+
+            for report in forensic_reports:
+                self.producer.send(forensic_topic, report)

requirements.txt

@@ -15,3 +15,4 @@ sphinx_rtd_theme
 collective.checkdocs
 wheel
 rstcheck
+kafka-python