feat(connectors): add the file connector with fail-closed path boundary

TDD step 3: read_csv executes a file-source extraction — boundary-checked
path resolution via commonpath on canonical paths (rejects .. traversal,
absolute paths, symlink escapes, and prefix-collision siblings), utf-8-sig
decoding, streaming max_rows cap as a typed error (never silent
truncation), ragged-row rejection, and verbatim cells (escaping stays in
the §5 renderers). New SourceError in the typed hierarchy.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01QeqhJpYQyghASjiJo5EhGg

src/llm_ingestion_okf/connectors.py

@@ -0,0 +1,67 @@
+"""Connectors: execute manifest extractions against sources (ingest-spec §4).
+
+Pure readers — no logging, no writes (materialization owns the §8 log).
+Input-shape failures raise SourceError fail-fast, never a leaked OSError.
+"""
+
+from __future__ import annotations
+
+import csv
+import os
+from pathlib import Path
+
+from .errors import SourceError
+
+
+def _safe_resolve(root: Path, relative: str) -> Path:
+    """Resolve `relative` against `root`, fail-closed (the OKF path rule).
+
+    Raises SourceError if the resolved path escapes `root` — `..` traversal,
+    an absolute query path, a symlink escape, or a prefix-collision sibling
+    (`/a/data-evil` vs `/a/data`; commonpath on canonical paths catches what
+    a naive startswith would not).
+    """
+    real_root = os.path.realpath(root)
+    candidate = os.path.realpath(os.path.join(real_root, relative))
+    try:
+        within = os.path.commonpath([real_root, candidate]) == real_root
+    except ValueError:
+        # Different drives / mixed absolute-relative -> not within.
+        within = False
+    if not within:
+        raise SourceError(f"extraction query escapes the source root: {relative!r}")
+    return Path(candidate)
+
+
+def read_csv(root: str | Path, query: str, *, max_rows: int) -> tuple[list[str], list[list[str]]]:
+    """Execute a `file`-source extraction: read the CSV at `query` inside `root`.
+
+    Boundary-checked fail-closed path resolution; utf-8-sig so a BOM never
+    leaks into the first header cell; streaming max_rows cap — exceeding it
+    is an ERROR the moment it happens, never silent truncation (§8);
+    ragged rows rejected, cells returned verbatim (escaping is §5 rendering).
+    """
+    root_path = Path(root)
+    if not root_path.is_dir():
+        raise SourceError(f"file-source root is not a directory: {root_path}")
+    resolved = _safe_resolve(root_path, query)
+    if not resolved.is_file():
+        raise SourceError(f"extraction query does not resolve to a file: {query!r}")
+    with resolved.open(encoding="utf-8-sig", newline="") as handle:
+        reader = csv.reader(handle)
+        header = next(reader, None)
+        if header is None:
+            raise SourceError(f"CSV has no header row: {query!r}")
+        rows: list[list[str]] = []
+        for row in reader:
+            if len(rows) >= max_rows:
+                raise SourceError(
+                    f"extraction {query!r} exceeds max_rows={max_rows} "
+                    "(error, never silent truncation — spec §8)"
+                )
+            if len(row) != len(header):
+                raise SourceError(
+                    f"ragged CSV row in {query!r}: expected {len(header)} cells, got {len(row)}"
+                )
+            rows.append(row)
+    return header, rows

src/llm_ingestion_okf/errors.py

@@ -13,3 +13,12 @@ class ManifestError(IngestError):
 
 class RenderError(IngestError):
     """A value cannot be rendered under the §5 body rules (never silent coercion)."""
+
+
+class SourceError(IngestError):
+    """An extraction failed against its source (ingest-spec §4, §8).
+
+    Covers fail-closed path-boundary violations, missing/malformed source
+    content, and max_rows cap violations — always typed, never a leaked
+    OSError and never silent truncation.
+    """

tests/test_file_connector.py

@@ -0,0 +1,154 @@
+"""The `file` connector (ingest-spec §4, §8).
+
+CSV under `root` with fail-closed boundary-checked path resolution,
+utf-8-sig decoding (BOM never leaks into the first header cell), streaming
+max_rows cap (error, never silent truncation), and ragged-row rejection.
+"""
+
+from __future__ import annotations
+
+import os
+from pathlib import Path
+
+import pytest
+
+from llm_ingestion_okf.connectors import read_csv
+from llm_ingestion_okf.errors import IngestError, SourceError
+
+
+def write_csv(root: Path, name: str, text: str) -> Path:
+    path = root / name
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_text(text, encoding="utf-8", newline="")
+    return path
+
+
+# --- error hierarchy ---
+
+
+def test_source_error_is_ingest_error() -> None:
+    assert issubclass(SourceError, IngestError)
+
+
+# --- happy path ---
+
+
+def test_reads_header_and_rows_in_source_order(tmp_path: Path) -> None:
+    write_csv(tmp_path, "orders.csv", "a,b\n1,x\n2,y\n")
+    header, rows = read_csv(tmp_path, "orders.csv", max_rows=10)
+    assert header == ["a", "b"]
+    assert rows == [["1", "x"], ["2", "y"]]
+
+
+def test_reads_csv_in_subdirectory(tmp_path: Path) -> None:
+    write_csv(tmp_path, "sub/orders.csv", "a\n1\n")
+    header, rows = read_csv(tmp_path, "sub/orders.csv", max_rows=10)
+    assert header == ["a"]
+    assert rows == [["1"]]
+
+
+def test_cells_are_verbatim_text(tmp_path: Path) -> None:
+    # No escaping at the connector layer — that is render_table's job.
+    write_csv(tmp_path, "t.csv", 'a\n"pipe|and\\slash"\n')
+    _, rows = read_csv(tmp_path, "t.csv", max_rows=10)
+    assert rows == [["pipe|and\\slash"]]
+
+
+def test_quoted_cell_with_embedded_newline_survives_verbatim(tmp_path: Path) -> None:
+    write_csv(tmp_path, "t.csv", 'a,b\n"line1\nline2",x\n')
+    _, rows = read_csv(tmp_path, "t.csv", max_rows=10)
+    assert rows == [["line1\nline2", "x"]]
+
+
+def test_bom_never_leaks_into_first_header_cell(tmp_path: Path) -> None:
+    write_csv(tmp_path, "t.csv", "\ufeffa,b\n1,2\n")
+    header, _ = read_csv(tmp_path, "t.csv", max_rows=10)
+    assert header == ["a", "b"]
+
+
+# --- boundary check: fail-closed path resolution (the OKF path rule) ---
+
+
+def test_dotdot_traversal_rejected(tmp_path: Path) -> None:
+    root = tmp_path / "root"
+    root.mkdir()
+    write_csv(tmp_path, "outside.csv", "a\n1\n")
+    with pytest.raises(SourceError):
+        read_csv(root, "../outside.csv", max_rows=10)
+
+
+def test_absolute_query_path_rejected(tmp_path: Path) -> None:
+    root = tmp_path / "root"
+    root.mkdir()
+    outside = write_csv(tmp_path, "outside.csv", "a\n1\n")
+    with pytest.raises(SourceError):
+        read_csv(root, str(outside), max_rows=10)
+
+
+def test_symlink_escape_rejected(tmp_path: Path) -> None:
+    root = tmp_path / "root"
+    root.mkdir()
+    write_csv(tmp_path, "outside.csv", "a\n1\n")
+    os.symlink(tmp_path / "outside.csv", root / "link.csv")
+    with pytest.raises(SourceError):
+        read_csv(root, "link.csv", max_rows=10)
+
+
+def test_prefix_collision_sibling_rejected(tmp_path: Path) -> None:
+    root = tmp_path / "data"
+    root.mkdir()
+    write_csv(tmp_path, "data-evil/t.csv", "a\n1\n")
+    with pytest.raises(SourceError):
+        read_csv(root, "../data-evil/t.csv", max_rows=10)
+
+
+# --- input-shape failures: typed errors, never leaked OSError ---
+
+
+def test_missing_root_rejected(tmp_path: Path) -> None:
+    with pytest.raises(SourceError):
+        read_csv(tmp_path / "nope", "t.csv", max_rows=10)
+
+
+def test_root_that_is_a_file_rejected(tmp_path: Path) -> None:
+    file_root = write_csv(tmp_path, "afile", "x")
+    with pytest.raises(SourceError):
+        read_csv(file_root, "t.csv", max_rows=10)
+
+
+def test_query_not_resolving_to_file_rejected(tmp_path: Path) -> None:
+    with pytest.raises(SourceError):
+        read_csv(tmp_path, "missing.csv", max_rows=10)
+
+
+def test_empty_csv_rejected(tmp_path: Path) -> None:
+    write_csv(tmp_path, "t.csv", "")
+    with pytest.raises(SourceError):
+        read_csv(tmp_path, "t.csv", max_rows=10)
+
+
+def test_ragged_row_rejected(tmp_path: Path) -> None:
+    write_csv(tmp_path, "t.csv", "a,b\n1\n")
+    with pytest.raises(SourceError):
+        read_csv(tmp_path, "t.csv", max_rows=10)
+
+
+# --- max_rows cap (spec §8): error, never silent truncation ---
+
+
+def test_row_count_at_cap_is_allowed(tmp_path: Path) -> None:
+    write_csv(tmp_path, "t.csv", "a\n1\n2\n")
+    _, rows = read_csv(tmp_path, "t.csv", max_rows=2)
+    assert len(rows) == 2
+
+
+def test_exceeding_cap_is_an_error(tmp_path: Path) -> None:
+    write_csv(tmp_path, "t.csv", "a\n1\n2\n3\n")
+    with pytest.raises(SourceError):
+        read_csv(tmp_path, "t.csv", max_rows=2)
+
+
+def test_header_does_not_count_toward_cap(tmp_path: Path) -> None:
+    write_csv(tmp_path, "t.csv", "a\n1\n")
+    _, rows = read_csv(tmp_path, "t.csv", max_rows=1)
+    assert rows == [["1"]]