Merge pull request certtools#2567 from sebix/fake-expert

add new bot to fake data

Author: sebix

CHANGELOG.md

@@ -43,6 +43,7 @@
 - `intelmq.bots.experts.securitytxt`:
   - Added new bot (PR#2538 by Frank Westers and Sebastian Wagner)
 - `intelmq.bots.experts.misp`: Use `PyMISP` class instead of deprecated `ExpandedPyMISP` (PR#2532 by Radek Vyhnal)
+- `intelmq.bots.experts.fake.expert`: New expert to fake data (PR#2567 by Sebastian Wagner).
 
 #### Outputs
 - `intelmq.bots.outputs.cif3.output`:

docs/user/bots.md

@@ -2682,6 +2682,39 @@ is `$portal_url + '/api/1.0/ripe/contact?cidr=%s'`.
 
 ---
 
+### Fake <div id="intelmq.bots.experts.fake.expert" />
+
+Adds fake data to events. Currently supports setting the IP address and network.
+
+For each incoming event, the bots chooses one random IP network range from the configured data file.
+It set's the first IP address of the range as `source.ip` and the network itself as `source.network`.
+To adapt the `source.asn` field accordingly, use the [ASN Lookup Expert](#asn-lookup).
+
+**Module:** `intelmq.bots.experts.fake.expert`
+
+**Parameters:**
+
+**`database`**
+
+(required, string) Path to a JSON file in the following format:
+```
+{
+    "ip_network": [
+        "10.0.0.0/8",
+        ...
+    ]
+}
+```
+
+**`overwrite`**
+
+(optional, boolean) Whether to overwrite existing fields. Defaults to false.
+
+For data consistency `source.network` will only be set if `source.ip` was set or overridden.
+If overwrite is false, `source.ip` was did not exist before but `source.network` existed before, `source.network` will still be overridden.
+
+---
+
 ### Field Reducer <div id="intelmq.bots.experts.field_reducer.expert" />
 
 The field reducer bot is capable of removing fields from events.

intelmq/bots/experts/fake/__init__.py

@@ -0,0 +1,46 @@
+# SPDX-FileCopyrightText: 2025 Institute for Common Good Technology, Sebastian Wagner
+#
+# SPDX-License-Identifier: AGPL-3.0-or-later
+
+from ipaddress import ip_network
+from random import choice
+from json import load as json_load
+
+from intelmq.lib.bot import ExpertBot
+
+
+class FakeExpertBot(ExpertBot):
+    """Add fake data"""
+
+    overwrite: bool = False
+    database: str = None  # TODO: should be pathlib.Path
+
+    def init(self):
+        with open(self.database) as database:
+            self.networks = json_load(database)['ip_network']
+
+    def process(self):
+        event = self.receive_message()
+        network = choice(self.networks)
+
+        updated = False
+        try:
+            updated = event.add('source.ip', ip_network(network)[1], overwrite=self.overwrite)
+        except IndexError:
+            updated = event.add('source.ip', ip_network(network)[0], overwrite=self.overwrite)
+        # For consistency, only set the network if the source.ip was set or overwritten, but then always overwrite it
+        if updated:
+            event.add('source.network', network, overwrite=True)
+
+        self.send_message(event)
+        self.acknowledge_message()
+
+    def check(parameters: dict):
+        try:
+            with open(parameters['database']) as database:
+                json_load(database)['ip_network']
+        except Exception as exc:
+            return [['error', exc]]
+
+
+BOT = FakeExpertBot

intelmq/bots/experts/fake/expert.py

@@ -0,0 +1,5 @@
+{
+    "ip_network": [
+        "10.0.0.0/8"
+    ]
+}

intelmq/tests/bots/experts/fake/__init__.py

@@ -0,0 +1,2 @@
+SPDX-FileCopyrightText: 2025 Institute for Common Good Technology, Sebastian Wagner
+SPDX-License-Identifier: AGPL-3.0-or-later

intelmq/tests/bots/experts/fake/data.json

@@ -0,0 +1,50 @@
+# SPDX-FileCopyrightText: 2025 Institute for Common Good Technology, Sebastian Wagner
+#
+# SPDX-License-Identifier: AGPL-3.0-or-later
+
+import unittest
+from json import loads as json_loads
+from ipaddress import ip_network, ip_address
+
+import pkg_resources
+
+import intelmq.lib.test as test
+from intelmq.bots.experts.fake.expert import FakeExpertBot
+
+FAKE_DB = pkg_resources.resource_filename('intelmq', 'tests/bots/experts/fake/data.json')
+EXAMPLE_INPUT = {"__type": "Event",
+                 "source.ip": "93.184.216.34",  # example.com
+                 }
+NETWOK_EXISTS = {"__type": "Event",
+                 "source.network": "93.184.216.0/24",
+                 }
+
+class TestFakeExpertBot(test.BotTestCase, unittest.TestCase):
+
+    @classmethod
+    def set_bot(cls):
+        cls.bot_reference = FakeExpertBot
+        cls.sysconfig = {'database': FAKE_DB}
+
+    def test_nochange(self):
+        self.input_message = EXAMPLE_INPUT
+        self.run_bot()
+        self.assertMessageEqual(0, EXAMPLE_INPUT)
+
+    def test_overwrite(self):
+        self.input_message = EXAMPLE_INPUT
+        self.run_bot(parameters={'overwrite': True})
+        msg = json_loads(self.get_output_queue()[0])
+        self.assertIn(ip_address(msg['source.ip']), ip_network("10.0.0.0/8"))
+        self.assertEqual(msg['source.network'], "10.0.0.0/8")
+
+    def test_network_exists(self):
+        self.input_message = NETWOK_EXISTS
+        self.run_bot(parameters={'overwrite': False})
+        msg = json_loads(self.get_output_queue()[0])
+        self.assertIn(ip_address(msg['source.ip']), ip_network("10.0.0.0/8"))
+        self.assertEqual(msg['source.network'], "10.0.0.0/8")
+
+
+if __name__ == '__main__':  # pragma: no cover
+    unittest.main()