From 2c99772bf6e6508448f7ef37a9c13caf065b67fc Mon Sep 17 00:00:00 2001 From: aryan hirapara <246931328+rootverdict@users.noreply.github.com> Date: Wed, 22 Jul 2026 09:55:19 +0000 Subject: [PATCH] SEC: escape spreadsheet formulas in CSV exports --- docs/user/bots.md | 12 +++++++ intelmq/bots/experts/csv_converter/expert.py | 7 ++++- intelmq/bots/outputs/smtp/output.py | 7 ++++- intelmq/bots/outputs/smtp_batch/output.py | 9 +++++- intelmq/lib/utils.py | 16 +++++++++- .../bots/experts/csv_converter/test_expert.py | 23 ++++++++++++++ .../tests/bots/outputs/smtp/test_output.py | 31 +++++++++++++++++++ intelmq/tests/lib/test_utils.py | 10 ++++++ 8 files changed, 111 insertions(+), 4 deletions(-) diff --git a/docs/user/bots.md b/docs/user/bots.md index 44e90971dd..a10381f948 100644 --- a/docs/user/bots.md +++ b/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`. diff --git a/intelmq/bots/experts/csv_converter/expert.py b/intelmq/bots/experts/csv_converter/expert.py index 67c8d0e9ec..8c7435a4b8 100644 --- a/intelmq/bots/experts/csv_converter/expert.py +++ b/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() diff --git a/intelmq/bots/outputs/smtp/output.py b/intelmq/bots/outputs/smtp/output.py index 8c62e77472..3d5bb3e91e 100644 --- a/intelmq/bots/outputs/smtp/output.py +++ b/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: diff --git a/intelmq/bots/outputs/smtp_batch/output.py b/intelmq/bots/outputs/smtp_batch/output.py index f07c2ba7b0..448256dade 100644 --- a/intelmq/bots/outputs/smtp_batch/output.py +++ b/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 = [] diff --git a/intelmq/lib/utils.py b/intelmq/lib/utils.py index 0a4922d703..0e995dac0b 100644 --- a/intelmq/lib/utils.py +++ b/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\.[0-9]+)?' r': (?P[A-Z]+) (?P.+)$') 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: """ diff --git a/intelmq/tests/bots/experts/csv_converter/test_expert.py b/intelmq/tests/bots/experts/csv_converter/test_expert.py index 9cee2f6554..7c4c07f55c 100644 --- a/intelmq/tests/bots/experts/csv_converter/test_expert.py +++ b/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() diff --git a/intelmq/tests/bots/outputs/smtp/test_output.py b/intelmq/tests/bots/outputs/smtp/test_output.py index 1865a45c6a..7085f1e7c0 100644 --- a/intelmq/tests/bots/outputs/smtp/test_output.py +++ b/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() diff --git a/intelmq/tests/lib/test_utils.py b/intelmq/tests/lib/test_utils.py index ddb34408a3..13ff5c5f72 100644 --- a/intelmq/tests/lib/test_utils.py +++ b/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],