Fix parser and export security controls (#115)

Co-authored-by: Johnny Wilson Dougherty <192861341+JohnnyWilson-Portfolio@users.noreply.github.com>

Author: Rohith27216

docs/parsers.md

@@ -31,8 +31,8 @@ domain-validation are a separate step (`fd.clean`), so parsing and rules stay de
 ### FHIR R4 JSON
 
 `fd.parse_domain(source, format="fhir")` accepts a **Bundle**, a single resource, a list
-of resources, a JSON string, or a file path, and flattens five resource types into frames
-whose columns line up with the healthcare validators:
+of resources, a JSON string, or a `pathlib.Path`, and flattens five resource types into
+frames whose columns line up with the healthcare validators:
 
 ```python
 result = fd.parse_domain(bundle_json, format="fhir")
@@ -64,9 +64,10 @@ patients = fd.clean_domain_file(
 )
 ```
 
-`fd.parse_domain` accepts a **path, raw text/bytes, or a file-like object**. Malformed
-input is recorded in `ParseResult.warnings` rather than raising, so a partial message is
-still usable.
+`fd.parse_domain` accepts **raw text/bytes, a `pathlib.Path`, or a file-like object**.
+String values are treated as raw content; use `clean_domain_file("admit.hl7", ...)` or
+pass a `Path` when you want filesystem input. Malformed input is recorded in
+`ParseResult.warnings` rather than raising, so a partial message is still usable.
 
 !!! note "Honest scope"
     Parsers are structural readers for the common parts of each format — HL7 MSH/PID/PV1/OBX,

src/freshdata/api.py

@@ -1012,9 +1012,11 @@ def profile(
 def parse_domain(source: Any, *, format: str) -> ParseResult:  # noqa: A002
     """Parse a raw message or file into DataFrames using the named *format* parser.
 
-    *source* may be a filesystem path, the raw text/bytes content, or a file-like
-    object. *format* is a registered parser name (``"hl7v2"``, ``"gpx"``, ``"sdmx"``,
-    ``"edifact"`` — see :func:`freshdata.parsers.available`).
+    *source* may be raw text/bytes content, a file-like object, or a
+    :class:`pathlib.Path` for filesystem input. String values are treated as content;
+    use :func:`clean_domain_file` for the convenience file-path workflow. *format* is a
+    registered parser name (``"hl7v2"``, ``"gpx"``, ``"sdmx"``, ``"edifact"`` — see
+    :func:`freshdata.parsers.available`).
 
     Returns a :class:`~freshdata.parsers.ParseResult` carrying the parsed frames,
     a suggested domain, metadata, and any audit warnings.
@@ -1044,7 +1046,8 @@ def clean_domain_file(
     through :func:`clean` with *domain* and any extra cleaning keyword arguments. When a
     parser yields several non-empty frames, pass ``frame=`` to pick one.
     """
-    result = parse_domain(path, format=format)
+    source = Path(path) if isinstance(path, str) and Path(path).exists() else path
+    result = parse_domain(source, format=format)
     if domain is None:
         # suggested_domain is advisory metadata, not an instruction to clean.
         return result

src/freshdata/integrations/dbt/__init__.py

@@ -20,7 +20,7 @@
 import logging
 import os
 from dataclasses import dataclass
-from pathlib import Path
+from pathlib import Path, PurePath
 from typing import TYPE_CHECKING, Any
 
 from .._core import OnLowScore, TrustGateError, TrustGateResult, evaluate_trust_gate
@@ -41,6 +41,20 @@
 )
 
 
+def _validate_audit_table_name(table: str) -> str:
+    path = PurePath(table)
+    if (
+        not table
+        or table in {".", ".."}
+        or path.is_absolute()
+        or path.name != table
+        or "/" in table
+        or "\\" in table
+    ):
+        raise ValueError(f"{table!r} is not a safe dbt model name for audit output")
+    return table
+
+
 def _read_table(conn_str: str, schema: str | None, table: str) -> pd.DataFrame:
     """Read ``schema.table`` from ``conn_str`` into a DataFrame."""
     import pandas as pd
@@ -89,6 +103,7 @@ def _split_table(self) -> tuple[str | None, str]:
         return None, self.model_name
 
     def _write_audit(self, table: str, result: TrustGateResult) -> Path:
+        table = _validate_audit_table_name(table)
         out_dir = Path(self.output_dir)  # type: ignore[arg-type]
         out_dir.mkdir(parents=True, exist_ok=True)
         path = out_dir / f"{table}_audit.json"
@@ -103,6 +118,8 @@ def run(self) -> TrustGateResult:
                 "No warehouse connection: pass conn_str or set FRESHDATA_WAREHOUSE_CONN."
             )
         schema, table = self._split_table()
+        if self.output_dir:
+            _validate_audit_table_name(table)
         df = _read_table(conn, schema, table)
         _, result = evaluate_trust_gate(
             df,

src/freshdata/integrations/exceptions.py

@@ -138,6 +138,14 @@ def _infer_format(path: str) -> str:
     return "csv"
 
 
+def _quote_duckdb_identifier(identifier: str) -> str:
+    if not isinstance(identifier, str) or not identifier:
+        raise ValueError("DuckDB table_name must be a non-empty string")
+    if "\x00" in identifier:
+        raise ValueError("DuckDB table_name must not contain NUL bytes")
+    return f'"{identifier.replace(chr(34), chr(34) * 2)}"'
+
+
 def write_exception_table(
     table: pd.DataFrame,
     path: str,
@@ -175,7 +183,7 @@ def write_exception_table(
         try:
             con.register("_freshdata_exceptions", table)
             con.execute(
-                f'CREATE OR REPLACE TABLE "{table_name}" AS '
+                f"CREATE OR REPLACE TABLE {_quote_duckdb_identifier(table_name)} AS "
                 "SELECT * FROM _freshdata_exceptions"
             )
         finally:

src/freshdata/parsers/base.py

@@ -10,14 +10,15 @@
 from __future__ import annotations
 
 import io
-import os
 from abc import ABC, abstractmethod
 from dataclasses import dataclass, field
 from pathlib import Path
 from typing import Any
 
 import pandas as pd
 
+_MAX_XML_BYTES = 10 * 1024 * 1024
+
 
 @dataclass
 class ParseResult:
@@ -76,10 +77,10 @@ class Parser(ABC):
 
     @abstractmethod
     def parse(self, source: Any) -> ParseResult:
-        """Parse *source* (path, text, bytes, or file-like) into a :class:`ParseResult`."""
+        """Parse *source* (Path, text, bytes, or file-like) into a :class:`ParseResult`."""
 
     def read_text(self, source: Any, *, encoding: str = "utf-8") -> str:
-        """Read *source* into text, accepting a path, str content, bytes, or file-like."""
+        """Read *source* into text, accepting a Path, str content, bytes, or file-like."""
         if isinstance(source, (bytes, bytearray)):
             return bytes(source).decode(encoding)
         if hasattr(source, "read"):
@@ -88,11 +89,6 @@ def read_text(self, source: Any, *, encoding: str = "utf-8") -> str:
         if isinstance(source, Path):
             return source.read_text(encoding=encoding)
         if isinstance(source, str):
-            # A short string that names an existing file is treated as a path;
-            # otherwise it is treated as the content itself.
-            if (len(source) < 4096 and "\n" not in source and "\r" not in source
-                    and os.path.exists(source)):
-                return Path(source).read_text(encoding=encoding)
             return source
         raise TypeError(f"cannot read a {type(source).__name__} source")
 
@@ -104,8 +100,29 @@ def open_binary(self, source: Any) -> io.BufferedIOBase | io.BytesIO:
             data = source.read()
             return io.BytesIO(data if isinstance(data, (bytes, bytearray))
                               else str(data).encode("utf-8"))
-        if isinstance(source, (str, Path)) and os.path.exists(str(source)):
+        if isinstance(source, Path):
             return open(source, "rb")  # noqa: SIM115 - caller consumes immediately
-        if isinstance(source, (str, Path)):
+        if isinstance(source, str):
             return io.BytesIO(str(source).encode("utf-8"))
         raise TypeError(f"cannot open a {type(source).__name__} source")
+
+    def open_safe_xml_binary(
+        self,
+        source: Any,
+        *,
+        max_bytes: int = _MAX_XML_BYTES,
+    ) -> io.BytesIO:
+        """Return bounded XML bytes with DTD/entity declarations rejected."""
+        stream = self.open_binary(source)
+        try:
+            data = stream.read(max_bytes + 1)
+        finally:
+            if hasattr(stream, "close"):
+                stream.close()
+
+        if len(data) > max_bytes:
+            raise ValueError(f"XML input exceeds {max_bytes} bytes")
+        lowered = data.lower()
+        if b"<!doctype" in lowered or b"<!entity" in lowered:
+            raise ValueError("XML DTD/entity declarations are not allowed")
+        return io.BytesIO(data)

src/freshdata/parsers/fhir.py

@@ -1,10 +1,10 @@
 """FHIR R4 JSON parser.
 
 Parses a FHIR R4 Bundle, a single resource, a list of resources, a JSON string, or a
-file path into flattened DataFrames keyed by resource: ``patient``, ``observation``,
-``encounter``, ``condition``, ``medication_request``. The flattened columns line up with
-the healthcare domain pack's resource validators, so a parsed frame can go straight into
-``fd.clean(frame, domain="healthcare")``.
+``pathlib.Path`` into flattened DataFrames keyed by resource: ``patient``,
+``observation``, ``encounter``, ``condition``, ``medication_request``. The flattened
+columns line up with the healthcare domain pack's resource validators, so a parsed frame
+can go straight into ``fd.clean(frame, domain="healthcare")``.
 
 Only predictable R4 fields are flattened; resource types the parser does not handle are
 counted and surfaced in :attr:`ParseResult.warnings` rather than dropped silently.
@@ -195,7 +195,7 @@ def parse(self, source: Any) -> ParseResult:
         )
 
     def _load_json(self, source: Any) -> Any:
-        """Return parsed JSON, accepting a dict/list directly or a path/text/bytes."""
+        """Return parsed JSON, accepting a dict/list directly or a Path/text/bytes."""
         if isinstance(source, (dict, list)):
             return source
         return json.loads(self.read_text(source))

src/freshdata/parsers/gpx.py

@@ -34,14 +34,18 @@ def parse(self, source: Any) -> ParseResult:
         warnings: list[str] = []
         rows: dict[str, list[dict[str, Any]]] = {v: [] for v in _POINT_KIND.values()}
 
-        stream = self.open_binary(source)
+        stream = None
         try:
+            stream = self.open_safe_xml_binary(source)
             root = ET.parse(stream).getroot()  # noqa: S314 - GPX files are local, trusted
+        except ValueError as exc:
+            return ParseResult(self.format, {v: pd.DataFrame() for v in _POINT_KIND.values()},
+                               self.suggested_domain, {}, [f"unsafe GPX XML: {exc}"])
         except ET.ParseError as exc:
             return ParseResult(self.format, {v: pd.DataFrame() for v in _POINT_KIND.values()},
                                self.suggested_domain, {}, [f"invalid GPX XML: {exc}"])
         finally:
-            if hasattr(stream, "close"):
+            if stream is not None and hasattr(stream, "close"):
                 stream.close()
 
         def _point(elem: ET.Element, kind: str, **extra: Any) -> None:

src/freshdata/parsers/sdmx.py

@@ -33,14 +33,18 @@ class SDMXParser(Parser):
 
     def parse(self, source: Any) -> ParseResult:
         warnings: list[str] = []
-        stream = self.open_binary(source)
+        stream = None
         try:
+            stream = self.open_safe_xml_binary(source)
             root = ET.parse(stream).getroot()  # noqa: S314 - local trusted SDMX files
+        except ValueError as exc:
+            return ParseResult(self.format, {"observations": pd.DataFrame()},
+                               None, {}, [f"unsafe SDMX XML: {exc} (audit only)"])
         except ET.ParseError as exc:
             return ParseResult(self.format, {"observations": pd.DataFrame()},
                                None, {}, [f"invalid SDMX XML: {exc} (audit only)"])
         finally:
-            if hasattr(stream, "close"):
+            if stream is not None and hasattr(stream, "close"):
                 stream.close()
 
         rows: list[dict[str, Any]] = []

tests/parsers/test_fhir.py

@@ -124,7 +124,7 @@ def test_json_string_and_path_inputs(tmp_path):
     assert fd.parse_domain(text, format="fhir").metadata["total_resources"] == 6
     p = tmp_path / "bundle.json"
     p.write_text(text)
-    assert fd.parse_domain(str(p), format="fhir").metadata["total_resources"] == 6
+    assert fd.parse_domain(p, format="fhir").metadata["total_resources"] == 6
 
 
 def test_invalid_json_warns_not_raises():

tests/parsers/test_gpx.py

@@ -60,6 +60,17 @@ def test_malformed_xml_returns_warning_not_exception():
     assert any("invalid GPX XML" in w for w in result.warnings)
 
 
+def test_doctype_entities_are_rejected():
+    entity_gpx = """<!DOCTYPE gpx [<!ENTITY x "expanded">]>
+<gpx version="1.1" xmlns="http://www.topografix.com/GPX/1/1">
+  <wpt lat="40.0" lon="-73.0"><name>&x;</name></wpt>
+</gpx>"""
+    result = fd.parse_domain(entity_gpx, format="gpx")
+
+    assert all(df.empty for df in result.frames.values())
+    assert any("unsafe GPX XML" in w for w in result.warnings)
+
+
 def test_empty_gpx_warns():
     result = fd.parse_domain('<gpx xmlns="http://www.topografix.com/GPX/1/1"/>', format="gpx")
     assert any("no waypoints" in w for w in result.warnings)