Merge pull request #1516 from mpurzynski/gdnew

A new version of the guardduty plugin and a dedicated worker

mq/esworker_guardduty.py

@@ -0,0 +1,166 @@
+#!/usr/bin/env python
+
+# This Source Code Form is subject to the terms of the Mozilla Public
+# License, v. 2.0. If a copy of the MPL was not distributed with this
+# file, You can obtain one at http://mozilla.org/MPL/2.0/.
+# Copyright (c) 2017 Mozilla Corporation
+
+
+import json
+import sys
+import socket
+from configlib import getConfig, OptionParser
+from datetime import datetime
+from mozdef_util.utilities.toUTC import toUTC
+from mozdef_util.utilities.logger import logger, initLogger
+
+from esworker_sns_sqs import taskConsumer
+from lib.plugins import sendEventToPlugins
+from lib.sqs import connect_sqs
+from mozdef_util.elasticsearch_client import ElasticsearchClient
+from mozdef_util.utilities.key_exists import key_exists
+
+
+# running under uwsgi?
+try:
+    import uwsgi
+
+    hasUWSGI = True
+except ImportError as e:
+    hasUWSGI = False
+
+
+class GDtaskConsumer(taskConsumer):
+    def build_submit_message(self, message):
+        # default elastic search metadata for an event
+        metadata = {"index": "events", "id": None}
+
+        event = {}
+
+        event["receivedtimestamp"] = toUTC(datetime.now()).isoformat()
+        event["mozdefhostname"] = self.options.mozdefhostname
+
+        if "tags" in event:
+            event["tags"].extend([self.options.taskexchange])
+        else:
+            event["tags"] = [self.options.taskexchange]
+
+        event["severity"] = "INFO"
+        event["source"] = "guardduty"
+        event["details"] = {}
+
+        event["details"] = message["details"]
+        if "hostname" in message:
+            event["hostname"] = message["hostname"]
+        if "summary" in message:
+            event["summary"] = message["summary"]
+        if "category" in message:
+            event["details"]["category"] = message["category"]
+        if "tags" in message:
+            event["details"]["tags"] = message["tags"]
+        event["utctimestamp"] = toUTC(message["timestamp"]).isoformat()
+        event["timestamp"] = event["utctimestamp"]
+        (event, metadata) = sendEventToPlugins(event, metadata, self.pluginList)
+        # Drop message if plugins set to None
+        if event is None:
+            return
+
+        self.save_event(event, metadata)
+
+    def on_message(self, message_raw):
+        if "Message" in message_raw:
+            message = json.loads(message_raw["Message"])
+            if key_exists('details.finding.action.actionType', message):
+                if message["details"]["finding"]["action"]["actionType"] == "PORT_PROBE":
+                    if "portProbeDetails" in message["details"]["finding"]["action"]["portProbeAction"]:
+                        for probe in message["details"]["finding"]["action"]["portProbeAction"]["portProbeDetails"]:
+                            isolatedmessage = message
+                            isolatedmessage["details"]["finding"]["probeevent"] = probe
+                            self.build_submit_message(isolatedmessage)
+                elif message["details"]["finding"]["action"]["actionType"] == "AWS_API_CALL":
+                    if "recentApiCalls" in message["details"]["finding"]["additionalInfo"]:
+                        message["details"]["finding"]["additionalInfo"]["apiCalls"] = message["details"]["finding"][
+                            "additionalInfo"
+                        ]["recentApiCalls"]
+                    for call in message["details"]["finding"]["additionalInfo"]["apiCalls"]:
+                        isolatedmessage = message
+                        isolatedmessage["details"]["finding"]["apicalls"] = call
+                        self.build_submit_message(isolatedmessage)
+                    else:
+                        self.build_submit_message(message)
+
+
+def esConnect():
+    """open or re-open a connection to elastic search"""
+    return ElasticsearchClient((list("{0}".format(s) for s in options.esservers)), options.esbulksize)
+
+
+def initConfig():
+    # capture the hostname
+    options.mozdefhostname = getConfig("mozdefhostname", socket.gethostname(), options.configfile)
+
+    # elastic search options. set esbulksize to a non-zero value to enable bulk posting, set timeout to post no matter how many events after X seconds.
+    options.esservers = list(getConfig("esservers", "http://localhost:9200", options.configfile).split(","))
+    options.esbulksize = getConfig("esbulksize", 0, options.configfile)
+    options.esbulktimeout = getConfig("esbulktimeout", 30, options.configfile)
+
+    # set to sqs for Amazon
+    options.mqprotocol = getConfig("mqprotocol", "sqs", options.configfile)
+
+    # rabbit message queue options
+    options.taskexchange = getConfig("taskexchange", "eventtask", options.configfile)
+    # rabbit: how many messages to ask for at once from the message queue
+    options.prefetch = getConfig("prefetch", 10, options.configfile)
+
+    # aws options
+    options.accesskey = getConfig("accesskey", "", options.configfile)
+    options.secretkey = getConfig("secretkey", "", options.configfile)
+    options.region = getConfig("region", "", options.configfile)
+
+    # How long to sleep between polling
+    options.sleep_time = getConfig("sleep_time", 0.1, options.configfile)
+
+
+def main():
+    if hasUWSGI:
+        logger.info("started as uwsgi mule {0}".format(uwsgi.mule_id()))
+    else:
+        logger.info("started without uwsgi")
+
+    if options.mqprotocol not in ("sqs"):
+        logger.error("Can only process SQS queues, terminating")
+        sys.exit(1)
+
+    sqs_queue = connect_sqs(
+        region_name=options.region,
+        aws_access_key_id=options.accesskey,
+        aws_secret_access_key=options.secretkey,
+        task_exchange=options.taskexchange,
+    )
+    # consume our queue
+    GDtaskConsumer(sqs_queue, es, options).run()
+
+
+if __name__ == "__main__":
+    # configure ourselves
+    parser = OptionParser()
+    parser.add_option(
+        "-c", dest="configfile", default=sys.argv[0].replace(".py", ".conf"), help="configuration file to use"
+    )
+    (options, args) = parser.parse_args()
+    initConfig()
+    initLogger(options)
+
+    # open ES connection globally so we don't waste time opening it per message
+    es = esConnect()
+
+    try:
+        main()
+    except KeyboardInterrupt as e:
+        logger.info("Exiting worker")
+        if options.esbulksize != 0:
+            es.finish_bulk()
+    except Exception as e:
+        if options.esbulksize != 0:
+            es.finish_bulk()
+        raise

mq/esworker_sns_sqs.py

@@ -53,7 +53,7 @@ class taskConsumer(object):
     def run(self):
         while True:
             try:
-                records = self.sqs_queue.receive_messages(MaxNumberOfMessages=options.prefetch)
+                records = self.sqs_queue.receive_messages(MaxNumberOfMessages=self.options.prefetch)
                 for msg in records:
                     msg_body = msg.body
                     try:
@@ -66,7 +66,7 @@ class taskConsumer(object):
                         logger.error("Invalid message, not JSON <dropping message and continuing>: %r" % msg_body)
                         msg.delete()
                         continue
-                time.sleep(options.sleep_time)
+                time.sleep(self.options.sleep_time)
             except (SSLEOFError, SSLError, socket.error):
                 logger.info("Received network related error...reconnecting")
                 time.sleep(5)

mq/plugins/guardDuty.py

@@ -3,87 +3,140 @@
 # file, You can obtain one at https://mozilla.org/MPL/2.0/.
 # Copyright (c) 2017 Mozilla Corporation
 
-from mozdef_util.utilities.key_exists import key_exists
+import os
+import yaml
+import jmespath
 from mozdef_util.utilities.toUTC import toUTC
-from mozdef_util.utilities.dot_dict import DotDict
+from platform import node
 
 
 class message(object):
     def __init__(self):
-        '''
+        """
             Plugin used to fix object type discretions with cloudtrail messages
-        '''
-        self.registration = ['guardduty']
-        self.priority = 10
+        """
+        self.registration = ["guardduty"]
+        self.priority = 3
+
+        try:
+            self.mozdefhostname = "{0}".format(node())
+        except:
+            self.mozdefhostname = "failed to fetch mozdefhostname"
+            pass
+
+        with open(os.path.join(os.path.dirname(__file__), "guardduty_mapping.yml"), "r") as f:
+            mapping_map = f.read()
+
+        yap = yaml.safe_load(mapping_map)
+        self.eventtypes = list(yap.keys())
+        self.yap = yap
+        del mapping_map
 
         # AWS guard duty sends dates as iso_8601 which ES doesn't appreciate
         # here's a list of date fields we'll convert to isoformat
-        self.date_keys = [
-            'details.finding.eventlastseen',
-            'details.finding.eventfirstseen',
-            'details.resource.instancedetails.launchtime',
-            'details.createdat',
-            'details.updatedat'
-        ]
-
-        # AWS guard duty can send IPs in a bunch of places
-        # Lets pick out some likely targets and format them
-        # so other mozdef plugins can rely on their location
-        self.ipaddress_keys = [
-            'details.finding.action.networkconnectionaction.remoteipdetails.ipaddressv4',
-            'details.finding.action.awsapicallaction.remoteipdetails.ipadrressv4'
-        ]
-
-    def convert_key_date_format(self, needle, haystack):
-        num_levels = needle.split(".")
-        if len(num_levels) == 0:
-            return False
-        current_pointer = haystack
-        for updated_key in num_levels:
-            if updated_key == num_levels[-1]:
-                current_pointer[updated_key] = toUTC(
-                    current_pointer[updated_key]).isoformat()
-                return haystack
-            if updated_key in current_pointer:
-                current_pointer = current_pointer[updated_key]
-            else:
-                return haystack
+        self.date_keys = ["gdeventcreatedts", "gdeventupdatedts", "gdeventfirstseents", "gdeventlastseents"]
 
     def onMessage(self, message, metadata):
-        if 'source' not in message:
+        if "source" not in message:
             return (message, metadata)
 
-        if not message['source'] == 'guardduty':
+        if not message["source"] == "guardduty":
             return (message, metadata)
 
-        # reformat the date fields to iosformat
+        if "details" not in message:
+            return (message, metadata)
+
+        newmessage = dict()
+        newmessage["receivedtimestamp"] = message["receivedtimestamp"]
+        newmessage["timestamp"] = message["timestamp"]
+        newmessage["utctimestamp"] = message["utctimestamp"]
+        newmessage["mozdefhostname"] = message["mozdefhostname"]
+        newmessage["tags"] = ["aws", "guardduty"] + message["tags"]
+        newmessage["category"] = "guardduty"
+        newmessage["source"] = "guardduty"
+        newmessage["customendpoint"] = ""
+        newmessage["details"] = {}
+        newmessage["details"]["type"] = message["details"]["finding"]["action"]["actionType"].lower()
+        newmessage["details"]["finding"] = message['details']["category"]
+        newmessage["summary"] = message["details"]["title"]
+        newmessage["details"]["resourcerole"] = message["details"]["finding"]["resourceRole"].lower()
+
+        # This is a hack to let the following code match and extract useful information about local network configuration
+        # Sometimes AWS does not feel like sending it at all or sends an empty list or a single element list or a multiple-elements list or a dictionary - so try to handle them all
+        if message["details"]["finding"]["action"]["actionType"] != "AWS_API_CALL":
+            if "networkInterfaces" in message["details"]["resource"]["instanceDetails"]:
+                nic = message["details"]["resource"]["instanceDetails"]["networkInterfaces"]
+                if isinstance(nic, list):
+                    if len(nic) > 0:
+                        message["details"]["resource"]["instanceDetails"]["networkInterfaces"] = nic[0]
+        if message["details"]["category"] in self.eventtypes:
+            for key in self.yap[newmessage["details"]["finding"]]:
+                mappedvalue = jmespath.search(self.yap[newmessage["details"]["finding"]][key], message)
+                # JMESPath likes to silently return a None object
+                if mappedvalue is not None:
+                    newmessage["details"][key] = mappedvalue
+
+        # reformat the date fields to isoformat
         for date_key in self.date_keys:
-            if key_exists(date_key, message):
-                if message.get(date_key) is None:
-                    continue
-                else:
-                    message = self.convert_key_date_format(date_key, message)
+            if date_key in newmessage["details"]:
+                newmessage["details"][date_key] = toUTC(newmessage["details"][date_key]).isoformat()
 
-        # convert the dict to a dot dict for saner deep key/value processing
-        message = DotDict(message)
-        # pull out the likely source IP address
-        for ipaddress_key in self.ipaddress_keys:
-            if 'sourceipaddress' not in message['details']:
-                if key_exists(ipaddress_key, message):
-                    message.details.sourceipaddress = message.get(
-                        ipaddress_key)
+        # Handle some special cases
 
-        # if we still haven't found what we are looking for #U2
-        # sometimes it's in a list
-        if 'sourceipaddress' not in message['details']:
-            if key_exists('details.finding.action.portprobeaction.portprobedetails', message) \
-                    and isinstance(message.details.finding.action.portprobeaction.portprobedetails, list):
+        # Propagate domain
+        if "miscinfo" in newmessage["details"]:
+            if "domain" in newmessage["details"]["miscinfo"]:
+                newmessage["details"]["query"] = newmessage["details"]["miscinfo"]["domain"]
 
-                # inspect the first list entry and see if it contains an IP
-                portprobedetails = DotDict(
-                    message.details.finding.action.portprobeaction.portprobedetails[0])
-                if key_exists('remoteipdetails.ipaddressv4', portprobedetails):
-                    message.details.sourceipaddress = portprobedetails.remoteipdetails.ipaddressv4
+        # Flatten tags
+        if "tags" in newmessage["details"]:
+            newmessage["details"]["awstags"] = []
+            for tagkve in newmessage["details"]["tags"]:
+                for k, v in tagkve.items():
+                    newmessage["details"]["awstags"].append(v.lower())
+            del newmessage["details"]["tags"]
 
-        # recovert the message back to a plain dict
-        return (dict(message), metadata)
+        # Find something that remotely resembles an FQDN
+        if "publicdnsname" in newmessage["details"]:
+            newmessage["hostname"] = newmessage["details"]["publicdnsname"]
+        elif "privatednsname" in newmessage["details"]:
+            newmessage["hostname"] = newmessage["details"]["privatednsname"]
+
+        # Flip IP addresses in we are the source of attacks
+        if (newmessage["details"]["finding"] == "UnauthorizedAccess:EC2/RDPBruteForce" or newmessage["details"]["finding"] == "UnauthorizedAccess:EC2/SSHBruteForce"):
+            if newmessage["details"]["direction"] == "OUTBOUND":
+                # could be more optimized here but need to be careful
+                truedstip = "0.0.0.0"
+                truesrcip = "0.0.0.0"
+                if "destinationipaddress" in newmessage["details"]:
+                    truedstip = newmessage["details"]["sourceipaddress"]
+                if "sourceipaddress" in newmessage["details"]:
+                    truesrcip = newmessage["details"]["destinationipaddress"]
+                newmessage["details"]["destinationipaddress"] = truedstip
+                newmessage["details"]["sourceipaddress"] = truesrcip
+                del newmessage["details"]["sourceport"]
+                del newmessage["details"]["destinationport"]
+
+        # Last resort in case we don't have any local IP address yet
+        # Fake it till you make it
+        attdir = {
+            "Recon:EC2/PortProbeUnprotectedPort": "INBOUND",
+            "CryptoCurrency:EC2/BitcoinTool.B!DNS": "INBOUND",
+            "Trojan:EC2/DGADomainRequest.B": "INBOUND",
+            "UnauthorizedAccess:IAMUser/TorIPCaller": "INBOUND",
+            "Persistence:IAMUser/ResourcePermissions": "INBOUND",
+            "Persistence:IAMUser/NetworkPermissions": "INBOUND",
+            "Persistence:IAMUser/UserPermissions": "INBOUND",
+        }
+        if "direction" not in newmessage["details"]:
+            newmessage["details"]["direction"] = attdir[newmessage["details"]["finding"]]
+        if newmessage["details"]["direction"] == "INBOUND":
+            if "destinationipaddress" not in newmessage["details"]:
+                if "publicip" in newmessage["details"]:
+                    newmessage["details"]["destinationipaddress"] = newmessage["details"]["publicip"]
+        if newmessage["details"]["direction"] == "OUTBOUND":
+            if "sourceipaddress" not in newmessage["details"]:
+                if "publicip" in newmessage["details"]:
+                    newmessage["details"]["sourceipaddress"] = newmessage["details"]["publicip"]
+
+        return (newmessage, metadata)