Add new bot: time filter

docs/user/bots.rst

@@ -3239,6 +3239,25 @@ This bot has certain limitations and is not a true threshold filter (yet). It wo
 Please note: Even if a message is sent, any further identical messages are dropped, if the time difference to the last message is less than the timeout! The counter is not reset if the threshold is reached.
 
 
+.. _intelmq.bots.experts.time_filter.expert:
+
+Time Filter
+^^^^^^^^^
+
+**Information**
+
+* `name:` `intelmq.bots.experts.time_filter.expert`
+* `lookup:` no
+* `public:` yes
+* `cache (redis db):` none
+* `description:` Time based filtering
+
+**Configuration Parameters**
+
+* `field`: `time.source`
+* `timespan`: `1d`
+
+
 .. _intelmq.bots.experts.tor_nodes.expert:
 
 Tor Nodes

intelmq/bots/experts/time_filter/expert.py

@@ -0,0 +1,58 @@
+# -*- coding: utf-8 -*-
+"""
+Time based filtering
+
+SPDX-FileCopyrightText: 2021 Marius Karotkis <marius.karotkis@gmail.com>
+SPDX-License-Identifier: AGPL-3.0-or-later
+"""
+
+from datetime import datetime, timedelta
+from dateutil import parser
+from intelmq.lib.bot import Bot
+from datetime import timezone
+from intelmq.lib.utils import get_timedelta
+
+
+class TimeFilterExpertBot(Bot):
+    """ Time based filtering """
+    field: str = 'time.source'
+    timespan: str = '1d'
+
+    __delta = None
+
+    def init(self):
+        if self.field:
+            timedelta_params = get_timedelta(self.timespan)
+            self.__delta = datetime.now(tz=timezone.utc) - timedelta(**timedelta_params)
+
+    def process(self):
+        event = self.receive_message()
+        event_time = self.__delta
+
+        if self.field in event:
+            try:
+                event_time = parser.parse(str(event.get(self.field)))
+            except ValueError:
+                self.process_message(event_time, event)
+                return
+            else:
+                self.process_message(event_time, event)
+                return
+        else:
+            # not found field
+            self.process_message(event_time, event)
+            return
+
+    def process_message(self, event_time, event):
+        event_time = event_time.replace(tzinfo=None)
+        self.__delta = self.__delta.replace(tzinfo=None)
+
+        if event_time > self.__delta:
+            self.send_message(event)
+        else:
+            self.logger.debug(f"Filtered out event with search field {self.field} and event time {event_time} .")
+
+        self.acknowledge_message()
+
+
+BOT = TimeFilterExpertBot

intelmq/lib/utils.py

@@ -856,7 +856,8 @@ def list_all_bots() -> dict:
 
     base_path = resource_filename('intelmq', 'bots')
 
-    botfiles = [botfile for botfile in pathlib.Path(base_path).glob('**/*.py') if botfile.is_file() and botfile.name != '__init__.py']
+    botfiles = [botfile for botfile in pathlib.Path(base_path).glob('**/*.py') if
+                botfile.is_file() and botfile.name != '__init__.py']
     for file in botfiles:
         file = Path(file.as_posix().replace(base_path, 'intelmq/bots'))
         mod = importlib.import_module('.'.join(file.with_suffix('').parts))
@@ -876,7 +877,8 @@ def list_all_bots() -> dict:
 
             bots[file.parts[2].capitalize()[:-1]][name] = {
                 "module": mod.__name__,
-                "description": "Missing description" if not getattr(mod.BOT, '__doc__', None) else textwrap.dedent(mod.BOT.__doc__).strip(),
+                "description": "Missing description" if not getattr(mod.BOT, '__doc__', None) else textwrap.dedent(
+                    mod.BOT.__doc__),
                 "parameters": keys,
             }
     return bots
@@ -916,3 +918,15 @@ def get_bots_settings(bot_id: str = None) -> dict:
     if 'global' in runtime_conf:
         del runtime_conf['global']
     return runtime_conf
+
+
+def get_timedelta(search_last):
+    m = re.search('(?P<number>^[1-9]+[0-9]*)(?P<map>[dmh])$', search_last)
+    if not m:
+        raise ValueError(f"ERROR: Incorrect search_last {search_last} parameter. Example 14h, 166d, 15m")
+    date_mapping = {
+        "d": "days",
+        "h": "hours",
+        "m": "minutes"
+    }
+    return {date_mapping[m.group('map')]: int(m.group("number"))}

intelmq/tests/bots/experts/time_filter/REQUIREMENTS.txt

@@ -0,0 +1,4 @@
+# SPDX-FileCopyrightText: 2021 Marius Karotkis <marius.karotkis@gmail.com>
+# SPDX-License-Identifier: AGPL-3.0-or-later
+
+time-machine

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

@@ -0,0 +1,121 @@
+# -*- coding: utf-8 -*-
+"""
+Time based filtering
+
+SPDX-FileCopyrightText: 2021 Marius Karotkis <marius.karotkis@gmail.com>
+SPDX-License-Identifier: AGPL-3.0-or-later
+"""
+
+import unittest
+
+import intelmq.lib.test as test
+from intelmq.bots.experts.time_filter.expert import TimeFilterExpertBot
+from intelmq.lib.exceptions import MissingDependencyError
+
+try:
+    import time_machine
+except ImportError:
+    time_machine = None
+
+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"
+}
+
+
+@test.skip_exotic()
+class TestFilterExpertBot(test.BotTestCase, unittest.TestCase):
+    """
+    A TestCase for TimeFilterExpertBot handling Reports.
+    """
+
+    @classmethod
+    def set_bot(cls):
+        if time_machine is None:
+            raise MissingDependencyError("time_machine")
+
+        cls.bot_reference = TimeFilterExpertBot
+        cls.input_message = EXAMPLE_INPUT_DROP
+        cls.sysconfig = {
+            'search_field': 'extra.lastonline',
+            'search_from': "1d"
+        }
+
+    @time_machine.travel("2021-05-05")
+    def test_expert_drop(self):
+        self.run_bot()
+        self.assertOutputQueueLen(0)
+
+    @time_machine.travel("2020-09-09")
+    def test_expert_pass(self):
+        self.input_message = EXAMPLE_INPUT_PASS
+        self.run_bot()
+        self.assertOutputQueueLen(1)
+
+    @time_machine.travel("2020-09-09")
+    def test_expert_pass_2(self):
+        self.input_message = EXAMPLE_INPUT_PASS_2
+        self.run_bot()
+        self.assertOutputQueueLen(1)
+
+    @time_machine.travel("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()