Add TAXII Collector bot and STIX Parser bot

laciKE · sebix · commit 316af6b03e77 · 2025-07-18T12:27:32.000+01:00
As a bare minimum, TAXII Collector currently collects only the objects of type indicator. These objects contain information about indicators and the detection patterns, e.g. in stix, pcre, sigma, snort, suricata, yara format. The pattern, pattern_type and valid_from properties are required, while confidence, description and labels are only optional properties. However, they are present in several TAXII feeds and could be used to determine classification.taxonomy and classification.type even without processing the relationships of the indicators (e.g. indicator indicates malware) STIX Parser is currently capable of parsing objects of type indicator (usually retrieved from the TAXII Collector). From the indicator objects, it extracts the detection pattern (currently only the single Observation Expressions in STIX format are supported). It supports IP addresses, Domains and URLs indicator values. Moreover, this parser also attempts to extract some optional properties of STIX objects such as description and labels, which can be useful for futher classification of the event with the Expert Bots TAXII Collector tests for missing parameters and mock the simple TAXII server providing minimal collection with simple indicator object STIX Parser tests fo indicator patterns parsing Improvements based on @sebix comments, collection title used as feed.code Fix codestyle in TAXII and STIX bots Fix Python 3.8 support in STIX Parser bot
diff --git a/intelmq/bots/collectors/taxii/REQUIREMENTS.txt b/intelmq/bots/collectors/taxii/REQUIREMENTS.txt
@@ -0,0 +1,4 @@
+# SPDX-FileCopyrightText: 2025 Ladislav Baco
+# SPDX-License-Identifier: AGPL-3.0-or-later
+
+taxii2-client>=2.3.0
diff --git a/intelmq/bots/collectors/taxii/__init__.py b/intelmq/bots/collectors/taxii/__init__.py
diff --git a/intelmq/bots/collectors/taxii/collector.py b/intelmq/bots/collectors/taxii/collector.py
@@ -0,0 +1,64 @@
+"""
+SPDX-FileCopyrightText: 2025 Ladislav Baco
+SPDX-License-Identifier: AGPL-3.0-or-later
+
+Get indicator objects from TAXII server
+
+Configuration parameters: taxii collection (feed) url, username and password.
+"""
+
+import datetime
+import json
+from requests.exceptions import HTTPError
+
+from intelmq.lib.bot import CollectorBot
+from intelmq.lib.exceptions import MissingDependencyError
+
+try:
+    import taxii2client.v21 as taxii2
+except ImportError:
+    taxii2 = None
+
+
+class TaxiiCollectorBot(CollectorBot):
+    """Collect data from TAXII Server"""
+    collection: str = None
+    username: str = None
+    password: str = None
+    rate_limit: int = 3600
+    time_delta: int = 3600
+
+    def init(self):
+        if taxii2 is None:
+            raise MissingDependencyError('taxii2-client')
+
+        if self.collection is None:
+            raise ValueError('No TAXII collection URL provided.')
+        if self.username is None:
+            raise ValueError('No TAXII username provided.')
+        if self.password is None:
+            raise ValueError('No TAXII password provided.')
+
+        self._date_after = datetime.datetime.now() - datetime.timedelta(seconds=int(self.time_delta))
+
+        self._taxii_collection = taxii2.Collection(self.collection, user=self.username, password=self.password)
+
+    def process(self):
+        try:
+            title = self._taxii_collection.title
+            self.logger.info('Collection title: %r.', title)
+
+            # get the indicator objects
+            objects = self._taxii_collection.get_objects(added_after=self._date_after, type='indicator').get('objects', [])
+            for obj in objects:
+                report = self.new_report()
+                report.add('raw', json.dumps(obj))
+                report.add('feed.url', self.collection)
+                report.add('feed.code', title)
+                self.send_message(report)
+
+        except HTTPError as e:
+            self.logger.error('Connection error: %r!', e)
+
+
+BOT = TaxiiCollectorBot
diff --git a/intelmq/bots/parsers/stix/__init__.py b/intelmq/bots/parsers/stix/__init__.py
diff --git a/intelmq/bots/parsers/stix/parser.py b/intelmq/bots/parsers/stix/parser.py
@@ -0,0 +1,79 @@
+"""
+SPDX-FileCopyrightText: 2025 Ladislav Baco
+SPDX-License-Identifier: AGPL-3.0-or-later
+
+Parse indicators objects in STIX format received from TAXII collector
+"""
+
+import json
+
+from intelmq.lib.bot import ParserBot
+
+
+class StixParserBot(ParserBot):
+    """Parse STIX indicators"""
+    parse = ParserBot.parse_json_stream
+    recover_line = ParserBot.recover_line_json_stream
+
+    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])
+                    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)
+
+    @staticmethod
+    def parse_stix_pattern(pattern):
+        """
+        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)
+
+
+BOT = StixParserBot
diff --git a/intelmq/tests/bots/collectors/taxii/test_collector_bot.py b/intelmq/tests/bots/collectors/taxii/test_collector_bot.py
@@ -0,0 +1,108 @@
+# SPDX-FileCopyrightText: 2016 Sebastian Wagner, 2025 Ladislav Baco
+#
+# SPDX-License-Identifier: AGPL-3.0-or-later
+
+# -*- coding: utf-8 -*-
+"""
+Test with reports, based on intelmq/tests/lib/test_collector_bot.py
+"""
+import unittest
+
+import re
+import requests_mock
+
+import intelmq.lib.bot as bot
+import intelmq.lib.test as test
+from intelmq.bots.collectors.taxii.collector import TaxiiCollectorBot
+
+
+EXAMPLE_REPORT = {'__type': 'Report',
+                  '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',
+                  'raw': 'eyJpZCI6ICJpbmRpY2F0b3ItLTAiLCAidHlwZSI6ICJpbmRpY2F0b3IiLCAic3BlY192ZXJzaW9uIjogIjIuMSIsICJjcmVhdGVkIjogIjE5NzAtMDEtMDFUMDA6MDA6MDAuMDAwWiIsICJtb2RpZmllZCI6ICIxOTcwLTAxLTAxVDAwOjAwOjAwLjAwMFoiLCAicGF0dGVybiI6ICJbdXJsOnZhbHVlID0gJ2h0dHA6Ly9leGFtcGxlLm9yZyddIiwgInBhdHRlcm5fdHlwZSI6ICJzdGl4IiwgInZhbGlkX2Zyb20iOiAiMTk3MC0wMS0wMVQwMDowMDowMFoifQ=='
+                  }
+
+def prepare_mocker(mocker):
+    mocker.get(
+        'http://localhost/feed/',
+        json={
+            'id': 'feed',
+            'title': 'feed stix2.1',
+            'can_read': True,
+            'can_write': False
+        },
+        headers={'Content-Type': 'application/taxii+json;version=2.1'}
+    )
+    mocker.get(
+        re.compile('http://localhost/feed/objects/.*'),
+        json={
+            'id': 'feed',
+            'title': 'feed stix2.1',
+            'can_read': True,
+            'can_write': False,
+            'more': False,
+            'objects': [{
+                'id': 'indicator--0',
+                'type': 'indicator',
+                'spec_version': '2.1',
+                'created': '1970-01-01T00:00:00.000Z',
+                'modified': '1970-01-01T00:00:00.000Z',
+                'pattern': "[url:value = 'http://example.org']",
+                'pattern_type': 'stix',
+                'valid_from': '1970-01-01T00:00:00Z'
+            }]},
+            headers={'Content-Type': 'application/taxii+json;version=2.1'}
+    )
+
+@test.skip_exotic()
+@requests_mock.Mocker()
+class TestTaxiiCollectorBot(test.BotTestCase, unittest.TestCase):
+    """
+    A TestCase for a TaxiiCollectorBot.
+    """
+
+    @classmethod
+    def set_bot(cls):
+        cls.bot_reference = TaxiiCollectorBot
+        cls.sysconfig = {'name': 'Taxii Feed',
+                         'provider': 'Taxii Provider',
+                         'documentation': 'Taxii Documentation',
+                         'collection': 'http://localhost/feed',
+                         'username': 'user',
+                         'password': 'pass'
+                         }
+
+    def test_event(self, mocker):
+        """ Test if correct Event has been produced. """
+        prepare_mocker(mocker)
+        self.run_bot()
+        self.assertMessageEqual(0, EXAMPLE_REPORT)
+
+    def test_missing_collection(self, mocker):
+        """ Test if missing collection is detected. """
+        with self.assertRaises(ValueError) as context:
+            self.run_bot(parameters={'collection': None})
+        exception = context.exception
+        self.assertEqual(str(exception), 'No TAXII collection URL provided.')
+
+    def test_missing_username(self, mocker):
+        """ Test if missing username is detected. """
+        with self.assertRaises(ValueError) as context:
+            self.run_bot(parameters={'username': None})
+        exception = context.exception
+        self.assertEqual(str(exception), 'No TAXII username provided.')
+
+    def test_missing_password(self, mocker):
+        """ Test if missing password is detected. """
+        with self.assertRaises(ValueError) as context:
+            self.run_bot(parameters={'password': None})
+        exception = context.exception
+        self.assertEqual(str(exception), 'No TAXII password provided.')
+
+
+if __name__ == '__main__':  # pragma: no cover
+    unittest.main()
diff --git a/intelmq/tests/bots/parsers/stix/test_parser_bot.py b/intelmq/tests/bots/parsers/stix/test_parser_bot.py
@@ -0,0 +1,110 @@
+# 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 import StixParserBot
+
+
+EXAMPLE_REPORT = {'__type': 'Report',
+                  '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',
+                  'raw': 'eyJpZCI6ICJpbmRpY2F0b3ItLTAiLCAidHlwZSI6ICJpbmRpY2F0b3IiLCAic3BlY192ZXJzaW9uIjogIjIuMSIsICJjcmVhdGVkIjogIjE5NzAtMDEtMDFUMDA6MDA6MDAuMDAwWiIsICJtb2RpZmllZCI6ICIxOTcwLTAxLTAxVDAwOjAwOjAwLjAwMFoiLCAicGF0dGVybiI6ICJbdXJsOnZhbHVlID0gJ2h0dHA6Ly9leGFtcGxlLm9yZyddIiwgInBhdHRlcm5fdHlwZSI6ICJzdGl4IiwgInZhbGlkX2Zyb20iOiAiMTk3MC0wMS0wMVQwMDowMDowMFoifQ=='
+                  }
+
+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=='
+                  }
+
+
+class TestStixParserBot(test.BotTestCase, unittest.TestCase):
+    """
+    A TestCase for a StixParserBot.
+    """
+
+    @classmethod
+    def set_bot(cls):
+        cls.bot_reference = StixParserBot
+        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_pattern_url(self):
+        """ Test if url pattern is parsed. """
+        indicator = self.bot_reference.parse_stix_pattern("[url:value = 'http://example.org']")
+        self.assertEqual(str(indicator[0]), 'source.url')
+        self.assertEqual(str(indicator[1]), 'http://example.org')
+
+    def test_pattern_url(self):
+        """ Test if domain pattern is parsed. """
+        indicator = self.bot_reference.parse_stix_pattern("[domain-name:value = 'example.org']")
+        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']")
+        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']")
+        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']")
+        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']")
+        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']")
+        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']")
+        self.assertEqual(str(indicator[0]), 'source.ip')
+        self.assertEqual(str(indicator[1]), 'fe:80::1')
+
+
+if __name__ == '__main__':  # pragma: no cover
+    unittest.main()
