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,75 @@
+# -*- 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:
+            raise ValueError("need to have the domaintools API installed. See https://github.com/domaintools/python_api.")
+
+        if not self.parameters.user:
+            raise ValueError("need to specify user for domaintools expert in runtime.conf.")
+
+        if not self.parameters.password:
+            raise ValueError("need to specify password for the user for domaintools expert in runtime.conf.")
+
+        self.api = API(self.parameters.user, self.parameters.password)
+
+        if not self.valid_credentials():
+            raise ValueError("invalid credentials found in runtime.conf.")
+
+    def valid_credentials(self):
+        resp = self.api.reputation(fqdn, include_reasons=False)
+        try:
+            resp['risk_score']
+            return True
+        except exceptions.NotAuthorizedException:
+            return False
+
+    def get_score(self, fqdn):
+        # don't include a reason in the JSON response
+        resp = self.api.reputation(fqdn, include_reasons=False)
+
+        try:
+            score = resp['risk_score']
+        except exceptions.NotFoundException:
+            score = None
+        except exceptions.BadRequestException:
+            raise
+
+        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
+
+            score = self.get_score(event.get(key_fqdn))
+
+            if score:
+                extra["domaintools_score_" + key_fqdn] = score
+
+        extra.update(event.get("extra"))
+        event.update("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': 'example01', 'password': 'mysecret'}
+
+    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()