feat: wire path validation into file loaders (Item 7)

CHANGELOG.md

@@ -5,6 +5,20 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## [Unreleased]
+
+### Security
+
+- `Pipeline.from_sql_files()` and `Pipeline.from_json_file()` now validate paths:
+  directory traversal, disallowed extensions, and symbolic links are rejected.
+
+### Changed
+
+- **BREAKING:** `Pipeline.from_sql_files()` and `Pipeline.from_json_file()` reject
+  symlinked paths, glob patterns that escape the directory, and files whose
+  extension is not `.sql`/`.json`. Pass `allow_symlinks=True` to opt back into
+  following symbolic links.
+
 ## [0.0.3] - 2025-12-29
 
 ### Added

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "clgraph"
-version = "0.0.3"
+version = "0.0.4"
 description = "Column lineage and pipeline dependency analysis for SQL"
 readme = "README.md"
 requires-python = ">=3.10"

src/clgraph/pipeline.py

@@ -494,13 +494,20 @@ def from_json(
         return create_from_json(data, apply_metadata=apply_metadata)
 
     @classmethod
-    def from_json_file(cls, file_path: str, apply_metadata: bool = True) -> "Pipeline":
+    def from_json_file(
+        cls,
+        file_path: str,
+        apply_metadata: bool = True,
+        *,
+        allow_symlinks: bool = False,
+    ) -> "Pipeline":
         """
         Create pipeline from JSON file exported by JSONExporter.
 
         Args:
             file_path: Path to JSON file
             apply_metadata: Whether to apply metadata from the JSON
+            allow_symlinks: If True, follow symbolic links (logs a security warning).
 
         Returns:
             Pipeline instance
@@ -514,7 +521,9 @@ def from_json_file(cls, file_path: str, apply_metadata: bool = True) -> "Pipelin
         """
         from .pipeline_factory import create_from_json_file
 
-        return create_from_json_file(file_path, apply_metadata=apply_metadata)
+        return create_from_json_file(
+            file_path, apply_metadata=apply_metadata, allow_symlinks=allow_symlinks
+        )
 
     @classmethod
     def _create_empty(cls, table_graph: "TableDependencyGraph") -> "Pipeline":
@@ -881,6 +890,8 @@ def from_sql_files(
         pattern: str = "*.sql",
         query_id_from: str = "filename",
         template_context: Optional[Dict[str, Any]] = None,
+        *,
+        allow_symlinks: bool = False,
     ) -> "Pipeline":
         """
         Create pipeline from SQL files in a directory.
@@ -893,6 +904,7 @@ def from_sql_files(
                 - "filename": Use filename without extension (default)
                 - "comment": Extract from first line comment (-- query_id: name)
             template_context: Optional dictionary of template variables
+            allow_symlinks: If True, follow symbolic links (logs a security warning).
 
         Returns:
             Pipeline instance
@@ -921,6 +933,7 @@ def from_sql_files(
             pattern=pattern,
             query_id_from=query_id_from,
             template_context=template_context,
+            allow_symlinks=allow_symlinks,
         )
 
     @classmethod

src/clgraph/pipeline_factory.py

@@ -276,25 +276,36 @@ def create_from_json(
 def create_from_json_file(
     file_path: str,
     apply_metadata: bool = True,
+    allow_symlinks: bool = False,
 ) -> "Pipeline":
     """
     Create Pipeline from JSON file exported by JSONExporter.
 
     Args:
         file_path: Path to JSON file
         apply_metadata: Whether to apply metadata from the JSON
+        allow_symlinks: If True, follow symbolic links (logs a security warning).
 
     Returns:
         Pipeline instance
     """
     import json
-    from pathlib import Path
+    import logging
 
-    path = Path(file_path)
-    if not path.exists():
-        raise FileNotFoundError(f"JSON file not found: {file_path}")
+    from .path_validation import PathValidator
 
-    with open(path) as f:
+    if allow_symlinks:
+        logging.getLogger(__name__).warning(
+            "SECURITY: allow_symlinks=True enables following symbolic links. "
+            "This may expose sensitive files outside the intended location."
+        )
+
+    validator = PathValidator()
+    resolved = validator.validate_file(
+        file_path, allowed_extensions=[".json"], allow_symlinks=allow_symlinks
+    )
+
+    with open(resolved) as f:
         data = json.load(f)
 
     return create_from_json(data, apply_metadata=apply_metadata)
@@ -306,6 +317,7 @@ def create_from_sql_files(
     pattern: str = "*.sql",
     query_id_from: str = "filename",
     template_context: Optional[Dict[str, Any]] = None,
+    allow_symlinks: bool = False,
 ) -> "Pipeline":
     """
     Create Pipeline from SQL files in a directory.
@@ -318,22 +330,37 @@ def create_from_sql_files(
             - "filename": Use filename without extension (default)
             - "comment": Extract from first line comment (-- query_id: name)
         template_context: Optional dictionary of template variables
+        allow_symlinks: If True, follow symbolic links (logs a security warning).
 
     Returns:
         Pipeline instance
     """
     import re
-    from pathlib import Path
 
-    sql_path = Path(sql_dir)
-    sql_files = sorted(sql_path.glob(pattern))
+    from .path_validation import PathValidator, _safe_read_sql_file
+
+    if allow_symlinks:
+        logging.getLogger(__name__).warning(
+            "SECURITY: allow_symlinks=True enables following symbolic links. "
+            "This may expose sensitive files outside the SQL directory."
+        )
+
+    validator = PathValidator()
+    resolved_dir = validator.validate_directory(sql_dir, allow_symlinks=allow_symlinks)
+    safe_pattern = validator.validate_glob_pattern(pattern, allowed_extensions=[".sql"])
+
+    sql_files = sorted(resolved_dir.glob(safe_pattern))
 
     if not sql_files:
         raise ValueError(f"No SQL files found in {sql_dir} matching {pattern}")
 
     queries = []
     for sql_file in sql_files:
-        sql_content = sql_file.read_text()
+        # Validate and read atomically to prevent TOCTOU (a validated file being
+        # swapped for a symlink before the read).
+        sql_content = _safe_read_sql_file(
+            sql_file, base_dir=resolved_dir, allow_symlinks=allow_symlinks
+        )
 
         if query_id_from == "filename":
             query_id = sql_file.stem

tests/test_path_validation_integration.py

@@ -0,0 +1,67 @@
+"""Integration tests: path validation reached through the public Pipeline API.
+
+These exercise the factory entry points a caller actually uses. The unit tests
+in test_path_validation.py cover PathValidator internals directly.
+"""
+
+import json
+from pathlib import Path
+
+import pytest
+
+from clgraph import Pipeline
+
+
+class TestFromJsonFilePathValidation:
+    def test_non_json_extension_rejected(self, tmp_path: Path):
+        bad = tmp_path / "data.txt"
+        bad.write_text("{}")
+        with pytest.raises(ValueError, match="Invalid file extension"):
+            Pipeline.from_json_file(str(bad))
+
+    def test_symlink_rejected_by_default(self, tmp_path: Path):
+        real = tmp_path / "real.json"
+        real.write_text(json.dumps({"columns": [], "edges": []}))
+        link = tmp_path / "link.json"
+        link.symlink_to(real)
+        with pytest.raises(ValueError, match="Symbolic links are not allowed"):
+            Pipeline.from_json_file(str(link))
+
+    def test_missing_file_raises_filenotfound(self, tmp_path: Path):
+        with pytest.raises(FileNotFoundError):
+            Pipeline.from_json_file(str(tmp_path / "nope.json"))
+
+
+class TestFromSqlFilesPathValidation:
+    def _write_sql(self, d: Path, name: str = "q.sql") -> None:
+        (d / name).write_text("SELECT 1 AS a")
+
+    def test_valid_directory_loads(self, tmp_path: Path):
+        self._write_sql(tmp_path)
+        pipeline = Pipeline.from_sql_files(str(tmp_path), dialect="bigquery")
+        assert pipeline is not None
+
+    def test_traversal_pattern_rejected(self, tmp_path: Path):
+        self._write_sql(tmp_path)
+        with pytest.raises(
+            ValueError, match="Glob pattern must not contain directory traversal components"
+        ):
+            Pipeline.from_sql_files(str(tmp_path), pattern="../*.sql")
+
+    def test_symlinked_dir_rejected_by_default(self, tmp_path: Path):
+        real_dir = tmp_path / "real"
+        real_dir.mkdir()
+        self._write_sql(real_dir)
+        link_dir = tmp_path / "link"
+        link_dir.symlink_to(real_dir, target_is_directory=True)
+        with pytest.raises(ValueError, match="Symbolic links are not allowed"):
+            Pipeline.from_sql_files(str(link_dir))
+
+    def test_symlinked_dir_allowed_with_optin(self, tmp_path: Path):
+        real_dir = tmp_path / "real"
+        real_dir.mkdir()
+        self._write_sql(real_dir)
+        link_dir = tmp_path / "link"
+        link_dir.symlink_to(real_dir, target_is_directory=True)
+        pipeline = Pipeline.from_sql_files(str(link_dir), allow_symlinks=True)
+        assert pipeline is not None

tests/test_pipeline_factories.py

@@ -215,9 +215,10 @@ def test_custom_pattern(self, tmp_path):
         pipeline = Pipeline.from_sql_files(str(tmp_path), pattern="*.sql")
         assert len(pipeline.table_graph.queries) == 1
 
-        # Load .txt files
-        pipeline = Pipeline.from_sql_files(str(tmp_path), pattern="*.txt")
-        assert len(pipeline.table_graph.queries) == 1
+        # Patterns for non-.sql extensions are rejected by the path validator:
+        # from_sql_files is restricted to .sql files by design.
+        with pytest.raises(ValueError, match="Invalid extension in pattern"):
+            Pipeline.from_sql_files(str(tmp_path), pattern="*.txt")
 
 
 class TestQueryIdGeneration: