Add new bot: cut string by delimeter, max length

intelmq/bots/experts/string_cut_delimiter/expert.py

@@ -0,0 +1,29 @@
+# -*- coding: utf-8 -*-
+"""
+Cut string if length is bigger than max
+"""
+from intelmq.lib.bot import Bot
+
+
+class StringCutDelimiterExpertBot(Bot):
+    string_delimiter: str = '.'
+    max_length: int = 200
+    field_for_cut: str = 'source.fqdn'
+
+    def init(self):
+        pass
+
+    def process(self):
+        event = self.receive_message()
+
+        if self.field_for_cut in event:
+            long_string = event[self.field_for_cut]
+            while (long_string.find(self.string_delimiter) != -1) and (len(long_string) > self.max_length):
+                long_string = long_string.split(self.string_delimiter, 1)[1]
+            event.change(self.field_for_cut, long_string)
+
+        self.send_message(event)
+        self.acknowledge_message()
+
+
+BOT = StringCutDelimiterExpertBot

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

@@ -0,0 +1,73 @@
+# -*- coding: utf-8 -*-
+"""
+Testing string cut by delimiter bot
+"""
+import unittest
+import intelmq.lib.test as test
+from intelmq.bots.experts.string_cut_delimiter.expert import StringCutDelimiterExpertBot
+
+EXAMPLE_INPUT = {
+    '__type': 'Event',
+    'feed.accuracy': 100.0,
+    'feed.name': 'MISP events',
+    'feed.provider': 'MISP BAE',
+    'time.observation': '2020-10-20T12:57:33+00:00',
+    'feed.url': 'https://sig01.threatreveal.com',
+    'source.fqdn': 'bing.com.google.com.digikala.com.myket.com.divar.ir.varzesh3.pw.aparat.com.torojoonemadaretkarkonkhasteshodamdigeenqadtestzadam.filterchipedaramodarovordibekeshbiroon.dollarshode20000tomanbaskondigeh.salavatemohammadibefres.soltane-tel-injas-heh.digital',
+    'extra.elastic_index': 'cti-2020-10',
+    'extra.elastic_id': 'VwVnSnUBXjJtaqsUSw8T'}
+
+EXAMPLE_OUTPUT = {
+    '__type': 'Event',
+    'feed.accuracy': 100.0,
+    'feed.name': 'MISP events',
+    'feed.provider': 'MISP BAE',
+    'time.observation': '2020-10-20T12:57:33+00:00',
+    'feed.url': 'https://sig01.threatreveal.com',
+    'source.fqdn': 'pw.aparat.com.torojoonemadaretkarkonkhasteshodamdigeenqadtestzadam.filterchipedaramodarovordibekeshbiroon.dollarshode20000tomanbaskondigeh.salavatemohammadibefres.soltane-tel-injas-heh.digital',
+    'extra.elastic_index': 'cti-2020-10',
+    'extra.elastic_id': 'VwVnSnUBXjJtaqsUSw8T'}
+
+EXAMPLE_INPUT_2 = {
+    '__type': 'Event',
+    'feed.accuracy': 100.0,
+    'feed.name': 'MISP events',
+    'feed.provider': 'MISP BAE',
+    'time.observation': '2020-10-20T12:57:33+00:00',
+    'feed.url': 'https://sig01.threatreveal.com',
+    'extra.elastic_index': 'cti-2020-10',
+    'extra.elastic_id': 'VwVnSnUBXjJtaqsUSw8T'}
+
+EXAMPLE_OUTPUT_2 = {
+    '__type': 'Event',
+    'feed.accuracy': 100.0,
+    'feed.name': 'MISP events',
+    'feed.provider': 'MISP BAE',
+    'time.observation': '2020-10-20T12:57:33+00:00',
+    'feed.url': 'https://sig01.threatreveal.com',
+    'extra.elastic_index': 'cti-2020-10',
+    'extra.elastic_id': 'VwVnSnUBXjJtaqsUSw8T'}
+
+
+class TestStringCutDelimiterExpertBot(test.BotTestCase, unittest.TestCase):
+    """
+    A TestCase for TestStringCutDelimiterExpertBot.
+    """
+
+    @classmethod
+    def set_bot(cls):
+        cls.bot_reference = StringCutDelimiterExpertBot
+
+    def test_event_cut(self):
+        self.input_message = EXAMPLE_INPUT
+        self.run_bot()
+        self.assertMessageEqual(0, EXAMPLE_OUTPUT)
+
+    def test_event_cut_without_field(self):
+        self.input_message = EXAMPLE_INPUT_2
+        self.run_bot()
+        self.assertMessageEqual(0, EXAMPLE_OUTPUT_2)
+
+
+if __name__ == '__main__':  # pragma: no cover
+    unittest.main()