Improve STIX patterns parsing

Use the official STIX2 Pattern Validator to get thecomparison expressions and extracts simple IoCs from them. Support for URLs, Domains, IPv4, IPv6 and also for MD5, SHA-1 and SHA-256 hashes. Small fixes and workarounds implemented to address certain anomalies in STIX data provided by some vendors (e.g. ETI) - SHA1 and SHA256 keywords accepted, invalid objects reported as Domains or URLs are dropped without throwing the exceptions

Author: laciKE

intelmq/bots/parsers/stix/REQUIREMENTS.txt

@@ -0,0 +1,4 @@
+# SPDX-FileCopyrightText: 2025 Ladislav Baco
+# SPDX-License-Identifier: AGPL-3.0-or-later
+
+stix2-patterns>=2.0.0

intelmq/bots/parsers/stix/parser.py

@@ -7,34 +7,48 @@
 
 import json
 
+
 from intelmq.lib.bot import ParserBot
 
+try:
+    import stix2patterns.v21.pattern as stix2_pattern
+except ImportError:
+    stix2_pattern = None
+
 
 class StixParserBot(ParserBot):
     """Parse STIX indicators"""
     parse = ParserBot.parse_json_stream
     recover_line = ParserBot.recover_line_json_stream
 
+    def init(self):
+        if stix2_pattern is None:
+            raise MissingDependencyError('stix2-patterns')
+
     def parse_line(self, line, report):
         """ Parse one STIX object of indicator type """
         object_type = line.get('type', '')
         if object_type == 'indicator':
-            event = self.new_event(report)
-            event.add('raw', json.dumps(line))
-            event.add('comment', line.get('description', ''))
-            event.add('extra.labels', line.get('labels', None))
-            event.add('time.source', line.get('valid_from', '1970-01-01T00:00:00Z'))
-            # classification will be determined by expert bot specific for given TAXII collection
-            event.add('classification.type', 'undetermined')
-
             pattern = line.get('pattern', '')
             # stix, pcre, sigma, snort, suricata, yara
             pattern_type = line.get('pattern_type', '')
 
             if pattern_type == 'stix':
-                indicator = self.parse_stix_pattern(pattern)
-                if indicator:
-                    event.add(indicator[0], indicator[1])
+                indicators = StixParserBot.parse_stix_pattern(pattern, self.logger)
+                for indicator_type, indicator_value in indicators:
+                    event = self.new_event(report)
+                    event.add('raw', json.dumps(line))
+                    event.add('comment', line.get('description', ''))
+                    event.add('extra.labels', line.get('labels', None))
+                    event.add('time.source', line.get('valid_from', '1970-01-01T00:00:00Z'))
+
+                    # IP address may be passed in Domain feeds or Domain may be passed in URL feeds
+                    # It violates the STIX format, however, in some sources it happens (e.g. in ETI)
+                    # Drop such events without failures and exceptions which slowing down the processing
+                    event.add(indicator_type, indicator_value, raise_failure=False)
+
+                    # classification can be overridden by vendor-specific parser below
+                    event.add('classification.type', 'undetermined')
                     self.parse_vendor_specific(event, line, report)
                     yield event
             else:
@@ -51,38 +65,84 @@ def parse_vendor_specific(self, event, line, report):
         return
 
     @staticmethod
-    def parse_stix_pattern(pattern):
+    def _get_value_from_comparison_expression(comparison, logger=None):
+        """
+        STIX Comparison Expressions:
+        https://docs.oasis-open.org/cti/stix/v2.1/os/stix-v2.1-os.html#_boiciucr9smf
+
+        comparison is a tuple obtained from stix2patterns.v21.pattern.Pattern(pattern).inspect().comparisons,
+        e.g. (['value'], '=', "'http://example.org'"), (['value'], '=', "'127.0.0.1/32'")
+        """
+        if len(comparison) != 3:
+            if logger:
+                logger.warning('Unexpected Comparison Expressions. Expression: {}'.format(comparison))
+            return
+
+        property_name, operator, value = comparison
+        supported_property_names = [['value'],
+                                    ['hashes', 'MD5'],
+                                    ['hashes', 'SHA-1'],
+                                    ['hashes', 'SHA-256'],
+                                    # Based on 10.7 Hashing Algorithm Vocabulary, these keys are not valid, but they are used in some feeds (e.g. ETI)
+                                    # https://docs.oasis-open.org/cti/stix/v2.1/os/stix-v2.1-os.html#_ths0b11wzxv3
+                                    ['hashes', 'SHA1'],
+                                    ['hashes', 'SHA256']
+                                    ]
+        if not (property_name in supported_property_names) or (operator != '=') or not value.startswith("'") or not value.endswith("'"):
+            if logger:
+                logger.info('Unsupported Comparison Expression. Only Comparison Expressions with "equal" comparison operator and "value" or "hashes" property are supported. Expression: {}'.format(comparison))
+            return
+
+        # remove single quotes from returned value
+        return value[1:-1]
+
+    @staticmethod
+    def parse_stix_pattern(pattern, logger=None):
         """
         STIX Patterning:
         https://docs.oasis-open.org/cti/stix/v2.1/os/stix-v2.1-os.html#_e8slinrhxcc9
         """
-        if pattern.count('[') != 1:
-            print('Unsupported Pattern Expression. Only single Observation Expression is supported. Pattern: {}'.format(pattern))
-            return
 
-        value = pattern.split("'")[1]
-        if pattern.startswith('[url:value = '):
-            return ('source.url', value)
-        if pattern.startswith('[domain-name:value = '):
-            return ('source.fqdn', value)
-        if pattern.startswith('[ipv4-addr:value = '):
-            # remove port, sometimes the port is present in ETI
-            value = value.split(':')[0]
-            # strip CIDR if IPv4 network contains single host only
-            value = value[:-3] if value.endswith('/32') else value
-            # check if pattern is in CIDR notation
-            if value.rfind('/') > -1:
-                return ('source.network', value)
-            else:
-                return ('source.ip', value)
-        if pattern.startswith('[ipv6-addr:value = '):
-            # strip CIDR if IPv6 network contains single host only
-            value = value[:-4] if value.endswith('/128') else value
-            # check if pattern is in CIDR notation
-            if value.rfind('/') > -1:
-                return ('source.network', value)
-            else:
-                return ('source.ip', value)
+        indicators = []
+        comparisons = stix2_pattern.Pattern(pattern).inspect().comparisons
+        for key in comparisons.keys():
+            comparison_expressions = comparisons.get(key, [])
+            for comparison in comparison_expressions:
+                value = StixParserBot._get_value_from_comparison_expression(comparison, logger)
+                if not value:
+                    pass
+                if key == 'url':
+                    indicators.append(('source.url', value))
+                elif key == 'domain-name':
+                    indicators.append(('source.fqdn', value))
+                elif key == 'ipv4-addr':
+                    # remove port, sometimes the port is present in ETI
+                    value = value.split(':')[0]
+                    # strip CIDR if IPv4 network contains single host only
+                    value = value[:-3] if value.endswith('/32') else value
+                    # check if pattern is in CIDR notation
+                    if value.rfind('/') > -1:
+                        indicators.append(('source.network', value))
+                    else:
+                        indicators.append(('source.ip', value))
+                elif key == 'ipv6-addr':
+                    # strip CIDR if IPv6 network contains single host only
+                    value = value[:-4] if value.endswith('/128') else value
+                    # check if pattern is in CIDR notation
+                    if value.rfind('/') > -1:
+                        indicators.append(('source.network', value))
+                    else:
+                        indicators.append(('source.ip', value))
+                elif key == 'file':
+                    if len(comparison) == 3 and len(comparison[0]) == 2 and comparison[0][0] == 'hashes':
+                        # converts MD5, SHA-1, SHA1, SHA-256, SHA256 to md5, sha1, sha256 used in IntelMQ
+                        hash_algo = comparison[0][1].lower().replace('-', '')
+                        indicators.append(('malware.hash.' + hash_algo, value))
+                else:
+                    if logger:
+                        logger.warning('Unsupported Object Type "{}" in Pattern Expression. Pattern: {}'.format(key, pattern))
+
+        return indicators
 
 
 BOT = StixParserBot

intelmq/tests/bots/parsers/stix/test_parser_bot.py

@@ -40,6 +40,7 @@
                  }
 
 
+@test.skip_exotic()
 class TestStixParserBot(test.BotTestCase, unittest.TestCase):
     """
     A TestCase for a StixParserBot.
@@ -58,52 +59,94 @@ def test_event(self):
 
     def test_pattern_url(self):
         """ Test if url pattern is parsed. """
-        indicator = self.bot_reference.parse_stix_pattern("[url:value = 'http://example.org']")
+        indicator = self.bot_reference.parse_stix_pattern("[url:value = 'http://example.org']")[0]
         self.assertEqual(str(indicator[0]), 'source.url')
         self.assertEqual(str(indicator[1]), 'http://example.org')
 
-    def test_pattern_url(self):
+    def test_pattern_domain(self):
         """ Test if domain pattern is parsed. """
-        indicator = self.bot_reference.parse_stix_pattern("[domain-name:value = 'example.org']")
+        indicator = self.bot_reference.parse_stix_pattern("[domain-name:value = 'example.org']")[0]
         self.assertEqual(str(indicator[0]), 'source.fqdn')
         self.assertEqual(str(indicator[1]), 'example.org')
 
     def test_pattern_ipv4(self):
         """ Test if ipv4 pattern is parsed. """
-        indicator = self.bot_reference.parse_stix_pattern("[ipv4-addr:value = '127.0.0.1']")
+        indicator = self.bot_reference.parse_stix_pattern("[ipv4-addr:value = '127.0.0.1']")[0]
         self.assertEqual(str(indicator[0]), 'source.ip')
         self.assertEqual(str(indicator[1]), '127.0.0.1')
 
     def test_pattern_ipv4_cidr(self):
         """ Test if ipv4 cidr pattern is parsed. """
-        indicator = self.bot_reference.parse_stix_pattern("[ipv4-addr:value = '127.0.0.0/8']")
+        indicator = self.bot_reference.parse_stix_pattern("[ipv4-addr:value = '127.0.0.0/8']")[0]
         self.assertEqual(str(indicator[0]), 'source.network')
         self.assertEqual(str(indicator[1]), '127.0.0.0/8')
 
     def test_pattern_ipv4_cidr_single_host(self):
         """ Test if ipv4 cidr with single host pattern is parsed. """
-        indicator = self.bot_reference.parse_stix_pattern("[ipv4-addr:value = '127.0.0.1/32']")
+        indicator = self.bot_reference.parse_stix_pattern("[ipv4-addr:value = '127.0.0.1/32']")[0]
         self.assertEqual(str(indicator[0]), 'source.ip')
         self.assertEqual(str(indicator[1]), '127.0.0.1')
 
     def test_pattern_ipv6(self):
         """ Test if ipv6 pattern is parsed. """
-        indicator = self.bot_reference.parse_stix_pattern("[ipv6-addr:value = '::1']")
+        indicator = self.bot_reference.parse_stix_pattern("[ipv6-addr:value = '::1']")[0]
         self.assertEqual(str(indicator[0]), 'source.ip')
         self.assertEqual(str(indicator[1]), '::1')
 
     def test_pattern_ipv6_cidr(self):
         """ Test if ipv6 cidr pattern is parsed. """
-        indicator = self.bot_reference.parse_stix_pattern("[ipv6-addr:value = 'fe:80::/10']")
+        indicator = self.bot_reference.parse_stix_pattern("[ipv6-addr:value = 'fe:80::/10']")[0]
         self.assertEqual(str(indicator[0]), 'source.network')
         self.assertEqual(str(indicator[1]), 'fe:80::/10')
 
     def test_pattern_ipv6_cidr_single_host(self):
         """ Test if ipv6 cidr with single host pattern is parsed. """
-        indicator = self.bot_reference.parse_stix_pattern("[ipv6-addr:value = 'fe:80::1/128']")
+        indicator = self.bot_reference.parse_stix_pattern("[ipv6-addr:value = 'fe:80::1/128']")[0]
         self.assertEqual(str(indicator[0]), 'source.ip')
         self.assertEqual(str(indicator[1]), 'fe:80::1')
 
+    def test_pattern_hash_md5(self):
+        """ Test if domain pattern is parsed. """
+        indicator = self.bot_reference.parse_stix_pattern("[file:hashes.MD5 = '44d88612fea8a8f36de82e1278abb02f']")[0]
+        self.assertEqual(str(indicator[0]), 'malware.hash.md5')
+        self.assertEqual(str(indicator[1]), '44d88612fea8a8f36de82e1278abb02f')
+
+    def test_pattern_hash_sha1(self):
+        """ Test if domain pattern is parsed. """
+        indicator = self.bot_reference.parse_stix_pattern("[file:hashes.'SHA-1' = '3395856ce81f2b7382dee72602f798b642f14140']")[0]
+        self.assertEqual(str(indicator[0]), 'malware.hash.sha1')
+        self.assertEqual(str(indicator[1]), '3395856ce81f2b7382dee72602f798b642f14140')
+
+        # Based on 10.7 Hashing Algorithm Vocabulary, keys SHA1 and SHA256 are not valid, but they are used in some feeds (e.g. ETI)
+        # https://docs.oasis-open.org/cti/stix/v2.1/os/stix-v2.1-os.html#_ths0b11wzxv3
+        indicator = self.bot_reference.parse_stix_pattern("[file:hashes.SHA1 = '3395856ce81f2b7382dee72602f798b642f14140']")[0]
+        self.assertEqual(str(indicator[0]), 'malware.hash.sha1')
+        self.assertEqual(str(indicator[1]), '3395856ce81f2b7382dee72602f798b642f14140')
+
+    def test_pattern_hash_sha256(self):
+        """ Test if domain pattern is parsed. """
+        indicator = self.bot_reference.parse_stix_pattern("[file:hashes.'SHA-256' = '275a021bbfb6489e54d471899f7db9d1663fc695ec2fe2a2c4538aabf651fd0f']")[0]
+        self.assertEqual(str(indicator[0]), 'malware.hash.sha256')
+        self.assertEqual(str(indicator[1]), '275a021bbfb6489e54d471899f7db9d1663fc695ec2fe2a2c4538aabf651fd0f')
+
+        # Based on 10.7 Hashing Algorithm Vocabulary, keys SHA1 and SHA256 are not valid, but they are used in some feeds (e.g. ETI)
+        # https://docs.oasis-open.org/cti/stix/v2.1/os/stix-v2.1-os.html#_ths0b11wzxv3
+        indicator = self.bot_reference.parse_stix_pattern("[file:hashes.SHA256 = '275a021bbfb6489e54d471899f7db9d1663fc695ec2fe2a2c4538aabf651fd0f']")[0]
+        self.assertEqual(str(indicator[0]), 'malware.hash.sha256')
+        self.assertEqual(str(indicator[1]), '275a021bbfb6489e54d471899f7db9d1663fc695ec2fe2a2c4538aabf651fd0f')
+
+    def test_complex_pattern1(self):
+        """ Test if complex pattern is parsed. """
+        indicators = self.bot_reference.parse_stix_pattern("[url:value = 'http://example.org' AND ipv4-addr:value = '127.0.0.1/32']")
+        self.assertEqual(('source.url', 'http://example.org') in indicators, True)
+        self.assertEqual(('source.ip', '127.0.0.1') in indicators, True)
+
+    def test_complex_pattern2(self):
+        """ Test if complex pattern is parsed. """
+        indicators = self.bot_reference.parse_stix_pattern("[url:value = 'http://example.org'] AND [ipv4-addr:value = '127.0.0.1/32']")
+        self.assertEqual(('source.url', 'http://example.org') in indicators, True)
+        self.assertEqual(('source.ip', '127.0.0.1') in indicators, True)
+
 
 if __name__ == '__main__':  # pragma: no cover
     unittest.main()