From 661acd525116e88334dca9032c027cb6a4f379e6 Mon Sep 17 00:00:00 2001 From: Miro <200482516+Mirochill@users.noreply.github.com> Date: Sat, 10 Oct 2026 23:28:55 +0200 Subject: [PATCH] fix: skip MISP search results without an IntelMQ object when updating timestamps Assisted-by: OpenAI Codex --- CHANGELOG.md | 1 + intelmq/bots/outputs/misp/output_api.py | 5 +- .../bots/outputs/misp/test_output_api.py | 87 +++++++++++++++++-- 3 files changed, 84 insertions(+), 9 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 77cba765b0..181b8b3f44 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -32,6 +32,7 @@ Please refer to the [NEWS](NEWS.md) for a list of changes which have an affect o #### Experts #### Outputs +- `intelmq.bots.outputs.misp.output_api`: Ignore matching MISP events without an `intelmq_event` object when updating timestamps, avoiding an `IndexError`. - `intelmq.bots.outputs.smtp_batch.output`: - Ignore parameter `bot_id` for parsing parameters (PR#2692 by Lukas Heindl, fixes #2666). - Fix CLI overwriting params over the defaults (PR#2699 by Edvard Rejthar). diff --git a/intelmq/bots/outputs/misp/output_api.py b/intelmq/bots/outputs/misp/output_api.py index ff3ca1685c..5a4b570b50 100644 --- a/intelmq/bots/outputs/misp/output_api.py +++ b/intelmq/bots/outputs/misp/output_api.py @@ -151,8 +151,11 @@ def process(self): def _update_misp_event(self, misp_event, intelmq_event): """Update timestamp on a found MISPEvent if it matches closely.""" + misp_objects = misp_event.get_objects_by_name(MISPOBJECT_NAME) + if not misp_objects: + return # As we insert only one MISPObject, we only examine the first one - misp_o = misp_event.get_objects_by_name(MISPOBJECT_NAME)[0] + misp_o = misp_objects[0] all_found = True for field in ['feed.provider'] + self.significant_fields: diff --git a/intelmq/tests/bots/outputs/misp/test_output_api.py b/intelmq/tests/bots/outputs/misp/test_output_api.py index 3f57d1403c..dffb7e540a 100644 --- a/intelmq/tests/bots/outputs/misp/test_output_api.py +++ b/intelmq/tests/bots/outputs/misp/test_output_api.py @@ -2,13 +2,84 @@ # # SPDX-License-Identifier: AGPL-3.0-or-later -import os -# import unittest +import datetime +import unittest +from unittest.mock import Mock -# import intelmq.lib.test as test -if os.environ.get('INTELMQ_TEST_EXOTIC'): - from intelmq.bots.outputs.misp.output_api import MISPAPIOutputBot # noqa +from intelmq.bots.outputs.misp.output_api import MISPOBJECT_NAME, MISPAPIOutputBot -# This file is a stub -# We cannot do much more as we are missing a mock MISP instance to use -# to initialise pymisp + +class TestMISPAPIOutputBot(unittest.TestCase): + def setUp(self): + # Exercise local event handling without a live MISP instance. + self.bot = MISPAPIOutputBot.__new__(MISPAPIOutputBot) + self.bot.logger = Mock() + self.bot.misp = Mock() + self.bot.significant_fields = ['source.fqdn'] + self.event_data = {'source.fqdn': 'example.org', 'feed.provider': 'Provider'} + + def misp_event(self, values=None): + event = Mock() + event.timestamp = datetime.datetime(2020, 1, 1) + if values is None: + event.get_objects_by_name.return_value = [] + else: + obj = Mock() + obj.get_attributes_by_relation.side_effect = ( + lambda field: [Mock(value=values[field])] if field in values else [] + ) + event.get_objects_by_name.return_value = [obj] + return event + + def test_no_intelmq_object_is_not_updated(self): + event = self.misp_event() + original_timestamp = event.timestamp + + self.bot._update_misp_event(event, self.event_data) + + event.get_objects_by_name.assert_called_once_with(MISPOBJECT_NAME) + self.bot.misp.update_event.assert_not_called() + self.assertEqual(event.timestamp, original_timestamp) + + def test_matching_object_is_updated(self): + event = self.misp_event(self.event_data) + original_timestamp = event.timestamp + + self.bot._update_misp_event(event, self.event_data) + + self.bot.misp.update_event.assert_called_once_with(event) + self.assertNotEqual(event.timestamp, original_timestamp) + + def test_mismatching_provider_is_not_updated(self): + event = self.misp_event(dict(self.event_data, **{'feed.provider': 'Other'})) + + self.bot._update_misp_event(event, self.event_data) + + self.bot.misp.update_event.assert_not_called() + + def test_missing_significant_attribute_is_not_updated(self): + event = self.misp_event({'feed.provider': 'Provider'}) + + self.bot._update_misp_event(event, self.event_data) + + self.bot.misp.update_event.assert_not_called() + + def test_process_continues_after_unrelated_search_result(self): + unrelated = self.misp_event() + matching = self.misp_event(self.event_data) + self.bot.misp.search.return_value = [unrelated, matching] + self.bot.misp_tag_for_bot = 'IntelMQ' + message = Mock() + message.to_dict.return_value = self.event_data + self.bot.receive_message = Mock(return_value=message) + self.bot.acknowledge_message = Mock() + + self.bot.process() + + self.bot.misp.update_event.assert_called_once_with(matching) + self.bot.acknowledge_message.assert_called_once_with() + self.bot.misp.add_event.assert_not_called() + + +if __name__ == '__main__': + unittest.main()