DomainTools Expert Bot (initial version)

intelmq/bots/BOTS

@@ -476,6 +476,14 @@
                 "redis_cache_ttl": "86400"
             }
         },
+        "Domaintools": {
+            "description": "Domaintools expert is a bot which queries domaintools.com for a scoring of a domain name",
+            "module": "intelmq.bots.experts.domaintools.expert",
+            "parameters": {
+                "user": "",
+                "password": ""
+            }
+        },
         "Field Reducer": {
             "description": "The field reducer bot is capable of removing fields from events.",
             "module": "intelmq.bots.experts.field_reducer.expert",

intelmq/bots/experts/domaintools/README.md

@@ -0,0 +1,13 @@
+# Domaintools expert
+
+This expert bot is an example on how to query domaintools.
+
+It does require an API from domaintools.
+
+Documentation on domaintools: https://www.domaintools.com/resources/api-documentation/
+Specifically, this bot can query a domain for the reputation score in domaintools: https://www.domaintools.com/resources/api-documentation/reputation/
+
+It will add the score in the extra field: ``extra.domaintools_score``.
+
+
+Authors: Juan Ortega Valiente, Aaron Kaplan

intelmq/bots/experts/domaintools/REQUIREMENTS.txt

@@ -0,0 +1 @@
+domaintools_api==0.1.7

intelmq/bots/experts/domaintools/expert.py

@@ -0,0 +1,58 @@
+# -*- coding: utf-8 -*-
+"""
+domaintools expert: query domtaintools.com to get a reputation score for a domain name
+
+"""
+from intelmq.lib.bot import Bot
+try:
+    from domaintools import API, exceptions
+except ImportError:
+    API = None
+
+
+class DomaintoolsExpertBot(Bot):
+
+    def init(self):
+        self.logger.info("Loading Domaintools expert.")
+        if (not API):
+            self.logger.exception("need to have the domaintools API installed. See https://github.com/domaintools/python_api")
+            self.stop()
+        if (not self.parameters.user):
+            self.logger.exception("need to specify user for domaintools expert in runtime.conf. Exiting")
+            self.stop()
+        if (not self.parameters.password):
+            self.logger.exception("need to specify password for the user for domaintools expert in runtime.conf. Exiting")
+            self.stop()
+        self.api = API(self.parameters.user, self.parameters.password)
+
+    def domaintools_get_score(self, fqdn):
+        score = None
+        if fqdn:
+            resp = self.api.reputation(fqdn, include_reasons=False)     # don't include a reason in the JSON response
+
+            try:
+                score = resp['risk_score']
+            except exceptions.NotFoundException:
+                score = None
+            except exceptions.BadRequestException:
+                score = None
+            return score
+
+    def process(self):
+        event = self.receive_message()
+        extra = {}
+
+        for key in ["source.", "destination."]:
+            key_fqdn = key + "fqdn"
+            if key_fqdn not in event:
+                continue        # can't query if we don't have a domain name
+            score = self.domaintools_get_score(event.get(key_fqdn))
+            if score is not None:
+                extra["domaintools_score"] = score
+                event.add("extra", extra)
+
+        self.send_message(event)
+        self.acknowledge_message()
+
+
+BOT = DomaintoolsExpertBot

intelmq/tests/bots/experts/domaintools/test_expert.py

@@ -0,0 +1,49 @@
+# -*- coding: utf-8 -*-
+"""
+Testing GethostbynameExpertBot.
+"""
+
+import unittest
+
+import intelmq.lib.test as test
+from intelmq.bots.experts.domaintools.expert import DomaintoolsExpertBot
+
+EXAMPLE_INPUT = {"__type": "Event",
+                 "source.fqdn": "google.com",
+                 "time.observation": "2015-01-01T00:00:00+00:00"
+                 }
+EXAMPLE_OUTPUT = {"__type": "Event",
+                  "source.fqdn": "google.com",
+                  "extra": '{"domaintools_score": 0}',
+                  "time.observation": "2015-01-01T00:00:00+00:00"
+                  }
+NONEXISTING_INPUT = {"__type": "Event",
+                     "source.fqdn": "example.invalid",
+                     "destination.fqdn": "example.invalid",
+                     "time.observation": "2015-01-01T00:00:00+00:00"
+                     }
+
+
+@test.skip_internet()
+class TestDomaintoolsExpertBot(test.BotTestCase, unittest.TestCase):
+    """
+    A TestCase for DomaintoolsExpertBot.
+    """
+
+    @classmethod
+    def set_bot(self):
+        self.bot_reference = DomaintoolsExpertBot
+        self.sysconfig = {'user': 'mkendrick_first2017', 'password': 'c0e4e-e2527-dc6af-824a4-229d5'}
+
+    def test_existing(self):
+        self.input_message = EXAMPLE_INPUT
+        self.run_bot()
+        self.assertMessageEqual(0, EXAMPLE_OUTPUT)
+
+    def test_non_existing(self):
+        self.input_message = NONEXISTING_INPUT
+        self.run_bot()
+        self.assertMessageEqual(0, NONEXISTING_INPUT)
+
+if __name__ == '__main__':  # pragma: no cover
+    unittest.main()