Add ESET STIX Parser bot

laciKE · laciKE · commit 15b678938ddc · 2025-05-03T01:43:25.000+02:00
Parser bot for enriching events from ESET Threat Intelligence, which were collected by TaxiiCollectorBot. It inherits from generic StixParserBot and implement vendor-specific parsing. ESET STIX Parser bot analyzes comment (based on original description of STIX Indicator object) and choose proper classification type and if possible, also fills the malware.name in the event.
diff --git a/intelmq/bots/parsers/stix/parser.py b/intelmq/bots/parsers/stix/parser.py
@@ -35,12 +35,21 @@ def parse_line(self, line, report):
                 indicator = self.parse_stix_pattern(pattern)
                 if indicator:
                     event.add(indicator[0], indicator[1])
+                    self.parse_vendor_specific(event, line, report)
                     yield event
             else:
                 self.logger.warning('Unexpected type of pattern expression: %r, pattern: %r', pattern_type, pattern)
         else:
             self.logger.warning('Unexpected type of STIX object: %r', object_type)
 
+    def parse_vendor_specific(self, event, line, report):
+        """
+        Parse vendor specific details from the STIX 2.1 Indicator object.
+        This method by default does nothing and it is called just before IntelMQ event is yielded.
+        If we need vendor-specific STIX parser, we can inherit from this class and override this one method.
+        """
+        return
+
     @staticmethod
     def parse_stix_pattern(pattern):
         """
diff --git a/intelmq/bots/parsers/stix/parser_eset.py b/intelmq/bots/parsers/stix/parser_eset.py
@@ -0,0 +1,111 @@
+# SPDX-FileCopyrightText: 2025 Ladislav Baco
+#
+# SPDX-License-Identifier: AGPL-3.0-or-later
+
+# -*- coding: utf-8 -*-
+"""
+Parser bot for ESET Threat Intelligence feeds
+This bot parses STIX Indicators objects received from TAXII collector
+Then it analyzes event's comments based on STIX indicator's description
+and it adds classification.type and malware family info
+It is recommended to apply TaxonomyExpertBot then to map the taxonomy
+"""
+
+import re
+
+from intelmq.bots.parsers.stix.parser import StixParserBot
+
+
+CLASSIFICATION_BY_STRING = {
+    'Host actively distributes high-severity malicious content in the form of executable code.': 'malware-distribution',
+    'Host actively distributes high-severity threat in the form of executable code.': 'malware-distribution',
+    'Host actively distributes high-severity threat in the form of malicious code.': 'malware-distribution',
+    'Host actively distributes high-severity threat in the form of script code.': 'malware-distribution',
+    'Host is known to be actively distributing adware or other medium-risk software.': 'malware-distribution',
+    'Host is known to be actively distributing high-severity mobile threats or low-risk software.': 'other',
+    'Host is known to be actively distributing threats or is of uncertain reputation.': 'other',
+    'Host is known to be distributing low-risk and potentially unwanted content.': 'other',
+    'Host actively distributes potentially unwanted or unsafe threat.': 'other',
+    'Host is known source of phishing or other fraudulent content.': 'phishing',
+    'Host is known source of active fraudulent content.': 'other',
+    'Host is used as command and control server.': 'c2-server',
+    'Web services scanning and attacks': 'scanner',
+    'RDP bruteforce IP': 'brute-force',
+    'SQL bruteforce IP': 'brute-force',
+    'SMB bruteforce IP': 'brute-force',
+    'MySQL bruteforce IP': 'brute-force',
+    'FTP bruteforce IP': 'brute-force'
+}
+
+CLASSIFICATION_REGEX = {
+    'C&C indicates that a botnet ([^ ]+) ([^ ]+) is present.': 'c2-server',
+    'C&C of ([^ ]+) ([^ ]+)': 'c2-server',
+    'Host is used as command and control server of ([^ ]+) ([^ ]+) malware family.': 'c2-server',
+    'Each of these file hashes indicates that a variant of ([^ ]+) ([^ ]+) is present.': 'malware',
+    '^[.* ]?([^ ]+) C&C server.*$': 'c2-server',
+    '^[.* ]?([^ ]+) backdoor.*$': 'malware',
+    '^[.* ]?([^ ]+) trojan.*$': 'malware',
+    '^[.* ]?([^ ]+) implant.*$': 'malware',
+    'Loader for ([^ ]+).*$': 'malware'
+}
+
+CLASSIFICATION_BY_REGEX = {}
+for (regex, classification_type) in CLASSIFICATION_REGEX.items():
+    CLASSIFICATION_BY_REGEX[re.compile(regex)] = classification_type
+
+
+class ESETStixParserBot(StixParserBot):
+    """Add classification.type and malware family to events"""
+
+    # Platform/Type.Family.Variant!Suffixes
+    # Type and suffixes are optional
+    _malware_naming_convention_pattern = re.compile(r'^([^/]*/)?([^\.]*\.)?([^\.]+)(\.[^!]*)(!.*)?$')
+
+    def parse_vendor_specific(self, event, line, report):
+        classification_type = event.get('classification.type', 'undetermined')
+        if classification_type == 'undetermined':
+            comment = event.get('comment', '')
+            classification_type, malware_name = self.classify(comment)
+            event.add('classification.type', classification_type, overwrite=True)
+            if malware_name:
+                event.add('malware.name', malware_name)
+        else:
+            # classification.type already present, do not change it
+            pass
+
+    @staticmethod
+    def classify(comment):
+        """ Classify comment and returns (classification_type, malware_name) """
+        classification_type = CLASSIFICATION_BY_STRING.get(comment, None)
+        if classification_type:
+            malware_name = None
+            return (classification_type, malware_name)
+
+        for (pattern, classification_type) in CLASSIFICATION_BY_REGEX.items():
+            match = pattern.match(comment)
+            if match:
+                malware_name = None
+                groups = match.groups()
+                if len(groups) > 0:
+                    malware = groups[0]
+                    malware_name = ESETStixParserBot.extract_malware_family(malware)
+                return (classification_type, malware_name)
+
+        return ('undetermined', None)
+
+    @staticmethod
+    def extract_malware_family(malware):
+        """ Extract malware family from the threat detection string """
+
+        match = ESETStixParserBot._malware_naming_convention_pattern.match(malware)
+        if match and len(match.groups()) == 5:
+            malware_name = match.groups()[2]
+        else:
+            # usually just malware family (or unknown naming convention)
+            malware_name = malware
+
+        # IntelMQ malware.name should be lowercase
+        return malware_name.lower()
+
+
+BOT = ESETStixParserBot
diff --git a/intelmq/tests/bots/parsers/stix/test_parser_bot.py b/intelmq/tests/bots/parsers/stix/test_parser_bot.py
@@ -27,17 +27,17 @@
                   }
 
 EXAMPLE_EVENT = {'__type': 'Event',
-                  'feed.name': 'Taxii Feed',
-                  'feed.code': 'feed stix2.1',
-                  'feed.provider': 'Taxii Provider',
-                  'feed.documentation': 'Taxii Documentation',
-                  'feed.accuracy': 100.0,
-                  'feed.url': 'http://localhost/feed',
-                  'source.url': 'http://example.org',
-                  'time.source': '1970-01-01T00:00:00+00:00',
-                  'classification.type': 'undetermined',
-                  'raw': 'eyJpZCI6ICJpbmRpY2F0b3ItLTAiLCAidHlwZSI6ICJpbmRpY2F0b3IiLCAic3BlY192ZXJzaW9uIjogIjIuMSIsICJjcmVhdGVkIjogIjE5NzAtMDEtMDFUMDA6MDA6MDAuMDAwWiIsICJtb2RpZmllZCI6ICIxOTcwLTAxLTAxVDAwOjAwOjAwLjAwMFoiLCAicGF0dGVybiI6ICJbdXJsOnZhbHVlID0gJ2h0dHA6Ly9leGFtcGxlLm9yZyddIiwgInBhdHRlcm5fdHlwZSI6ICJzdGl4IiwgInZhbGlkX2Zyb20iOiAiMTk3MC0wMS0wMVQwMDowMDowMFoifQ=='
-                  }
+                 'feed.name': 'Taxii Feed',
+                 'feed.code': 'feed stix2.1',
+                 'feed.provider': 'Taxii Provider',
+                 'feed.documentation': 'Taxii Documentation',
+                 'feed.accuracy': 100.0,
+                 'feed.url': 'http://localhost/feed',
+                 'source.url': 'http://example.org',
+                 'time.source': '1970-01-01T00:00:00+00:00',
+                 'classification.type': 'undetermined',
+                 'raw': 'eyJpZCI6ICJpbmRpY2F0b3ItLTAiLCAidHlwZSI6ICJpbmRpY2F0b3IiLCAic3BlY192ZXJzaW9uIjogIjIuMSIsICJjcmVhdGVkIjogIjE5NzAtMDEtMDFUMDA6MDA6MDAuMDAwWiIsICJtb2RpZmllZCI6ICIxOTcwLTAxLTAxVDAwOjAwOjAwLjAwMFoiLCAicGF0dGVybiI6ICJbdXJsOnZhbHVlID0gJ2h0dHA6Ly9leGFtcGxlLm9yZyddIiwgInBhdHRlcm5fdHlwZSI6ICJzdGl4IiwgInZhbGlkX2Zyb20iOiAiMTk3MC0wMS0wMVQwMDowMDowMFoifQ=='
+                 }
 
 
 class TestStixParserBot(test.BotTestCase, unittest.TestCase):
@@ -56,7 +56,6 @@ def test_event(self):
         self.run_bot()
         self.assertMessageEqual(0, EXAMPLE_EVENT)
 
-
     def test_pattern_url(self):
         """ Test if url pattern is parsed. """
         indicator = self.bot_reference.parse_stix_pattern("[url:value = 'http://example.org']")
diff --git a/intelmq/tests/bots/parsers/stix/test_parser_eset_bot.py b/intelmq/tests/bots/parsers/stix/test_parser_eset_bot.py
@@ -0,0 +1,125 @@
+# SPDX-FileCopyrightText: 2025 Ladislav Baco
+#
+# SPDX-License-Identifier: AGPL-3.0-or-later
+
+# -*- coding: utf-8 -*-
+"""
+Test with example reports (STIX objects usually collected from TAXII server)
+"""
+import unittest
+
+import re
+import requests_mock
+
+import intelmq.lib.bot as bot
+import intelmq.lib.test as test
+from intelmq.bots.parsers.stix.parser_eset import ESETStixParserBot
+
+
+EXAMPLE_REPORT = {'__type': 'Report',
+                  'feed.name': 'Botnet feed',
+                  'feed.code': 'botnet stix 2.1',
+                  'feed.provider': 'ESET',
+                  'feed.documentation': 'https://help.eset.com/eti_portal/en-US/botnet-feed.',
+                  'feed.accuracy': 100.0,
+                  'feed.url': 'https://taxii.eset.com/taxii2/643f4eb5-f8b7-46a3-a606-6d61d5ce223a/collections/0abb06690b0b47e49cd7794396b76b20/',
+                  'raw': 'eyJpZCI6ICJpbmRpY2F0b3ItLTAiLCAidHlwZSI6ICJpbmRpY2F0b3IiLCAic3BlY192ZXJzaW9uIjogIjIuMSIsICJjcmVhdGVkIjogIjE5NzAtMDEtMDFUMDA6MDA6MDAuMDAwWiIsICJtb2RpZmllZCI6ICIxOTcwLTAxLTAxVDAwOjAwOjAwLjAwMFoiLCAicGF0dGVybiI6ICJbdXJsOnZhbHVlID0gJ2h0dHA6Ly9leGFtcGxlLm9yZyddIiwgInBhdHRlcm5fdHlwZSI6ICJzdGl4IiwgInZhbGlkX2Zyb20iOiAiMTk3MC0wMS0wMVQwMDowMDowMFoiLCAiZGVzY3JpcHRpb24iOiAiQyZDIGluZGljYXRlcyB0aGF0IGEgYm90bmV0IFdpbjMyL1NweS5MdW1tYVN0ZWFsZXIuQiB0cm9qYW4gaXMgcHJlc2VudC4iLCAibGFiZWxzIjogWyJtYWxpY2lvdXMtYWN0aXZpdHkiXX0='
+                  }
+
+EXAMPLE_EVENT = {'__type': 'Event',
+                 'feed.name': 'Botnet feed',
+                 'feed.code': 'botnet stix 2.1',
+                 'feed.provider': 'ESET',
+                 'feed.documentation': 'https://help.eset.com/eti_portal/en-US/botnet-feed.',
+                 'feed.accuracy': 100.0,
+                 'feed.url': 'https://taxii.eset.com/taxii2/643f4eb5-f8b7-46a3-a606-6d61d5ce223a/collections/0abb06690b0b47e49cd7794396b76b20/',
+                 'source.url': 'http://example.org',
+                 'time.source': '1970-01-01T00:00:00+00:00',
+                 'classification.type': 'c2-server',
+                 'malware.name': 'lummastealer',
+                 'comment': 'C&C indicates that a botnet Win32/Spy.LummaStealer.B trojan is present.',
+                 'extra.labels': ['malicious-activity'],
+                 'raw': 'eyJpZCI6ICJpbmRpY2F0b3ItLTAiLCAidHlwZSI6ICJpbmRpY2F0b3IiLCAic3BlY192ZXJzaW9uIjogIjIuMSIsICJjcmVhdGVkIjogIjE5NzAtMDEtMDFUMDA6MDA6MDAuMDAwWiIsICJtb2RpZmllZCI6ICIxOTcwLTAxLTAxVDAwOjAwOjAwLjAwMFoiLCAicGF0dGVybiI6ICJbdXJsOnZhbHVlID0gJ2h0dHA6Ly9leGFtcGxlLm9yZyddIiwgInBhdHRlcm5fdHlwZSI6ICJzdGl4IiwgInZhbGlkX2Zyb20iOiAiMTk3MC0wMS0wMVQwMDowMDowMFoiLCAiZGVzY3JpcHRpb24iOiAiQyZDIGluZGljYXRlcyB0aGF0IGEgYm90bmV0IFdpbjMyL1NweS5MdW1tYVN0ZWFsZXIuQiB0cm9qYW4gaXMgcHJlc2VudC4iLCAibGFiZWxzIjogWyJtYWxpY2lvdXMtYWN0aXZpdHkiXX0='
+                 }
+
+
+class TestESETStixParserBot(test.BotTestCase, unittest.TestCase):
+    """
+    A TestCase for an ESETStixParserBot.
+    """
+
+    @classmethod
+    def set_bot(cls):
+        cls.bot_reference = ESETStixParserBot
+        cls.sysconfig = {}
+
+    def test_event(self):
+        """ Test if correct Event has been produced. """
+        self.input_message = EXAMPLE_REPORT
+        self.run_bot()
+        self.assertMessageEqual(0, EXAMPLE_EVENT)
+
+    def test_classification_by_string(self):
+        """ Test if correct classification based on string is returned. """
+        classification_type, malware_name = self.bot_reference.classify('Host actively distributes high-severity malicious content in the form of executable code.')
+        self.assertEqual(str(classification_type), 'malware-distribution')
+        self.assertEqual(malware_name, None)
+
+        classification_type, malware_name = self.bot_reference.classify('Host is known source of phishing or other fraudulent content.')
+        self.assertEqual(str(classification_type), 'phishing')
+        self.assertEqual(malware_name, None)
+
+        classification_type, malware_name = self.bot_reference.classify('Host is used as command and control server.')
+        self.assertEqual(str(classification_type), 'c2-server')
+        self.assertEqual(malware_name, None)
+
+        classification_type, malware_name = self.bot_reference.classify('Web services scanning and attacks')
+        self.assertEqual(str(classification_type), 'scanner')
+        self.assertEqual(malware_name, None)
+
+        classification_type, malware_name = self.bot_reference.classify('RDP bruteforce IP')
+        self.assertEqual(str(classification_type), 'brute-force')
+        self.assertEqual(malware_name, None)
+
+    def test_classification_by_regex(self):
+        """ Test if correct classification based on regex is returned. """
+        classification_type, malware_name = self.bot_reference.classify('C&C indicates that a botnet Win32/Spy.LummaStealer.B trojan is present.')
+        self.assertEqual(str(classification_type), 'c2-server')
+        self.assertEqual(str(malware_name), 'lummastealer')
+
+        classification_type, malware_name = self.bot_reference.classify('C&C of Win32/Spy.LummaStealer.B trojan')
+        self.assertEqual(str(classification_type), 'c2-server')
+        self.assertEqual(str(malware_name), 'lummastealer')
+
+        classification_type, malware_name = self.bot_reference.classify('Host is used as command and control server of Win32/Emotet.BN trojan malware family.')
+        self.assertEqual(str(classification_type), 'c2-server')
+        self.assertEqual(str(malware_name), 'emotet')
+
+        classification_type, malware_name = self.bot_reference.classify('WizardNet backdoor.')
+        self.assertEqual(str(classification_type), 'malware')
+        self.assertEqual(str(malware_name), 'wizardnet')
+
+        classification_type, malware_name = self.bot_reference.classify('Loader for Emotet')
+        self.assertEqual(str(classification_type), 'malware')
+        self.assertEqual(str(malware_name), 'emotet')
+
+    def test_unknown_classification(self):
+        """ Test if undetermined classification is returned when comment contains something unexpected. """
+        classification_type, malware_name = self.bot_reference.classify('Example of unexpected comment.')
+        self.assertEqual(str(classification_type), 'undetermined')
+        self.assertEqual(malware_name, None)
+
+    def test_malware_family_name_extraction(self):
+        """ Test if correct malwae family name is extracted from the given malware string. """
+        malware_name = self.bot_reference.extract_malware_family('Win32/Spy.LummaStealer.B')
+        self.assertEqual(str(malware_name), 'lummastealer')
+
+        malware_name = self.bot_reference.extract_malware_family('Win32/Rescoms.B')
+        self.assertEqual(str(malware_name), 'rescoms')
+
+        malware_name = self.bot_reference.extract_malware_family('Emotet')
+        self.assertEqual(str(malware_name), 'emotet')
+
+
+if __name__ == '__main__':  # pragma: no cover
+    unittest.main()
