From 0ea140836a33f9673af69a3c16fa227debef2d05 Mon Sep 17 00:00:00 2001 From: HC-ONLINE Date: Sat, 17 Oct 2026 12:09:01 -0500 Subject: [PATCH] fix(export): neutralize formula-like cells in CSV output --- src/ciberwebscan/export/csv.py | 44 +++++- tests/unit/export/test_csv.py | 240 +++++++++++++++++++++++++++++++++ 2 files changed, 282 insertions(+), 2 deletions(-) diff --git a/src/ciberwebscan/export/csv.py b/src/ciberwebscan/export/csv.py index 5393605..1ad31cb 100644 --- a/src/ciberwebscan/export/csv.py +++ b/src/ciberwebscan/export/csv.py @@ -25,6 +25,33 @@ logger = logging.getLogger(__name__) +# Leading characters that spreadsheets may interpret as formulas (CWE-1236). +FORMULA_PREFIXES: tuple[str, ...] = ("=", "+", "-", "@", "\t", "\r") + + +def neutralize_formula(value: Any) -> Any: + """ + Neutralize CSV formula injection (CWE-1236). + + Prepends a single quote to string values starting with a character that + spreadsheets may interpret as the start of a formula. Non-string values + (numbers, bools, None) are returned unchanged: they are never formula + vectors when written by the csv module. + + Note: the neutralized value keeps the original text after the apostrophe, + so consumers that strip a leading quote recover the raw value. + + Args: + value: Cell value to check. + + Returns: + The value, prefixed with an apostrophe if it starts with a character + in FORMULA_PREFIXES; otherwise the value unchanged. + """ + if isinstance(value, str) and value[:1] in FORMULA_PREFIXES: + return f"'{value}" + return value + def flatten_dict( data: dict[str, Any], @@ -174,9 +201,19 @@ def write_header(self, meta: dict[str, Any] | None = None) -> None: if self.columns: writer = self._get_writer() - writer.writeheader() + self._writeheader(writer) self._header_written = True + def _writeheader(self, writer: csv.DictWriter) -> None: + """ + Write the header row, neutralizing formula-like column names. + + Column names may embed third-party content (e.g. flattened HTTP + header names from a scanned target), so the header row is subject + to the same neutralization as data rows. + """ + writer.writerow({col: neutralize_formula(col) for col in self.columns or []}) + def write_item(self, item: Any) -> None: """ Write a single row to the CSV. @@ -200,6 +237,9 @@ def write_item(self, item: Any) -> None: max_depth=self.max_flatten_depth, ) + # Neutralize CSV formula injection (CWE-1236) + data = {key: neutralize_formula(value) for key, value in data.items()} + # Auto-detect columns from first item if self.columns is None: self.columns = list(data.keys()) @@ -208,7 +248,7 @@ def write_item(self, item: Any) -> None: if not self._header_written: self._started = True writer = self._get_writer() - writer.writeheader() + self._writeheader(writer) self._header_written = True else: writer = self._get_writer() diff --git a/tests/unit/export/test_csv.py b/tests/unit/export/test_csv.py index e101384..76debbb 100644 --- a/tests/unit/export/test_csv.py +++ b/tests/unit/export/test_csv.py @@ -19,10 +19,14 @@ csv_to_dicts, export_to_csv, flatten_dict, + neutralize_formula, ) pytestmark = pytest.mark.unit +# Leading characters spreadsheets may interpret as formulas (CWE-1236). +DANGEROUS_PREFIXES = ("=", "+", "-", "@", "\t", "\r") + # ============================================================================= # Test Fixtures # ============================================================================= @@ -526,3 +530,239 @@ def test_finalize_is_idempotent(self, tmp_path: Path): reader = csv.DictReader(output.open()) rows = list(reader) assert len(rows) == 1 + + +# ============================================================================= +# CSV Formula Injection Neutralization (CWE-1236 / BUG-008) +# ============================================================================= + + +def _read_cells(stream: StringIO, column: str = "value") -> list[str]: + """Parse CSV output and return the values of *column* via csv.DictReader.""" + stream.seek(0) + return [row[column] for row in csv.DictReader(stream)] + + +class TestNeutralizeFormula: + """Tests for the neutralize_formula helper.""" + + @pytest.mark.parametrize("prefix", DANGEROUS_PREFIXES) + def test_dangerous_prefix_gets_apostrophe(self, prefix: str): + """Strings starting with a dangerous character are prefixed.""" + assert neutralize_formula(f"{prefix}1+1") == f"'{prefix}1+1" + + @pytest.mark.parametrize("value", ["normal", "hello world", "a=b", "x+y"]) + def test_safe_strings_unchanged(self, value: str): + """Safe strings (dangerous chars in non-leading position) are untouched.""" + assert neutralize_formula(value) == value + + @pytest.mark.parametrize("value", ["+34 600", "- trailing"]) + def test_leading_plus_or_minus_strings_neutralized(self, value: str): + """Strings starting with + or - are neutralized even if human-written. + + Known trade-off of the OWASP recommendation: values such as phone + numbers starting with '+' lose their raw form in CSV output. + """ + assert neutralize_formula(value) == f"'{value}" + + @pytest.mark.parametrize("value", [42, -5, 3.14, True, None]) + def test_non_strings_unchanged(self, value): + """Non-string values are never neutralized (numbers are not formulas).""" + assert neutralize_formula(value) is value + + def test_empty_string_unchanged(self): + """Empty string is unchanged.""" + assert neutralize_formula("") == "" + + +class TestCSVFormulaNeutralization: + """Regression tests: cells must not retain unneutralized formula prefixes.""" + + @pytest.mark.parametrize("prefix", DANGEROUS_PREFIXES) + def test_six_dangerous_prefixes_neutralized(self, prefix: str): + """BUG-008: =, +, -, @, TAB and CR prefixed cells are neutralized in output.""" + stream = StringIO() + raw = f"{prefix}1+1" + + with CSVExporter(stream=stream, columns=["value"]) as exporter: + exporter.write_item({"value": raw}) + + (cell,) = _read_cells(stream) + assert not cell.startswith(DANGEROUS_PREFIXES) + assert cell == f"'{raw}" + + def test_mixed_row_neutralizes_only_dangerous(self): + """Only dangerous cells are modified within a mixed row.""" + stream = StringIO() + item = { + "eq": "=1+1", + "plus": "+1+1", + "minus": "-2+3", + "at": "@SUM(1+1)", + "tab": "\t=1+1", + "cr": "\r=1+1", + "normal": "safe text", + "number": -5, + "empty": "", + } + + with CSVExporter(stream=stream) as exporter: + exporter.write_item(item) + + stream.seek(0) + row = next(csv.DictReader(stream)) + assert row["eq"] == "'=1+1" + assert row["plus"] == "'+1+1" + assert row["minus"] == "'-2+3" + assert row["at"] == "'@SUM(1+1)" + assert row["tab"] == "'\t=1+1" + assert row["cr"] == "'\r=1+1" + assert row["normal"] == "safe text" + assert row["number"] == "-5" + assert row["empty"] == "" + + @pytest.mark.parametrize( + "value", + [ + "plain text", + "line1\nline2", + "has, comma", + 'has "quotes"', + "a=b internal", + "x+y internal", + "trailing minus -", + " leading space", + ], + ) + def test_legitimate_values_preserved(self, value: str): + """Ordinary values (including internal dangerous chars) survive intact.""" + stream = StringIO() + + with CSVExporter(stream=stream, columns=["value"]) as exporter: + exporter.write_item({"value": value}) + + (cell,) = _read_cells(stream) + assert cell == value + + def test_negative_numbers_as_numbers_preserved(self): + """Negative numbers written as numbers are not string-neutralized.""" + stream = StringIO() + + with CSVExporter(stream=stream, columns=["n"]) as exporter: + exporter.write_item({"n": -2}) + exporter.write_item({"n": -2.5}) + + assert _read_cells(stream, "n") == ["-2", "-2.5"] + + def test_nested_values_neutralized_after_flatten(self): + """Values under nested keys (e.g. scrape.title) are neutralized.""" + stream = StringIO() + item = { + "url": "https://example.com", + "scrape": { + "title": '=HYPERLINK("http://evil","click")', + "headers": {"X-Test": "=1+1"}, + }, + } + + with CSVExporter(stream=stream, flatten=True) as exporter: + exporter.write_item(item) + + stream.seek(0) + row = next(csv.DictReader(stream)) + assert row["scrape.title"] == '\'=HYPERLINK("http://evil","click")' + assert not row["scrape.headers.X-Test"].startswith("=") + + def test_list_joined_values_neutralized(self): + """Joined list values starting with a dangerous prefix are neutralized.""" + stream = StringIO() + + with CSVExporter(stream=stream, columns=["tags"]) as exporter: + exporter.write_item({"tags": ["=1+1", "safe"]}) + + (cell,) = _read_cells(stream, "tags") + assert cell == "'=1+1; safe" + + def test_no_flatten_still_neutralized(self): + """Neutralization does not depend on the flatten flag.""" + stream = StringIO() + + with CSVExporter(stream=stream, flatten=False, columns=["v"]) as exporter: + exporter.write_item({"v": "=1+1"}) + + (cell,) = _read_cells(stream, "v") + assert cell == "'=1+1" + + def test_dangerous_column_names_neutralized(self): + """Header row cells with formula-like column names are neutralized.""" + stream = StringIO() + + with CSVExporter(stream=stream, columns=["=cmd", "safe"]) as exporter: + exporter.write_header() + exporter.write_item({"=cmd": "x", "safe": "y"}) + + stream.seek(0) + rows = list(csv.reader(stream)) + assert rows[0][0] == "'=cmd" + assert rows[0][1] == "safe" + assert rows[1] == ["x", "y"] + + def test_streaming_multiple_items(self, sample_items): + """Streaming writes neutralize every row, not only the first.""" + stream = StringIO() + hostile_items = [ + {**item, "title": f"=cmd|' /C calc'!A{i}"} + for i, item in enumerate(sample_items) + ] + + with CSVExporter(stream=stream) as exporter: + for item in hostile_items: + exporter.write_item(item) + assert exporter._items_written == 3 + + stream.seek(0) + titles = [row["title"] for row in csv.DictReader(stream)] + assert len(titles) == 3 + assert all(t.startswith("'=") for t in titles) + + def test_export_to_csv_neutralizes(self, tmp_path: Path): + """The export_to_csv convenience wrapper also neutralizes.""" + output = tmp_path / "out.csv" + + export_to_csv([{"v": "@SUM(1+1)"}], output) + + rows = list(csv.DictReader(output.open())) + assert rows[0]["v"] == "'@SUM(1+1)" + + def test_export_report_server_neutralized(self, tmp_path: Path, mock_report): + """BUG-008 path: hostile Server header reaching the server column.""" + mock_report.fingerprint.server = "=cmd|' /C calc'!A1" + output = tmp_path / "report.csv" + + CSVExporter(output_path=output).export_report(mock_report) + + rows = list(csv.DictReader(output.open())) + assert rows[0]["server"] == "'=cmd|' /C calc'!A1" + # Non-dangerous report fields are intact + assert rows[0]["target_url"] == "https://example.com" + + def test_output_remains_valid_csv(self): + """Neutralization preserves CSV structure (quoting, delimiter, rows).""" + stream = StringIO() + items = [ + {"v": "=1+1", "note": "has, comma"}, + {"v": "'=1+1", "note": 'already "quoted" text'}, + {"v": "\r=1+1", "note": "multi\nline"}, + ] + + with CSVExporter(stream=stream) as exporter: + for item in items: + exporter.write_item(item) + + stream.seek(0) + rows = list(csv.DictReader(stream)) + assert len(rows) == 3 + assert rows[0]["note"] == "has, comma" + assert rows[1]["v"] == "'=1+1" # apostrophe-leading value is not re-prefixed + assert rows[1]["note"] == 'already "quoted" text' + assert rows[2]["note"] == "multi\nline"