fix: neutralize CSV formula injection in spreadsheet-bound exports (OWASP)

No CSV output in the project escaped spreadsheet formula triggers: a cell
like =cmd|' /C calc'!A0 or =HYPERLINK(...) in untrusted input executed
when an exported file was opened in Excel/Sheets/LibreOffice. The review
queue export is the sharpest edge - it exists to be opened by human
reviewers in spreadsheets and embeds values from the input data.

Fix:
- new freshdata._util.sanitize_csv_formulas: copies the frame and
  prefixes ' to string cells and column labels starting with
  = + - @ <tab> <cr> (OWASP trigger set); numeric cells (incl. negative
  numbers) untouched; unchanged columns keep their dtype.
- export_review_queue: sanitizes csv output BY DEFAULT (spreadsheet-bound
  artifact), opt-out via sanitize_formulas=False; jsonl/parquet untouched.
- fd.clean_csv: byte-exact by default, opt-in sanitize_formulas=True
  applies only to the written file, never the returned frame.
- streaming CLI (stream/stream-kafka): opt-in --sanitize-formulas, also
  covers the --quarantine exceptions export.

15 regression tests in tests/test_csv_formula_sanitize.py; the file
import-errors on main (sanitizer absent) and the flags did not exist.

Author: JyothiSwaroop000

CHANGELOG.md

@@ -39,6 +39,14 @@ adheres to [Semantic Versioning](https://semver.org/).
   materialization — pipeline stages currently collect intermediates
   eagerly, so peak memory during cleaning matches eager output. The DuckDB
   handle path is the measured lower-peak-memory route (#52, #53).
+- **CSV formula-injection protection** (OWASP): `export_review_queue` now
+  neutralizes spreadsheet formula payloads in CSV exports **by default**
+  (string cells and column labels starting with `= + - @ <tab> <cr>` get a
+  leading `'`; opt out with `sanitize_formulas=False`) — review queues are
+  built to be opened by humans in spreadsheets. `fd.clean_csv` and the
+  streaming CLI keep byte-exact output by default and gain an explicit
+  opt-in (`sanitize_formulas=True` / `--sanitize-formulas`) covering the
+  cleaned output and the quarantine export. JSONL/Parquet are never altered.
 
 ### Added
 - `benchmarks/bench_outofcore.py`: subprocess-isolated peak-RSS evidence for

src/freshdata/_util.py

@@ -82,3 +82,31 @@ def stringlike_columns(df: pd.DataFrame) -> list:
 
 def _is_stringlike_dtype(dtype: object) -> bool:
     return pd.api.types.is_object_dtype(dtype) or isinstance(dtype, pd.StringDtype)
+
+
+#: Leading characters Excel/Sheets/LibreOffice interpret as a formula
+#: (OWASP CSV-injection guidance).
+_FORMULA_PREFIXES = ("=", "+", "-", "@", "\t", "\r")
+
+
+def _formula_guard(value: object) -> object:
+    if isinstance(value, str) and value.startswith(_FORMULA_PREFIXES):
+        return "'" + value
+    return value
+
+
+def sanitize_csv_formulas(df: pd.DataFrame) -> pd.DataFrame:
+    """Copy of *df* safe to open in a spreadsheet: string cells (and column
+    labels) starting with ``= + - @ <tab> <cr>`` are prefixed with ``'`` so
+    they render as text instead of executing as formulas. Non-string cells
+    (including negative numbers) are untouched.
+    """
+    out = df.copy()
+    for i, dtype in enumerate(out.dtypes):
+        if _is_stringlike_dtype(dtype) or isinstance(dtype, pd.CategoricalDtype):
+            column = out.iloc[:, i]
+            guarded = column.astype(object).map(_formula_guard)
+            if not guarded.equals(column.astype(object)):
+                out.isetitem(i, guarded)
+    out.columns = pd.Index([_formula_guard(c) for c in out.columns])
+    return out

src/freshdata/api.py

@@ -10,6 +10,7 @@
 import pandas as pd
 
 from ._reportframe import ReportFrame
+from ._util import sanitize_csv_formulas
 from .adapters.polars import from_pandas, to_pandas
 from .cleaner import Cleaner, run_pipeline
 from .config import CleanConfig, merge_options
@@ -423,6 +424,7 @@ def clean_csv(
     policy: object | None = None,
     strict: bool = False,
     profile: object | None = None,
+    sanitize_formulas: bool = False,
     **options: object,
 ) -> pd.DataFrame | tuple[pd.DataFrame, CleanReport]:
     """Read a CSV file, clean it, and optionally write the result to disk.
@@ -433,6 +435,12 @@ def clean_csv(
         Path to the input CSV file.
     output_path:
         Optional path to write the cleaned CSV.
+    sanitize_formulas:
+        If ``True``, string cells (and column labels) in the **written**
+        file that start with ``= + - @ <tab> <cr>`` are prefixed with ``'``
+        so spreadsheets render them as text instead of executing them
+        (OWASP CSV-injection guidance). Off by default to keep the output
+        byte-exact; the returned DataFrame is never altered.
     return_report:
         If True, return ``(cleaned_df, CleanReport)``.
     read_csv_kwargs:
@@ -473,7 +481,8 @@ def clean_csv(
     )
     cleaned_df = cast(pd.DataFrame, result[0] if return_report else result)
     if output_path is not None:
-        cleaned_df.to_csv(output_path, **{"index": False, **(to_csv_kwargs or {})})
+        to_write = sanitize_csv_formulas(cleaned_df) if sanitize_formulas else cleaned_df
+        to_write.to_csv(output_path, **{"index": False, **(to_csv_kwargs or {})})
     return result
 
 

src/freshdata/enterprise/entity_resolution.py

@@ -36,6 +36,7 @@
 
 import pandas as pd
 
+from .._util import sanitize_csv_formulas
 from ..adapters.polars import from_pandas, to_pandas
 from .config import (  # noqa: F401  (configs re-exported for discoverability)
     BlockingRule,
@@ -1424,11 +1425,18 @@ def export_review_queue(
     *,
     format: str | None = None,
     config: ReviewQueueConfig | None = None,
+    sanitize_formulas: bool = True,
 ) -> Path:
     """Write a review queue to *path* as ``csv``, ``jsonl``, or ``parquet``.
 
     *report* may be a freshly-built :class:`ReviewQueueReport` or a raw
     :class:`EntityResolutionReport` (in which case a queue is built first).
+
+    Review queues exist to be opened by humans in spreadsheets, so the
+    ``csv`` format neutralizes formula-injection payloads by default: string
+    cells starting with ``= + - @ <tab> <cr>`` are prefixed with ``'``
+    (OWASP CSV-injection guidance). Pass ``sanitize_formulas=False`` for a
+    byte-exact export. Other formats are never altered.
     """
     queue = (
         report
@@ -1443,7 +1451,10 @@ def export_review_queue(
             for it in queue.items:
                 fh.write(json.dumps(it.to_dict(), default=str) + "\n")
     elif fmt == "csv":
-        queue.to_frame().to_csv(out, index=False)
+        frame = queue.to_frame()
+        if sanitize_formulas:
+            frame = sanitize_csv_formulas(frame)
+        frame.to_csv(out, index=False)
     else:  # parquet
         queue.to_frame().to_parquet(out, index=False)
     return out

src/freshdata/streaming/_cli.py

@@ -17,6 +17,7 @@
 
 import pandas as pd
 
+from .._util import sanitize_csv_formulas
 from ._cleaner import StreamingCleaner
 
 
@@ -34,10 +35,11 @@ def _read_chunks(path: str, batch_size: int) -> Iterator[pd.DataFrame]:
 class _BatchWriter:
     """Append cleaned batches to one CSV or Parquet file without buffering them all."""
 
-    def __init__(self, path: str | None) -> None:
+    def __init__(self, path: str | None, sanitize_formulas: bool = False) -> None:
         self.path = path
         self.fmt = None if path is None else ("parquet"
                     if path.lower().endswith((".parquet", ".pq")) else "csv")
+        self.sanitize_formulas = sanitize_formulas
         self._pq_writer: Any = None
         self._csv_header = True
 
@@ -53,6 +55,8 @@ def write(self, df: pd.DataFrame) -> None:
                 self._pq_writer = pq.ParquetWriter(self.path, table.schema)
             self._pq_writer.write_table(table)
         else:
+            if self.sanitize_formulas:
+                df = sanitize_csv_formulas(df)
             df.to_csv(self.path, mode="w" if self._csv_header else "a",
                       header=self._csv_header, index=False)
             self._csv_header = False
@@ -109,7 +113,8 @@ def _timeseries_config(args: argparse.Namespace) -> Any:
     return TimeSeriesCleanConfig(**kwargs)
 
 
-def _write_exceptions(cleaner: StreamingCleaner, path: str | None) -> None:
+def _write_exceptions(cleaner: StreamingCleaner, path: str | None,
+                      sanitize_formulas: bool = False) -> None:
     """Persist any quarantined (late/anomalous) rows to *path* (CSV or Parquet)."""
     if not path:
         return
@@ -119,12 +124,15 @@ def _write_exceptions(cleaner: StreamingCleaner, path: str | None) -> None:
     if path.lower().endswith((".parquet", ".pq")):
         exc.to_parquet(path, index=False)
     else:
+        if sanitize_formulas:
+            exc = sanitize_csv_formulas(exc)
         exc.to_csv(path, index=False)
 
 
 def _run_stream(cleaner: StreamingCleaner, batches: Iterator[pd.DataFrame],
                 writer: _BatchWriter, report_dir: str | None, quiet: bool,
-                quarantine_path: str | None = None) -> int:
+                quarantine_path: str | None = None,
+                sanitize_formulas: bool = False) -> int:
     if report_dir:
         os.makedirs(report_dir, exist_ok=True)
     for cleaned, report in cleaner.clean_batches(batches):
@@ -137,7 +145,7 @@ def _run_stream(cleaner: StreamingCleaner, batches: Iterator[pd.DataFrame],
             print(json.dumps(report.streaming))
     writer.close()
     final = cleaner.finalize()
-    _write_exceptions(cleaner, quarantine_path)
+    _write_exceptions(cleaner, quarantine_path, sanitize_formulas=sanitize_formulas)
     if report_dir:
         with open(os.path.join(report_dir, "summary.json"), "w") as fh:
             json.dump(final.to_dict(), fh, default=str)
@@ -148,17 +156,21 @@ def _run_stream(cleaner: StreamingCleaner, batches: Iterator[pd.DataFrame],
 
 def cmd_stream(args: argparse.Namespace) -> int:
     cleaner = StreamingCleaner(**_stream_options(args))
+    sanitize = getattr(args, "sanitize_formulas", False)
     return _run_stream(cleaner, _read_chunks(args.input, args.batch_size),
-                       _BatchWriter(args.output), args.report, args.quiet,
-                       getattr(args, "quarantine", None))
+                       _BatchWriter(args.output, sanitize_formulas=sanitize),
+                       args.report, args.quiet,
+                       getattr(args, "quarantine", None),
+                       sanitize_formulas=sanitize)
 
 
 def cmd_stream_kafka(args: argparse.Namespace) -> int:
     cleaner = StreamingCleaner(**_stream_options(args))
     batches = cleaner.clean_kafka(  # validates the kafka dependency
         topic=args.topic, bootstrap_servers=args.bootstrap_servers,
         batch_size=args.batch_size, max_batches=args.max_batches)
-    writer = _BatchWriter(args.output)
+    writer = _BatchWriter(args.output,
+                          sanitize_formulas=getattr(args, "sanitize_formulas", False))
     if args.report:
         os.makedirs(args.report, exist_ok=True)
     for cleaned, report in batches:
@@ -234,6 +246,9 @@ def add_stream_subparsers(subparsers: argparse._SubParsersAction) -> None:
     s = subparsers.add_parser("stream", help="clean a CSV/Parquet file in micro-batches")
     s.add_argument("input")
     s.add_argument("-o", "--output")
+    s.add_argument("--sanitize-formulas", action="store_true",
+                   help="prefix ' to string cells starting with = + - @ tab/CR in "
+                        "CSV outputs so spreadsheets render them as text (OWASP)")
     s.add_argument("--batch-size", "--chunksize", type=int, default=100_000, dest="batch_size")
     s.add_argument("--report", metavar="DIR", help="directory for per-batch + summary JSON")
     s.add_argument("--target-column")
@@ -252,6 +267,9 @@ def add_stream_subparsers(subparsers: argparse._SubParsersAction) -> None:
     k.add_argument("--batch-size", type=int, default=10_000)
     k.add_argument("--max-batches", type=int)
     k.add_argument("-o", "--output")
+    k.add_argument("--sanitize-formulas", action="store_true",
+                   help="prefix ' to string cells starting with = + - @ tab/CR in "
+                        "CSV outputs so spreadsheets render them as text (OWASP)")
     k.add_argument("--report", metavar="DIR")
     k.add_argument("--target-column")
     k.add_argument("--id-columns", nargs="*", default=())