SEC: Escape spreadsheet formulas in CSV exports

docs/user/bots.md

@@ -2519,6 +2519,10 @@ parameter `single_key` of the output bot and set it to `output`.
 
 (optional, string) Defaults to `,`.
 
+**`escape_csv_injection`**
+
+(optional, boolean) Prefix values starting with spreadsheet formula characters with a single quote. Defaults to `true`.
+
 **`fieldnames`**
 
 (required, string) Comma-separated list of field names, e.g. `"time.source,classification.type,source.ip"`.
@@ -5369,6 +5373,10 @@ Default: `{subject: False, body: False, attachment: False}`
 
 (required, string) Sender's e-mail of the outgoing messages.
 
+**`escape_csv_injection`**
+
+(optional, boolean) Prefix values starting with spreadsheet formula characters with a single quote. Defaults to `true`.
+
 
 **`gpg_key`**
 
@@ -5449,6 +5457,10 @@ Sends a MIME Multipart message containing the text and the event as CSV for ever
 
 (optional, string/array of strings) Array of field names (or comma-separated list) to be included in the email. If empty, no attachment is sent - this can be useful if the actual data is already in the body (parameter `text`) or the `subject`.
 
+**`escape_csv_injection`**
+
+(optional, boolean) Prefix values starting with spreadsheet formula characters with a single quote. Defaults to `true`.
+
 **`mail_from`**
 
 (optional, string) Sender's e-email address. Defaults to `cert@localhost`.

intelmq/bots/experts/csv_converter/expert.py

@@ -6,12 +6,14 @@
 import csv
 import io
 from intelmq.lib.bot import ExpertBot
+from intelmq.lib.utils import sanitize_csv_value
 
 
 class CSVConverterExpertBot(ExpertBot):
     """Convert data to CSV"""
     fieldnames: str = "time.source,classification.type,source.ip"  # TODO: could maybe be List[str]
     delimiter: str = ','
+    escape_csv_injection: bool = True
 
     def init(self):
         self.fieldnames = self.fieldnames.split(',')
@@ -23,7 +25,10 @@ def process(self):
         writer = csv.writer(out, delimiter=self.delimiter)
         row = []
         for field in self.fieldnames:
-            row.append(event[field])
+            value = event[field]
+            if self.escape_csv_injection:
+                value = sanitize_csv_value(value)
+            row.append(value)
         writer.writerow(row)
         event['output'] = out.getvalue().rstrip()
 

intelmq/bots/outputs/smtp/output.py

@@ -11,12 +11,14 @@
 from email.mime.text import MIMEText
 
 from intelmq.lib.bot import OutputBot
+from intelmq.lib.utils import sanitize_csv_value
 from typing import Optional
 
 
 class SMTPOutputBot(OutputBot):
     """Send single events as CSV attachment in dynamically formatted e-mails via SMTP"""
     fieldnames: str = "classification.taxonomy,classification.type,classification.identifier,source.ip,source.asn,source.port"
+    escape_csv_injection: bool = True
     mail_from: str = "cert@localhost"
     mail_to: str = "{ev[source.abuse_contact]}"
     smtp_host: str = "localhost"
@@ -53,7 +55,10 @@ def process(self):
                                     quoting=csv.QUOTE_MINIMAL, delimiter=";",
                                     extrasaction='ignore', lineterminator='\n')
             writer.writeheader()
-            writer.writerow(event)
+            row = event
+            if self.escape_csv_injection:
+                row = {field: sanitize_csv_value(event[field]) for field in self.fieldnames}
+            writer.writerow(row)
             attachment = csvfile.getvalue()
 
         with self.smtp_class(self.smtp_host, self.smtp_port, **self.kwargs) as smtp:

intelmq/bots/outputs/smtp_batch/output.py

@@ -21,6 +21,7 @@
 from intelmq.lib.bot import Bot
 from intelmq.lib.cache import Cache
 from intelmq.lib.exceptions import MissingDependencyError
+from intelmq.lib.utils import sanitize_csv_value
 
 try:
     from envelope import Envelope
@@ -61,6 +62,7 @@ class SMTPBatchOutputBot(Bot):
     alternative_mails: Optional[str] = None
     bcc: Optional[list] = None
     email_from: str = ""
+    escape_csv_injection: bool = True
     gpg_key: Optional[str] = None
     gpg_pass: Optional[str] = None
     mail_template: str = ""
@@ -313,7 +315,12 @@ def prepare_mails(self) -> Iterable[Mail]:
                     row["raw"] = b64decode(row["raw"]).decode("utf-8").strip().replace("\n", r"\n").replace("\r", r"\r")
                 except (ValueError, KeyError):  # not all events have to contain the "raw" field
                     pass
-                rows_output.append(OrderedDict({self.fieldnames_translation[k]: row[k] for k in ordered_keys}))
+                rows_output.append(OrderedDict({
+                    self.fieldnames_translation[k]: (
+                        sanitize_csv_value(row[k]) if self.escape_csv_injection else row[k]
+                    )
+                    for k in ordered_keys
+                }))
 
             # prepare headers for csv attachment
             ordered_fieldnames = []

intelmq/lib/utils.py

@@ -14,6 +14,7 @@
 log
 reverse_readline
 parse_logline
+sanitize_csv_value
 """
 import base64
 import collections
@@ -62,7 +63,7 @@
            'reverse_readline', 'error_message_from_exc', 'parse_relative',
            'RewindableFileHandle',
            'file_name_from_response',
-           'list_all_bots', 'get_global_settings',
+           'list_all_bots', 'get_global_settings', 'sanitize_csv_value',
            ]
 
 # Used loglines format
@@ -82,12 +83,25 @@
                 r'(?P<thread_id>\.[0-9]+)?'
                 r': (?P<log_level>[A-Z]+) (?P<message>.+)$')
 RESPONSE_FILENAME = re.compile("filename=(.+)")
+CSV_FORMULA_PREFIXES = ('=', '+', '-', '@', '\t', '\r', '\n')
 
 
 class Parameters:
     pass
 
 
+def sanitize_csv_value(value: Any) -> Any:
+    """Neutralize strings which spreadsheet programs may treat as formulas.
+
+    The value is escaped only when it starts with a known formula trigger.
+    Non-string values are returned unchanged so CSV writers retain their
+    existing number and ``None`` handling.
+    """
+    if isinstance(value, str) and value.startswith(CSV_FORMULA_PREFIXES):
+        return "'" + value
+    return value
+
+
 def decode(text: Union[bytes, str], encodings: Sequence[str] = ("utf-8",),
            force: bool = False) -> str:
     """

intelmq/tests/bots/experts/csv_converter/test_expert.py

@@ -3,6 +3,7 @@
 # SPDX-License-Identifier: AGPL-3.0-or-later
 
 # -*- coding: utf-8 -*-
+import json
 import unittest
 
 import intelmq.lib.test as test
@@ -39,6 +40,28 @@ def test_delimiter(self):
         self.run_bot(parameters={'delimiter': ';'})
         self.assertMessageEqual(0, DELIMITER_OUT)
 
+    def test_csv_injection_is_escaped_by_default(self):
+        self.input_message = {
+            **EXAMPLE_INPUT,
+            'event_description.text': '=HYPERLINK("https://example.invalid")',
+        }
+        self.run_bot(parameters={'fieldnames': 'event_description.text'})
+        self.assertEqual(
+            '"\'=HYPERLINK(""https://example.invalid"")"',
+            json.loads(json.loads(self.get_output_queue()[0])['output']),
+        )
+
+    def test_csv_injection_escape_can_be_disabled(self):
+        self.input_message = {
+            **EXAMPLE_INPUT,
+            'event_description.text': '=1+1',
+        }
+        self.run_bot(parameters={
+            'escape_csv_injection': False,
+            'fieldnames': 'event_description.text',
+        })
+        self.assertEqual('=1+1', json.loads(json.loads(self.get_output_queue()[0])['output']))
+
 
 if __name__ == '__main__':  # pragma: no cover
     unittest.main()

intelmq/tests/bots/outputs/smtp/test_output.py

@@ -88,6 +88,37 @@ def test_no_attachment(self):
         self.assertIn(('Content-Type', 'text/plain; charset="us-ascii"'),
                       SENT_MESSAGE[0].get_payload()[0]._headers)
 
+    def test_csv_injection_is_escaped_by_default(self):
+        self.input_message = {
+            **EVENT,
+            'event_description.text': '=HYPERLINK("https://example.invalid")',
+        }
+        with unittest.mock.patch('smtplib.SMTP.send_message', new=send_message):
+            with unittest.mock.patch('smtplib.SMTP.close'):
+                self.run_bot(parameters={'fieldnames': 'event_description.text'})
+
+        self.assertEqual(
+            'event_description.text\n"\'=HYPERLINK(""https://example.invalid"")"\n',
+            SENT_MESSAGE[0].get_payload()[1].get_payload(),
+        )
+
+    def test_csv_injection_escape_can_be_disabled(self):
+        self.input_message = {
+            **EVENT,
+            'event_description.text': '=1+1',
+        }
+        with unittest.mock.patch('smtplib.SMTP.send_message', new=send_message):
+            with unittest.mock.patch('smtplib.SMTP.close'):
+                self.run_bot(parameters={
+                    'escape_csv_injection': False,
+                    'fieldnames': 'event_description.text',
+                })
+
+        self.assertEqual(
+            'event_description.text\n=1+1\n',
+            SENT_MESSAGE[0].get_payload()[1].get_payload(),
+        )
+
 
 if __name__ == '__main__':  # pragma: no cover
     unittest.main()

intelmq/tests/lib/test_utils.py

@@ -62,6 +62,16 @@ def new_get_runtime() -> dict:
 
 class TestUtils(unittest.TestCase):
 
+    def test_sanitize_csv_value(self):
+        for prefix in ('=', '+', '-', '@', '\t', '\r', '\n'):
+            value = prefix + 'formula'
+            with self.subTest(prefix=repr(prefix)):
+                self.assertEqual("'" + value, utils.sanitize_csv_value(value))
+
+        for value in ('text', ' =formula', "'=formula", '', 42, None):
+            with self.subTest(value=value):
+                self.assertIs(value, utils.sanitize_csv_value(value))
+
     def test_decode_byte(self):
         """Tests if the decode can handle bytes."""
         self.assertEqual(SAMPLES['normal'][1],