Add new bot: time filter

intelmq/bots/experts/time_filter/expert.py

@@ -0,0 +1,55 @@
+# -*- coding: utf-8 -*-
+from datetime import datetime, timedelta
+from dateutil import parser
+from intelmq.lib.utils import get_timedelta
+from intelmq.lib.bot import Bot
+from datetime import timezone
+
+
+class TimeFilterExpertBot(Bot):
+    search_field: str = 'time.source'
+    search_from: str  = '1d'
+
+    not_after = None
+
+    def init(self):
+        self.search_field = self.search_field
+        if self.search_from:
+            timedelta_params = get_timedelta(self.search_from)
+            self.not_after = datetime.now(tz=timezone.utc) - timedelta(**timedelta_params)
+
+    def process(self):
+        event = self.receive_message()
+        # time based filtering
+        if self.search_field in event:
+            try:
+                event_time = parser.parse(str(event.get(self.search_field)))
+            except ValueError:
+                event_time = self.not_after
+                self.process_message(event_time, event)
+                return
+            else:
+                self.process_message(event_time, event)
+                return
+        else:
+            # not found field
+            event_time = self.not_after
+            self.process_message(event_time, event)
+            return
+
+    def process_message(self, event_time, event):
+        event_time = event_time.replace(tzinfo=None)
+        self.not_after = self.not_after.replace(tzinfo=None)
+
+        if event_time < self.not_after:
+            self.acknowledge_message()
+            self.logger.debug(
+                f"Filtered out event with search field {self.search_field} and event time {event_time} .")
+            return
+        else:
+            self.send_message(event)
+            self.acknowledge_message()
+            return
+
+
+BOT = TimeFilterExpertBot

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

@@ -0,0 +1,106 @@
+# -*- coding: utf-8 -*-
+
+import unittest
+from freezegun import freeze_time
+
+import intelmq.lib.test as test
+from intelmq.bots.experts.time_filter.expert import TimeFilterExpertBot
+
+EXAMPLE_INPUT_DROP = {
+    "__type": "Event",
+    "feed.accuracy": 90.0,
+    "feed.name": "Feodo Tracker IPs",
+    "feed.provider": "abuse.ch",
+    "feed.url": "https://feodotracker.abuse.ch/downloads/ipblocklist.csv",
+    "time.observation": "2020-10-13T06:14:49+00:00",
+    "raw": "dGVzdA==",
+    "extra.firstseen": "2020-10-11T02:10:59+00:00",
+    "source.port": 447,
+    "extra.lastonline": "2020-08-13T00:00:00+00:00",
+    "malware.name": "trickbot",
+    "time.source": "2020-10-13T00:00:00+00:00"
+}
+EXAMPLE_INPUT_PASS = {
+    "__type": "Event",
+    "feed.accuracy": 90.0,
+    "feed.name": "Feodo Tracker IPs",
+    "feed.provider": "abuse.ch",
+    "feed.url": "https://feodotracker.abuse.ch/downloads/ipblocklist.csv",
+    "time.observation": "2020-10-13T06:14:49+00:00",
+    "raw": "dGVzdA==",
+    "extra.firstseen": "2020-10-11T02:10:59+00:00",
+    "source.port": 447,
+    "extra.lastonline1": "2020-09-13T00:00:00+00:00",
+    "malware.name": "trickbot",
+    "time.source": "2020-10-13T00:00:00+00:00"
+}
+EXAMPLE_INPUT_PASS_2 = {
+    "__type": "Event",
+    "feed.accuracy": 90.0,
+    "feed.name": "Feodo Tracker IPs",
+    "feed.provider": "abuse.ch",
+    "feed.url": "https://feodotracker.abuse.ch/downloads/ipblocklist.csv",
+    "time.observation": "2020-10-13T06:14:49+00:00",
+    "raw": "dGVzdA==",
+    "extra.firstseen": "2020-10-11T02:10:59+00:00",
+    "source.port": 447,
+    "extra.lastonline": "",
+    "malware.name": "trickbot",
+    "time.source": "2020-10-13T00:00:00+00:00"
+}
+EXAMPLE_INPUT_PASS_3 = {
+    "__type": "Event",
+    "feed.accuracy": 90.0,
+    "feed.name": "Feodo Tracker IPs",
+    "feed.provider": "abuse.ch",
+    "feed.url": "https://feodotracker.abuse.ch/downloads/ipblocklist.csv",
+    "time.observation": "2020-10-13T06:14:49+00:00",
+    "raw": "dGVzdA==",
+    "extra.firstseen": "2020-10-11T02:10:59+00:00",
+    "source.port": 447,
+    "extra.lastonline": "2020-09-13",
+    "malware.name": "trickbot",
+    "time.source": "2020-10-13T00:00:00+00:00"
+}
+
+
+class TestFilterExpertBot(test.BotTestCase, unittest.TestCase):
+    """
+    A TestCase for TimeFilterExpertBot handling Reports.
+    """
+
+    @classmethod
+    def set_bot(cls):
+        cls.bot_reference = TimeFilterExpertBot
+        cls.input_message = EXAMPLE_INPUT_DROP
+        cls.sysconfig = {
+            'search_field': 'extra.lastonline',
+            'search_from': "1d"
+        }
+
+    @freeze_time("2021-05-05")
+    def test_expert_drop(self):
+        self.run_bot()
+        self.assertOutputQueueLen(0)
+
+    @freeze_time("2020-09-09")
+    def test_expert_pass(self):
+        self.input_message = EXAMPLE_INPUT_PASS
+        self.run_bot()
+        self.assertOutputQueueLen(1)
+
+    @freeze_time("2020-09-09")
+    def test_expert_pass_2(self):
+        self.input_message = EXAMPLE_INPUT_PASS_2
+        self.run_bot()
+        self.assertOutputQueueLen(1)
+
+    @freeze_time("2020-09-09")
+    def test_expert_pass_3(self):
+        self.input_message = EXAMPLE_INPUT_PASS_3
+        self.run_bot()
+        self.assertOutputQueueLen(1)
+
+
+if __name__ == '__main__':  # pragma: no cover
+    unittest.main()