diff --git a/tests/inbox_frontend.py b/tests/inbox_frontend.py index 4a5119c..b2055ab 100644 --- a/tests/inbox_frontend.py +++ b/tests/inbox_frontend.py @@ -17,6 +17,8 @@ Slice 2a covers the text formats — ``.txt`` and ``.md`` (stdlib only). ``.zip` """ from __future__ import annotations +import csv +import io import stat import zipfile from dataclasses import dataclass @@ -41,6 +43,12 @@ _MATERIALIZE_PREFIX = "uploads" # The text formats this slice reads directly (no parser dependency). _TEXT_SUFFIXES = {".txt", ".md"} +# CSV cells leading with any of these parse as a formula in a spreadsheet — the +# CSV-injection / DDE vector (RCE when a human opens the file). Leading whitespace +# does not defuse it, so it is stripped before the check. Numeric cells that lead +# with '-'/'+' are the accepted false-positive (README honest-limits). +_FORMULA_LEADS = ("=", "+", "-", "@") + # Zip self-safety caps (OWASP LLM10). Bounded so a decompression bomb is refused # before its uncompressed bytes are read into memory. Defaults are generous for a # document inbox; tests pass small caps to exercise the gate. @@ -87,6 +95,63 @@ def _is_symlink_entry(info: zipfile.ZipInfo) -> bool: return stat.S_ISLNK(info.external_attr >> 16) +def _materialize_text(rel_name, text, source_type, bundle, provenance): + """Add one text concept to the bundle under the materialize prefix.""" + concept_path = _materialize_path(rel_name) + bundle[concept_path] = text + provenance.append(Provenance(concept_path, rel_name, source_type)) + + +def _is_formula_cell(cell: str) -> bool: + stripped = cell.lstrip("\t\r\n ") + return bool(stripped) and stripped[0] in _FORMULA_LEADS + + +def _ingest_csv(rel_name, text, bundle, provenance, rejected): + """Materialize CSV cell text as a concept and flag formula-injection cells. + + The raw cell text becomes the concept body so a prompt-injection *phrase* in + a cell is caught by the stage-2 scan; formula-lead cells are a spreadsheet + threat the guard would not recognize, so the front-end refuses them here. + """ + formula_cells = [ + cell for row in csv.reader(io.StringIO(text)) for cell in row if _is_formula_cell(cell) + ] + if formula_cells: + rejected.append( + (rel_name, f"CSV formula-injection lead in {len(formula_cells)} cell(s): {formula_cells[0][:24]!r}") + ) + _materialize_text(rel_name, text, "csv", bundle, provenance) + + +def _ingest_regular_file(fs_path, rel_name, bundle, provenance, rejected, *, strict): + """Dispatch one on-disk file by suffix. ``strict`` raises on an unsupported + suffix (a top-level drop); a folder walk passes ``strict=False`` to skip it.""" + suffix = Path(rel_name).suffix.lower() + if suffix == ".csv": + text = Path(fs_path).read_text(encoding="utf-8", errors="replace") + _ingest_csv(rel_name, text, bundle, provenance, rejected) + elif suffix in _TEXT_SUFFIXES: + text = Path(fs_path).read_text(encoding="utf-8", errors="replace") + _materialize_text(rel_name, text, suffix.lstrip("."), bundle, provenance) + elif strict: + raise ValueError(f"unsupported upload format in this slice: {Path(rel_name).name!r}") + + +def _extract_folder(root, bundle, provenance, rejected): + """Walk a dropped folder, materializing its text/CSV members (relative paths + preserved, so a reserved-name member trips the guard's path gate).""" + root = Path(root) + for fs_path in sorted(root.rglob("*")): + if fs_path.is_symlink(): + rejected.append((str(fs_path.relative_to(root)), "symlink refused (container threat)")) + continue + if not fs_path.is_file(): + continue + rel_name = fs_path.relative_to(root).as_posix() + _ingest_regular_file(fs_path, rel_name, bundle, provenance, rejected, strict=False) + + def _extract_zip(path, bundle, provenance, rejected, max_entry_bytes, max_total_bytes): """Read a ``.zip`` in memory, materializing its text entries; refuse bombs, symlinks and oversize entries at the front-end (container threats).""" @@ -128,25 +193,23 @@ def extract_inbox( ) -> InboxExtract: """Read dropped files and materialize them into an OKF bundle + provenance. - ``paths`` is an iterable of file paths. Each ``.txt`` / ``.md`` becomes one - concept (a ``.md`` keeps its OKF frontmatter verbatim). A ``.zip`` is read in - memory and its text entries materialized, with bomb/symlink/oversize entries - refused into ``InboxExtract.rejected``. Other suffixes are not handled yet. + ``paths`` is an iterable of file/folder paths. Each ``.txt`` / ``.md`` becomes + one concept (a ``.md`` keeps its OKF frontmatter verbatim); a ``.csv`` is + materialized and its formula-lead cells refused; a ``.zip`` is read in memory + and its text entries materialized, with bomb/symlink/oversize entries refused; + a folder is walked member-by-member. Refusals land in ``InboxExtract.rejected``. """ bundle: dict = {} provenance: list = [] rejected: list = [] for path in paths: path = Path(path) - suffix = path.suffix.lower() - if suffix == ".zip": + if path.is_dir(): + _extract_folder(path, bundle, provenance, rejected) + elif path.suffix.lower() == ".zip": _extract_zip(path, bundle, provenance, rejected, max_entry_bytes, max_total_bytes) - elif suffix in _TEXT_SUFFIXES: - concept_path = _materialize_path(path.name) - bundle[concept_path] = path.read_text(encoding="utf-8") - provenance.append(Provenance(concept_path, path.name, suffix.lstrip("."))) else: - raise ValueError(f"unsupported upload format in this slice: {path.name!r}") + _ingest_regular_file(path, path.name, bundle, provenance, rejected, strict=True) return InboxExtract(bundle, tuple(provenance), tuple(rejected)) diff --git a/tests/test_okf_inbox_uploads.py b/tests/test_okf_inbox_uploads.py index 5bfb2e6..719b433 100644 --- a/tests/test_okf_inbox_uploads.py +++ b/tests/test_okf_inbox_uploads.py @@ -153,3 +153,74 @@ def test_zip_symlink_entry_is_refused(tmp_path): assert any("link.md" in n for n, _reason in extracted.rejected) assert "uploads/link.md" not in extracted.bundle assert verdict == "REJECT" + + +# --- slice 2c: .csv formula injection + folder walk ------------------------- +# CSV cells that lead with =, +, -, @ are formula-injection vectors (RCE/DDE when +# a human opens the file in a spreadsheet). The front-end owns that format threat; +# a prompt-injection *phrase* in a cell is materialized into the concept text and +# caught by the stage-2 scan instead. A folder is walked member-by-member. + + +def test_csv_formula_injection_cell_is_flagged(tmp_path): + p = _write(tmp_path, "data.csv", "name,note\nAlice,=cmd|'/c calc'!A1\nBob,ok\n") + extracted, _result, verdict = receive([p]) + assert any("data.csv" in n for n, _r in extracted.rejected) + assert verdict == "REJECT" + + +def test_csv_formula_leading_whitespace_bypass_is_flagged(tmp_path): + # a tab/space before the '=' still parses as a formula in a spreadsheet. + p = _write(tmp_path, "sneaky.csv", 'a,b\n1,"\t=HYPERLINK(\'http://evil\')"\n') + extracted, _result, verdict = receive([p]) + assert extracted.rejected != () + assert verdict == "REJECT" + + +def test_clean_csv_admits(tmp_path): + p = _write(tmp_path, "clean.csv", "name,count\nAlice,3\nBob,5\n") + extracted, _result, verdict = receive([p]) + assert extracted.rejected == () + assert verdict == "ADMIT" + assert "uploads/clean.md" in extracted.bundle + + +def test_csv_formula_detach_proof(tmp_path): + # The same file with plain cells has no formula flag -> ADMIT, so the flag is + # the formula content, not the .csv suffix or the filename. + p = _write(tmp_path, "data.csv", "name,note\nAlice,calc\nBob,ok\n") + extracted, _result, verdict = receive([p]) + assert extracted.rejected == () + assert verdict == "ADMIT" + + +def test_csv_injection_phrase_in_cell_is_caught_by_the_guard(tmp_path): + # not a formula — a prompt injection sitting in a cell. It rides the + # materialized concept text into the stage-2 scan (T1), not the formula check. + p = _write(tmp_path, "notes.csv", "id,note\n1," + _INJECTION + "\n") + extracted, result, verdict = receive([p]) + assert extracted.rejected == () # no formula lead + assert result.disposition is Disposition.FAIL_SECURE # caught by the guard + assert verdict == "REJECT" + + +def test_folder_upload_reserved_member_is_rejected(tmp_path): + folder = tmp_path / "bundle" + (folder / "tables").mkdir(parents=True) + (folder / "tables" / "users.md").write_text("---\ntype: t\n---\nclean.\n", encoding="utf-8") + (folder / "index.md").write_text("---\ntype: t\n---\nlisting.\n", encoding="utf-8") + _extracted, result, verdict = receive([folder]) + by_path = {c.path: c for c in result.concepts} + assert "uploads/index.md" in by_path # reserved basename + assert by_path["uploads/index.md"].disposition is Disposition.FAIL_SECURE # T4 + assert verdict == "REJECT" + + +def test_clean_folder_admits(tmp_path): + folder = tmp_path / "bundle" + (folder / "tables").mkdir(parents=True) + (folder / "tables" / "users.md").write_text("---\ntype: t\n---\nclean.\n", encoding="utf-8") + (folder / "notes.txt").write_text("A routine note.", encoding="utf-8") + extracted, _result, verdict = receive([folder]) + assert set(extracted.bundle) == {"uploads/tables/users.md", "uploads/notes.md"} + assert verdict == "ADMIT"